JBoss.orgCommunity Documentation

Chapter 2. Messaging

2.1. Messaging Overview
2.2. Messaging API Basics
2.2.1. Sending Messages with the Client Bus
2.2.2. Receiving Messages on the Server Bus / Server Services
2.2.3. Sending Messages with the Server Bus
2.2.4. Receiving Messages on the Client Bus/ Client Services
2.2.5. Local Services
2.3. Single-Response Conversations & Pseudo-Synchronous Messaging
2.4. Sender Inferred Subjects
2.5. Broadcasting
2.6. Client-to-Client Communication
2.6.1. Relay Services
2.7. Message Routing Information
2.8. Handling Errors
2.8.1. Handling global message transport errors
2.9. Asynchronous Message Tasks
2.10. Repeating Tasks
2.11. Queue Sessions
2.11.1. Lifecycle
2.11.2. Scopes
2.12. Client Logging and Error Handling
2.13. Wire Protocol (J.REP)
2.13.1. Payload Structure
2.13.2. Message Routing
2.13.3. Bus Management and Handshaking Protocols
2.14. Conversations
2.15. WebSockets
2.15.1. Configuring the sideband server
2.15.2. Deploying with JBoss AS 7.1.2 (or higher)
2.15.3. JSR-356 WebSocket support (Deploying to WildFly 8.0 or higher)
2.15.4. WebSocket Security
2.16. Bus Lifecycle
2.16.1. Turning Server Communication On and Off
2.16.2. Observing Bus Lifecycle State and Communication Status
2.17. Shadow Services
2.18. Debugging Messaging Problems

This section covers the core messaging concepts of the ErraiBus messaging framework.

ErraiBus provides a straight-forward approach to a complex problem space. Providing common APIs across the client and server, developers will have no trouble working with complex messaging scenarios such as building instant messaging clients, stock tickers, to monitoring instruments. There’s no more messing with RPC APIs, or unwieldy AJAX or COMET frameworks. We’ve built it all in to one concise messaging framework. It’s single-paradigm, and it’s fun to work with.

It’s important to understand the concept of how messaging works in ErraiBus. Here are some important facts you’ll need to know:

It can be tempting to think of ErraiBus simply as a client-server communication platform, but there is a plethora of possibilities for using ErraiBus purely within the GWT client context, such as a way to advertise and expose components dynamically, to get around the lack of reflection in GWT.

So keep that in mind when you run up against problems in the client space that could benefit from runtime federation.

Plugin Tip

Use the Errai Forge Addon Add Errai Features command and select Errai Messaging to follow along with this section.

Manual Setup

Checkout the Manual Setup Section for instructions on how to manually add messaging to your project.

The MessageBuilder is the heart of the messaging API in ErraiBus. It provides a fluent / builder API, that is used for constructing messages. All three major message patterns can be constructed from the MessageBuilder.

Components that want to receive messages need to implement the MessageCallback interface.

But before we dive into the details, let’s look at some use cases.

In order to send a message from a client you need to create a Message and send it through an instance of MessageBus . In this simple example we send it to the subject HelloWorldService.

public class HelloWorld implements EntryPoint {


  // Get an instance of the RequestDispatcher
  private RequestDispatcher dispatcher = ErraiBus.getDispatcher();
  public void onModuleLoad() {
    Button button = new Button("Send message");
    button.addClickHandler(new ClickHandler() {
      public void onClick(ClickEvent event) {
        // Send a message to the 'HelloWorldService'.
        MessageBuilder.createMessage()
          .toSubject("HelloWorldService") // (1)
          .signalling() // (2)
          .noErrorHandling() // (3)
          .sendNowWith(dispatcher); // (4)
        });
        [...]
      }
   }
}

In the above example we build and send a message every time the button is clicked. Here’s an explanation of what’s going on as annotated above:

Important

An astute observer will note that access to the RequestDispatcher differs within client code and server code. Because this client code does not run within a container, access to the RequestDispatcher and MessageBus is provided statically using the ErraiBus.get() and ErraiBus.getDispatcher() methods. See the section on Errai IOC and Errai CDI for using ErraiBus from a client-side container.

When using Errai IOC or CDI, you can also use the Sender<T> interface to send messages.

In the following example we extend our server side component to reply with a message when the callback method is invoked. It will create a message and address it to the subject ' HelloWorldClient ':

@Service

