Overview

Quick Start Examples

Queued Messages

Sending a message from a client to a named service is a simple and important use of BAM. Typical applications include message queuing, chat text, game updates, Atom/RSS updates, pub/sub messaging, and event notification for administration consoles. In out example, a servlet sends a message to an internal logging service.

The sending agent calls message() to send a message with the jid address of the target agent (to), and a message payload. The broker routes the message based on the address and calls the message() method on the target service to process the message. As the service processes the message, the client returns from its message() call and continues processing. If a new message arrives for the service, the broker will queue the message until the service is ready to process it. The queue isolates the producing thread from the service, improving response time and even allowing to be processed on a separate machine or cluster as a batch job.

The message can be any serializable object appropriate to the service, either a custom Java model bean, or an XML string, or a defined protocol message, like a Jabber IM message. This flexibility lets services define messages appropriate to the application, and avoids tying the service into knots trying to conform to a restricted encoding like SOAP. If remoting is used, the remoting protocol might restrict the possible messages. HMTP (Hessian) will allow any serializable object, while XMPP (Jabber) is restricted to XML and ImMessage.

Since BAM messages are addressible and routed through the broker, clients have flexibility in choosing their destination at runtime. BAM messages use a JID (Jabber ID) for addressing, which looks like service@domain or service@domain/resource. The second service@domain/resource is used for dynamic agent, e.g. a user logged into messaging with a cellphone.

Writing a BAM client involves the following steps:

1. Create a BamClient
2. Sending messages

In the example, the servlet creates a BamClient and sends the message. The special jid syntax "service@" uses the local domain as a destination.

package example;

import javax.servlet.*;
import com.caucho.bam.BamClient;

public class TestClient extends GenericServlet
{
  private BamClient _client = new BamClient();

  public void service(ServletRequest req, ServletResponse response)
  {
    _client.message("test@", "Hello, world!");
  }
}

Writing a BAM message/queuing service involves the following steps:

1. Implementing BamService (usually by extending SimpleBamService)
2. Configuring the BamService using <bam-service>, which will automatically register the service with the BamBroker.
3. Receiving messages by overriding BamStream methods in SimpleBamService.

By configuring <bam-service>, the service automatically gains a queuing ability. The broker will queue the message and spawn a new thread before calling the service's message, in order to isolate the receiver from the sender. Advanced applications can disable the queue if appropriate.

package example;

import com.caucho.bam.SimpleBamService;
import java.io.Serializable;
import java.util.logging.*;

public class LogService extends SimpleBamService
{
  private static final Logger log
    = Logger.getLogger(LogService.class.getName());

  @Override
  public void message(String to, String from, Serializable value)
  {
    log.info(this + " message from=" + from + " value=" + value);
  }
}

The BAM configuration the service configured with Resin IoC.

<web-app xmlns="http://caucho.com/ns/resin">

    <bam-service name="test" class="example.LogService"/>

</web-app>

Client queryGet (RPC) example

Remote calls in BAM can query or update a service based on the type of the query message. Since the query is typed, a service can be defined by the set of query types it understands and even mixin multiple capabilities, like implementing both a chat and a pub/sub service. In this example, we just query a service for some basic information.

We'll use a remote client to show how BAM can be used as a remote service as well a local organization. HmtpClient, which extends the same BamConnection API as the local BamClient implements BAM using Hessian as the wire protocol. For local messages, we could use BamClient instead. Once the connection is established, the remaining code is identical.

When you create a HmtpClient, you'll send it the URL of the HMTP service, then call connect() and login to authenticate. The login() method will register an agent with the broker, letting the client send and receive messages. The example sends a single TestQuery query to the test@localhost service.

package example;

import com.caucho.hmtp.client.HmtpClient;

public class TestClient
{
  public static void main(String []args)
    throws Exception
  {
    HmtpClient client = new HmtpClient("http://localhost:8080/hmtp");
    client.connect();
    client.login("user@localhost", null);

    Object value = client.queryGet("test@localhost", new TestQuery());

    System.out.println(value);

    client.close();
  }
}

To implement the server side of an RPC call, the service implements queryGet or querySet and examines the query to see if it understands the query class. To simplify the query dispatching, SimpleBamService introspects the methods looking for @QueryGet annotations and creating a map of query types, in this case TestQuery.

For a query, the service must always send a QueryResult message or a QueryError message with the same id back to the caller to match responses to the calls. If the service understands the query, it will send a result message and return true. If it does not understand the query, it will return false, which tells the broker to send a query error message.

