feat(bloc): bloc-level event concurrency

[fakalit]

I just wanted the bring up this feature request that doesn't exist in the original bloc ecosystem and listen to your perspective on it.

As things are, it's not possible to apply transformers on the stream that includes all events for a given bloc without using ad-hoc, non-declarative solutions that are hard to read on felangel/bloc. Considering all event handlers mutate the same state which belongs to a given bloc instance, this shortcoming eliminates one of the primary use cases of sequential event handling, concurrency, and using event driven architectures in the first place: write blocks.

I.e: If one event increases the stock levels, and the other consumes the stock, we are unable to ensure the second event is processed after the first one by leveraging the expressed bloc API. The suggested workaround to this problem is doing something similar to this:

class ExampleBloc extends Bloc<ExampleEvent, ExampleState> {

  constructor() {
    super(new ExampleInitial());  
    on<ExamepleEvent>(_onEvent, transformer: sequential());
  }

  async  _onEvent(event: ExampleEvent, emit: Emitter<ExampleState>) {
      if (event is ExamepleEventOne) { return await _onExamepleEventOneevent, emit);}
      else if (event is ExamepleEventTwo) { return await _onExamepleEventTwo(event, emit);}
      else if (event is ExamepleEventThree) { return await _onExamepleEventThree(event, emit);}
  }

....
}

While this is usable, it's in my opinion clunky and needs commentary on why the code was organised this way to be readable. I don't have any suggested APIs for this use case but still wanted to bring it to your attention.

Best

[jacobtipp]

I just wanted the bring up this feature request that doesn't exist in the original bloc ecosystem and listen to your perspective on it.

As things are, it's not possible to apply transformers on the stream that includes all events for a given bloc without using ad-hoc, non-declarative solutions that are hard to read on felangel/bloc. Considering all event handlers mutate the same state which belongs to a given bloc instance, this shortcoming eliminates one of the primary use cases of sequential event handling, concurrency, and using event driven architectures in the first place: write blocks.

I.e: If one event increases the stock levels, and the other consumes the stock, we are unable to ensure the second event is processed after the first one by leveraging the expressed bloc API. The suggested workaround to this problem is doing something similar to this:

class ExampleBloc extends Bloc<ExampleEvent, ExampleState> {

  constructor() {
    super(new ExampleInitial());  
    on<ExamepleEvent>(_onEvent, transformer: sequential());
  }

  async  _onEvent(event: ExampleEvent, emit: Emitter<ExampleState>) {
      if (event is ExamepleEventOne) { return await _onExamepleEventOneevent, emit);}
      else if (event is ExamepleEventTwo) { return await _onExamepleEventTwo(event, emit);}
      else if (event is ExamepleEventThree) { return await _onExamepleEventThree(event, emit);}
  }

....
}

While this is usable, it's in my opinion clunky and needs commentary on why the code was organised this way to be readable. I don't have any suggested APIs for this use case but still wanted to bring it to your attention.

Best

Unfortunately, I don't have an alternative that comes to mind. Having a base event that narrows down multiple child events behind a sequential transformer provides a high level of consistency. Race conditions can become really ugly when you have multiple handlers trying to coordinate between events.

[fakalit]

Unfortunately, I don't have an alternative that comes to mind. Having a base event that narrows down multiple child events behind a sequential transformer provides a high level of consistency. Race conditions can become really ugly when you have multiple handlers trying to coordinate between events.

The goal is not to have multiple handlers for the same event and coordinate between them, but being able to apply transformers on the rather abstract base event level rather than concrete child events. Being able to define sequentiality across different sub-event types, such as being able to make the handler for 'ExampleEventTwo' to wait until the handler for the 'ExampleEventOne' finished processing an event it has received is a critical use case for any event-sourcing or CQRS-like approach to handling the business layer. This also applies for other transformers like restartable or even custom transformers which can be used to define behaviours like throttling. As long as all of the event handlers operate over the same state -which is the case for the bloc pattern here- the need for the ability to define common event transformers that applies to all events that can mutate a given common state instance becomes a must, if the architecture is to be used at scale.

Unfortunately the above example I shared which acts as a valid work around for this use-case in felangel/bloc does not work in the bloc-state library, as it has (thankfully) stricter type check requirements and it doesn't allow adding events to bloc instances without explicitly registering a handler for the sub event classes like 'ExampleEventTwo'.

if (!this._eventMap.has(Object.getPrototypeOf(event).constructor)) {
      throw new StateError(`
        add(${event}) was called without a registered event handler.
        Make sure to register a handler via on(${event}, (event, emit) {...})
      `);
}

Meaning, we can not add events with the type of 'ExampleEventTwo' if we only registered a handler like 'on < ExamepleEvent > '. So with the current version I'm not sure if there is a way to cover this use-case in any way.

