About Us Products Articles News Contact Support

Programming Distributed Systems with YAMI4

6.4 Publish-Subscribe Messaging

Simple publish-subscribe scenarios are supported with the concept of value publisher.

Publish-subscribe messaging can be appropriate when the distributed system has one or more well-identified sources of data and there are programs that are interested in these data values. If in such a system the source or publisher is not particularly interested in who might need the data, then publish-subscribe approach can make it much easier to implement the whole system. Many examples of such architectures can be seen in control systems where many programs are dedicated to measurements - for example, a temperature measurement server can periodically poll its physical temperature sensors and publish current values without any regard to who might be listening at the given moment, while client applications can express their interest in some or all of the values. Publish-subscribe examples are also found in the enterprise world, where traders announce prices of various items while others passively wait for announcements to take further decisions based on the received values.

Independently of the actual application, the publish-subscribe messaging can be implemented either with the central messaging service or with proper logic embedded in the publisher program. Both approaches are supported in YAMI4 - the centralized messaging service can be managed independently and is appropriate in large-scale installations, while the latter approach is much easier to deploy, since even though the publisher program does not need to be aware of who is currently subscribed, the communication is still of the peer-to-peer kind and does not rely on external brokers.

The centralized message broker is described in a separate part of this book.

In order to benefit from the peer-to-peer publish-subscribe support in YAMI4, the publisher program can set up a special object that automatically interprets the "subscribe" and "unsubscribe" commands from remote agents and keeps the list of active subscribers. Whenever a new value is ``published'' via this object, all active subscribers receive the update message with that new value. The subscribers, in turn, express their interest in receiving updates by sending the subscribe and unsubscribe messages with the name of their own objects, where the updates are to be delivered. This means that named objects need to be managed on both sides: at the publisher side the named object represents the source data value (like temperature or stock price), whereas on the subscriber side the named object represents the consumer of that data. In many cases these names will be equal.

The subscriber program can establish new subscription with the "subscribe" message that has to be sent to the publisher, to the object that was automatically registered at the publisher's agent. This message can contain a parameters object with zero, one or all of these fields:

"destination_object" (string), with the intended object name where the updates should be sent; if not provided, the same object name that is used for the value publisher will be used for subscription updates.
"destination_target" (string), with the intended target where the updates should be sent; this field essentially allows to establish subscriptions on behalf of other agents, but in typical cases will not be provided at all, which means that the subscribing agent is to receive subscription updates.

Note:

It is not possible to establish multiple subscriptions for the same value publisher with the same subscriber target. This is intended to protect the system from sending multiple identical messages between two agents. A new subscription with the same subscriber target will replace the existing one.

After receiving the subscription request, the value publisher automatically registers the request in its internal data structure and replies to confirm the subscription. It is also possible for the data source component to intercept this process and implement additional processing of such request or provide customized response to the subscriber.

If any communication error happens during the transmission of updates, the problematic subscribers are removed from the list of active subscribers. This means that it is up to the subscriber to recover its subscription if it was restarted.

Subscriptions can be also established and revoked by explicit calls to dedicated functions named subscribe() and unsubscribe() on the value publisher object. This is intended to help in implementation of subscription scenarios that are driven by means other than regular and single messages from remote agents.

It should be noted that even though the typical publish-subscribe systems do not focus on command-response interactions, YAMI4 implements the publish-subscribe scheme on top of and in addition to command-response, which means that both paradigms are equally possible in a single program and the integration of these two concepts allows a single logical entity (a YAMI4 object) to take part in the two forms of communication at the same time. In fact, ``subscribe'' and ``unsubscribe'' are regular command-response messages.

Note:

The simple publish-subscribe service does not provide automatic renewal of subscriptions when the publisher is switched off and subsequently restarted. Providing such a renewal in a general way would require the introduction of external discovery, directory or broking service and is therefore outside of the scope of the peer-to-peer messaging library. A simplified implementation of such recovery scheme can be provided by the publisher itself, which can inspect its list of subscribers, store it in a persistent way and reestablish all subscriptions after restart. Simple solutions that rely on the subscriber to be proactive in the recovery are also possible and can be based on periodic probing of the publishing side.

One of the more challenging problems that can be encountered in publish-subscribe messaging is related to the situation where clients (subscribers) do not process the stream of updates properly. Two different problems have to be addressed:

Subscriber definitively disconnects or terminates - how does it affect the delivery of updates to other subscribers?
Subscriber hangs or processes incoming updates not fast enough, but their network connections are still in a proper state, which leads to the accumulation of unconfirmed messages at the publisher side - what is the impact on other subscribers and the publisher itself?

The first problem above is handled automatically by the publisher - a problematic subscriber is immediately dropped from the set and no connection is recreated automatically. This process does not introduce any artificial delays and therefore the timing of message delivery to other subscribers is not affected.

The other problem is less obvious, as the hiccups at the subscriber side might be temporary or can result from the temporary burst of messages produced by the publisher. Most of the time such problems are recoverable, unless the backlog accumulates beyond reasonable limits. In order to properly distinguish between such different situations, the management of such subscriptions can be handled by user, in a way that is most appropriate for the given system - this is done with the overflow reaction callback, which can be installed by user when the value publisher is created. Then, every time when the overflow condition is detected for any of the subscribers, the user-provided callback is notified with the connection name and target object name and the user is asked for his decision about how to proceed with the overflowing message. The following decisions are possible:

Wait for some previously sent message, so that the queue associated with the given subscriber can be consumed to create a place for new message. This strategy ensures that subscribers will ultimately receive all published messages (as long as they are alive and can participate in communication without I/O errors), but at the price of possible delay that might affect other subscribers - that is, if one subscriber consumes messages slowly, others will notice the slow-down as well. This is the default strategy that is used by the value publisher when no user-provided overflow callback is registered.
Abandon this single message and proceed to other subscribers, who might be ready to receive it. This strategy temporarily penalizes slow subscribers, but allows them to consume their pending messages and receive later updates properly.
Abandon the whole subscription. This strategy penalizes slow subscribers by removing them from the set, so that no further messages will be sent to them until they explicitly subscribe again.

The overflow management is essential in high-frequency or real-time systems, but in low-frequency systems where all subscribers always consume messages before new ones are published, the default strategy will be appropriate.

The following sections present examples of publish-subscribe messaging in each programming language.