Previous Up Next

Chapter 22  Publish and Subscribe Services

22.1  What is Publish and Subscribe?

The default communication model in CORBA is a call from one client to (an object in) one server. This is often one-to-one or point-to-point communication. In contrast, publish and subscribe (often abbreviated to pub-sub) communication is where one application “publishes” (that is sends) a message on a particular topic, and all the other applications that have “subscribed” to this topic receive the message. This is a form of one-to-many communication, and it is intrinsically asynchronous because the application that publishes a message does not wait to get responses from those applications that receive the message.

Computer mailing lists are good analogy for pub-sub communication. For example, let us assume that the ACME company has mailing lists for staff in different departments: eng-staff@acme.com reaches the Engineering staff, sales-staff@acme.com reaches the Sales staff, and so on. The following points are worth noting:

22.1.1  Emulating Different Communication Models

Some middleware systems, such as TIBCO Rendezvous, are based on pub-sub communication. Some other middleware systems, such as IBM MQ Series, are based on asynchronous, point-to-point communication. Some other middleware systems, such as CORBA, are based on synchronous, point-to-point communication.

Each of these three kinds of middleware can emulate the other two kinds. For example, CORBA can provide one-to-one, asynchronous communication with IDL oneway calls or CORBA Messaging (Chapter 16). And CORBA can provide pub-sub communication with the various CORBA Services discussed in this chapter.

22.1.2  CORBA Services for Publish and Subscribe

The CORBA Event Service (Section 22.2) provides a very basic form of pub-sub communication. The Notification Service (Section 22.3) extends the Event Service in ways that provide a much richer form of pub-sub communication. Finally, the Telecom Log Service (Section 22.4) extends the Notification Service with the ability to permanently log messages so that they can be replayed.

CORBA uses terminology that is different to what has been used in the discussion so far. Instead of publisher and subscriber, CORBA uses the terms supplier and consumer. Instead of topic or mailing list, CORBA uses the term event channel.

22.2  Event Service

The CORBA Event Service offers both a “push” and a “pull” model of communication. The push model is similar to how I have described pub-sub systems, so I discuss the push model first, in Section 22.2.1 and then discuss the pull model in Section 22.2.2.

22.2.1  The Push Model

Figure 22.1 shows the IDL definitions relevant to the push model of the Event Service. For conciseness, the raises clauses on operations have been omitted. Figure 22.2 shows graphically how the various interfaces interact with each other.


module CosEventComm {
  interface PushConsumer {
    void push(in any data); 
    void disconnect_push_consumer();
  };
  interface PushSupplier {
    void disconnect_push_supplier();
  };
};
module CosEventChannelAdmin {
  interface ProxyPushConsumer
    : CosEventComm::PushConsumer
  {
    void connect_push_supplier( 
        in CosEventComm::PushSupplier push_supplier);
  };
  interface ProxyPushSupplier
    : CosEventComm::PushSupplier
  {
    void connect_push_consumer( 
        in CosEventComm::PushConsumer push_consumer);
  };
  interface ConsumerAdmin {
    ProxyPushSupplier obtain_push_supplier();
    ProxyPullSupplier obtain_pull_supplier();
  };
  interface SupplierAdmin {
    ProxyPushConsumer obtain_push_consumer();
    ProxyPullConsumer obtain_pull_consumer();
  };
  interface EventChannel {
    ConsumerAdmin for_consumers();
    SupplierAdmin for_suppliers();
    void destroy();
  };
};
Figure 22.1: IDL for the Event Service push model
Figure 22.2: The Push Model of the Event Service

A consumer application must implement the PushConsumer interface. This has a push() operation that is invoked to pass it an any (Section 15.3) containing arbitrary data related to an event. The PushConsumer interface also has an operation called disconnect_push_consumer(), which is invoked if the Event Service wants to disconnect itself from the consumer, for example, when an EventChannel is being destroy()ed.

A supplier application must implement the PushSupplier interface. This is, in effect, a callback interface (Section 1.4.2.2). The Event Service invokes the disconnect_push_supplier() operation if it wants to disconnect itself from the supplier, for example, when an EventChannel is being destroy()ed.

An EventChannel is the initial point of contact for the Event Service. This interface just splits its functionality among the SupplierAdmin and ConsumerAdmin objects, and so provides operations that allow applications to access these objects.

SupplierAdmin is a factory interface (Section 1.4.2.1). Its obtain_push_supplier() operation creates a ProxyPushConsumer. The use of “Proxy” in this name has nothing to do with code that is generated by an IDL compiler (Section 1.4.5). Rather, a ProxyPushConsumer is a delegation object within the Event Service: a supplier invokes push() on a ProxyPushConsumer object and it, in turn, invokes (or arranges for something else within the Event Service to invoke) push() on each PushConsumer object in consumer applications.

