Quick Start

Server sendMessage 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 sendMessage with the jid address of the target agent (to), and a message payload. The receiving agent implements the sendMessage method to receive messages. Since HMTP is a streaming API like java.io.OutputStream, the sending method is the same as the receiving method, sendMessage.

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 HmtpConnectionFactory (usually with @In)
2. Creating a HmtpConnection (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 HmtpConnection from a HmtpConnectionFactory 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 HmtpConnectionFactory.

package example;

import javax.servlet.*;
import javax.webbeans.In;
import com.caucho.hmtp.HmtpConnectionFactory;
import com.caucho.hmtp.HmtpConnection;

public class TestServlet extends GenericServlet
{
  @In HmtpConnectionFactory _factory;

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

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

Writing a HMTP Service involves the following steps:

1. Implementing HmtpService (usually by extending GenericService)
2. Getting the HmtpBroker (either with @In or done by GenericService automatically)
3. Registering the HmtpService with HmtpBroker using addService (done by GenericService automatically)
4. Receiving messages to the implemented HmtpAgentStream (or overriding HmtpStream 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 sendMessage, 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 sendMessage(String to, String from, Serializable value)
  {
    log.info(this + " sendMessage 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 class="com.caucho.hemp.broker.HempBroker"/>

    <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 HmtpConnection 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 HmtpBroker, implemented by HempBroker
2. Any registered HmtpService, e.g. the TestService
3. The exposed HMTP service, implemented with HempServlet

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

    <bean class="com.caucho.hemp.broker.HempBroker"/>

    <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.hmtp;

public interface HmtpConnection
{
  String getJid();

  boolean isClosed();
  void close();
  
  void setMessageHandler(HmtpMessageStream handler);
  void setQueryHandler(HmtpQueryStream handler);
  void setPresenceHandler(HmtpPresenceStream handler);

  void sendMessage(String to, Serializable value);

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

  void queryGet(String to, Serializable query, HmtpQueryCallback callback);
  void querySet(String to, Serializable query, HmtpQueryCallback 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, HmtpError error);
  
  HmtpStream getBrokerStream();
}

package com.caucho.hmtp;

public interface HmtpConnectionFactory
{
  HmtpConnection getConnection(String uid, String password);
}

package com.caucho.hmtp;

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

Remote Client API

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

package com.caucho.hmtp;

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

  public void connect() throws IOException;

  public void login(String uid, String password);

  // HmtpConnection methods
  String getJid();

  boolean isClosed();
  void close();
  
  void setMessageHandler(HmtpMessageStream handler);
  void setQueryHandler(HmtpQueryStream handler);
  void setPresenceHandler(HmtpPresenceStream handler);

  void sendMessage(String to, Serializable value);

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

  void queryGet(String to, Serializable query, HmtpQueryCallback callback);
  void querySet(String to, Serializable query, HmtpQueryCallback 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, HmtpError error);
  
  HmtpStream getBrokerStream();
}

Protocol(Packet) API

package com.caucho.hmtp;

public interface HmtpMessageStream
{
  public void sendMessage(String to, String from, Serializable value);
  
  public void sendMessageError(String to, String from, Serializable value,
			       HmtpError error);
}

package com.caucho.hmtp;

public interface HmtpQueryStream
{
  boolean sendQueryGet(long id, String to, String from, Serializable query);
  boolean sendQuerySet(long id, String to, String from, Serializable query);

  void sendQueryResult(long id, String to, String from, Serializable value);
  void sendQueryError(long id, String to, String from, Serializable query,
                      HmtpError error);
}

package com.caucho.hmtp;

public interface HmtpPresenceStream
{
  void sendPresence(String to, String from, Serializable []data);
  void sendPresenceUnavailable(String to, String from, Serializable []data);
  void sendPresenceProbe(String to, String from, Serializable []data);
  void sendPresenceSubscribe(String to, String from, Serializable []data);
  void sendPresenceSubscribed(String to, String from, Serializable []data);
  void sendPresenceUnsubscribe(String to, String from, Serializable []data);
  void sendPresenceUnsubscribed(String to, String from, Serializable []data);
  void sendPresenceError(String to, String from, Serializable []data,
                         HmtpError error);
}

package com.caucho.hmtp;

public interface HmtpStream
  extends HmtpMessageStream, HmtpQueryStream, HmtpPresenceStream
{
}

package com.caucho.hmtp;

public interface HmtpAgentStream extends HmtpStream
{
  Strin getJid();
}

Service APIs

package com.caucho.hmtp;

public interface HmtpBroker extends HmtpConnectionFactory
{
  HmtpStream getBrokerStream();

  void addService(HmtpService service);
  void removeService(HmtpService service);

  void addServiceManager(ServiceManager manager);
}

package com.caucho.hmtp;

public interface HmtpService
{
  public String getJid();
  
  public HmtpAgentStream getAgentStream();
  
  public HmtpAgentStream findAgent(String jid);

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

  public HmtpAgentStream getAgentFilter(HmtpAgentStream stream);
  public HmtpStream getBrokerFilter(HmtpStream stream);
}

package com.caucho.hmtp;

public interface HmtpServiceManager
{
  public HmtpService findService(String jid);
}