Marble.js 3.0 - continuous streams

continuous-http-streams.ts

const foo$ = r.pipe(
  r.applyMeta({ continuous: true }),
  r.matchPath('/'),
  r.matchType('GET'),
  r.useEffect((req$, ctx) => {
    const reqBus$ = useContext(HttpRequestBusToken)(ctx.ask);
    const terminate$ = reqBus$.pipe(filter(req => req.url === '/flush'));

    return req$.pipe(
      bufferWhen(() => terminate$),
      mergeMap(buffer => from(buffer)),
      mergeMap(request => of(request).pipe(
        mergeMap(processData),
        map(body => ({ body, request })),
        catchError(error => of({
          request,
          status: HttpStatus.BAD_REQUEST,
          body: { error: { message: error.message }}
        })),
      )),
    );
  }));