public class HelloWorldService implements MessageCallback {
  private RequestDispatcher dispatcher;
  @Inject
  public HelloWorldService(RequestDispatcher dispatcher) {
    dispatcher = dispatcher;
  }
  public void callback(CommandMessage message) {
    // Send a message to the 'HelloWorldClient'.
    MessageBuilder.createMessage()
      .toSubject("HelloWorldClient") // (1)
      .signalling()                  // (2)
      .with("text", "Hi There")      // (3)
      .noErrorHandling()             // (4)
      .sendNowWith(dispatcher);      // (5)
    });
  }
}

The above example shows a service which sends a message in response to receiving a message. Here’s what’s going on:

It is possible to contruct a message and a default response handler as part of the MessageBuilder API. It should be noted, that multiple replies will not be possible and will result an exception if attempted. Using this aspect of the API is very useful for doing simple psuedo-synchronous conversive things.

You can do this by specifying a MessageCallback using the repliesTo() method in the MessageBuilder API after specifying the error handling of the message.

MessageBuilder.createMessage()

  .toSubject("ConversationalService").signalling()
  .with("SomeField", someValue)
  .noErrorHandling()
  .repliesTo(new MessageCallback() {
    public void callback(Message message) {
      System.out.println("I received a response");
    }
   })

See the next section on how to build conversational services that can respond to such messages.

It is possible for the sender to infer, to whatever conversational service it is calling, what subject it would like the reply to go to. This is accomplished by utilizing the standard MessageParts.ReplyTo message part. Using this methodology for building conversations is generally encouraged.

Consider the following client side code:

  MessageBuilder.createMessage()

    .toSubject("ObjectService").signalling()
    .with(MessageParts.ReplyTo, "ClientEndpoint")
    .noErrorHandling().sendNowWith(dispatcher);

And the conversational code on the server (for service ObjectService ):

  MessageBuilder.createConversation(message)

    .subjectProvided().signalling()
    .with("Records", records)
    .noErrorHandling().reply();

In the above examples, assuming that the latter example is inside a service called "ObjectService" and is referencing the incoming message that was sent in the former example, the message created will automatically reference the ReplyTo subject that was provided by the sender, and send the message back to the subject desired by the client on the client that sent the message.

Broadcasting messages to all clients listening on a specific subject is quite simple and involves nothing more than forgoing use of the reply API. For instance:

MessageBuilder.createMessage().

    .toSubject("MessageListener")
    .with("Text", "Hello, from your overlords in the cloud")
    .noErrorHandling().sendGlobalWith(dispatcher);

If sent from the server, all clients currently connected, who are listening to the subject "MessageListener" will receive the message. It’s as simple as that.

Communication from one client to another client is not directly possible within the bus federation, by design. This isn’t to say that it’s not possible. But one client cannot see a service within the federation of another client. We institute this limitation as a matter of basic security. But many software engineers will likely find the prospects of such communication appealing, so this section will provide some basic pointers on how to go about accomplishing it.

Every message that is sent between a local and remote (or server and client) buses contain session routing information. This information is used by the bus to determine what outbound queues to use to deliver the message to, so they will reach their intended recipients. It is possible to manually specify this information to indicate to the bus, where you want a specific message to go.

You can obtain the SessionID directly from a Message by getting the QueueSession resource:

   QueueSession sess = message.getResource(QueueSession.class, Resources.Session.name());

   String sessionId = sess.getSessionId();

You can extract the SessionID from a message so that you may use it for routing by obtaining the QueueSession resource from the Message. For example:

...

  public void callback(Message message) {
   QueueSession sess = message.getResource(QueueSession.class, Resources.Session.name());
   String sessionId = sess.getSessionId();
    // Record this sessionId somewhere.
    ...
  }

The SessionID can then be stored in a medium, say a Map, to cross-reference specific users or whatever identifier you wish to allow one client to obtain a reference to the specific SessionID of another client. In which case, you can then provide the SessionID as a MessagePart to indicate to the bus where you want the message to go.

  MessageBuilder.createMessage()

    .toSubject("ClientMessageListener")
    .signalling()
    .with(MessageParts.SessionID, sessionId)
    .with("Message", "We're relaying a message!")
    .noErrorHandling().sendNowWith(dispatcher);

By providing the SessionID part in the message, the bus will see this and use it for routing the message to the relevant queue.

It may be tempting however, to try and include destination SessionIDs at the client level, assuming that this will make the infrastructure simpler. But this will not achieve the desired results, as the bus treats SessionIDs as transient. Meaning, the SessionID information is not ever transmitted from bus-to-bus, and therefore is only directly relevant to the proximate bus.

