Publish/Subscribe Push MEP: Implementation Guidance

Last updated: JULY 2nd, 2019

1. Introduction
2. Technology Agnostic Service Design
3. Service Implementation
4. References

Introduction

This document intends to serve as implementation guidance of the Publish/Subscribe Push Message Exchange Pattern [7] using SWIM-TI Yellow Profile Service Bindings [6]. The document is structured as follows:

In section 2, we provide a general overview of Publish/Subscribe services in a technology agnostic manner.
In section 3, the technology agnostic service design is realised in a concrete implementation using WS-Light and AMQP Messaging Service Bindings [3].

The proposed service design and implementation is intended to provide a basic template for implementers that can be customised or complemented in multiple ways to suit specific needs.

Glossary of Terms

MEP:	Message Exchange Pattern
UUID:	Universally Unique IDentifier
REST:	Representational State Transfer
AMQP:	Advanced Message Queuing Protocol

Technology Agnostic Service Design

Publish/Subscribe Service Design

In its general form a Publish/Subscribe service can be understood as consisting of at least 2 interfaces:

A subscription interface: Containing operations that relate to the management of the service subscription (e.g. subscribe, unsubscribe, pause…).
A publish interface: Dealing with the actual publication of data to its consumers.

Figure 1: Publish Subscribe Service: Interfaces

Specific details of the implementation of the service subscription interface may vary. For example;

The subscribe , unsubscribe and pause operations may be merged into a single overloaded operation that performs either depending on the request payload.
The pause subscription operation may altogether not exist if said functionality is not needed.
Additional operations can be added to support extended functionality, for instance:
- A getSubscription/getSubscriptions operation to obtain details of an existing subscription(s),
- A getTopics operation that lists the possible topics a Consumer can subscribe to,
- Etc…

What is usually common across implementations is that the Subscribe interface is implemented using a Request/Response pattern ^[1] as subscribing to the service typically entails some form of approval by the service provider which in turn provides back certain information detailing the means to use to obtain the publications. This interface is provided by the Service Provider and required by the Consumer.

On the other hand, the Publish interface is most commonly implemented using a Fire-and-Forget MEP [7] as it fully leverages the advantages of Publish/Subscribe, minimizing interactions between providers and consumers and preventing unnecessary workload on consumer side, which would otherwise need to perform periodic requests (polling) to get the latest updates. This interface is provided by the Consumer that needs to be able to accept reception of the publish operation and required by the Service Provider who will invoke it.

Particular technical implementations may rely on additional interfaces and operations to manage different aspects of the service like; connection establishment between Consumer and Service, delivery state of messages, heartbeats to communicate the status of each participant, etc... These interfaces are implementation specific and vary from technology to technology and thus are best documented as part of the technology specific service description (see section 3).

Subscription Model

The Subscribe interface needs to provide a model that specifies what defines a subscription. There is no standardized way to do so and different services might have different needs that may justify variation in the subscription models. In this section we provide a simple and extensible subscription model and common alternative variations.

A subscription can be described as depicted in Figure 2.

Figure 2: Subscription class (non-normative).

id: Unique identifier of a subscription, (e.g. a UUIDv4).
topic: A string that matches under certain predefined rules a topic or set of topics of the service. It may take values in an enumeration list or might be defined using a tree structure and allow the use of wildcard operators.
durable: Expresses the durability of the subscription, a subscription is durable if messages are preserved while subscribers are offline. Sometimes can be accompanied by a time-to-live attribute that indicates the expiration time of non-delivered messages.
qos: Expresses the reliability Quality of Service of the subscription.
status: A description of the state of a subscription, usually taking values in a predefined enumeration list. The possible states must be accompanied by a state diagram indicating the possible transitions and its triggers.

Subscription States and Lifecycle

The subscription model defines the following subscription states:

ACTIVE: The subscription has been created and messages are being queued in the subscriber´s queue.
PAUSED: The subscription has been created but messages are not being queued in the subscriber’s queue.
DELETED: The subscription and the queue have been deleted.

The operations of the Subscribe interface allow the Consumer to manage the state of a subscription, the following state diagram depicts the lifecycle of a subscription and the different state transitions associated with its operations.

Subscription Filtering

Filtering mechanisms provide a way for consumers to declare interest in a subset of publications, helping reduce unnecessary processing and bandwidth consumption for both provider and consumers.

