The Sapo-Broker 4.0 User Guide

In order to start using the Sapo-Broker you just need to have an installation of Java SDK 1.6 or higher and the "java" command in your PATH. You can quickly check for java version by opening a shell and typing:

The installation process is very easy. These are the required steps:

From a shell type:

At this moment you should have an instance of the Sapo-Broker up and running. There is nothing stopping you from using a single instance of Sapo-Broker, but since one of it's main goals is to be used in a distributed fashion you really want to have at least two instances to play with, read the User Guide to know how to setup and connect multiple instances.

The distribution includes a couple of sample scripts that illustrate a common use case for Sapo-Broker, event propagation, among others. These scripts include a producer that sends messages with 100 random alpha numeric characters at the rate of one per second and a consumer that prints the message content to the console every time it receives a message.

To run the producer open a shell and type:

To run the consumer:

Besides the Linux scripts, there are also available windows batch files counterparts.

To help the choice of library and enlighten about what features are supported by witch libraries a client compatibility matrix is presented.

Specific information about each client and their features, as well some samples, should be available. Poke around.

In this section examples of publishing and consuming messages are going to be presented for several languages, namely Java , C# (.NET) , Python , Perl , PHP.

BrokerClient bk = new BrokerClient("127.0.0.1", 3323);
String msg = RandomStringUtils.randomAlphanumeric(messageLength);
NetBrokerMessage brokerMessage = new NetBrokerMessage("Hello World!!".getBytes("UTF-8"));
bk.publishMessage(brokerMessage, "/topic/foo");

public class Consumer implements BrokerListener
{
	public static void main(String[] args) throws Throwable
	{
		Consumer consumer = new Consumer();
		BrokerClient bk = new BrokerClient("localhost", 3323);
		NetSubscribe subscribe = new NetSubscribe("/topic/foo", DestinationType.TOPIC);
		bk.addAsyncConsumer(subscribe, consumer);
	}

	@Override
	public boolean isAutoAck()
	{
		return false;
	}

	@Override
	public void onMessage(NetNotification notification)
	{
		byte[] payload = notification.getMessage().getPayload();
		System.out.println(String.format("Message destination: %s%n Payload: %s", notification.getDestination(), new String(payload)));
	}
}

BrokerClient brokerClient = new BrokerClient(new HostInfo("localhost", 3323));
NetBrokerMessage brokerMessage = new NetBrokerMessage(System.Text.Encoding.UTF8.GetBytes("Hello World!!"));
brokerClient.Publish(brokerMessage, "/topic/foo");

BrokerClient brokerClient = new BrokerClient(HostInfo("localhost", 3323));
Subscription subscription = new Subscription("/topic/foo", NetAction.DestinationType.TOPIC);
subscription.OnMessage += delegate(NetNotification notification)
{
	 System.Console.WriteLine("Message received: {0}", System.Text.Encoding.UTF8.GetString(notification.Message.Payload));
};
brokerClient.Subscribe(subscription);

#!/usr/bin/env python

from Broker.Messages import Message, Publish
from Broker.Transport import TCP, UDP
from Broker.Codecs import Codec
from Broker.Clients import Minimal

server='localhost'
destination = '/python/tests'
destination_type = 'QUEUE'
N=10000

broker = Minimal(codec=Codec(), transport=TCP(host=server))

for n in xrange(N):
    message = Message(payload='Message number %d' % n)
    publish = Publish(destination=destination, destination_type=destination_type, message=message)
    broker.send(publish)

#!/usr/bin/env python


from Broker.Messages import Message, Subscribe, Acknowledge
from Broker.Transport import TCP, UDP
from Broker.Codecs import Codec
from Broker.Clients import Minimal

server='localhost'
destination = '/python/tests'
destination_type = 'QUEUE'

broker = Minimal(codec=Codec(), transport=TCP(host=server))

broker.send(Subscribe(destination=destination, destination_type=destination_type))
for n in xrange(N):
    message = broker.receive()
    broker.send(Acknowledge(message_id=message.message.id, destination=message.destination))
    print message.message.payload

use SAPO::Broker::Clients::Simple;

use strict;
use warnings;

my $broker = SAPO::Broker::Clients::Simple->new();
my %options = (
	'destination_type' => 'QUEUE',
	'destination' => '/perl/tests',
);

my $N = $ARGV[0] || 100;

for my $n (1..$N){
	$broker->publish(%options, 'payload' => "payload $n");
}

use lib ('../lib');

use SAPO::Broker::Clients::Simple;
use Data::Dumper;

use strict;
use warnings;

my $broker = SAPO::Broker::Clients::Simple->new();
my %options = (
	'destination_type' => 'QUEUE',
	'destination' => '/perl/tests',
	'auto_acknowledge' => 1
);

