Messaging and Streaming > Publish-Subscribe > Message Filtering

By default subscribers receive all messages published into destination. In order to restrict message delivery to a specific subscriber or to a group of subscribers, WebORB supports the concept of message selectors. A selector can be either a query expressed using the syntax of SQL-92 or a class implementing the weborb.messaging.v3.IMessageSelector interface. Selectors are set by the subscribers, as a result, each subscriber can have its own selector (query or IMessageSelector object).

Selector Query

A selector query must be expressed using the SQL-92 syntax and formatted as the condition part of the SQL's WHERE clause. A query condition must reference the headers of the published messages. When a message is published and a subscriber has a selector query, WebORB executes the query on the headers of the published message. If the result of the query is true, the message is delivered to the subscriber. Consider the following example where subscriber will receive only the messages containing the "CITYNAME" header with the value of "Dallas":

Creating a Flex consumer with a selector:

var consumer:Consumer = new Consumer();
consumer.destination = "SamplePollingDestination";
consumer.addEventListener( MessageEvent.MESSAGE, gotMessage );

// the selector property is set to a query.
// WebORB will check if a published message contains the
// "CITYNAME" header with the value "Dallas". If it does,
// the message will be delivered to this consumer
consumer.selector = "CITYNAME = 'Dallas'";
consumer.subscribe();

Publishing a message with a header:

var producer:Producer = new Producer();
producer.destination = "SamplePollingDestination";
var message:AsyncMessage = new AsyncMessage();
message.body = "Home Sweet Home";
message.headers[ "CITYNAME" ] = "Dallas"
producer.send( message );

Importing WebORB.js and setting up the URLs

Creating a JavaScript consumer with a selector:

var myConsumer = new Consumer("SamplePollingDestination", new Async(messageReceived, handleFault));
// the selector property is set to a query.
// WebORB will check if a published message contains the
// "CITYNAME" header with the value "Dallas". If it does,
// the message will be delivered to this consumer
myConsumer.selector = "CITYNAME = 'Dallas'";
myConsumer.subscribe();

Publishing a message with headers:

var myProducer = new Producer("SamplePollingDestination");
var headers = new Object();
headers[ "CITYNAME" ] = "Dallas";
myProducer.send( "Home Sweet Home", undefined, undefined, headers );

Applications using the WeborbClient API must make sure to reference weborbclient.jar. Creating a consumer with a selector:

WeborbClient consumer = new WeborbClient("http://localhost/weborb5/weborb.aspx", "SamplePollingDestination");
IResponder responder = new MyResponder();
// the selector is a query passed into the subscribe() call.
// WebORB will check if a published message contains the
// "CITYNAME" header with the value "Dallas". If it does,
// the message will be delivered to this consumer
String selector = "CITYNAME = 'Dallas'";
consumer.subscribe( responder, null, selector );

Publishing a message with headers:

WeborbClient producer = new WeborbClient("http://localhost/weborb5/weborb.aspx", "SamplePollingDestination");
Hashtable headers = new Hashtable();
headers.put( "CITYNAME", "Dallas" );
producer.publish( "Home Sweet Home", headers );

Creating a .NET consumer:

WeborbClient consumer = new WeborbClient( "http://localhost/weborb5/weborb.aspx", "SamplePollingDestination" );
SubscribeResponder responder = new SubscribeResponder(
message => Dispatcher.BeginInvoke(() =>
{
IAdaptingType[] body = message.GetBody();
object message = body[0].defaultAdapt();
System.Console.WriteLine( message + "\n" );
}),
fault => Dispatcher.BeginInvoke(() => System.Console.WriteLine( fault.Message + "\n" )));
// the selector is a query passed into the subscribe() call.
// WebORB will check if a published message contains the
// "CITYNAME" header with the value "Dallas". If it does,
// the message will be delivered to this consumer
String selector = "CITYNAME = 'Dallas'";
consumer.Subscribe( responder, null, selector );

Creating a .NET publisher:

public static string WeborbUrl = "http://localhost/weborb5/weborb.aspx"; // Can also change to "rtmpt://localhost:port/root";
private WeborbClient publisher = new WeborbClient( WeborbUrl, "SamplePollingDestination" );
AsyncMessage asyncMessage = new AsyncMessage();
asyncMessage.headers = new Dictionary<object, object> {{"CITYNAME", "Dallas"}};
asyncMessage.body = "Home Sweet Home";
publisher.Publish( asyncMessage );

Selector Class

Message selector is a class responsible for providing message filtering before WebORB delivers messages to the subscribers. Selectors can alter messages or instruct WebORB to completely ignore a message and skip the delivery. A selector class must implement the following interface:

using System;
using Weborb.V3Types;

namespace Weborb.Messaging.PubSub
{
public interface IMessageSelector
{
Object processClientMessage( V3Message message );
Object processServerMessage( Object message );
void setClientID( String clientID );
String getClientID();
}
}

Messages entering WebORB can come either from remote clients (Flex, Flash, AJAX, remote .NET or Java process using the WeborbClient API) or can be published by the code hosted in the same memory space where WebORB Messaging server is running. Before WebORB delivers an incoming message to subscribers, it checks if the destination where the message is published to has a selector. If the selector is set, WebORB invokes either the processClientMessage or processServerMessage methods, as follows:

•	processClientMessage is invoked if the message has arrived from a remote client.

•	processServerMessage is called if the message is published internally from code running in the same process.

Both methods can modify the published message and return a changed object. Alternatively, a completely different object can be returned from the method and will be published by WebORB. To cancel message delivery, method implementations should return null.

The argument in the processClientMessage method is a WebORB internal representation of the published message. An instance of Weborb.V3Types.V3Message class contains message headers and message body which can be accessed as shown below:

public Object processClientMessage( V3Message message )
{
System.Collections.IDictionary headers = message.headers;

Object sentMessage = message.body.body;
// cast message to the type you expect
// make any changes to the message if needed and return the message
// alternatively, return any other message to be sent to the subscribers
// return null if the message delivery needs to be cancelled
return sentMessage;
}