Pub/Sub Messaging-Subscribe to a Topic

In Pub/Sub messaging model, subscriber (or client application) can register to a particular Topic through a subscription. So, NCache provides multiple types of Pub/Sub Topic subscriptions.

Note

This feature is also available in the NCache Community Edition, except for Multiple Subscriptions.

Prerequisites

To learn about the standard prerequisites required to work with all NCache client-side features, please refer to the given page on Client-Side API Prerequisites.
For API details, refer to: ICache, CacheItem, ITopic, IDurableTopicSubscription, CreateSubscription, CreateDurableSubscription,TopicName, SubscriptionName, IMessagingService, GetTopic, SubscriptionPolicy, UnSubscribe, MessageEventArgs, DeliveryMode, TopicSearchOptions, ITopicSubscription, MessageDeliveryFailure, PayLoad.

Install either of the following NuGet packages in your .NET client application:
- Enterprise: Install-Package Alachisoft.NCache.SDK -Version 4.9.1.0
- Professional: Install-Package Alachisoft.NCache.Professional.SDK -Version 4.9.1.0
Create a new Console Application.
Make sure that the data being added is serializable.
Add NCache References by locating %NCHOME%\NCache\bin\assembly\4.0 and adding Alachisoft.NCache.Web and Alachisoft.NCache.Runtime as appropriate.
Include the Alachisoft.NCache.Web.Caching and Alachisoft.NCache.Runtime.Caching namespaces in your application.
To learn more about the NCache Legacy API, please download the NCache 4.9 documents available as a .zip file on the Alachisoft Website.

Methods to Create a Subscription

Here we describe how you can create Non-durable Subscription, Durable Subscription, Multiple subscriptions, and Asynchronous subscription in Pub/Sub Messaging Model.

Non-Durable Subscriptions

The ITopic/Topic interface facilitates creating a Non-durable Subscription and publishing of messages against the Topic. The create subscription method registers a Non-durable Subscription against a Topic if the Topic exists. It allows the subscriber to register MessageReceivedCallback against the Topic so that it can receive the published messages.

The following code sample does the following:

Get existing Topic of interest, i.e., NewBeverages.
Create subscription for each Topic.
Register events for subscribers to receive messages once published to the Topic.

// Precondition: Cache is already connected

// NewBeverages is the name of the Topic created, beforehand
string topicName = "NewBeverages";

// Get the Topic
ITopic topic = cache.MessagingService.GetTopic(topicName);

// If Topic exists, Create subscription
if (topic != null)
{
    // Create and register subscribers for order Topic
    // Message received callback is specified
    ITopicSubscription subscription = topic.CreateSubscription(MessageReceived);
}

// Precondition: Cache is already connected

// NewBeverages is the name of the Topic created beforehand
String topicName = "NewBeverages";

// Get the Topic
Topic ordertopic = cache.getMessagingService().getTopic(topicName);

// If Topic exists, Create subscription
if (ordertopic != null) {
    // Create and register subscribers for Topic
    ordertopic.createSubscription(new PubSubMessageReceivedListener());
    System.out.println("Subscriber registered for Topic: " + ordertopic.getName());
} else {
    System.out.println("Topic does not exist.");
}

# Precondition: Cache is already connected

# NewBeverages is the name of the Topic created, beforehand
topic_name = "NewBeverages"

# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name)

# If Topic exists, Create subscription
if order_topic is not None:
    # Create and register subscribers for the Topic
    # Message received callback is specified
    order_subscriber = order_topic.create_subscription(on_message_received)

// This is an async method

// Precondition: Cache is already connected

// NewBeverages is the name of the Topic created, beforehand
let topicName = "NewBeverages";

// NewBeverages is the name of the Topic created, beforehand
let messagingService = await cache.getMessagingService();
let ordertopic = await messagingService.getTopic(topicName);

// If Topic exists, Create subscription
if (ordertopic != null) 
{
    // Create and register subscribers for Topic
    // Message received callback is specified
    let messageListener = new ncache.MessageReceivedListener(onMessageReceived);
    let ordersubscriber = await ordertopic.createSubscription(messageListener);
}

// Using NCache Enterprise 4.9.1
// Precondition: Cache is already connected