The initialization of a supplier application involves the following steps. It first connects to the EventChannel. It invokes for_suppliers() to access the SupplierAdmin and then calls obtain_push_consumer() to create a ProxyPushConsumer object. Finally, the supplier calls connect_push_supplier() to register its own PushSupplier object. At this point, the supplier is fully connected to the Event Service and can call push() on the ProxyPushConsumer.

The initialization of a consumer application mirrors that of a supplier. It first connects to the EventChannel. Then it invokes for_consumers() to access the ConsumerAdmin and then calls obtain_push_supplier() to create a ProxyPushSupplier object. Finally, the consumer application calls connect_push_consumer() to register its own PushConsumer object. At this point, the consumer is fully connected to the Event Service and so its push() operation will be invoked whenever an event occurs.

22.2.2  The Pull Model

The push model is so called because it pushes data towards (proxy) consumers. Conversely, the pull model is so called because it pulls data from (proxy) suppliers. The additional IDL interfaces required for this are shown in Figure 22.3.


module CosEventComm {
  ...
  interface PullSupplier {
    any pull();
    any try_pull(out boolean has_event);
    void disconnect_pull_supplier();
  };
  interface PullConsumer {
    void disconnect_pull_consumer();
  };
};
module CosEventChannelAdmin {
  ...
  interface ProxyPullConsumer
    : CosEventComm::PullConsumer
  {
    void connect_pull_supplier( 
        in CosEventComm::PullSupplier pull_supplier);
  };
  interface ProxyPullSupplier
    : CosEventComm::PullSupplier
  {
    void connect_pull_consumer( 
        in CosEventComm::PullConsumer pull_consumer);
  };
};
Figure 22.3: IDL for the Event Service pull model

The pull() operation defined on PullSupplier is a blocking operation. The try_pull() operation is a non-blocking version. It returns immediately, and the has_event out parameter indicates whether the returned any has event data in it or is empty.

The Event Service specification defines additional interfaces that make it possible to transmit data using strongly-typed APIs rather than having to package up the data inside an any. However, the Typed Event Service was difficult for vendors to implement (due to some immaturity in CORBA at the time) and so there were very few implementations of it. For that reason, I do not discuss the Typed Event Service in this book.

The 80/20 principle [Koc00] applies to software products: “80% of people use just 20% of a product’s capabilities”. Of course, the percentages are not always 80 and 20, but the principle that most people use just a small subset of a product’s capabilities is true. This principle applies to the Event Service. Most people use just the push model of event communication; very few people use the pull model or the typed push/pull models.

22.2.3  Limitations of the Event Service

The Event Service suffers from several limitations, as I discuss in this section.

One limitation is that the Event Service does not define a factory interface (Section 1.4.2.1) for creating event channels. This means that an Event Service implementation might have just one event channel. Conceptually, this is similar to a company deciding that it will have just one internal mailing list. The result is that subscribers receive all messages, even those in which they have no interest. Many implementations of the Event Service overcome this limitation by providing their own proprietary factory interface. However, use of such proprietary APIs obviously hinders source-code portability.

Another limitation is that the Event Service specification does not define what quality of service should be provided by an implementation. For example:

The Event Service specification deliberately refrained from defining what quality of service should be offered. This was done with the hope of encouraging different vendors to compete by offering different qualities of service. However, this strategy backfired for two reasons.

First, most implementations of the Event service held all messages and information about connected suppliers/consumers in in-memory data-structures rather than in a file or a database. This meant that there was not much competition based on different qualities of service.

Second, an application might want more than one quality of service at the same time. For example, assuming that a vendor provided an EventChannel factory as a proprietary enhancement, a supplier application might want to send some messages on one EventChannel with a particular quality of service and some more messages on a different EventChannel that had another quality of service. However, most vendors offered only a single quality of service that applied to all the EventChannel objects.

The above limitations means that, unfortunately, the Event Service is unsuitable for the needs of most applications.

22.3  Notification Service

The Notification Service is sometimes referred to as “the Event Service on steroids”, “the Event Service++” or simply “what the Event Service should have been in the first place”. As I discuss in the following subsections, the Notification Service removes all the limitations of the Event Service that were discussed in Section 22.2.3, and adds additional functionality. The result is a publish-subscribe system that is very flexible, and scalable.

22.3.1  IDL Interfaces

