Quick Start

Server message example

Sending a message from one agent to another is a basic HMTP use case. Typical applications include chat text, game updates, Atom/RSS updates, pub/sub messaging, and event notification for administration consoles. In this example, a servlet send a message to an internal logging service.

The sending agent sends a message using message with the jid address of the target agent (to), and a message payload. The receiving agent implements the message method to receive messages. Since HMTP is a streaming API like java.io.OutputStream, the sending method is the same as the receiving method, message.

HMTP messages use a JID (Jabber ID) for addressing, which looks like user@domain or user@domain/resource. The second user@domain/resource is used for dynamic agent, e.g. a user logged into messaging with a cellphone.

The message payload can be any serializable object. Because Hessian is available for many languages, the message can easily interoperate with RIA platforms like Flash.

Since HMTP uses a broker/agent or hub-and-spoke messaging model, all messages go through the broker to be routed to the target agent. Each service and client will register an agent with the broker and receive a unique JID for the routing.

Writing a HMTP client involves the following steps:

1. Obtaining the BamConnectionFactory (usually with @In)
2. Creating a BamConnection (which automatically registers an agent)
3. Optionally: setting message handlers to receive messages
4. Sending messages
5. Close the connection

In the example, the servlet creates a BamConnection from a BamConnectionFactory and sends the message. Opening a connection registers an agent with the broker. When the connection closes, it will unregister the agent. The Web Beans @javax.webbeans.In injection annotation provides access to the BamConnectionFactory.

package example;

import javax.servlet.*;
import javax.webbeans.In;
import com.caucho.bam.BamConnectionFactory;
import com.caucho.bam.BamConnection;

public class TestServlet extends GenericServlet
{
  @In BamConnectionFactory _factory;

  public void service(ServletRequest req, ServletResponse response)
  {
    BamConnection conn = _factory.getConnection("demo@localhost", null);

    try {
      conn.message("test@localhost", "Hello, world!");
    } finally {
      conn.close();
    }
  }
}

Writing a HMTP Service involves the following steps:

1. Implementing BamService (usually by extending GenericService)
2. Getting the BamBroker (either with @In or done by GenericService automatically)
3. Registering the BamService with BamBroker using addService (done by GenericService automatically)
4. Receiving messages to the implemented BamStream (or overriding BamStream methods in GenericService)
5. Sending messages to other agents using the broker's getBrokerStream.

By extending GenericService, 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.hemp.broker.GenericService;
import java.io.Serializable;
import java.util.logging.*;

public class LogService extends GenericService
{
  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 HMTP configuration consists of the broker and the service configured with Resin IoC. Because the LogService extends GenericService, it will automatically register itself with the broker.

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

    <bean name="test@localhost" class="example.LogService"/>

</web-app>

Client queryGet (RPC) example

Remote calls are another primary use for HMTP. In this example, we just query a service for some basic information. In HMTP, queries are bidirectional: the server can also query the client. And the application can also use the messaging in the previous example.

To implement the server side of an RPC call, the service implements sendQueryGet or sendQuerySet, and examines the query to see if it understands the query class. If the service does not understand the query, it will return false (or call the super method for GenericService). If a query returns true, it must either send a query response or query error with the same id, so the waiting client will always receive a response.

The type-based query system gives enormous flexibility in creating services. Services can be mash-ups of capabilities just by adding new query types.

package example;

import com.caucho.hemp.broker.GenericService;
import java.io.Serializable;
import java.util.logging.*;

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

  @Override
  public boolean sendQueryGet(long id, String to, String from,
                              Serializable query)
  {
    if (query instanceof TestQuery) {
      getBrokerStream().sendQueryResult(id, to, from, "hello response");

      return true;
    }
    else {
      return super.sendQueryGet(id, to, from, query);
    }
  }
}

The remote client, HmtpClient, extends the BamConnection API to provide extra calls for connecting and logging on to the server. The message passing calls are 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.

This example sends a RPC query to test@localhost. RPC calls in HMTP are typed. Each query class might execute a different query on the server. In this case we create a trivial TestQuery.

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();
  }
}

The configuration for a remote service now has three components:

1. The BamBroker, implemented by HempBroker
2. Any registered BamService, e.g. the TestService
3. The exposed HMTP service, implemented with HempServlet

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

    <bean name="test@localhost" class="example.TestService"/>

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

</web-app>

Brokered Agent Messaging (BAM)

XMPP (Jabber)

Packet types

Message Packets

The main Message packet contains a target ("to"), a sender ("from"), and a payload ("value"). In HMTP, the payload can be any serializable value. Example messages could be IM text messages, administration console graph, game updates, or updated stock quotes. Since HMTP 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

Addressing (JIDs)

API

Client API

package com.caucho.bam;

public interface BamConnection
{
  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();
}

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 void message(String to, String from, Serializable value);
  
  public void messageError(String to, String from, Serializable value,
                           BamError error);
			       
  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);
			       
  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 BamStream findAgent(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);
}