Pub/Sub Topics Behavior and Properties

A topic is an entity containing the message itself, along with additional subscriber and publisher information and is stored in the cache. The topic contains the message store which stores the actual data objects published by the publisher in a queue. It also internally maintains a list of all subscribers subscribing to it.

Once the message is published to the topic, it fires an event, which the topic relays to subscribers based upon 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:

PubSub Topics in NCache

Topic Behavior

Temporary Client Disconnection

In case of temporary client application disconnection and then auto re-connection, all topic related information such as client subscriptions and failure event notifications are re-registered to topic without any interruption on client application end.

Topic Deletion Notification

In case a topic is deleted, it is a forceful deletion and deletes all messages and related meta-info form 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 only available in NCache 5.2 and onward.

NCache introduces topic level priority that helps you priorities the lowest and highest 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 created with lower priority contain unimportant data which results in evicting these messages 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 of enabled eviction or expiration, the topics with least priority are evicted first and the ones with the highest are evicted last.

Important

Priority of a topic can be specified at the time of topic creation only and cannot be modified afterwards.

Note

To use Maven packages for NCache Professional Edition, change the <artifactId> as shown below:
- <artifactId>ncache-professional-client</artifactId>
To use Node.js API in NCache Professional Edition, install and include the ncache-professional-client npm package in your Node.js application.

Managing Topics

Pre-Requisites for Managing Topics

Install the following NuGet packages in your application:
- Alachisoft.NCache.SDK
Include the following namespace in your application:
- Alachisoft.NCache.Client
- Alachisoft.NCache.Runtime
- Alachisoft.NCache.Runtime.Caching
- Alachisoft.NCache.Runtime.Exceptions
The application must be connected to cache before performing the operation.
Cache must be running.
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.

Add the following Maven dependencies in your pom.xml file:

<dependency>
    <groupId>com.alachisoft.ncache</groupId>
    <artifactId>ncache-client</artifactId>
    <version>5.2.0</version>
</dependency>

Import the following packages in your application:
- import com.alachisoft.ncache.client.*;
- import com.alachisoft.ncache.runtime.exceptions.*;
The application must be connected to cache before performing the operation.
Cache must be running.
Make sure that the data being added is serializable.
For API details, refer to: Cache, CacheItem, Topic, 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.

Include the following modules in your application:
- const ncache = require('ncache-client')
The application must be connected to cache before performing the operation.
Cache must be running.
For API details, refer to: Cache, CacheItem, 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 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
}

try 
{
    // Precondition: Cache is already connected 

    // Mention the name of the topic
    String topicName = "orderTopic";

    // Create the topic
    Topic topic = cache.getMessagingService().createTopic(topicName);
} 
catch (OperationFailedException ex) 
{
    // Exception can occur due to:
    // Connection Failures
    // Operation Timeout
    // Operation performed during state transfer
} 
catch(Exception ex) 
{
    // Any generic exception like IllegalArgumentException or NullPointerException
}

// This is an async method
try 
{
    // Precondition: Cache is already connected 

    // Mention the name of the topic
    let topicName = "orderTopic";

    // Create the topic
    let topic = await ncache.getMessagingService().createTopic(topicName);
} 
catch (error) 
{
    // Handle any errors
}

Recommendation: 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 code below creates a topic with priority as highest so that the messages of the topic are published earliest.

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
}

try
{
    // Precondition: Cache is already connected 

    // Mention name of the topic
    String topicName = "orderTopic";

    // Create topic with priority
    Topic topic = cache.getMessagingService().createTopic(topicName, 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 IllegalArgumentException or NullPointerException
}

// This is an async method
try 
{
    // Precondition: Cache is already connected 

    // Mention name of the topic
    let topicName = "orderTopic";

    // Create topic with priority
    let topic = await ncache.getMessagingService().createTopic(topicName, ncache.TopicPriority.High);
} 
catch (error) 
{
    // Handle any errors
}

Get Topic

GetTopic method fetches an instance of the specified topic from the cache. If the topic exists, it is returned, otherwise null is returned.

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
}

try 
{
    // Precondition: Cache is already connected 

    // Mention name of the topic 
    String topicName = "orderTopic";

    // Get the topic
    Topic orderTopic = cache.getMessagingService().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) 
{
    // Exception can occur due to:
    // Connection Failures
    // Operation Timeout
    // Operation performed during state transfer
} 
catch (Exception ex) 
{
    // Generic exception like IllegalArgumentException or NullPointerException
}

// This is an async method
try 
{
    // Precondition: Cache is already connected 

    // Mention name of the topic 
    let topicName = "orderTopic";

    // Get the topic
    let orderTopic = await ncache.getMessagingService().getTopic(topicName);

    // Verify successful topic retrieval 
    if (!(orderTopic == null)) 
    {
        // orderTopic will be used for receiving and/or publishing messages
    }
    else 
    {
        // No topic exists
    }
} 
catch (error) 
{
    // Handle any errors
}

Recommendation: 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 unregisters 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 code sample deletes the existing topic orderTopic from pubSubCache 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
}

try 
{
    // Precondition: Cache is already connected 

    // Mention the topic to be deleted
    String topicName = "orderTopic";

    // Delete the topic
    cache.getMessagingService().deleteTopic(topicName);

    // Callback will be triggered if registered
} 
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 IllegalArgumentException or NullPointerException
}

try 
{
    // Precondition: Cache is already connected 

    // Mention the topic to be deleted
    let topicName = "orderTopic";

    // Delete the topic
    ncache.getMessagingService().deleteTopic(topicName);

    // Callback will be triggered if registered
} 
catch (error) 
{
    // Handle any errors
}

Note

This feature is only available from NCache 5.0 SP2 onwards.

You can also delete a topic asynchronously using the DeleteTopicAsync method. 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 code shows the asynchronous deletion of topic orderTopic.

.NET/.NET Core
Java

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
}

try 
{
    // Precondition: Cache is connected

    // Mention the topic to be deleted
    String topicName = "orderTopic";

    // Delete the topic asynchronously
    TimeScheduler.Task task = (TimeScheduler.Task) cache.getMessagingService().deleteTopicAsync(topicName);

    // Use tasks to perform further operations according to business logic
    // Callback will be triggered if registered
} 
catch (OperationFailedException exception) 
{
    if (exception.getErrorCode() == 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 exception) 
{
    // Any other generic exception like IllegalArgumentException or NullPointerException
}

Recommendation: 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 subscribers 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`	Topic level expiration, if message level expiration is not provided, this 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 sample application for Pub/Sub on GitHub.

Pub/Sub Topics Behavior and Properties

Topic Behavior

Temporary Client Disconnection

Topic Deletion Notification

Topic Priority

Note

Important

Note

Managing Topics

Create Topic

Create Topic with Priority

Note

Get Topic

Delete Topic

Note

Properties of ITopic Interface

Additional Resources

See Also