Use Stomp in your PHP application

Introduction

This project is a PHP Stomp clients that besides it implements the Stomp protocol fully, adds some ActiveMQ specific features that could make your messaging from PHP easier.

Getting Started

Obtain the source of the library by downloading the distribution and add its content to your include_path. Alternatively, you can grab the source and add src/main folder to your include_path.

Examples are located in src/examples folder. Before running them, be sure you have installed this library properly and you have started ActiveMQ broker (recommended version 5.5.0 or higher) with Stomp connector enabled.

You can start the first example by running

cd examples
php first.php

Also, be sure to check comments in the particular examples for some special configuration steps (if needed).

First Example

The simplest example shows how you can connect to the ActiveMQ message broker and send/receive a message from the queue.

<?
    // include a library
    require_once("Stomp.php");
    // make a connection
    $con = new Stomp("tcp://localhost:61613");
    // connect
    $con->connect();
    // send a message to the queue
    $con->send("/queue/test", "test");
    echo "Sent message with body 'test'\n";
    // subscribe to the queue
    $con->subscribe("/queue/test");
    // receive a message from the queue
    $msg = $con->readFrame();

    // do what you want with the message
    if ( $msg != null) {
        echo "Received message with body '$msg->body'\n";
        // mark the message as received in the queue
        $con->ack($msg);
    } else {
        echo "Failed to receive a message\n";
    }

    // disconnect
    $con->disconnect();
?>

After the execution this script should print

$ php first.php 
Sent message with body 'test'
Received message with body 'test'

and it shows how easy it is to do asynchronous messaging using Stomp.

Transactions

Stomp transactions allow you to send messages and acknowledgments in atomic operations. For more information on transactions, take a look at the following article.

Now let's go through the basics of transaction usage in PHP Stomp client. To start the transaction, you should use begin() method with transaction id as a parameter.

<?
    $stomp->begin("tx1");
?>    

In the similar way you can commit or abort your transactions

<?
    $con->commit("tx1");    
    $stomp->abort("tx1");    
?>

To send a message in a transaction, you should pass an extra transaction header, like this

        
<?
    $stomp->send("/queue/transactions", "message text", array("transaction" => "tx1"));
?>

To send an acknowledgment in a transaction, pass transaction id as an extra argument to the ack() method, like this

<?
    $stomp->ack($message, "tx1");
?>

The full example of the transaction usage with PHP Stomp client can be found in examples/transactions.php.

Durable Subscribers

Durable topic subscribers are not part of the Stomp protocol, but an ActiveMQ feature that allow topic subscribers to receive messages sent to the topic while there were offline.

Stomp protocol implementation in ActiveMQ enables this feature for Stomp subscribers as well. To use this feature, first you need to define a client id your consumer will use, like this

<?
    $consumer = new Stomp("tcp://localhost:61613");
    $consumer->clientId = "test";
?>

Next, you client need to provide this id while subscribing to the topic. This is done by passing special activemq.subcriptionName header along with the SUBSCRIBE frame. But once, you've set the client id to your consumer, PHP Stomp client will do this work for you. So you can subscribe to the topic, just as you'd normally do

<?
    $consumer->subscribe("/topic/test");
?>

One more important thing is that you need to send persistent messages to the topic in you want them delivered to the delivered to the durable consumers that are currently offline. You can do that by passing a persistent header along with your messages

<?
    $producer->send("/topic/test", "test1", array('persistent'=>'true'));
?>    

The full example of the PHP Stomp client durable subscribers can be found in examples/durable.php.

Connectivity

This Stomp client supports various connectivity methods helping you achieve more flexibility in how you connect to your broker (or brokers)

Failover

The Failover transport allows you to set URIs of multiple brokers and define basic parameters of how the client should attempt to connect to those brokers (in case it fails to connect for the first time). More information on the Failover transport and how ActiveMQ implements it could be found at the following URL.

PHP Stomp client uses the same configuration syntax as ActiveMQ. So instead of connecting to the broker by using standard broker URL, such as

<?
    $this->con = new Stomp("tcp://localhost:61613");
?>    

you can use composite URLs. For example