Asynchronous messaging necessitates the need for asynchronous error handling. Luckily, support for handling errors is built directly into the MessageBuilder API, utilizing the ErrorCallback interface. In the examples shown in previous exceptions, error handing has been glossed over with aubiquitous usage of the noErrorHandling() method while building messaging. We chose to require the explicit use of such a method to remind developers of the fact that they are responsible for their own error handling, requiring you to explicitly make the decision to forego handling potential errors.

As a general rule, you should always handle your errors . It will lead to faster and quicker identification of problems with your applications if you have error handlers, and generally help you build more robust code.

MessageBuilder.createMessage()

    .toSubject("HelloWorldService")
    .signalling()
    .with("msg", "Hi there!")
    .errorsHandledBy(new ErrorCallback() {
      public boolean error(Message message, Throwable throwable) {
        throwable.printStackTrace();
          return true;
      }
    })
    .sendNowWith(dispatcher);

The addition of error handling at first may put off developers as it makes code more verbose and less-readable. This is nothing that some good practice can’t fix. In fact, you may find cases where the same error handler can appropriately be shared between multiple different calls.

ErrorCallback error = new ErrorCallback() {

  public boolean error(Message message, Throwable throwable) {
    throwable.printStackTrace();
    return true;
  }
}
MessageBuilder.createMessage()
  .toSubject("HelloWorldService")
  .signalling()
  .with("msg", "Hi there!")
  .errorsHandledBy(error)
  .sendNowWith(dispatcher);

The error handler is required to return a boolean value. This is to indicate whether or not Errai should perform the default error handling actions it would normally take during a failure. You will almost always want to return true here, unless you are trying to explicitly surpress some undesirably activity by Errai, such as automatic subject-termination in conversations. But this is almost never the case.

In some applications, it may be necessary or desirable to delay transmission of, or continually stream data to a remote client or group of clients (or from a client to the server). In cases like this, you can utilize the replyRepeating() , replyDelayed() , sendRepeating() and sendDelayed() methods in the MessageBuilder.

Delayed TasksSending a task with a delay is straight forward. Simply utilize the appropriate method (either replyDelayed() or sendDelayed()).

  MessageBuilder.createConversation(msg)

    .toSubject("FunSubject")
    .signalling()
    .noErrorHandling()
    .replyDelayed(TimeUnit.SECONDS, 5); // sends the message after 5 seconds.

or

   MessageBuilder.createMessage()

     .toSubject("FunSubject")
     .signalling()
     .noErrorHandling()
     .sendDelayed(requestDispatcher, TimeUnit.SECONDS, 5); // sends the message after 5 seconds.

