Pub/Sub Topics Behavior and Properties
A topic is an entity containing the message itself, along with additional subscriber and publisher information, and it is stored in the cache. It contains the message store, which stores the actual data objects being published by the publisher in a queue. It also internally maintains a list of all subscribers subscribing to it. The topic also contains a list of publishers.
Once the message is published to the topic, it fires an event, which the topic relays to subscribers based upon the preference of message delivery option so that they can get the message as needed. The following figure illustrates the role of a topic as an intermediary channel for publishers and subscribers:
Topic Behavior
Temporary Subscriber Disconnection
In case of temporary subscriber disconnection and then auto re-connection, all topic related information such as subscriptions and failure event notifications are re-registered to topic without any interruption on subscriber end.
Topic Deletion Notification
In case of topic deletion, it is forceful and deletes all messages and related meta-info from the cache. Hence, the subscriber and publisher must be notified of this deletion because of the following reasons:
The subscriber might be waiting for incoming messages from the registered topic. Once the topic does not exist, the subscribers can then handle their execution accordingly through event notification and prevent infinite waiting state.
The publisher can prevent sending messages to a non-existing topic and handle any pending messages and future execution accordingly.
Topic Priority
Note
This feature is available in NCache 5.2 and onward.
NCache introduces topic-level priority that helps you prioritize the lowest and high priority topics based on their importance. Topics containing critical or important messages can be created with a high priority which means that those messages are delivered first. Similarly, topics containing unimportant data can be created with a lower priority which means that those messages are evicted first.
Topics created with no priority specified will be created with the default priority i.e. Normal. Priority is specified using the TopicPriority
property as Low
, Normal
, and High
. In case eviction is enabled, the topics with least priority are evicted first, and the ones with the high priority are evicted last.
Important
Priority of a topic can be specified at the time of topic creation only and cannot be modified afterwards.
Prerequisites
- Install either of the following NuGet packages in your application based on your NCahce edition:
- Enterprise: Alachisoft.NCache.SDK
- Professional: Alachisoft.NCache.Professional.SDK
- Include the following namespace in your application:
Alachisoft.NCache.Client
Alachisoft.NCache.Runtime
Alachisoft.NCache.Runtime.Caching
Alachisoft.NCache.Runtime.Exceptions
- Cache must be running.
- The application must be connected to cache before performing the operation.
- Make sure that the data being added is serializable.
- For API details, refer to: ICache, CacheItem, ITopic, CreateTopic, GetTopic, DeleteTopic.
- To ensure the operation is fail safe, it is recommended to handle any potential exceptions within your application, as explained in Handling Failures.
- To handle any unseen exceptions, refer to the Troubleshooting section.
Create Topic
CreateTopic
method creates topic in cache with specified name. If the topic already exists, an instance of the topic is returned as ITopic
. Whenever a message is published on a topic, it is delivered based on message preference to subscribers that are registered on that topic. Following example creates a topic orderTopic.
try
{
// Pre-condition: Cache is already connected
// Mention the name of the topic
string topicName = "orderTopic";
// Create the topic
ITopic topic = cache.MessagingService.CreateTopic(topicName);
}
catch (OperationFailedException ex)
{
// Exception can occur due to:
// Connection Failures
// Operation Timeout
// Operation performed during state transfer
}
catch (Exception ex)
{
// Any other generic exception like ArgumentNullException or ArgumentException
// Topic name is null/empty
}
Note
To ensure the operation is fail safe, it is recommended to handle any potential exceptions within your application, as explained in Handling Failures.
Create Topic with Priority
Note
This feature is only available in NCache 5.2 and onward.
The following example creates a topic, orderTopic, with priority as high to avoid early eviction (if eviction is enabled).
try
{
// Pre-condition: Cache is already connected
// Mention the name of the topic
string topicName = "orderTopic";
// Create the topic with priority
ITopic topic = cache.MessagingService.CreateTopic(topicName, Alachisoft.NCache.Runtime.Messaging.TopicPriority.High);
}
catch (OperationFailedException ex)
{
// Exception can occur due to:
// Connection Failures
// Operation Timeout
// Operation performed during state transfer
}
catch (Exception ex)
{
// Any other generic exception like ArgumentNullException or ArgumentException
// Topic name is null/empty
}
Get Topic
GetTopic method fetches an instance of the specified topic from the cache. If the topic exists, it is returned, otherwise exception is thrown. The following example gets an existing topic orderTopic from the cache.
try
{
// Pre-condition: Cache is already connected
// Mention the name of the topic
string topicName = "orderTopic";
// Get the topic from the cache
ITopic orderTopic = cache.MessagingService.GetTopic(topicName);
// Verify successful topic retrieval
if (orderTopic != null)
{
// orderTopic will be used for receiving and/or publishing messages
}
else
{
// No topic exists
}
}
catch (OperationFailedException ex)
{
if (ex.ErrorCode == NCacheErrorCodes.TOPIC_NOT_FOUND)
{
// Specified topic does not exist
}
else
{
// Exception can occur due to:
// Connection Failures
// Operation Timeout
// Operation performed during state transfer
}
}
catch (Exception ex)
{
// Any other generic exception like ArgumentNullException or ArgumentException
// Topic name is null/empty
}
Note
To ensure the operation is fail safe, it is recommended to handle any potential exceptions within your application, as explained in Handling Failures.
Delete Topic
DeleteTopic method un-registers the topic from cache and removes all messages associated with that topic. If registered, a topic deletion callback OnTopicDeleted
will be triggered upon this method call.
The following example deletes the existing topic orderTopic from cache and removes all messages associated with that topic. If the OnTopicDeleted
callback is registered, it will be triggered upon this method call.
try
{
// Pre-condition: Cache is already connected
// Define the topic to be deleted
string topicName = "orderTopic";
// Delete the topic "orderTopic"
cache.MessagingService.DeleteTopic(topicName);
// Callback will be triggered if registered
}
catch (OperationFailedException ex)
{
if (ex.ErrorCode == NCacheErrorCodes.TOPIC_NOT_FOUND)
{
// Topic does not exist
}
else
{
// Exception can occur due to:
// Connection Failures
// Operation Timeout
// Operation performed during state transfer
}
}
catch (Exception ex)
{
// Any other generic exception like ArgumentNullException or ArgumentException
// Topic name is null/empty
}
Note
To ensure the operation is fail safe, it is recommended to handle any potential exceptions within your application, as explained in Handling Failures.
Delete Topic Asynchronously
DeleteTopicAsync
method deletes the topic asynchronously. Whenever a topic is deleted asynchronously, Task
is returned to the user for performing further tasks without waiting for the topic to be deleted. The following example shows the asynchronous deletion of topic orderTopic
.
try
{
// Pre-condition: Cache is already connected
// Define the topic to be deleted
string topicName = "orderTopic";
// Delete the topic "orderTopic"
Task task = cache.MessagingService.DeleteTopicAsync(topicName);
// Use task to perform further operations according to business logic
// Callback will be triggered if registered
}
catch (OperationFailedException ex)
{
if (ex.ErrorCode == NCacheErrorCodes.TOPIC_NOT_FOUND)
{
// Topic does not exist
}
else
{
// Exception can occur due to:
// Connection Failures
// Operation Timeout
// Operation performed during state transfer
}
}
catch (Exception ex)
{
// Any other generic exception like ArgumentNullException or ArgumentException
// Topic name is null/empty
}
Note
To ensure the operation is fail safe, it is recommended to handle any potential exceptions within your application, as explained in Handling Failures.
Properties of ITopic Interface
Members | Type | Description |
---|---|---|
MessageDeliveryFailure |
MessageDeliveryFailureCallback |
Event on topic so that publisher receives all failed messages that are not delivered to any subscriber or may be messages are expired or evicted before delivery. |
Name |
string |
Name of the topic specified during topic creation. |
OnTopicDeleted |
TopicDeletedCallback |
Event to handle topic deletion by publisher and subscriber. |
ExpirationTime |
TimeSpan |
If message level expiration is not provided, then this topic level expiration expires messages in the topic by default. The value is TimeSpan.MaxValue by default. |
IsClosed |
bool |
Check whether topic is disposed, before performing any operation. |
Dispose |
IDisposable |
Removes registered topic’s subscription from cache server. |
Additional Resources
NCache provides a sample application for Pub/Sub on GitHub.
See Also
Event Notifications in Cache
Pub/Sub Messages
Publish Messages to Topic
Subscribe for Topic Messages
Continuous Query