Class MessageClientAbstract

Receives messages published to the given topic.

You can subscribe to multiple topics simultaneously by using wildcard segments in the topic. If a segment begins with a colon (:), then the segment acts as a placeholder for any segment value. Substituted segment values are then available via the params property of the received message.

const topic: string = 'myhome/:room/temperature';

Beans.get(MessageClient).observe$(topic).subscribe((message: TopicMessage) => {
  console.log(message.params);
});
Copy

If the received message has the MessageHeaders.ReplyTo header field set, the publisher expects the receiver to send one or more replies to that ReplyTo topic. If streaming responses, you can use the takeUntilUnsubscribe operator to stop replying when the requestor unsubscribes.

const topic: string = 'myhome/livingroom/temperature';

Beans.get(MessageClient).observe$(topic).subscribe((request: TopicMessage) => {
  const replyTo = request.headers.get(MessageHeaders.ReplyTo);
  sensor$
    .pipe(takeUntilUnsubscribe(replyTo))
    .subscribe(temperature => {
      Beans.get(MessageClient).publish(replyTo, `${temperature}°C`);
    });
});
Copy

Convenience API for handling messages.

Unlike observe$, messages are passed to a callback function rather than emitted from an Observable. Response(s) can be returned directly from the callback. It supports error propagation and request termination. Using this method over observe$ significantly reduces the code required to respond to requests.

For each message received, the specified callback function is called. When used in request-response communication, the callback function can return the response either directly or in the form of a Promise or Observable. Returning a Promise allows the response to be computed asynchronously, and an Observable allows to return one or more responses, e.g., for streaming data. If the callback function returns no value (void), returns undefined, or returns a Promise that resolves to undefined, communication is terminated immediately without a response. If the callback returns an Observable, all its emissions are transported to the requestor and communication is not terminated until the Observable completes. Termination of communication always completes the requestor's Observable. If the callback throws an error, or the returned Promise or Observable errors, the error is transported to the requestor, erroring the requestor's Observable.

Publishes a message to the given topic. The message is transported to all consumers subscribed to the topic.

A message can be marked as "retained" by setting the PublishOptions.retain flag to true. It instructs the broker to store this message and deliver it to new subscribers, even if they subscribe after the message has been published. The broker stores one retained message per topic. To delete a retained message, send a retained message without payload to the topic. Deletion messages are not transported to subscribers.

Sends a request to the given topic and receives one or more replies.

A request can be marked as "retained" by setting the RequestOptions.retain flag to true. It instructs the broker to store this request and deliver it to new subscribers, even if they subscribe after the request has been sent. Retained requests are not replaced by later retained requests/ messages and remain in the broker until the requestor unsubscribes.

If not marking the request as "retained", at least one subscriber must be subscribed to the topic. Otherwise, the request is rejected.

Allows observing the number of subscriptions on a topic. The Observable never completes.