Reusable Observables

[gNechitailo]

Is your feature request related to a problem? Please describe.
I have read the description to this project and I like the idea of functional programming related to Web development. The point I didn't get is that you call HTTP single-event based. HTTP has a set of declared endpoints with a declared processing logic.
I researched a bit about RxJS and found out that creating observable every time you want to process an event is WAY MORE expensive than using the same observable and pushing new events to it.
I have also used a debugger a bit with Marble.js and saw that it really creates a new observable for each request. The nature of EffectFactory about that.

Describe the solution you'd like
I think if marble used effectFactory callback upon application startup to create a processing pipe Once instead of calling it on each request, it would be much more performant.
Maybe you'll need some extra RxJS operators to perform routing, but I can't tell for sure (didn't dive deep to marble.js sources).

Additional context
Maybe there are there fundamental problems about it?
If not, I'd like to make some help with this subject (maybe make a version for this).

[JozefFlakus]

Hi @gNechitailo,
Thanks for the interest! Good spot. The way how the routing in @marblejs/core was build had two approaches:

1. stream based routing (without static routing table - dynamic routing) Marble 0.5
2. router based (streams like callbacks but with static routing build on startup) Marble 1.0

The diagram below shows how how the server reacted to high load of incoming request (req/sec) depending of endpoints total number. As you can see it is quite performant right now and it was the main aim when releasing stable version 1.

The first iteration was built like this:

1: raw dynamic

const getUsers$: Effect = req$ =>
  req$.pipe(
    filter(req => req.url === '/user'),
    filter(req => req.method === 'GET'),
    mergeMap(Dao.getUsers),
    map(users => ({ body: users })),
  );

2: dynamic

const getUser$: Effect = req$ =>
  req$.pipe(
    matchPath('/user/:id'),
    matchType('GET'),
    map(req => req.params.id),
    mergeMap(Dao.getUser),
    map(user => ({ body: user })),
  );

3: factory - EffectFactory / r.pipe

const getUser$ = EffectFactory
  .matchPath('/user/:id'),
  .matchType('GET'),
  .use(req$ => req$.pipe(
    map(req => req.params.id),
    mergeMap(Dao.getUser),
    map(user => ({ body: user })),
  ));

The mechanism that you describing would be an evolution of the current solution. Basically on the server startup the built routing table should be reflected as a stream with proper operators acting as filters and matchers for method and routing paths. This is indeed possible but we have to be very careful, especially we have to be sure the stream will be always active - it cannot be closed due to errors.

The significant benefit of this would be in Context dependency injection, eg. you will be able to resolve the declared dependencies inside the callback (instead of stream), so the dips will be always injected after bootraping.

The idea of this always running steam I'm already implementing in upcoming @marblejs/messaging module in version 3 but recently I'm thinking about such an improvement in core module quite a lot.

tl;dr I have to crate a POC of this solution if it is possible to reflect the collected routing table as a static stream.

[gNechitailo]

Wow! Thank you for the detailed answer!
Keep up the good work!
I'm not sure, but if I were you, I'd possibly use the following approach:
Instead of creating one large pipe of a single stream, I would take events from one Observable (routing) and then use Subjects (created for each effect) to put event to.
I think this could help with situations where errors can break everything (so routing stream never die).

[JozefFlakus]

In general, the proposed approach has to be tested, it can turn out that in the end, it will be slower than the current implementation, but the memory overhead will be lower (which is not so bad right now).

[JozefFlakus]

@gNechitailo checkout PR number #188 - it will be interesting for you.

[gNechitailo]

Wow! Amazing! This is my favorite library now :)
Tried to start this work myself, but you were much faster (it wasn't that easy).

[JozefFlakus]

Glad to hear that! I believe that it can be optimized even more, but right now it is a good base for future improvements.

[JozefFlakus]

@gNechitailo It turned out that the implementation previously introduced was buggy so I had to revert to older implementation. If you would like to be interested in problem-solving - here is the issue #206 . Any help highly appreciated! :)

[gNechitailo]

Okay, I will try to take a look when I have spare time!