The id matches responses to the corresponding queries. Since BAM is a bidirectional streaming architecture, queries can be unordered and start from either direction. The id turns this unordered mess into a coherent request-response pattern for RPC-style calls.

package example;

import com.caucho.bam.SimpleBamService;
import com.caucho.bam.annotation.QueryGet;

public class TestService extends SimpleBamService
{
  @QueryGet
  public boolean testQueryGet(long id, String to, String from,
                              TestQuery query)
  {
    getBrokerStream().sendQueryResult(id, to, from, "hello response");

    return true;
  }
}

The configuration for a service now has two components:

1. Any registered BamService, e.g. the TestService
2. The exposed HMTP service protocol, implemented with HempServlet

The BAM service itself does not know or care that it's being called remotely. The remote HMTP servlet exists only so the remote client can login to the local broker.

<web-app xmlns="http://caucho.com/ns/resin">

    <bam-service name="test" class="example.TestService"/>

    <servlet-mapping url-pattern="/hmtp"
                     servlet-class="com.caucho.hemp.servlet.HempServlet"/>

</web-app>

Brokered Agent Messaging (BAM)

Addressing (JIDs)

bam-service

Resin Services

JMS

The JMS service forwards BAM messages to a JMS queue, wrapping each message in a JMS ObjectMessage. Applications can use this BAM to JMS bridge to queue messages from a Java client.

<web-app xmlns="http://caucho.com/ns/resin">

  <jms-connection-factory name="jms_cf" uri="resin:"/>
  <jms-queue name="queue" uri="memory:"/>

  <bam-service name="jms" uri="caucho.jms:">
    <init>
      <connection-factory>${jms_cf}</connection-factory>
      <queue>${queue}</queue>
    </init>
  </bam-service>
  
</web-app>

Logging

The logging service adds BAM messages to a java.util.logging Logger. An application could use the logging service as a chat-room recording or debugging service.

<web-app xmlns="http://caucho.com/ns/resin">

  <bam-service name="log" uri="caucho.log:">
    <init>
      <name>com.foo.chat</name>
      <level>info</level>
    </init>
  </bam-service>
  
</web-app>

Mail

The mail service sends BAM messages to an email address. This can be used to notify any issues with an application that may need administration.

<web-app xmlns="http://caucho.com/ns/resin">

  <bam-service name="mail" uri="caucho.mail:">
    <init>
      <to>bam@foo.com</name>
      <subject>BAM Notification</subject>
      <properties>
        mail.from=bamservice@foo.com
      </properties>
    </init>
  </bam-service>
  
</web-app>

PHP

The php service configures a PHP script as a service handler.

<web-app xmlns="http://caucho.com/ns/resin">

  <bam-service name="php" uri="caucho.php:">
    <init>
      <script>WEB-INF/php/php-service.php</script>
    </init>
  </bam-service>
  
</web-app>

<?php

bam_message($to, $from, $message)
{
  resin_debug($message);
}

bam_dispatch();
  
?>

Protocols

Local (JVM calls)

Local JVM applications can use BAM to organize internal applications like queuing consumers to replace JMS, or service-oriented architectures. Because BAM passes messages by reference, it avoids the cpu and memory overhead of serialization, improving performance.

The local clients can take advantage of BAM's federated addressing and can send messages to foreign machines while logging into the local broker.

HMTP (Hessian)

HMTP (Hessian Message Transport Protocol) is a streaming mode of the Hessian protocol supporting BAM. Since Hessian can serialize any Java object, it can support all of BAM's capabilities for remote services and since Hessian has been ported to many languages including Flash/ActionScript, multilanguage clients can use BAM directly though HMTP.

XMPP (Jabber)

BAM is an adaptation of the XMPP (Jabber) instant messaging protocol. Where XMPP (Xml Messaging and Presence Protocol) is based on XML, HMTP (Hessian Message Transport Protocol) is based on Hessian. Because BAM is designed to follow XMPP, its architecture and protocols are essentially identical until the very lowest layer.

Because of the close relationship to XMPP, you may want to browse the XMPP specifications for a deeper understanding of how HMTP works. Since XMPP is only a wire protocol, not an API, it does not include all of the HMTP classes, but the architecture remains the same.

The primary advantages HMTP offers over XMPP include the performance advantages of Hessian over XML, and more importantly a more strict layering than XMPP provides. Because the payloads of the HMTP messages are all Serializable, applications have enormous flexibility in developing their own messages using application objects. In contrast, XMPP messages are always XML, so applications are not only restricted to XML data, but also must create their own XML parsers and formatters.