If you're interested, I can try and experiment whether providing a surface level API like the following, and then applying a transformer directly to the _eventSubject$ could work without breaking any of the existing behaviours and tests and share the results with you. If you see this as outside of the scope of the project, I can go ahead with a fork as well.

class ExampleBloc extends Bloc<ExampleEvent, ExampleState> {

  constructor() {
    super(new ExampleInitial(), sequential()); 

    on<ExampleEventOne>(this._onExampleEventOne);
    on<ExampleEventTwo>(this._onExampleEventTwo);
    on<ExampleEventThree>(this._onExampleEventThree);
  }

  async  _onExampleEventOne(event: ExampleEvent, emit: Emitter<ExampleState>) {
  //...
  }

  async  _onExampleEventOne(event: ExampleEvent, emit: Emitter<ExampleState>) {
  //...
  }

// ....
}

Once again, thanks for the great work and the amazing library you put together.

[jacobtipp]

@fakalit feel free to fork off the main branch. I'll need to add some contribution docs to the monorepo soon.

install

1. download pnpm npm i -g pnpm
2. pnpm i in the root directory
3. git commits need to be in conventional-commit format. pnpm commit can be used to help generate a commit for you.

apps

projects in the /apps/* directory are web apps that can be ran which hot-reload after any changes in /packages/* occur. Very useful for debugging and testing new features.

example commands:

pnpm nx serve bloc-hacker-news pnpm nx serve bloc-todos

[jacobtipp]

@fakalit Just a heads up, there are quite a few new changes that will be in the next release, some of which are breaking changes. I don't think any of the breaking changes affect this logic, you can checkout the next branch to see those changes.

Would something like this fix the issue?

Basically just recursively walk up the prototype chain, if the current constructor exists in the WeakMap of events, return true, otherwise keep searching until constructor is null then return false. I think _eventMap needs to be a WeakSet instead of WeakMap, I'll have to look into that a little more.

  private checkPrototype(event: Event, pass = false): boolean {
    let constructor = Object.getPrototypeOf(event)
    if (!pass) {
      constructor = constructor.constructor
    }

    if (this._eventMap.has(constructor)) {
      return true
    }

    return constructor === null ? false : this.checkPrototype(constructor, true)
  }

  /**
   * Adds an event to the BLoC's stream of events.
   *
   * @param event - The event to add.
   *
   * @throws if there is no registered event handler for the given event.
   *
   * @returns The instance of the Bloc.
   */
  add(event: Event) {
    if (!this.checkPrototype(event)) {
      throw new StateError(`
        add(${event}) was called without a registered event handler.
        Make sure to register a handler via on(${event}, (event, emit) {...})
      `);
    }

    try {
      this.onEvent(event);
      this._eventSubject$.next(event);
    } catch (error) {
      this.onError(error as Error);
      throw error;
    }

    return this;
  }

[fakalit]

That would indeed solve the problem, but it might also be meaningul to pair that with allowing to register abstract classes as events to blocs. Which is something that we can't do right now. Overall, once again, many thanks for looking into the problem.

That aside, here's also an example implementation of what I talking about in my previous post. I'm not confident in the approach I followed, as I'm sure there would be better ways to cover the same use-case or there is a chance that I broke some behaviour that's not covered in the tests. But it should be at least helpful to communicate what I had in mind. Let me know your take on it.

https://github.com/Zira-Games/bloc-state/commit/b5c3c8273e908bdc9c7890d044821a1ca34c0663

[jacobtipp]

@fakalit 👉 https://github.com/jacobtipp/bloc-state/pull/91 👈

• has changes that allow checking for ancestors
• you can now register abstract class event handlers this.on(event: Event, () => ....)

You can check out those changes in next. It is also an available channel on npm. npm i @jacobtipp/bloc@next

I'll be sure to check out your branch sometime today.

[jacobtipp]

That would indeed solve the problem, but it might also be meaningul to pair that with allowing to register abstract classes as events to blocs. Which is something that we can't do right now. Overall, once again, many thanks for looking into the problem.

That aside, here's also an example implementation of what I talking about in my previous post. I'm not confident in the approach I followed, as I'm sure there would be better ways to cover the same use-case or there is a chance that I broke some behaviour that's not covered in the tests. But it should be at least helpful to communicate what I had in mind. Let me know your take on it.

Zira-Games@b5c3c82

I see what you mean about having a single transformer for all event handlers. You would no longer need the abstract class event handler, the bloc-level transformer would be used for all events.

I'll have to spend more time looking over your implementation. For now, you should be able to use an abstract class event handler that can work with multiple subclass events.

[fakalit]

Thanks. Having feature parity on this with bloc makes porting our app a lot more straightforward!