Marble.js 3.0 - microservice example

microservice.ts

// 📄 publisher - fib.effect.ts

const fib$ = r.pipe(
  r.matchPath('/fib/:number'),
  r.matchType('GET'),
  r.useEffect((req$, ctx) => {
    const client = useContext(ClientToken)(ctx.ask);

    return req$.pipe(
      validateRequest,
      map(req => req.params.number),
      mergeMap(number => forkJoin(
        client.send({ type: 'FIB', payload: number + 0 }),
        client.send({ type: 'FIB', payload: number + 1 }),
        client.send({ type: 'FIB', payload: number + 2 }),
      )),
      map(body => ({ body })),
    );
  }),
);

// 📄 consumer - fib.effect.ts

const microservice = createMicroservice({
  transport: Transport.AMQP,
  options: {
    host: 'amqp://localhost:5672',
    queue: 'fib_queue',
  },
  listener: messagingListener({
    effects: [fib$],
  }),
});

const fib$: MsgEffect = event$ =>
  event$.pipe(
    matchEvent('FIB'),
    act(eventValidator$(t.number)),
    act(event => of(event).pipe(
      map(event => fib(event.payload)),
      map(payload => reply(event)({ payload })),
    )),
  );