[EPIC] Support messaging/eventing style APIs

[raymondfeng]

Currently LoopBack supports RPC style APIs such as REST or gRPC. We would like to extend that with messaging/eventing capabilities.

Use cases

• Expose endpoints to accept a stream of messages/events and/or return a stream of messages/events (such as SSE)
• Connect to backend resources (such as Kafka or MQTT Broker) to produce/consume (including pub/sub) messages/events
• Define business functions to produce/consume a stream of messages/events (synchronously or asynchronously)
• Create serverless actions (such as IBM Cloud Functions and AWS Lambda) with LoopBack 4 as event handlers
• Observe events and react on them (hooks? TBD)
  • https://github.com/ReactiveX/rxjs?
  • https://rxjs-dev.firebaseapp.com/guide/observable
• Make it easy to implement web hooks

Protocols

• WebSocket
• HTTP/2 with SSE
• MQTT

Mapping to LoopBack 4 concepts

• Servers: listen on WebSocket, MQTT, and HTTP/2 protocols
• Connectors: subscribe to an event source or message broker or publish messages to backend systems
• Controllers: create methods (with decorators) to produce/consume/process a stream of events.
• Events: define models representing various types of events

Examples

app.server(WebSocketServer, config);

app.datasource({name: 'kafka', ...};

export class MoveEvent {
  gameId: string;
  sequence: number;
  player: string;
  shape: 'o' | 'x';
  move: [number, number];
}

export class GameStatus {
  id: string;
  board: ('o' | 'x' )[][];
  winner: string;
  status: string;
}

export class TicTakToeController {
  async createGame(player1: string, player2: string): number {}

  // Should we support event by event, a stream of events, or both
  handleMoves(move: AsyncIterator<MoveEvent>): AsyncIterator<GameStatus> {}
 
  async handleMove(move: MoveEvent): GameStatus {}
}

export class MyEventController {
  @subscribe('order.created')
  async onNewOrders(@event('order.created') orderCreatedEvent: OrderCreated) {}

  @publish('invoice.created')
  generateInvoices(): AsyncIterator<InvoiceCreated> {}
}

[mikaeljensen]

I think this is going to be very important for the future of LB4.

Change-stream was IMHO one of the most underrated features in Loopback 2/3 and I think one of the reasons why it did not see major adoption was that the documentation was lackluster and that it was not pushed as selling points commercially/marketing-wise for the many benefits that it could provide.

In our case, we are currently running fairly large systems and use both LB2 and LB3 and the only reason why we are not running all clients on change streams instead of REST push/pull, is it was a lot harder and more time consuming to implement. Concerns included cross-cluster synchronization (propagating events on one process, to streams managed in other processes/servers), problems getting filters to work as expected, lack of good client side support for observable's as base data foundation and developer experience with said and related technology and patterns.

Why this is important:
Competing frameworks do it (this list https://loopback.io/resources/ and many more (LB2/3 included))
Most modern MBaaS's implement realtime push-data in some form (API Connect included)
Client side adaption is skyrocketing with the observable pattern becoming more and more core to client side and especially frameworks like Angular and React
Realtime/subscription/stream data is not an isolated set of use cases, but handled a good chuck of current REST/pull/promised-based use cases better than RPC do.

If LB4 nails this good, this could be HUGE.

[dhmlau]

Discussed with @raymondfeng, #1925 can be the first POC.

[raymondfeng]

See another PoC - https://github.com/raymondfeng/loopback4-example-websocket

[lancedouglas1]

I'd like to provide a real-world example that I'm looking for the correct framework to build upon.
A Flux architecture

1. Clients login via REST API and receive one or more authenticated websocket contexts (I really like socketcluster FWIW)
2. Clients connect to the socket-server with their personalized-context and receive initial state of their context
3. Client UI is manipulated by the state provided on the socket
4. Client UI can only make RPC calls to API (not picky if exposed via REST or WS)
5. Socket-server watches change streams on the applicable datasources and emits ACL-filtered & diff-only context changes to applicable socket clients.
6. When the Client socket receives a message, it is processed down to adjust UX as applicable

Example:

• /login returns {url:"/trade?session=abXd3", ws:[{global:"20aafd023aef", hifi:"220dkavd23"} ]}
• UI loads 'url' and opens a websocket to the 'global' context.
• The URL is a fairly static template UX and the 'global' context emits data that populates the UI with User-Session-Specific data like name and current stock quotes, and populates local objects with preferences
• UX code determines that the latency with the server is very low, and that based on user preferences provided in 'global' it begins further enhancing the UI with the data emitted on the 'hifi' channel on the socket that provides animations instructions
• The user clicks on a form-submit button to complete a stock-trade which triggers an ajax call to the REST API (all ACL credentials and context are populated with data from 'global').
• The socket server witnesses the change and emits the new stock data to display via 'global' and an animation and sound trigger via 'hifi'.
• Other subscribers (systems, regulators, users) also receive emitted events from those changes, if applicable

I'm fairly certain we can build this as an extension with hooks into the session management as well as the ability to watch changestreams on the datasources. But the challenge I see is in having a native all-data-sources.watch() solution with applicable relationships clarified unless this was a more native feature.

Thanks for all your hard work.

[ElHupeke]

I'm still using Fireloop (on top of LB3) for realtime updates. It is great and would love to use LB4 if it got these feature of Fireloop. I saw Fireloop mentioned before, but when database is updated, I want all clients connected to app be updated automatically. Will there be a realtime option in LB4 or do I need te look for alternatives?