<?
    $this->con = 
     new Stomp("failover://(tcp://localhost:61614,tcp://localhost:61613)?randomize=false");
?>

Now instead of trying to connect to the single broker, the client will execute reconnection logic if needed. For the previous example, if we don't have a broker connector running on port 61614 it will try to connect on port 61613. If you look at PHP , you'll find messages similar to these

Warning: fsockopen(): unable to connect to tcp://localhost:61614 
(Connection refused) in
 /home/dejan/development/stomp/src/main/php4/Stomp.php on line 201

Warning: Could not connect to localhost:61614 (1/10) in 
/home/dejan/development/stomp/src/main/php4/Stomp.php on line 204

This connectivity option allows you to use this client in a clustered broker environment

Currently, you can only use the randomize parameter to specify whether you want to use random algorithm when choosing broker from the list. The client will try to connect 10 time before it fails. The more connection options will be implemented soon.

SSL

You can instruct this client to use SSL when connecting to brokers that support this feature. For example, you can add another transport connector for ActiveMQ like this

    <transportConnector name="stomp+ssl" uri="stomp+ssl://localhost:61612"/>

and then you can connect to it by using the following syntax for the broker URL

<?
    $this->con = new Stomp("ssl://localhost:61612");
?>    

SSL connections can be used as a part of failover URLs. For example

<?
    $this->con = 
      new Stomp("failover://(tcp://localhost:61614,ssl://localhost:61612)?randomize=false");
?>      

The full example of the PHP Stomp client connectivity options can be found in examples/connectivity.php.

Handling Errors

Synchronous Operations

By default this client is configured to work in a synchronous mode, which means that for every command (except CONNECT) the client will wait for the receipt. You can set the client to work in the asynchronous mode, by setting the sync property of the StompConnection object to false.

<?
    $conn = new Stomp("tcp://localhost:61613");
    $conn->sync = false;
?>    

You can also perform single operations in desired mode, no matter of the global client settings, by passing additional parameter to the appropriate method

<?
    $this->conn->send("/queue/test", "test", null, true);
?>    

Some things you have to consider when deciding whether to use synchronous or asynchronous mode * When you use asynchronous mode broker errors will be ignored * Asynchronous method allows you to send messages faster, since will not wait for broker's response

Exceptions

Basically there are two kind of errors you can experience when using this client: errors while trying to connect to a broker and errors after sending a certain command to the broker (like authorization errors for example).

Broker Errors

In case you use the client (or at least the current operation) in a synchronous mode, the client will wait for the broker to send a receipt for every synchronous command. In case that error is received instead of the receipt, the client will throw StompException. In case you have sent a command in an asynchronous mode the errors will be ignored. The more examples on how to handle exception will be shown in the following section.

Security

PHP client is able to work with secured brokers, correctly applying their authentication and authorization policies. For more information on how to secure ActiveMQ broker see this page.

Authentication

When connecting to the broker you can specify username and password which will be used to check your privileges against broker's security mechanism.

<?
    try {
        $con->connect("dejan", "test");
    } catch (StompException $e) {
        echo "dejan cannot connect\n";
        echo $e->getMessage() . "\n";
        echo $e->getDetails() . "\n\n\n";
    }
?>    

In case of unsuccessful authentication, the connect() method will throw StompException.

dejan cannot connect
User name or password is invalid.
java.lang.SecurityException: User name or password is invalid.
    at org.apache.activemq.security.SimpleAuthenticationBroker.addConnection(SimpleAuthenticationBroker.java:52)
    at org.apache.activemq.broker.BrokerFilter.addConnection(BrokerFilter.java:82)
    at org.apache.activemq.broker.MutableBrokerFilter.addConnection(MutableBrokerFilter.java:89)
    at org.apache.activemq.broker.TransportConnection.processAddConnection(TransportConnection.java:666)
    at org.apache.activemq.broker.jmx.ManagedTransportConnection.processAddConnection(ManagedTransportConnection.java:83)
    at org.apache.activemq.command.ConnectionInfo.visit(ConnectionInfo.java:134)
    at org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:297)
    at org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:175)
    at org.apache.activemq.transport.TransportFilter.onCommand(TransportFilter.java:68)
    at org.apache.activemq.transport.stomp.StompTransportFilter.sendToActiveMQ(StompTransportFilter.java:78)
    at org.apache.activemq.transport.stomp.ProtocolConverter.sendToActiveMQ(ProtocolConverter.java:135)
    at org.apache.activemq.transport.stomp.ProtocolConverter.onStompConnect(ProtocolConverter.java:487)
    at org.apache.activemq.transport.stomp.ProtocolConverter.onStompCommand(ProtocolConverter.java:187)
    at org.apache.activemq.transport.stomp.StompTransportFilter.onCommand(StompTransportFilter.java:67)
    at org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:84)
    at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:203)
    at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:185)
    at java.lang.Thread.run(Thread.java:613)