A repeating task is sent using one of the MessageBuilder’s repeatXXX() methods. The task will repeat indefinitely until cancelled (see next section).

  MessageBuilder.createMessage()

    .toSubject("FunSubject")
    .signalling()
    .withProvided("time", new ResourceProvider<String>() {
       SimpleDateFormat fmt = new SimpleDateFormat("hh:mm:ss");
       public String get() {
         return fmt.format(new Date(System.currentTimeMillis());
       }
     }
     .noErrorHandling()
     .sendRepeatingWith(requestDispatcher, TimeUnit.SECONDS, 1); //sends a message every 1 second

The above example sends a message very 1 second with a message part called "time", containing a formatted time string. Note the use of the withProvided() method; a provided message part is calculated at the time of transmission as opposed to when the message is constructed.

Cancelling an Asynchronous TaskA delayed or repeating task can be cancelled by calling the cancel() method of the AsyncTask instance which is returned when creating a task. Reference to the AsyncTask object can be retained and cancelled by any other thread.

AsyncTask task = MessageBuilder.createConversation(message)

  .toSubject("TimeChannel").signalling()
  .withProvided(TimeServerParts.TimeString, new ResourceProvider<String>() {
     public String get() {
       return String.valueOf(System.currentTimeMillis());
     }
   }).defaultErrorHandling().replyRepeating(TimeUnit.MILLISECONDS, 100);
    ...
  // cancel the task and interrupt it's thread if necessary.
  task.cancel(true);

The ErraiBus maintains it’s own seperate session management on-top of the regular HTTP session management. While the queue sessions are tied to, and dependant on HTTP sessions for the most part (meaning they die when HTTP sessions die), they provide extra layers of session tracking to make dealing with complex applications built on Errai easier.

ErraiBus implements a JSON-based wire protocol which is used for the federated communication between different buses. The protocol specification encompasses a standard JSON payload structure, a set of verbs, and an object marshalling protocol. The protocol is named J.REP. Which stands for JSON Rich Event Protocol.

All wire messages sent across are assumed to be JSON arrays at the outermost element, contained in which, there are 0..n messages. An empty array is considered a no-operation, but should be counted as activity against any idle timeout limit between federated buses.


Here we see an example of a J.REP payload containing two messages. One bound for an endpoint named "SomeEndpoint" and the other bound for the endpoint "SomeOtherEndpoint" . They both include a payload element "Value" which contain strings. Let’s take a look at the anatomy of an individual message.


The message shows a very vanilla J.REP message. The keys of the JSON Object represent individual message parts , with the values representing their corresponding values. The standard J.REP protocol encompasses a set of standard message parts and values, which for the purposes of this specification we’ll collectively refer to as the protocol verbs.

The following table describes all of the message parts that a J.REP capable client is expected to understand:

PartRequiredJSON TypeDescription

ToSubject

Yes

String

Specifies the subject within the bus, and its federation, which the message should be routed to.

CommandType

No

String

Specifies a command verb to be transmitted to the receiving subject. This is an optional part of a message contract, but is required for using management services

ReplyTo

No

String

Specifies to the receiver what subject it should reply to in response to this message.

Value

No

Any

A recommended but not required standard payload part for sending data to services

PriorityProcessing

No

Number

A processing order salience attribute. Messages which specify priority processing will be processed first if they are competing for resources with other messages in flight. Note: the current version of ErraiBus only supports two salience levels (0 and >1). Any non-zero salience in ErraiBus will be given the same priority relative to 0 salience messages

ErrorMessage

No

String

An accompanying error message with any serialized exception

Throwable

No

Object

If applicable, an encoded object representing any remote exception that was thrown while dispatching the specified service

Federation between buses requires management traffic to negotiate connections and manage visibility of services between buses. This is accomplished through services named ClientBus and ServerBus which both implement the same protocol contracts which are defined in this section.

Both bus services share the same management protocols, by implementing verbs (or commands) that perform different actions. These are specified in the protocol with the CommandType message part. The following table describes these commands:


PartRequiredJSON TypeDescription

CapabilitiesFlags

Yes

String

A comma delimited string of capabilities the bus is capable of us

Subject

Yes

String

The subject to subscribe or unsubscribe from

SubjectsList

Yes

Array

An array of strings representing a list of subjects to subscribe to

Conversations are message exchanges which are between a single client and a service. They are a fundmentally important concept in ErraiBus, since by default, a message will be broadcast to all client services listening on a particular channel.

When you create a reply with an incoming message, you ensure that the message you are sending back is received by the same client which sent the incoming message. A simple example:

@Service

public class HelloWorldService implements MessageCallback {
  public void callback(CommandMessage message) {
    // Send a message to the 'HelloWorldClient' on the client that sent us the
    // the message.
    MessageBuilder.createConversation(message)
      .toSubject("HelloWorldClient")
      .signalling()
      .with("text", "Hi There! We're having a reply!")
      .noErrorHandling().reply();
    });
  }
}

Note that the only difference between the example in the previous section and this is the use of the createConversation() method with MessageBuilder.

ErraiBus has support for WebSocket-based communication. When WebSockets are enabled, capable web browsers will attempt to upgrade their COMET-based communication with the server-side bus to use a WebSocket channel.

There are three different ways the bus can enable WebSockets. The first uses a sideband server, which is a small, lightweight server which runs on a different port from the application server. The second is native JBoss AS 7-based integration and the third is to rely in JSR-356 support in WildFly. Of course, you only need to configure one of these three options!

This is an alternative approach to the sideband server described in the previous chapter. Make sure to NOT configure both! It is currently necessary to use the native connector in JBoss AS for WebSockets to work. So the first step is to configure your JBoss AS instance(s) to use the native connector by changing the domain/configuration/standalone.xml or domain/configuration/domain.xml file as follows:


<subsystem xmlns="urn:jboss:domain:web:1.1" default-virtual-server="default-host" native="false">

to:


<subsystem xmlns="urn:jboss:domain:web:1.1" default-virtual-server="default-host" native="true">

You will then need to configure the servlet in your application’s web.xml which will provide WebSocket upgrade support within AS7.

Add the following to the web.xml :


<context-param>
  <param-name>websockets-enabled</param-name>
  <param-value>true</param-value>
</context-param>

<context-param>
  <param-name>websocket-path-element</param-name>
  <param-value>in.erraiBusWS</param-value>
</context-param>