Packet types

Message Packets

The main Message packet contains a target ("to"), a sender ("from"), and a payload ("value"). In BAM, the payload can be any serializable value. Example messages could be IM text messages, queued tasks, administration console graph, game updates, or updated stock quotes. Since BAM is bidirectional, messages can flow to and from any client.

• Message - sends a message to a resource
• MessageError - sends a message error to a resource

Query Packets

Query packages are RPC get and set packets with a matching response or error. Because the query will always have a matching response packet or an error packet, clients can either block for the result or attach a callback.

Like the other packets, queries are bidirectional, so a service can query a client as well as the usual client querying the server.

Query packets have an associated id field to match requests with responses. The client will increment the id for each new query.

• QueryGet - sends an information request
• QuerySet - sends an action query
• QueryResponse - returns a response
• QueryError - returns an error

Presence Packets

Presence packets send specialized information for subscription notification. Many applications will not need to use any presence packets at all.

• Presense - sends a presence (login) notification
• PresenseUnavailable - sends unavailable (logout) notification
• PresenseProbe - query probe for IM clients
• PresenseSubscribe - request to subscribe to a service
• PresenseSubscribed - acknowledgement of a subscription
• PresenseUnsubscribe - notification of an unsubscription
• PresenseUnsubscribed - notification of an unsubscription
• PresenseError - error message

API

Client API

package com.caucho.bam;

public class BamClient implements BamConnection
{
  public BamClient();
  public BamClient(String uid, String password);
  public BamClient(String uid, String password, String resource);

  String getJid();

  // message
  void message(String to, Serializable value);

  // rpc
  Serializable queryGet(String to, Serializable query);
  Serializable querySet(String to, Serializable query);
  void queryGet(String to, Serializable query, BamQueryCallback callback);
  void querySet(String to, Serializable query, BamQueryCallback callback);

  // presence
  void presence(Serializable data);
  void presence(String to, Serializable data);
  void presenceUnavailable(Serializable data);
  void presenceUnavailable(String to, Serializable data);
  void presenceProbe(String to, Serializable data);
  void presenceSubscribe(String to, Serializable data);
  void presenceSubscribed(String to, Serializable data);
  void presenceUnsubscribe(String to, Serializable data);
  void presenceUnsubscribed(String to, Serializable data);
  void presenceError(String to, Serializable data, BamError error);
  
  // callback handler for receiving messages
  void setStreamHandler(BamStream handler);

  // raw stream to return rpc responses
  BamStream getBrokerStream();
}

package com.caucho.bam;

public interface BamConnection
{
  String getJid();

  boolean isClosed();
  void close();

  // message
  void message(String to, Serializable value);

  // rpc
  Serializable queryGet(String to, Serializable query);
  Serializable querySet(String to, Serializable query);
  void queryGet(String to, Serializable query, BamQueryCallback callback);
  void querySet(String to, Serializable query, BamQueryCallback callback);

  // presence
  void presence(Serializable data);
  void presence(String to, Serializable data);
  void presenceUnavailable(Serializable data);
  void presenceUnavailable(String to, Serializable data);
  void presenceProbe(String to, Serializable data);
  void presenceSubscribe(String to, Serializable data);
  void presenceSubscribed(String to, Serializable data);
  void presenceUnsubscribe(String to, Serializable data);
  void presenceUnsubscribed(String to, Serializable data);
  void presenceError(String to, Serializable data, BamError error);

  // callback handler for receiving messages
  void setStreamHandler(BamStream handler);
  
  // raw stream to return rpc responses
  BamStream getBrokerStream();
}

package com.caucho.bam;

public interface BamConnectionFactory
{
  BamConnection getConnection(String uid, String password);
  
  BamConnection getConnection(String uid, String password, String resourceId);
}

package com.caucho.bam;

public interface BamQueryCallback
{
  void onQueryResult(String to, String from, Serializable value);
  
  void onQueryError(String to, String from, Serializable value,
                    BamError error);
}

Remote Client API

HmtpClient is the remote client API for Java clients. Most of the methods are extended from BamConnection. The additional method provide some control for connection and login. Once the client is logged in, applications will typically use BamConnection methods to send messages and set handlers to receive messages.

package com.caucho.bam;

public class HmtpClient implements BamConnection
{
  public HmtpClient(String url);

  public void connect() throws IOException;

  public void login(String uid, String password);

  // BamConnection methods
  String getJid();