// NewBeverages is the name of the Topic created, beforehand
string topicName = "NewBeverages";

// Get Order Topic
ITopic topic = cache.MessagingService.GetTopic(topicName);

// If Topic exists, Create subscription
if (topic != null)
{
    // Create and register subscribers for Topic
    // Message received callback is specified
    ITopicSubscription ordSubscriber = topic.CreateSubscription(MessageReceived);

}

Note

To ensure the operation is fail-safe, it is recommended to handle any potential exceptions within your application, as explained in Handling Failures.

Durable Subscriptions

The IDurableTopicSubscription interface facilitates creating Durable subscriptions and publishing messages against the Topic if it exists. It allows the subscriber to register a MessageReceivedCallback against the Topic, so that it can receive the published messages.

Shared Durable Subscriptions

The following code demonstrates the Shared Subscription policy.

// Precondition: Cache is already connected

// DiscountedBeverages is the name of the Topic created, beforehand
string topicName = "DiscountedBeverages";

string subscriptionName = "DiscountedBeveragesSubscription";

// Get the Topic
ITopic ordertopic = cache.MessagingService.GetTopic(topicName);

if (ordertopic != null)
{
    // Create and register subscribers for Topic
    // Message received callback is specified
    // The subscription policy is shared which means that the subscription can have more than one subscribers
    IDurableTopicSubscription ordersubscription = ordertopic.CreateDurableSubscription (subscriptionName, SubscriptionPolicy.Shared, MessageReceived, TimeSpan.FromMinutes(20));
}

// Precondition: Cache is already connected

// DiscountedBeverages is the name of the Topic created beforehand
String topicName = "DiscountedBeverages";
Topic ordertopic = cache.getMessagingService().getTopic(topicName);

// Get the Topic
if (ordertopic != null) {
    // Create and register subscribers for Topic
    // The subscription policy is shared which means that the subscription can have more than one subscribers
    System.out.println("Topic retrieved: " + ordertopic.getName());
    ordertopic.createDurableSubscription(subscriptionName, SubscriptionPolicy.Shared, new PubSubMessageReceivedListener(), TimeSpan.FromMinutes(20));
    System.out.println("Durable subscriber registered for Topic: " + ordertopic.getName());
} else {
    System.out.println("Topic does not exist.");
}

# Precondition: Cache is already connected

# DiscountedBeverages is the name of the Topic created, beforehand
topic_name = "DiscountedBeverages";

subscription_name = "DiscountedBeveragesSubscription";

# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name);

if order_topic is not None:
    # Create and register subscribers for Topic
    # MessageReceived callback is specified below
    # The subscription policy is Shared which means that the subscription can have more than one subscribers
    order_Subscriber = order_topic.create_durable_subscription(subscription_name, SubscriptionPolicy.SHARED, on_message_received, TimeSpan.from_minutes(20));

// Precondition: Cache is already connected

// DiscountedBeverages is the name of the Topic created, beforehand
let topicName = "DiscountedBeverages";

let subscriptionName = "DiscountedBeveragesSubscription";

// Get the Topic
let messagingService = await cache.getMessagingService();
let orderTopic = await messagingService.getTopic(topicName);

let messageListener = new ncache.MessageReceivedListener(onMessageReceived);

if (orderTopic !=null)
{
    // Create and register subscribers for Topic
    // Message received callback is specified below
    // The subscription policy is Shared which means that the subscription can have more than one subscribers
    let timeSpan = new TimeSpan(0, 20, 0);
    let orderSubscriber = orderTopic.createDurableSubscription(subscriptionName, SubscriptionPolicy.Shared, messageListener, timeSpan, DeliveryMode.Reliable);
}

Exclusive Durable Subscriptions

The following code demonstrates the Exclusive Subscription policy.

// Precondition: Cache is already connected

// orderTopic is the name of the Topic created beforehand
string topicName = "orderTopic";

string subscriptionName = "orderTopicName";

// Get the Topic
ITopic orderTopic = cache.MessagingService.GetTopic(topicName);

// Create and register subscribers for Topic
// The subscription policy is exclusive which means that the subscription can have only one subscriber
IDurableTopicSubscription orderSubscriber = orderTopic.CreateDurableSubscription(subscriptionName, SubscriptionPolicy.Exclusive, MessageReceived, TimeSpan.FromMinutes(20));