This will tell the bus to enable web sockets support. The websocket-path-element specified the path element within a URL which the client bus should request in order to negotiate a websocket connection. For instance, specifying in.erraiBusWS as we have in the snippit above, will result in attempted negotiation at http://<your_server>:<your_port>/<context_path>/in.erraiBusWS. For this to have any meaningful result, we must add a servlet mapping that will match this pattern:


<servlet>
  <servlet-name>ErraiWSServlet</servlet-name>
  <servlet-class>org.jboss.errai.bus.server.servlet.JBossAS7WebSocketServlet</servlet-class>
  <load-on-startup>1</load-on-startup>
</servlet>

<servlet-mapping>
  <servlet-name>ErraiWSServlet</servlet-name>
  <url-pattern>*.erraiBusWS</url-pattern>
</servlet-mapping>

Do not remove the regular ErraiBus servlet mappings!

When configuring ErraiBus to use WebSockets on JBoss AS, you do not remove the existing servlet mappings for the bus. The WebSocket servlet is in addition to your current bus servlet. This is because ErraiBus always negotiates WebSocket sessions over the COMET channel.

Important dependency

Also make sure to deploy the required errai-bus-jboss7-websocket.jar to your server. If you’re using Maven simply add the following dependency to your pom.xml file:


    <dependency>
        <groupId>org.jboss.errai</groupId>
        <artifactId>errai-bus-jboss7-websocket</artifactId>
        <version>${errai.version}</version>
    </dependency>

Errai provides two implementations for this:

Make sure to add the following project dependency:


<!-- For JSR-356 without depending on Weld -->
<dependency>
    <groupId>org.jboss.errai</groupId>
    <artifactId>errai-bus-jsr356-websocket</artifactId>
    <version>${errai.version}</version>
</dependency>

<!-- For JEE environment with Weld-->
<dependency>
    <groupId>org.jboss.errai</groupId>
    <artifactId>errai-bus-jsr356-websocket-weld</artifactId>
    <version>${errai.version}</version>
</dependency>

To configure ErraiBus that WebSocket communication should be used, define the following in your web.xml


<context-param>
    <param-name>websockets-enabled</param-name>
    <param-value>true</param-value>
</context-param>

You can also define filters when using the JSR-356 WebSocket implementation. These filters will be executed for each received ErraiBus message on the server. Your filters need to implement org.jboss.errai.bus.server.websocket.jsr356.filter.WebSocketFilter and must be configured in your appliations’s web.xml as an ordered comma separated list:


<context-param>
    <param-name>errai-jsr-356-websocket-filter</param-name>
    <param-value>foo.bar.FooFilter,foo.bar.BarFilter</param-value>
</context-param>

Errai supports Websocket security (wss) for two deployment scenarios.

Shadow Services is a Service that will get invoked when there is no longer a connection with the server. This is particular helpful when developing an application for mobile. To create a Shadow Service for a specific Services all you have to do is annotate a new client side implementation with the @ShadowService:

@ShadowService

public class SignupShadowService implements MessageCallback {
   @Override
   public void callback(Message message) {
   }
}

Also when you have a RPC based Service you can just add @ShadowService on a client side implementation to configure it to be the service to get called when there is no network:

@ShadowService

public class SignupServiceShadow implements SignupService {
   @Override
   public User register(User newUserObject, String password) throws RegistrationException {
   }
}

In this shadow service we can create logic that will deal with the temporary connection loss. For instance you could save the data that needs to get send to the server with JPA on the client and then when the bus get online again sent the data to the server.

Errai includes a bus monitoring application, which allows you to monitor all of the message exchange activity on the bus in order to help track down any potential problems It allows you to inspect individual messages to examine their state and structure.

To utilize the bus monitor, you’ll need to include the _errai-tools _ package as part of your application’s dependencies. When you run your application in development mode, you will simply need to add the following JVM options to your run configuration in order to launch the monitor: -Derrai.tools.bus_monitor_attach=true


The monitor provides you a real-time perspective on what’s going on inside the bus. The left side of the main screen lists the services that are currently available, and the right side is the service-explorer, which will show details about the service.

To see what’s going on with a specific service, simply double-click on the service or highlight the service, then click "Monitor Service…". This will bring up the service activity monitor.


The service activity monitor will display a list of all the messages that were transmitted on the bus since the monitor became active. You do not need to actually have each specific monitor window open in order to actively monitor the bus activity. All activity on the bus is recorded.

The monitor allows you select individual messages, an view their individual parts. Clicking on a message part will bring up the object inspector, which will allow you to explore the state of any objects contained within the message, not unlike the object inspectors provided by debuggers in your favorite IDE. This can be a powerful tool for looking under the covers of your application.