  boolean isClosed();
  void close();
  
  void setStreamHandler(BamStream handler);

  void message(String to, Serializable value);

  Serializable queryGet(String to, Serializable query);
  Serializable querySet(String to, Serializable query);

  void queryGet(String to, Serializable query, BamQueryCallback callback);
  void querySet(String to, Serializable query, BamQueryCallback callback);

  void presence(Serializable data);
  void presence(String to, Serializable data);
  void presenceUnavailable(Serializable data);
  void presenceUnavailable(String to, Serializable data);
  void presenceProbe(String to, Serializable data);
  void presenceSubscribe(String to, Serializable data);
  void presenceSubscribed(String to, Serializable data);
  void presenceUnsubscribe(String to, Serializable data);
  void presenceUnsubscribed(String to, Serializable data);
  void presenceError(String to, Serializable data, BamError error);
  
  BamStream getBrokerStream();
}

Protocol(Packet) API

package com.caucho.bam;

public interface BamStream
{
  public String getJid();

  // message
  public void message(String to, String from, Serializable value);
  public void messageError(String to, String from, Serializable value,
                           BamError error);

  // rpc
  boolean queryGet(long id, String to, String from, Serializable query);
  boolean querySet(long id, String to, String from, Serializable query);
  void queryResult(long id, String to, String from, Serializable value);
  void queryError(long id, String to, String from, Serializable query,
                  BamError error);

  // presence
  void presence(String to, String from, Serializable data);
  void presenceUnavailable(String to, String from, Serializable data);
  void presenceProbe(String to, String from, Serializable data);
  void presenceSubscribe(String to, String from, Serializable data);
  void presenceSubscribed(String to, String from, Serializable data);
  void presenceUnsubscribe(String to, String from, Serializable data);
  void presenceUnsubscribed(String to, String from, Serializable data);
  void presenceError(String to, String from, Serializable data,
                     BamError error);
}

Service APIs

package com.caucho.bam;

public interface BamBroker extends BamConnectionFactory
{
  BamStream getBrokerStream();

  void addService(BamService service);
  void removeService(BamService service);

  void addServiceManager(ServiceManager manager);
}

package com.caucho.bam;

public interface BamService
{
  public String getJid();
  
  public BamStream getAgentStream();
  
  public boolean startAgent(String jid);
  public boolean stopAgent(String jid);

  public void onAgentStart(String jid);
  public void onAgentStop(String jid);

  public BamStream getAgentFilter(BamStream stream);
  public BamStream getBrokerFilter(BamStream stream);
}

package com.caucho.bam;

public interface BamServiceManager
{
  public BamService findService(String jid);
}

package com.caucho.bam;

abstract public class SimpleBamService implements BamService, BamStream
{
  public String getJid();

  // message
  public void message(String to, String from, Serializable value);
  public void messageError(String to, String from, Serializable value,
                           BamError error);

  // rpc
  boolean queryGet(long id, String to, String from, Serializable query);
  boolean querySet(long id, String to, String from, Serializable query);
  void queryResult(long id, String to, String from, Serializable value);
  void queryError(long id, String to, String from, Serializable query,
                  BamError error);

  // presence
  void presence(String to, String from, Serializable data);
  void presenceUnavailable(String to, String from, Serializable data);
  void presenceProbe(String to, String from, Serializable data);
  void presenceSubscribe(String to, String from, Serializable data);
  void presenceSubscribed(String to, String from, Serializable data);
  void presenceUnsubscribe(String to, String from, Serializable data);
  void presenceUnsubscribed(String to, String from, Serializable data);
  void presenceError(String to, String from, Serializable data,
                     BamError error);

  // BamService methods
  public BamStream getAgentStream();
  
  public boolean startAgent(String jid);
  public boolean stopAgent(String jid);

  public void onAgentStart(String jid);
  public void onAgentStop(String jid);

  public BamStream getAgentFilter(BamStream stream);
  public BamStream getBrokerFilter(BamStream stream);
}

package com.caucho.bam.annotation;

public @interface Message {}
public @interface MessageError {}

public @interface QueryGet {}
public @interface QuerySet {}
public @interface QueryResult {}
public @interface QueryError {}

public @interface Presence {}
public @interface PresenceProbe {}
public @interface PresenceUnavailable {}
public @interface PresenceSubscribe {}
public @interface PresenceSubscribed {}
public @interface PresenceUnsubscribe {}
public @interface PresenceUnsubscribed {}
public @interface PresenceError {}