// Precondition: Cache is already connected

// Get the Topic
Topic orderTopic = cache.getMessagingService().getTopic(topicName);

if (orderTopic != null) {
    System.out.println("Topic retrieved: " + orderTopic.getName());

    // Create and register subscribers for Topic
    // The subscription policy is exclusive which means that the subscription can have only one subscriber
    orderTopic.createDurableSubscription(subscriptionName,SubscriptionPolicy.Exclusive, new PubSubMessageReceivedListener(),TimeSpan.FromMinutes(20));
    System.out.println("Durable subscriber registered for Topic: " + orderTopic.getName());
} else {
    System.out.println("Topic does not exist.");
}

# Precondition: Cache is already connected

# orderTopic is the name of the Topic created beforehand
topic_name = "orderTopic"
subscription_name = "orderTopicName"

# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name)

# Create and register subscribers for orderTopic
# The subscription policy is exlusive which means that the subscription can have more than one subscriber
order_subscriber = order_topic.create_durable_subscription(subscription_name, SubscriptionPolicy.EXCLUSIVE, on_message_received, TimeSpan.from_minutes(20))

// Precondition: Cache is already connected

// orderTopic is the name of the Topic created beforehand
 let topicName = "orderTopic";

let subscriptionName = "orderTopicName";

// Get the Topic
let messagingService = await cache.getMessagingService();
let orderTopic = await messagingService.getTopic(topicName);

let messageListener = new ncache.MessageReceivedListener(onMessageReceived);

// Create and register subscribers for order Topic
// Message received callback is specified below
let timeSpan = new TimeSpan(0, 20, 0);
let orderSubscriber = orderTopic.createDurableSubscription(subscriptionName, SubscriptionPolicy.Exclusive, messageListener, timeSpan, DeliveryMode.Reliable);

Multiple Subscriptions

Using this method of subscription, subscribers can provide patterns to subscribe to multiple Topics with a single call. For this, it is important that the Topics matching the pattern already exist on the server. Once the subscription(s) is/are successfully created on the pattern-based Topic, the subscribers receive messages published on Topics that match the pattern. Also, if a Topic is created after the pattern-based-subscription has been registered against the pattern, it will register that subscriber against that Topic. Similarly, regarding unregistering from a Topic, this method unsubscribes the subscriber against all the matching Topics with provided pattern without affecting any other subscription using this call.

Note

A subscriber can only get pattern-based Topic and is not allowed to create it.
A pattern can be used by the publisher only to receive failure notifications.

Supported Wildcards

Pattern-based method of subscription supports the following three wildcards:

* : Zero or many characters. For example, bl* subscribes to black, blue, and blur, etc.
? : Any one character. For example, h?t subscribes to hit, hot and hat, etc.
[] : Range of characters. For example, b[ae]t subscribes to bet and bat, but not bit.

Creating a Subscription with Wildcard

The following example creates a subscription by providing a pattern based on which the Topic is subscribed matching the pattern.

// Precondition: Cache is already connected

// Create Topic name for all Topics with suffix Beverages
// Only ? * [] wildcards supported
string topicName = "*Beverages";

// Get the Topic
ITopic topic = cache.MessagingService.GetTopic(topicName, TopicSearchOptions.ByPattern);

// If Topic exists, Create subscription
if (topic != null)
{
    // Create and register subscribers for Topic
    ITopicSubscription subscription = topic.CreateSubscription(MessageReceived);
}

// Precondition: Cache is already connected

// Define Topic name pattern
String topicNamePattern = "*ages";

// Get all Topics that fulfill the pattern
Topic topic = cache.getMessagingService().getTopic(topicNamePattern, TopicSearchOptions.ByPattern);

if (topic != null) {
    System.out.println("Matching Topic retrieved: " + topic.getName());

    // Create and register notifications for Topic
    topic.createSubscription(new PubSubMessageReceivedListener());
    System.out.println("Subscription created for Topic: " + topic.getName());
} else {
    System.out.println("No matching Topic found.");
}

# Precondition: Cache is already connected