$broker->subscribe(%options);

my $N = $ARGV[0] || 100;

for my $n (1..$N){
	my $message = $broker->receive();
	print Dumper($message);
}

#!/usr/bin/env php
<?php

$broker = broker_init("127.0.0.1", 3323, 0, 1);
$msg = "Hello, world!";
broker_publish($broker, "/test/foo", $msg, strlen($msg));
broker_destroy($broker);
?>

#!/usr/bin/env php
<?php

$broker = broker_init("127.0.0.1", 3323, 0, 1);
$ret = broker_subscribe_topic($broker, "/test/foo");

// consume without auto-ack
while (1) {
	if (($msg = broker_receive($broker, 1000, true)) !== false)
    	echo "Got message: " . print_r($msg, true) . "\n";
}

?>

Agents publish statistics regarding the number of consumers subscribed in each topic and queue.

When an unhandled exception occurs, and it's not I/O related, agents publish a fault error. The message content conforms to SOAP Fault message format.

Fault messages are published to topic:

/system/faults/[agent name]

A fault message can also be published when the maximum number of allowed queues or distinct topic subscriptions is reached . Here is a sample message content for each fault type:

The maximum number of queues (500) has been reached.

The maximum number of distinct subscriptions (500) has been reached.

There are three main configuration sections in global configuration file: Domain, Messages and Access Control. The later is divided in two parts as discussed in Access control section.

The "domain" element contains all agentes that are part of the broker cloud. For each agent an IP:PORT must be defined.

This would be the Domain section of a configuration used to setup a cloud with two nodes:

<domain>
	<peer>
		<ip>192.168.0.10</ip>
		<port>3315</port>
	</peer>
	<peer>
		<ip>192.168.0.11</ip>
		<port>3315</port>
	</peer>
</domain>

Keep in mind that the configuration must be the same for all cloud agents.

In the "messages" section it is possible to define the maximum message size, the maximum number of queues/subscriptions and default message TTL.

See the SSL Configuration guide for help on how to setup the SSL transport.

Every agent as an associated security policy, except in the case of security policy absence in witch it defaults to "allow everything". In the case of policy inheritance, eventually with multiple degrees, a stack of entries is built with the most specific ACLs at the top and entries inserted by declaration order within each ACL. The result is an ordered entry stack where the most specific/priority come first, so evaluating a client request against the full access control is sequential evaluation of entries. The first entry witch results in a positive evaluation is applied.

The entry stack is optimized for evaluation, although maintaining the described model. (More on this we I have the time).

In Security policy configuration sample a security configuration sample is presented. As said before, this configuration aspect is defined in global configuration file.

The sample presents a scenario where the default security policy mandates that no client can produce messages to topic or queue whose named begin with "/system/". The default policy also dictates that destination names beginning with "/private/" are forbidden for all clients. These security settings apply to all agents using the default security policy.

At the bottom of the sample a section (XML element) entitled "agents" is present. Here is the place to associate specific security policies with agents. The sample states that an agent named "agent007" must used the "anotherPolicy" security policy. This policy inherits from the default security, adds some entries to the ACL and in a specific case overrides the default policy. The added entries specify that message production to the queue named "/specialQueue" is denied to all except those clients that simultaneous possess the role "VIP_role" and are using a type of communication channel that offers confidentiality (such as SSL). The declaration details are self-explanatory.

<security-policies>
   <policies>
      <policy policy-name="default">
         <acl>
            <entry action="DENY" destination-type="TOPIC QUEUE" destination="/system/.*"  privilege="WRITE">
               <condition condition-type="ALWAYS" />
            </entry>
            <entry action="DENY" destination-type="TOPIC QUEUE TOPIC_AS_QUEUE" destination="/private/.*"  privilege="WRITE READ">
               <condition condition-type="ALWAYS" />
            </entry>
         </acl>
      </policy>
      <policy policy-name="anotherPolicy" inherits="default">
         <acl>
            <entry action="DENY" destination-type="TOPIC QUEUE TOPIC_AS_QUEUE" destination="/private/.*"  privilege="READ">
               <condition condition-type="ADDRESS">
                  <address  mask="24">10.11.12.0</address>
               </condition>
            </entry>
            <entry action="PERMIT" destination-type="QUEUE" destination="/specialQueue"  privilege="WRITE">
               <condition condition-type="AND">
                  <condition condition-type="ROLE">
                     <role>VIP_role</role>
                  </condition>
                  <condition condition-type="CHANNELTYPE">
                     <channel-type>CONFIDENTIALITY</channel-type>
                  </condition>
               </condition>
            </entry>
            <entry action="DENY" destination-type="QUEUE" destination="/specialQueue"  privilege="WRITE">
               <condition condition-type="ALWAYS" />
            </entry>
         </acl>
         </policy>
         <policy policy-name="very special policy" inherits="special policy">
         <acl>
            <entry action="PERMIT" destination-type="TOPIC" destination="/topic/sensitive"  privilege="READ WRITE">
               <condition condition-type="AND">
                  <condition condition-type="ROLE">
                     <role>brk_writer_role</role>
                  </condition>
                  <condition condition-type="CHANNELTYPE">
                     <channel-type>CONFIDENTIALITY</channel-type>
                  </condition>
               </condition>
            </entry>       
           </acl>
         </policy>
      </policies>
     <agents>
         <agent agent-name="agent007">
            <agent-policy policy-name="anotherPolicy" />
         </agent>
      </agents>