The Notification Service is backwards compatible with the Event Service. This backwards compatibility is achieved by having the IDL interfaces of the Notification Service inherit from those of the Event Service. Furthermore, the naming conventions of the IDL interfaces for the Notification Service closely mirror those of the Event Service. This backwards compatibility and similar naming conventions provide two important benefits:

The Notification Service defines 38 IDL interfaces. This is in addition to the 11 IDL interfaces defined in the Event Service, from which interfaces in the Notification Service inherit. That is 49 interfaces in total! This is an amazingly high number of interfaces but, thankfully, most developers use just a small fraction of these interfaces. Many of the interfaces provide administration-type functionality and most vendors provide either command-line utilities or a graphical program that interacts with these administration-type interfaces, so that developers do not need to write any code to do so.

22.3.2  StructuredEvent

The Notification Service allows event data to be transmitted as an any. This is for backwards compatibility with the Event Service. However, the Notification Service allows event data to be transmitted in a different format, called a StructuredEvent, which is shown in Figure 22.4. Actually, this figure shows a slightly simplified IDL. In particular, some typedef declarations have been removed in order to make it more concise. For example, the type of the variable_header field in EventHeader is really a typedef of a typedef of the sequence shown.


module CosNotification {
  struct Property {
    string  name;
    any     value;
  };
  struct EventType {
    string  domain_name;
    string  type_name;
  };
  struct FixedEventHeader {
    EventType  event_type;
    string     event_name;
  }; 
  struct EventHeader {
    FixedEventHeader    fixed_header;
    sequence<Property>  variable_header;
  };
  struct StructuredEvent {
    EventHeader         header;
    sequence<Property>  filterable_data;
    any                 remainder_of_body;
  }; 
  typedef sequence<StructuredEvent> EventBatch;
  ...
};
Figure 22.4: Pseudo IDL for the Notification Service event data

The EventType embedded in the header of a StructuredEvent contains two string fields. The domain_name should be set to identify a particular vertical industry, for example, "Telecomms", while the type_name should be set to uniquely identify a type of event within that domain, for example, CommunicationsAlarm. The rest of an EventHeader consists of a sequence of Propertys, which are name-value pairs. Programmers can place whatever name-value pairs they want in this sequence, but it is intended to be used to express a desired quality of service (QoS) for controlling message delivery, for example, a message priority or a delivery timeout. The Notification Service could have used strongly typed fields in EventHeader to specify the QoS. However, there are two benefits to specifying them as weakly-typed name-value pairs. One benefit is that it reduces the size of the event header if, as is often the case, a supplier is happy with default QoS values. Another benefit is that it allows future revisions of the Notification Service to define additional QoS name-value pairs in a backwards-compatible manner; likewise, it allows vendors to define additional, proprietary QoS() name-value pairs.

After the header, the filterable_data part of a StructuredEvent provides another sequence of name-value pairs. It is in this sequence that users are expected to place (most/all of) their event data. Representing the event data as name-value pairs is a trade-off between the awkwardness of a single unnamed any (as was provided by the Event Service) and a non-extensible but compile-time-safe struct with fixed fields. The intention of the OMG is that different vertical domains will define which name-value pairs should be specified for particular kinds of events. For example, representatives from different telecommunications companies might get together to define the name-value pairs for different kinds of events, such as CommunicationsAlarm, relevant to that industry.1 These name-value pairs are called filterable_data because they can be accessed by filters (Section 22.3.4).

The final field in a StructuredEvent is an any. This field allows you to store any data that is unlikely to be of use to filters, possibly because the data is a large “blob” of data, such as the contents of a file.

22.3.3  EventBatch

The Notification Service can send and receive an EventBatch (Figure 22.4), which is a sequence of StructurdEvent. This enables applications with high throughput requirements to send or receive events in bulk. As an example of this, let us suppose a producer application produces 1000 events per second. Instead of the producer pushing 1000 events individually every second, this application could populate a sequence of 1000 structured events and push that sequence once per second.

A consumer can receive events in an EventBatch instead of receiving individual events. As I will discuss in Section 22.3.7, applications can specify various quality-of-services for their interactions with the Notification Service. A consumer application can use this to specify its desired batch size and pacing interval. The pacing interval is the maximum delay between delivery of events. For example, let us suppose a consumer sets the batch size to 1000 events and the pacing interval to 10 seconds. After 10 seconds, even if the channel has received only 300 events, it will send that batch to the consumer, rather than waiting until it has received all 1000 events before sending them to the consumer.