Subject-based Filtering

In Publish/Subscribe services a topic is a descriptor of the subject of the information exchanged. Topics classify the publications in a number of different subjects, during subscription Consumers can declare interest in a number of topics which allows them to filter the publication received to that particular subset.

Topic Enumerations

An enumeration provides the simplest mechanism to define the list of subjects to which a consumer can subscribe.

The following figure illustrates an example topic enumeration with three possible topics that classify the publications of a service according to their subject.

Topic Trees

Topic trees provide a rich mechanism to classify the publications of a service in a structured way.

A topic tree can be defined as follows:

Define a reserved character that acts as word separator. E.g.
- Use the character “/” as a word separator.
Define a tree structure of topics using the word separator character. E.g.:
/weather/precipitation/LEMD
/weather/precipitation/LFPG
/weather/wind/LEMD
/weather/wind/LFPG
Define a number of wildcard characters that can be used to match sub-branches of tree structure. E.g.:
- “*” as a wildcard that matches any single word of a topic
- “#” as a wildcard that matches any number of consecutive words of a topic.

With this topic structure the consumers can declare interest to specific data by subscribing to topics using the semantics defined. E.g.:

Subscribing to “ /weather/#” declares interest to precipitation and wind data for all airports.
Subscribing to “/weather/wind/*” declares interested to wind data for all airports.
Subscribing to “/weather/precipitation/LFPG” declares interest to precipitation data for the Paris Charles de Gaulle Airport.

Content-based Filtering

Content-based filtering allows consumers to specify their interest into specific publications based on the runtime evaluation of certain predicates against the content of the publication message. They provide the most flexible and general filtering mechanism but the most difficult to implement and with expensive runtime costs.

For example, a consumer may declare interest in every publication that contains a particular keyword or publications where a particular attribute of the payload is bigger than a concrete value.

Dynamic Behaviour

The following sequence diagram shows the nominal flow of a subscription, subsequent publications and final unsubscription from the service:

Figure 5: Publish/Subscribe sequence diagram

The Consumer performs a subscription request, filling the appropriate request message.
The Service Provider processes the request (payload validation, authentication and authorization processing occur at this moment) and provides the appropriate response message.
When a new message is available it is published to the Service Consumer.
Finally, the Service Consumer unsubscribes from the service which stops further publications.

This nominal flow represents the basic Publish/Subscribe interaction that can be extended in a straight-forward manner to include additional operations like pause and resume subscription. This nominal flow can also be extended to a Brokered Publish/Subscribe Push Message Exchange Pattern, in which case the broker or brokers would appear as intermediaries between Service and Consumer.

Service Implementation

Topic Publish/Subscribe Service Implementation

A Topic-based Publish/Subscribe service can be implemented in the following manner:

Using WS-Light Service Binding for the implementation of the Subscribe interface,
Using AMQP Messaging Service Binding for the implementation of the Publish interface.

In this section we will provide detailed implementation guidance to realize the generic service design outlined in section 2.1 using the proposed Service Bindings.

Subscription

The Subscribe interface can be implemented using a REST design. The REST resources represent:

A collection of topics: /topics/
A specific topic: /topics/<topic>
The collection of all subscriptions: /subscriptions/
A specific subscription: /subscriptions/<subscription_id>

The following table summarizes a mapping between the technology agnostic interface design (to the left) and the REST resource endpoint (to the right) with the corresponding HTTP method.

Design	Implementation
Operation	Resource endpoint	HTTP method
subscribe	/subscriptions/	POST
pause	/subscriptions/<subscription_id>	PUT
resume	/subscriptions/<subscription_id>	PUT
unsubscribe	/subscriptions/<subscription_id>	DELETE

Additional operations to retrieve the list of available topics and the current state of a particular subscription can be included in a simple manner:

Design	Implementation
Operation	Resource endpoint	HTTP method
getTopics	/topics/	GET
getTopic	/topics/<topic>	GET
getSubscription	/subscriptions/<subscription_id>	GET

This operation allows a Consumer to request the subscription to a particular Topic of the Service.

Description:

The Consumer invokes a “POST” on the /subscriptions/ endpoint.
The Service Provider applies Authentication, Authorization and Data Validation controls and determine the result of the subscription request.
The Service Provider provides a Response with the result of the request including (if successful) the appropriate queue where the messages will be published.

Post-Conditions: If successful, the /<queue> and /subscriptions/<subscription_id> endpoints are created.

Example Request Message

Headers

Accept: application/json, application/xml

Content-Type : application/json

Accept-Encoding: gzip

Body

{“topic”: <topic>,

“qos”: “EXACTLY_ONCE”,

“durable”: true}

Example Response Message

Headers

Status-Code: 201

Reason-Phrase: Created

Content-Type: application/json

Body

{“topic”: <topic>,

"subscription_id": <subscription_id>,

"queue": <queue>,

"subscription_status": "PAUSED",

“qos”: “EXACTLY_ONCE”,

“durable”: true}

Error Codes

The following error codes may be of interest to implementers, covering common failures applicable to this resource invocation. Implementers are free to use a suitable subset of the ones proposed or complement it with additional ones to cover particular implementation needs.

400 Bad Request: Indicated for malformed requests on the part of the client.

401 Unauthorized: Indicated for authentication errors (e.g. the authentication is not provided, the credentials used do not exists, the authentication mechanism used is not valid…).

403 Forbidden: Indicated for authorization errors (e.g. the authenticated credentials are not allowed to perform this operation on the resource).

405 Method Not Allowed: The HTTP method used is not supported by the resource.

406 Not Acceptable: None of the Content-Types provided in the request Accept header are supported.

415 Unsupported Media Type: The Content-Type of the request is not valid for this operation.

500 Internal Server Error: Catch-all error for unexpected internal errors during the processing of the request.

pause ^[2]

This operation allows a Consumer to pause the publishing of messages for a particular topic subscription.

Description:

Pre-Conditions: The Consumer has successfully subscribed to a particular <topic>.

The Consumer invokes the “PUT” method on the /subscriptions/<subscription_id> endpoint. The body of the message includes a change to the subscription_status to “PAUSED”.
The Service Provider may apply Authentication and Authorization controls to determine the result of the subscription request.
The Service Provider provides a Response with the result of the request including (if successful) the new subscription_status .

Post-Conditions: If successful, the subscription is paused.

Example Request Message

Headers

Accept: application/json, application/xml

Accept-Encoding: gzip

Body

{“topic”: <topic>,

"subscription_status": "PAUSED",

“qos”: “EXACTLY_ONCE”,

“durable”:true}

Example Response Message

Headers

Status-Code: 200

Reason-Phrase: Accepted

Content-Type: application/json

Body

{“topic”: <topic>,

"subscription_id": <subscription_id>,

"queue": <queue>,

"subscription_status": "PAUSED",

“qos”: “EXACTLY_ONCE”,

“durable”:true}

Error Codes

400 Bad Request: Indicated for malformed requests on the part of the client.

401 Unauthorized: Indicated for authentication errors (e.g. the authentication is not provided, the credentials used do not exists, the authentication mechanism used is not valid…).

403 Forbidden: Indicated for authorization errors (e.g. the authenticated credentials are not allowed to perform this operation on the resource).

404 Not Found: The requested resource cannot be found (e.g. the <subscription_id> does not exist).

405 Method Not Allowed: The HTTP method used is not supported by the resource.

406 Not Acceptable: None of the Content-Types provided in the request Accept header are supported.

415 Unsupported Media Type: The Content-Type of the request is not valid for this operation.

500 Internal Server Error: Catch-all error for unexpected internal errors during the processing of the request.

resume

Due to the stateless implementation of the API, this operation is performed in exactly the same way as the pause changing the subscription_status in the body of the message to “ACTIVE”.

Description:

Pre-Conditions: The Consumer has successfully subscribed to a particular <topic>.

The Consumer invokes the “PUT” method on the /subscriptions/<subscription_id> endpoint. The body of the message includes a change to the subscription_status to “ACTIVE”.
The Service Provider may apply Authentication and Authorization controls to determine the result of the subscription request.
The Service Provider provides a Response with the result of the request including (if successful) the new subscription_status .

Post-Conditions: If successful, the subscription is paused.

Example Request Message

Headers

Accept: application/json, application/xml

Accept-Encoding: gzip

Body

{“topic”: <topic>,

"subscription_status": "ACTIVE",