As you can see, you can use getMessage() method is used to get short message of the error. To get the full stack trace from the broker (if provided), you can use getDetails() method.

Authorization

If you use synchronous operations you can handle authorization errors when you try to access (read/write) certain destinations.

<?
    try {
        $con->send("/queue/test", "test");
        echo "Guest sent message with body 'test'\n";
    } catch (StompException $e) {
        echo "guest cannot send\n";
        echo $e->getMessage() . "\n";
        echo $e->getDetails() . "\n\n\n";
    }
?>    

In case of error, you can expect the following output

guest cannot send
User guest is not authorized to write to: queue://test
java.lang.SecurityException: User guest is not authorized to write to: queue://test
    at org.apache.activemq.security.AuthorizationBroker.send(AuthorizationBroker.java:176)
    at org.apache.activemq.broker.MutableBrokerFilter.send(MutableBrokerFilter.java:133)
    at org.apache.activemq.broker.TransportConnection.processMessage(TransportConnection.java:443)
    at org.apache.activemq.command.ActiveMQMessage.visit(ActiveMQMessage.java:631)
    at org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:297)
    at org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:175)
    at org.apache.activemq.transport.TransportFilter.onCommand(TransportFilter.java:68)
    at org.apache.activemq.transport.stomp.StompTransportFilter.sendToActiveMQ(StompTransportFilter.java:78)
    at org.apache.activemq.transport.stomp.ProtocolConverter.sendToActiveMQ(ProtocolConverter.java:135)
    at org.apache.activemq.transport.stomp.ProtocolConverter.onStompSend(ProtocolConverter.java:247)
    at org.apache.activemq.transport.stomp.ProtocolConverter.onStompCommand(ProtocolConverter.java:173)
    at org.apache.activemq.transport.stomp.StompTransportFilter.onCommand(StompTransportFilter.java:67)
    at org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:84)
    at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:203)
    at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:185)
    at java.lang.Thread.run(Thread.java:613)

The full example of the PHP Stomp client security and exceptions handling can be found in examples/security.php.

Message Transformation

You can use PHP Stomp client to send your arrays as Map Messages. This functionality only works with ActiveMQ at the moment, but it is something that will be addressed with upcoming Stomp 1.1 specification. You can learn more about existing ActiveMQ message transformation functionality here.

If you want to send a map message, you should add transformation header to your send() command, like this

<?
    $body = array("city"=>"Belgrade", "name"=>"Dejan");
    $header = array();
    $header['transformation'] = 'jms-map-json';
    $mapMessage = new StompMessageMap($body, $header);
    $con->send("/queue/test", $mapMessage);
?>    

Currently, only jms-map-json value is supported and it will convert your array to JSON. On the ActiveMQ side, the message will be transformed to JMS MapMessage.

If you wish to receive map messages in your PHP client, you should provide transformation header to your subscribe command, like this

<?
    $con->subscribe("/queue/test", array('transformation' => 'jms-map-json'));
    $msg = $con->readFrame();

    if ( $msg != null) {
        echo "Received array: "; 
        print_r($msg->map);
        // mark the message as received in the queue
        $con->ack($msg);
    } else {
        echo "Failed to receive a message\n";
    }
?>    

In this case, broker will transform JMS MapMessage to JSON, which will converted to PHP array by the client.

The full example of the PHP Stomp client security and exceptions handling can be found in examples/transformation.php.