A discussion of how an application specifies if it wants to supply/consume events in the form of individual anys, individual StructuredEvents or as an EventBatch is deferred until Section 22.3.5.

22.3.4  Filters

A filter is an object wrapper around a collection of constraints (conditions). Filters can be applied to messages as they pass through the Notification Service. Syntactically, there are two different kinds of filter: Filter and MappingFilter. A FilterFactory is used to create both kinds of filter. 


module CosNotifyFilter {
	  interface FilterFactory {
	    Filter        create_filter(...) raises(...);
	    MappingFilter create_mapping_filter(...)
	                                    raises(...);
	  };
};

22.3.4.1  Filters to Remove Messages

The purpose of the Filter type is to delete messages before they are transmitted to consumers. Doing this saves consumer applications from having to examine messages to see if they are relevant. It also saves on network traffic because deleted messages are not transmitted.

The constraints within a Filter are expressed in Extended Trader Constraint Language (ETCL), which, as its name suggests, is an enhancement of the Trader Constraint Language (TCL) that is used with the Trading Service (Chapter 20). A constraint is a boolean expression that is written in terms of the name-value pairs within a StructuredEvent (Section 22.3.2). A constraint can also refer to the event’s domain_name and type_name.

As previously mentioned, a Filter is a wrapper around a collection (represented as a sequence) of constraints. If any constraint within a Filter evaluates to true then the message is allowed to pass through. In other words, a message is discarded only if if all the constraints evaluate to false.

Filters can be attached to all combinations of proxy push/pull/consumer/supplier objects. However, it is more common to attach filters to the consumer proxies rather than the supplier proxies because it it more natural to want to filter what consumers receive rather than to filter what suppliers send.

Filters can also be attached to SupplierAdmin and ConsumerAdmin objects. The benefits of doing this are discussed in Section 22.3.5.

You can attach multiple filters to a proxy or admin object. If this you do this then a message is discarded only if all the constraints in all the filters evaluate to false.

22.3.4.2  Filters for Message Timeouts and Priorities

The header of a StructuredEvent can contain name-value pairs that specify a priority and/or deadline for message delivery. These entries, if present, are specified by the supplier of the event. The MappingFilter type allows consumers—actually the ConsumerAdmin and supplier proxies that act on behalf of consumers—to override the priority and/or deadline associated with a message. The name MappingFilter is not very intuitive; OverrideFilter might have been more intuitive.

A consumer admin or supplier proxy can have two MappingFilter objects associated with it. One of these is used to override a message’s priority and the other is used to override a message’s delivery deadline. Section 22.3.4.1 mentioned that a Filter is an object wrapper around a collection of constraints. In contrast, a MappingFilter is an object wrapper around a collection of constraint-value pairs. If a constraint evaluates to true then its associated value is used to override the message’s priority or delivery deadline. Because different IDL types are used to express priorities and delivery deadlines, the value is wrapped inside an any.

22.3.5  ConsumerAdmin and SupplierAdmin

Figure 22.5 shows the create-style operations on the SupplierAdmin and ConsumerAdmin interfaces. Each of these operations takes a ClientType parameter that indicates if the supplier/consumer will handle event data as an any, StructuredEvent or a EventBatch. The create-style operation then creates a type-specific proxy object for the desired type. Each of these proxy types is a sub-type of ProxySupplier or ProxyConsumer.


module CosNotifyChannelAdmin {
  ...
  enum ClientType {ANY_EVENT, STRUCTURED_EVENT,
                              SEQUENCE_EVENT};
  interface ConsumerAdmin
    : CosNotifyFilter::FilterAdmin,
      // rest of inheritance clause omitted
  {
    attribute CosNotifyFilter::MappingFilter
                                         priority_filter;
    attribute CosNotifyFilter::MappingFilter
                                         lifetime_filter;
    ProxySupplier obtain_notification_pull_supplier(
                in  ClientType ctype, ...) raises (...);
    ProxySupplier obtain_notification_push_supplier(
                in  ClientType ctype, ...) raises (...);
    ...
  };
  interface SupplierAdmin 
    : CosNotifyFilter::FilterAdmin,
      // rest of inheritance clause omitted
  {
    ProxyConsumer obtain_notification_pull_consumer(
                in  ClientType ctype, ...) raises (...);
    ProxyConsumer obtain_notification_push_consumer(
                in  ClientType ctype, ...) raises (...);
    ...
  };
};
Figure 22.5: IDL for Notification Service Admin objects

