Added support for onMessage callback in PubSub subscribe

[necessarylion]

Objective

Added support for onMessage callback in PubSub subscribe function

Description

• Support onMessage callback

subscriber.subscribe(['chat'],
  onMessage: (dynamic message, String? channel) {
    print(message);
    print(channel);
  },
);

• Fix fews typo in PubSub

[ra1u]

Thank you for this.

subscribe looks similar to listen available on Stream.
Can you provide some rationale about use-case that is now supported with this PR?

Kind regards, Luka

[necessarylion]

Thank you for this.

subscribe looks similar to listen available on Stream. Can you provide some rationale about use-case that is now supported with this PR?

Kind regards, Luka

@ra1u Thanks for your quick review.

Yes, it is kind of similar to listen on stream. I added support for onMessage callback, so that I can listen directly for new message for subscribed channels only. It is useful and more understandable when multiple channels are subscribed on different files/functions.

eg.

I have 2 function that subscribe to a specific channel and process the job differently.

function1() {
  subscriber.subscribe('websocket_emit_event', onMessage: (message, channel) {
     // do something
  });
}

function2() {
   subscriber.subscribe('websocket_remove_event', onMessage: (message, channel) {
     // do something else
   });
}

[ra1u]

Other option for similar goal is to create new streams that will return only messages where some predicate holds.

pubsub.subscribe(["ch1","ch2"]);
stream_splitter = StreamSplitter(pubsub.getStream());
stream_ch1 = stream_splitter.split().where((message) => message[1] == "ch1");
stream_ch2 = stream_splitter.split().where((message) => message[1] == "ch2");

This now gets you 2 streams where each sends only messages from appropriate channel.

To get this PR trough, we would like to have code that handles all scenarios that our users will face, not just "subscribe" but also "psubscribe" and unsubscribe and punsubscribe. We need to think about what happens with handlers when (p)unsubscribe is called.

[necessarylion]

Closing since it can be handle using StreamSplitter