</security-policies>
				

Authentication modules, also called validation providers, are configured at the global configuration file. There are defined the full name of type implementing pt.com.broker.auth.AuthInfoValidator (more on this later) in the element class and some provider specific provider parameters, under provider-params. The class must be in agent's classpath in order to be loaded.

Each "provider-name" must be unique and known by both the broker agent and the client.

Each of the available implementations are described later.

<credential-validators>
   <credential-validator provider-name="SapoSTS">
      <class>pt.com.broker.auth.saposts.SapoSTSAuthInfoValidator</class>
      <provider-params>
         <sts>
            <sts-location>https://services.sapo.pt/STS/</sts-location>
            <sts-username>username</sts-username>
            <sts-password>password</sts-password>
         </sts>
      </provider-params>
   </credential-validator>
   <credential-validator provider-name="BrokerRolesDB">
      <class>pt.com.broker.auth.jdbc.JdbcAuthInfoValidator</class>
      <provider-params>
         <db-roles>
            <driver-class>org.postgresql.Driver</driver-class>
            <database-url>jdbc:postgresql://localhost/BROKER_ROLES</database-url>
            <database-username>username</database-username>
            <database-password>password</database-password>
         </db-roles>
      </provider-params>
   </credential-validator>
</credential-validators>
				

At the heart of an authorization provider lies an implementation of the interface  pt.com.broker.auth.AuthInfoValidator. This interface and reaming relevant types are presented at Figure 5.

The initialization method of AuthInfoValidator gives the provider an opportunity to realize any actions  required to properly validate client's authorization information and translate that information in client roles. The ProviderInfo is used to inform the provider what is the name of the authentication type it will be performing, the class name (redundant) and the parameter specified in the configuration file. These arguments are supplied as an XML representation object.

The AuthInfoValidator.validate() method is invoked after the reception of an AuthInfo message. Depending on the authentication type the AuthInfo object contains a combination of a user identifier, an authentication token and list of user roles. The authentication token is a binary value, so, in case of representing a string such as a password must be UTF-8 encoded. In response to an authentication request a result is produced (AuthValidationResult) indicating whether the credentials are valid (a reason for failure should be supplied in case their not), and the list of roles associated with the user in case of valid credentials.

Figure 2.1. Authorization Provider relevant types

Currently, there are two authenticate modules implemented. One interacts with Sapo STS and it's used internally; the other was implemented as a proof-of-concept attesting broker authentication extensibility model's flexibility and provides user name-password database authentication.

CREATE TABLE users
(
 username varchar(50),
 password varchar(50)
);
ALTER TABLE users ADD PRIMARY KEY (username);

CREATE TABLE roles
(
 username varchar(50),
 role varchar(50)
);
ALTER TABLE roles ADD PRIMARY KEY (username, role);
ALTER TABLE roles ADD FOREIGN KEY (username) REFERENCES users(username);

					

Sapo-Broker agents use SSL connections as message transport to client authentication messages and to support all communications that require authentication (role based access control). This type of connection is based on asymmetric cryptography operations and X.509 certificates. In order generate your own key pair and X.509 certificates a small tutorial on how to use SUN's keytool is provided.

<ssl>
   <broker-ssl-port>3390</broker-ssl-port>
   <keystore-location>/home/server/agent/certs2</keystore-location>
   <keystore-password>serverkspw</keystore-password>
   <key-password>serverpw</key-password>
</ssl>
					

Clients, once they receive the message may persist it and send the respective Acknowledge with an Action Identifier. If the connection with the agent is lost or the Accept message is not received the message should not be processed, but if this not the case the client can process the message. Furthermore, the client is responsible for process locally persisted messages, and this behaviour should be ensured whenever a client restarts.

To guarantee that each produced message is processed only once a global acknowledge record may be implemented. After receiving a message, clients query the global acknowledge system to determine if the message was already processed, if not, they process it and afterwards send the due Acknowledge message and inform the global acknowledge system that a given message was already processed. The  last step may be only done if the acknowledge fails.