A Publish/Subscribe Mechanism for Web Services by Dmitri Tcherevik
WSJ Vol 03 Issue 2 - pg.14

	


Listing 1
// Retrieve the namespace handle from the catalog
// INamespace eventSource = (INamespace) root.find("File 
// System");
// Create an event filter
IQuery filter = (IQuery)eventSource.create("FileCreatedEvent", ItemType.IT_Query);
filter.setQueryExp("path like '%.doc'");
// Create an event listener object
MyEventListener listener = new MyEventListener();
// Register the filter and the listener object with the runtime system
EventBroker.subscribe(listener, filter);

Listing 2
// Retrieve the namespace handle from the catalog
INamespace eventSource = (INamespace) root.find("File System");
// Create an event object and set its properties
IEvent event = (IEvent) eventSource.create("FileCreatedEvent", ItemType.IT_Event);
event.setPropertyValue("path", "/Docs/Important Paper.doc");
// Broadcast the event to local and remote listeners
EventBroker.raise(event);

Listing 3
<env:Envelope xmlns:env="http://schemas.xmlsoap.org/soap/envelope"
              xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    <env:Body>
        <broker:raise xmlns:broker="http://eventcentral.com">
            <ev:FileCreatedEvent 
                xmlns:ev="http://eventcentral.com/Root/File System">
                <path xsi:type="xsd:string">/Docs/Important Paper.doc</path>
            </ev:FileCreatedEvent>
            <ev:FileDestroyedEvent 
                xmlns:ev="http://eventcentral.com/Root/File System">
                <path xsi:type="xsd:string">/Docs/Junk Paper.doc</path>
            </ev:FileCreatedEvent>
        </broker:raise>
    </env:Body>
</env:Envelope>

Listing 4
public SOAPMessage onMessage(SOAPMessage requestMsg)
{
    ... 
    SOAPEnvelope requestEnv = requestMsg.getSOAPPart().getEnvelope();
    SOAPHeader requestHeader = requestEnv.getHeader();
    SOAPBody requestBody = requestEnv.getBody(); 
    SOAPElement raiseElem = (SOAPElement) requestBody.getChildElements().next();
    
    // Decode events and forward them to the runtime system
    IEvent [] events = SOAPCodec.decodeEventList(requestEnv, raiseElem);
    for (int k = 0; k < events.length; k ++)
    {
        // Raise the event in the runtime system
        EventBroker.raise(events[k]);
    }
    // Compose the response message
    SOAPMessage responseMsg = messageFactory.createMessage();
    ...
    return responseMsg;
}

Listing 5
public class EventListener implements IEventListener
{
    private String endpoint; // end point of a web based listener 
    public void handleEvent(IEvent event)
    {
        ...
        SOAPConnection connection = connectionFactory.createConnection(); 
        SOAPMessage requestMsg = messageFactory.createMessage();
        SOAPEnveloper soapEnv = requestMsg.getSOAPPart().getEnvelope();
        SOAPBody soapBody = soapEnv.getBody();
        SOAPBodyElement command = soapBody.addBodyElement(
             soapEnv.createName("raise", "broker", "http://eventcentral.com"));
        SOAPCodec.encodeEventList(soapEnv, command, new IEvent [] { event });
        requestMsg.saveChanges();
        SOAPMessage responseMsg = connection.call(requestMsg, endpoint);
        connection.close();
        ...
    }
    ...
}

Listing 6
public class Subscription
{
     public String endpoint; // an endpoint used to receive events 
     public String nsppath; // path of a namespace containing the event class
     public String classname; // name of an event class 
     public String filter; // a query expression for filtering events
}
public interface IEventManager
{
    public void subscribe(Subscription [] subscriptions);
    public void unsubscribe(String endpoint); 
}

Listing 7
public class EventManager implements IEventManager
{
    public void subscribe(Subscription [] subs)
    {
        ...
        for (int k = 0; k < subs.length; k ++)
        {
            // locate the event source namespace in the catalog
            INamespace nsp = root.find(subs[k].nsppath);
            // create an event filter 
            IQuery q = (IQuery) nsp.create(subs[k].classname, ItemType.IT_Query);
            q.setQueryExp(subs[k].filter);
            // create an event listener object
            EventListener lsn = new EventListener(subs[k].endpoint);
            // register the listener object with the runtime system
            EventBroker.subscribe(lsn, q);
        }
    }
    ...
}