# Create Topic name for all Topics with suffix Beverages
# Only ? * [] wildcards supported
topic_name = "*Beverages*"

# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name, TopicSearchOptions.BY_PATTERN)

if order_topic is not None:
    # Create and register subscribers for Topic
    order_subscriber = order_topic.create_subscription(on_message_received)

// Precondition: Cache is already connected

// Create Topic name for all Topics with suffix Beverages
// Only ? * [] wildcards supported
let topicName = "*Beverages";

// Get the Topic
let messagingService = await cache.getMessagingService();
let orderTopic = await messagingService.getTopic(topicName, TopicSearchOptions.ByPattern);

let messageListener = new ncache.MessageReceivedListener(onMessageReceived);

// If Topic exists, Create subscription
if (orderTopic!=null)
{
    // Create and register subscribers for Topic
    let orderSubscriber = orderTopic.createSubscription(messageListener);
}

Asynchronous Subscriptions

Delivery mode can be specified while creating subscriptions for ordered messages and can be sync or async. The following example creates subscription using CreateSubscription method with sync delivery mode. It is recommended to use sync mode for ordered messages and async mode otherwise to achieve high performance.

While creating a subscription, you can specify delivery mode that can be synchronous or asynchronous. In particular, synchronous delivery mode can be used for ordered messages, whereas if you are not using ordered messages, asynchronous mode improves performance.

Note

By default, the delivery mode of messages is set to synchronous.

// Precondition: Cache is already connected

// NewBeverages is the name of the Topic created beforehand
string topicName = "NewBeverages";

// Get the Topic
ITopic topic = cache.MessagingService.GetTopic(topicName);

// If Topic exists, Create subscription
if (topic != null)
{
    // Create and register subscribers for Topic
    // Message received callback is specified
    // DeliveryMode is set to async
    ITopicSubscription subscription = topic.CreateSubscription(MessageReceived, DeliveryMode.Async);
}

// Precondition: Cache is already connected

// NewBeverages is the name of the Topic created beforehand
String topicName = "NewBeverages";

// Get the Topic
Topic topic = cache.getMessagingService().getTopic(topicName);

if (topic != null) 
{
    // Create and register subscribers for Topic
    // Message received callback is specified
    // DeliveryMode is set to async
    topic.createSubscription(new PubSubMessageReceivedListener(),DeliveryMode.Async);
} else 
{
    System.out.println("Topic does not exist.");
}

# Precondition: Cache is already connected

# NewBeverages is the name of the Topic created beforehand
topic_name = "NewBeverages"

# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name)

if order_topic is not None:
    # Create and register subscribers for Topic
    # Message received callback is specified
    # DeliveryMode is set to async
    order_subscriber = order_topic.create_subscription(on_message_received, DeliveryMode.SYNC)

// Precondition: Cache is already connected

// This is an async method

// NewBeverages is the name of the Topic created beforehand
let topicName = "NewBeverages";

// Get the Topic
let messagingService = await cache.getMessagingService();
let orderTopic = await messagingService.getTopic(topicName);

let messageListener = new ncache.MessageReceivedListener(onMessageReceived);

if (orderTopic != null) 
{

    // Create and register subscribers for Topic
    // Message received callback is specified
    let subscription = await orderTopic.createSubscription(messageListener, DeliveryMode.Sync);

}

Note

You can also unsubscribe from a Topic.

Additional Resources

NCache provides sample application for Pub/Sub on GitHub.

Note

Prerequisites

Methods to Create a Subscription

Non-Durable Subscriptions

Note

Durable Subscriptions

Shared Durable Subscriptions

Exclusive Durable Subscriptions

Multiple Subscriptions

Note

Supported Wildcards

Creating a Subscription with Wildcard

Asynchronous Subscriptions

Note

Note

Additional Resources

See Also

Contact Us

Pub/Sub Messaging: Subscribe to a Topic

Note

Prerequisites

Methods to Create a Subscription

Non-Durable Subscriptions

Note

Durable Subscriptions

Shared Durable Subscriptions

Exclusive Durable Subscriptions

Multiple Subscriptions

Note

Supported Wildcards

Creating a Subscription with Wildcard

Asynchronous Subscriptions

Note

Note

Additional Resources

See Also

Contact Us