The SupplierAdmin and ConsumerAdmin interfaces both inherit
from FilterAdmin, which provides operations for associating Filter objects with the admin object. In addition to this, the ConsumerAdmin has two MappingFilter attributes, for overriding the priority and delivery deadline of messages. The ability to associate filters with a ConsumerAdmin makes it possible to do filtering for a group of consumers rather than individually for each consumer. Not only is this more convenient from a maintenance point of view, it is also an important optimization because a filter is evaluated once for the entire group of consumers rather than being evaluated repeatedly, once per consumer. Providing filters in a SupplierAdmin is of less utility (simply because filtering is typically a consumer-side issue) but the capability is provided for the sake of symmetry.

22.3.6  EventChannel

One of the limitations of the Event Service was that it defined just a single event channel; it did not define a factory (Section 1.4.2.1) that could be used to create additional event channels. The Notification Service removes this limitation. Figure 22.6 shows the IDL definitions for EventChannel and EventChannelFactory.


module CosNotifyChannelAdmin {
  ...
  interface EventChannel
    : // inheritance clause omitted
  {
    readonly attribute ConsumerAdmin
                                  default_consumer_admin;
    readonly attribute SupplierAdmin
                                  default_supplier_admin;
    readonly attribute CosNotifyFilter::FilterFactory 
                                  default_filter_factory;
    ConsumerAdmin new_for_consumers(...);
    SupplierAdmin new_for_suppliers(...);
    ...
  };
  interface EventChannelFactory
  {
    EventChannel create_channel(...) raises(...);
    ...
  };
};
Figure 22.6: IDL for Notification Service event channel and factory

The create_channel() operation defined on the factory interface is used to create a new EventChannel. The parameters passed to this operation (which have been omitted from the IDL shown in Figure 22.6) are used to specify the QoS of the newly created channel. An overview of the available QoS is provided in Section 22.3.7.

Each EventChannel provides a set of pre-created ConsumerAdmin, SupplierAdmin and FilterFactory objects. These objects are accessible through the default_<...> attributes. However, the new_for_consumers() and new_for_suppliers() operations allow additional admin objects to be created. By creating different admin objects for different groups of suppliers/consumers, it is possible for the Notification Service to offer a different QoS (Section 22.3.7) for different groups of suppliers/consumers. Also, this arrangement makes it possible for, say, a ConsumerAdmin object to filter messages on behalf of a group of consumers. Such group-level filtering is much more efficient (and hence more scalable) than repeated filtering at the granularity of individual proxies.

22.3.7  Quality of Service (QoS)

The Notification Service allows users to choose a quality of service for the transmission of events. The following list briefly explains some of the more important quality of services available with the Notification Service:

Quality of service for a channel is indicated through name-value pairs. QoS can be specified when creating an EventChannel (Figure 22.6). This default QoS can then be overridden at the level of a ConsumerAdmin or SupplierAdmin object and/or at the level of individual proxy objects.

22.4  Telecom Log Service

In the Notification Service (Section 22.3), an EventChannel created with the persistent QoS stores each event in a persistent store (for example, a file or database) until it has delivered the event to all consumers. At that point, the EventChannel deletes the event from the persistent store. Some organizations prefer to keep a permanent record of events. They can do this with the Telecom Log Service, so called because it was defined by companies in the telecommunications industry, but its functionality is useful to organizations in other industries too.

The most important interfaces in the Telecom Log Service are NotifyLogFactory, which is a factory (Section 1.4.2.1) that creates NotifyLog objects, which inherit from the EventChannel interface of the Notification Service.

When a NotifyLog object (actually, one of its consumer proxy objects) receives an event from a supplier, it passes the event on to consumers, if any, and it also stores the event in a persistent store. The NotifyLog keeps events in its persistent store, even after it has delivered the events to all subscribed consumers.

Because the Telecom Log Service leverages the infrastructure of the Notification Service, a NotifyLog can have Filter objects (Section 22.3.4.1) associated with it. This means that a log object can be selective about which events it records.

The NotifyLog interface has operations that allow you to iterate over the events in its persistent store. You can query() events that match a particular constraint. You can also retrieve() the events that occurred since a specified time.

The Telecom Log Service gives you a lot of control over the persistent store of a NotifyLog object. For example:

1
Broadly speaking, this passing of data through weakly-typed, name-value pairs is similar in concept to the exchange of documents in XML format. Theoretically, an XML document could contain arbitrary elements (which are conceptually similar to name-value pairs). However, XML schemas can be defined that specify which elements should be present in an XML document. In both XML document exchange and transmission of messages through the Notification Service, it is people adhering to documented conventions that increases the likelihood of a particular document/message containing all the elements/name-value pairs that it is supposed to contain.
Previous Up Next