27 Performing Basic Topic Publish and Subscribe Operations

You can use the Coherence APIs to perform basic topic publish and subscribe operations.

Security Prerequisites

Classify subscriber filters and converters as remote executable code and include their captured data types in the application security metadata. Deploy those definitions on the required members before creating durable subscriber groups. Persisted definitions can be read again during recovery, so keep the necessary classes and metadata available for replay.

For the complete configuration and policy, see Classifying Remote Executable Objects, Limiting Resource Use and Protecting Persisted Data in Securing Oracle Coherence.

Publisher and Subscriber Completion

Await publisher publish futures to observe delivery failures, and close publishers and subscribers owned by the application. Subscriber.receive returns CompletableFuture<Subscriber.Element<V>>; process the element successfully before committing its position. Cancellation of a pending receive does not mean an already delivered element was processed or committed. Keep durable group identity and filter/converter configuration stable during restarts.

This chapter includes the following sections:

• Overview of Topics API
  The com.tangosol.net.NamedTopic<V> interface is the initial interface used by applications to publish and subscribe to the topic. It has factory methods to create a Publisher and a Subscriber. It also has methods to manage subscriber groups.
• Getting a Topic Instance
  To get a reference to a NamedTopic instance, applications can use the Session API or the CacheFactory API.
• Using NamedTopic Type Checking
  Coherence includes the ability to request strongly-typed NamedTopic instances when using either the Session or CacheFactory API by using explicit types.
• Publishing to a Topic
  An active Publisher, one that has not been closed, asynchronously sends a value to a topic using the send method and returns a CompletableFuture.
• Subscribing Directly to a Topic
  An active subscriber receives all values delivered to a topic.
• Releasing Subscriber Resources
  A subscriber can be closed either by explicitly invoking the Subscriber.close() operation or by creating the subscriber with a try-with-resources statement.
• Subscribing to a Subscriber Group
  A topic value either is retained until it is received by a subscriber group member or expires. By default, there is no expiry for topic values.
• Subscriber Groups for a NamedTopic
  Subscriber group can be created in two ways.
• Subscribing to Specific Channels
  Data in a Coherence topic is partitioned into channels.
• Destroying a Topic’s Subscriber Group
  A subscriber group life span is independent of its subscriber group members.
• Committing Messages
  To indicate a message has been processed and should not be redelivered, the subscriber must commit the message. A subscriber does not have to commit every message received. Committing a message automatically commits any messages received by the subscriber before the committed message.
• Managing the Resources Used by a Topic
  Both Publisher and Subscriber have resources associated with them and the resources must be closed when no longer needed.
• Understanding Topics Flow Control
  Given that topics Publisher.send and Subscriber.receive methods provide an asynchronous (non-blocking) way of submitting data exchange requests, there is a default flow control mechanism to efficiently manage these requests.
• Managing the Publisher Flow Control to Place Upper Bound on Topics Storage
  Use publisher flow control and topic storage limits when publishers can outpace subscribers.

Overview of Topics API

The com.tangosol.net.NamedTopic<V> interface is the initial interface used by applications to publish and subscribe to the topic. It has factory methods to create a Publisher and a Subscriber. It also has methods to manage subscriber groups.

The com.tangosol.net.Publisher<V> interface has an asynchronous publish method that publishes a value to a topic. The method returns CompletableFuture<Publisher.Status> to enable tracking when the send completes or if an exception is thrown. The returned Publisher.Status contains details of the published message

The com.tangosol.net.Subscriber<V> interface has a receive method that returns a CompleteFuture<Element<V>>. A Subscriber either subscribes to a subscriber group (using the Subscriber.inGroup (subscriberGroupName) option or subscribes directly to a topic as an anonymous subscriber.

• Publisher Creation Options
  
• Subscriber Creation Options
  

Publisher Creation Options

The following table lists the Publisher creation options to be used with NamedTopic.createPublisher(Publisher.Option... option).

Publisher Option	Description
OnFailure.Stop	Default. If an individual Publisher.send(V) invocation fails, stop any further publishing and close the Publisher.
OnFailure.Continue	If an individual Publisher.send(V) invocation fails, skip that value and continue to publish other values.
FailOnFull.enabled()	When the storage size of the unprocessed values on the topic exceeds a configured high-units value, the CompletableFuture returned from the Publisher.send(V) invocation should complete exceptionally. Overrides the default to block completing until the operation completes after space becomes available on the topic.
OrderBy.thread()	Default. Ensures that all values sent from the same thread are stored sequentially.
OrderBy.none()	Enforces no specific ordering between sent values allowing for the greatest level of parallelism.
OrderBy.id(int)	Ensures ordering of sent values across all threads which share the same ID.
OrderBy.value(ToIntFunction)	Computes the unit-of-order based on the application of this method on sent value.
OrderBy.roundRobin()	Publishes messages to each channel in the topic sequentially, looping back around to channel zero, after “channel count” messages have been published.

Subscriber Creation Options

The following table lists the Subscriber creation options to be used with NamedTopic.createSubscriber(Subscriber.Option... option).

Subscriber Option	Description
Name.of(String)	Specifies a subscriber group name to join as a member. For each value delivered to a subscriber group, only one member of the group receives the value.
Filtered.by(Filter)	Only Subscriber.receive() values matching this Filter. Only one Filter per subscription group.
Convert.using(Function)	Converts topic value using provided Function prior to Subscriber.receive(). Only one Converter per subscription group.
CompleteOnEmpty.enabled()	When no values left to Subscriber.receive(), returned CompletableFuture.get() returns null element. By default, returned CompletableFuture.get() blocks until next topic value is available to return.
ChannelOwnershipListeners.of(ChannelOwnershipListener)	Adds a ChannelOwnershipListener to receive events when the channels assigned to the subscriber change.

The com.tangosol.net.Subscriber<V> interface has a receive method that returns a CompleteFuture<Element<V>>. A Subscriber either subscribes to a subscriber group (using Subscriber.Name.of (subscriberGroupName) or directly subscribes to a topic.

Getting a Topic Instance

To get a reference to a NamedTopic instance, applications can use the Session API or the CacheFactory API.The Session API is the preferred approach for getting a topic because it offers a concise set of methods that allows for the use and injection of Coherence sessions that are non-static. In contrast, the CacheFactory API exposes many static methods that require knowledge of internal Coherence concepts and services. Lastly, the Session API allows for a more efficient lifecycle and integration with other frameworks, especially frameworks that use injection.

The following examples demonstrates creating a session using a default session provider and then get a reference to a NamedTopic instance using the Session.getTopic method. The name of the topic is included as a parameter.

import com.tangosol.net.topic.NamedTopic;
import com.tangosol.net.ValueTypeAssertion;

// session is the application’s started Session.
NamedTopic<String> topic = session.getTopic("topic",
        ValueTypeAssertion.withType(String.class));

Or

// Use this alternative when the integration owns a ConfigurableCacheFactory.
NamedTopic<String> topic = ccf.ensureTopic("topic",
        ValueTypeAssertion.withType(String.class));

To bypass generics type checking, omit optional <value-type> definition in topic-mapping configuration and omit ValueTypeAssertion option as a parameter for Session.getTopic or ConfigurableCacheFactory.ensureTopic methods.

Using NamedTopic Type Checking

Coherence includes the ability to request strongly-typed NamedTopic instances when using either the Session or CacheFactory API by using explicit types.By default, NamedTopic<Object> instances return. This is the most flexible mechanism to create and use NamedTopic instances; however, it is the responsibility of the application to ensure the expected value types when interacting with topic instances.

The following example stores objects of any type for use by an application:

// session is the application’s started Session.
NamedTopic<Object> topic = session.getTopic("MyTopic");

or,

// ccf is the integration’s configured ConfigurableCacheFactory.
NamedTopic<Object> topic = ccf.ensureTopic("MyTopic",
        ValueTypeAssertion.withRawTypes());

The getTopic method can be used to request a NamedTopic instance of a specific type, including if necessary, without type checking. The ValueTypeAssertion interface is used with the getTopic method to assert the correctness of the type of values used with a NamedTopic instance. The method can be used to assert that a topic should use raw types. For example:

NamedTopic<Object> topic = session.getTopic("MyTopic",
 ValueTypeAssertion.withRawTypes());

likewise when using CacheFactory,

NamedTopic<String> topic = ccf.ensureTopic("MyTopic",
        ValueTypeAssertion.withRawTypes());

For stronger type safety, you can create a topic and explicitly assert the value types to be used by the topic. For example, to create a topic and assert that values must be of type String, an application can use:

NamedTopic<String> topic = session.getTopic("MyTopic",
        ValueTypeAssertion.withType(String.class));

A NamedTopic instance is not required to adhere to the asserted type and may choose to disregard it; however, a warning message is emitted at compile-time. For example:

// Deliberately raw Java reference: runtime assertion still checks String values.
NamedTopic topic = session.getTopic("MyTopic",
        ValueTypeAssertion.withType(String.class));

Likewise, an application may choose to assert that raw types be used and a name cache instance may use specific types. However, in both cases, this could lead to errors if types are left unchecked. For example:

NamedTopic<String> topic = session.getTopic("MyTopic",
 ValueTypeAssertion.withRawTypes());

For the strongest type safety, specific types can also be declared as part of a topic definition in the cache configuration file. A runtime error occurs if an application attempts to use types that are different than those configured as part of the topic definition. The following examples configures a topic that only support values that are of type String.

<topic-mapping>
   <topic-name>MyTopic</topic-name>
   <scheme-name>distributed-topic</scheme-name>
   <value-type>String</value-type>
</topic-mapping>

Lastly, an application can choose to disable explicit type checking. If type checking is disabled, then the application is responsible for ensuring type safety.

NamedTopic<Object> topic = session.getTopic("MyTopic",
   ValueTypeAssertion.withoutTypeChecking());

Publishing to a Topic

An active Publisher, one that has not been closed, asynchronously sends a value to a topic using the send method and returns a CompletableFuture. For example, an asynchronous call is turned into a synchronous call by calling join on the CompletableFuture.

See CompletableFuture for the numerous options available for working with the completion of many outstanding asynchronous operations.

// topic is a NamedTopic<String> from the application’s started Session.
try (Publisher<String> publisher = topic.createPublisher()) {
    CompletableFuture<Publisher.Status> published = publisher.publish("someValue");
    Publisher.Status status = published.join();
    System.out.println(status);
}

• Requirements for Topic Values
  
• Awaiting Completion of Outstanding Publisher Sends
  
• Releasing Publisher Resources
  

Requirements for Topic Values

Topic values must be serializable (for example, java.io.Serializable or Coherence Portable Object Format serialization). See Using Portable Object Format.

Awaiting Completion of Outstanding Publisher Sends

If a publisher sends a number of values to a topic without waiting for each send to complete, a Publisher.flush().join() invocation synchronously waits for all pending Publisher.send() to complete. Note that even when one or more of the outstanding operations complete exceptionally, this invocation always completes successfully. Use the CompletableFuture methods for detecting completing exceptionally when it is necessary to detect when a value is not published successfully to the topic.

Releasing Publisher Resources

One releases publisher resources by explicitly invoking Publisher.close() or by creating the publisher with a try-with-resources statement as shown in Example 27-1. Publisher.close() is a blocking operation and does not return until the close is complete. Once the Publisher is closed, it is no longer active (see Publisher.isActive()), all future send operations on that Publisher fail with an IllegalStateException stating the Publisher is no longer active. However, all outstanding CompletableFutures for outstanding send operations occurring prior to the close call are allowed to complete. Just as Publisher.flush.join(), the close may complete successfully or exceptionally, and no exceptions are thrown from the close for any outstanding CompletableFutures that complete exceptionally.

Subscribing Directly to a Topic

An active subscriber receives all values delivered to a topic. A subscriber is active from its creation until it is closed.

// topic is the NamedTopic<String> obtained from the started application Session.
try (Subscriber<String> subscriber = topic.createSubscriber()) {
    CompletableFuture<Subscriber.Element<String>> pending = subscriber.receive();
    String received = pending.get().getValue();
    System.out.println(received);
}

Releasing Subscriber Resources

A subscriber can be closed either by explicitly invoking the Subscriber.close() operation or by creating the subscriber with a try-with-resources statement.For details, see Example 27-2.

Subscriber close is a blocking operation and does not return until the operation is completed. After close is called on a Subscriber, all future receive calls on the inactive Subscriber fail with an IllegalStateException. Most of the outstanding CompletableFuture(s) from receive calls will complete exceptionally, while some of them may complete.

Subscribing to a Subscriber Group

A topic value either is retained until it is received by a subscriber group member or expires. By default, there is no expiry for topic values.

See configuration <expiry-delay> subelement of <paged-topic-scheme>.

try (Subscriber<String> subscriber = topic.createSubscriber(
        Subscriber.inGroup("subscriberGroup"))) {
    Subscriber.Element<String> element = subscriber.receive().get();
    System.out.println(element.getValue());
    Subscriber.CommitResult committed = element.commit();
    System.out.println(committed);
}

Example setup and lifecycle

This method fragment assumes a started Session and NamedTopic<String> named topic. Process successfully before committing, inspect commit outcomes, and make external side effects idempotent where redelivery is possible. Closing this subscriber does not destroy its durable group.

Subscriber Groups for a NamedTopic

Subscriber group can be created in two ways.

There are:

• Statically configure durable subscriber group(s) on a <topic-mapping> element in the coherence-cache-config file. See Example 22-3.
• Dynamically create a subscriber group when a Subscriber is created with the Subscriber.Name.of(aSubscriberGroupName) option to join a subscriber group, and the subscriber group does not exist yet on the topic.

The method NamedTopic.getSubscriberGroups returns both configured and dynamically created subscriber groups.

NamedTopic<String> topic            = ….;
Set<String>        subscriberGroups = topic.getSubscriberGroups();

Subscribing to Specific Channels

Data in a Coherence topic is partitioned into channels. A full explanation of channels is given in the About Channels topic.

By default, the server manages allocations of channels to subscribers to ensure that every channel is owned by a subscriber in a subscriber group. Each time a new subscriber subscribes to a group or unsubscribes from a group the channels are reallocated.

Application code can manually select the channel or channels to which a subscriber subscribes. This allows applications more control over assigning subscribers to channels but with the caveat that it is then up to the application to ensure that the channels subscribed to correspond to the channels being published to; otherwise a subscriber may not receive any messages or alternatively the messages published by a publisher may never be received by a subscriber.

• If a subscriber is manually allocated channels, then it is allocated those channels and only those channels.
• If multiple subscribers in a subscriber group are manually allocated the same channels, then only one of them is assigned as the channel owner. The choice of subscriber is not deterministic. For example, if Subscriber-A and Subscriber-B both try to subscribe to channel 1, then only one owns the channel. If Subscriber-A is assigned ownership, then Subscriber-B receives no messages from channel 1. If Subscriber-A is closed, then channel 1 is reassigned to Subscriber-B.
• It is possible to mix subscribers with manual allocations and subscribers with auto-allocations. For example if Subscriber-A was manually subscribed to channels 0, 1 and 2 and Subscriber-B had no manual allocations then Subscriber-B would automatically be assigned all the remaining channels.

A subscriber can be manually allocated channels using the Subscriber.subscribeTo() option when it is created. For example, a subscriber could subscribe to channel 1 as shown in the following:

Subscriber<MyMessage> subscriber = topic.createSubscriber(Subscriber.subscribeTo(1));

A subscriber could subscribe to multiple channels 0, 1, 3 and 16 as shown in the following:

Subscriber<MyMessage> subscriber = topic.createSubscriber(Subscriber.subscribeTo(0, 1, 3, 16));

Note:

Any invalid channel numbers used in the Subscriber.subscribeTo() method is ignored. If all the channel identifiers passed to Subscriber.subscribeTo() are invalid, the subscriber is not subscribed to any channels.

Destroying a Topic’s Subscriber Group

A subscriber group life span is independent of its subscriber group members. The code fragment below stops accumulating values for a subscriber group and releases all unconsumed values.

All existing subscriber group members will be impacted by this operation and all outstanding asynchronous receive operations for the subscriber group members are cancelled.

NamedTopic<String> topic            = ….;
topic.destroySubscriberGroup(“subscriberGroup”);

Committing Messages

When using subscriber groups, messages are durable and will be delivered with an “at least once” guarantee. This means that a message will be redelivered if the subscriber disconnects for any reason or the subscriber’s channels are re-allocated to another subscriber. To indicate a message has been processed and should not be redelivered, the subscriber must commit the message. A subscriber does not have to commit every message received. Committing a message automatically commits any messages received by the subscriber before the committed message. Messages can be committed synchronously or asynchronously.

To synchronously commit a message, use the commit() method on the Element provided by the CompletableFuture that is returned by the subscriber’s receive() method.

For example, the code later in this topic creates a subscriber for the topic named “test-topic” in the subscriber group named “test”. A message is received, processed, and then committed. The returned Subscriber.CommitResult can be used to determine information about the commit operation, such as success or failure.

Subscriber<String> subscriber = session.createSubscriber("test-topic", Subscriber.inGroup("test"));
CompletableFuture<Subscriber.Element<String>> future = subscriber.receive();
Subscriber.Element<String> element = future.get();
String message = element.getValue();
// process message...
Subscriber.CommitResult result = element.commit();

To commit a message asynchronously use the element’s commitAsync() method.

For example, the code later in this topic calls the element’s commitAsync() method. The CompletableFuture returned is completed when the commit operation is completed.

CompletableFuture<Subscriber.CommitResult> future = element.commitAsync();

Messages can also be committed using just a Channel and a Position, using the Subscriber’s commit methods.

For example, in the following code, a Subscriber receives an element, and the message’s Channel and Position are obtained from the element. After processing, the message is committed using the Channel and Position.

Subscriber<String> subscriber = session.createSubscriber("test-topic", Subscriber.inGroup("test"));
CompletableFuture<Subscriber.Element<String>> future = subscriber.receive();
Subscriber.Element<String> element = future.get();

int channel = element.getChannel();
Position position = element.getPosition();

String message = element.getValue();

// process message...

Subscriber.CommitResult result = subscriber.commit(channel, position);

Managing the Resources Used by a Topic

Both Publisher and Subscriber have resources associated with them and the resources must be closed when no longer needed.

Both Publisher and Subscriber can be used with try-with-resource pattern. This usage pattern ensures close is called for both of these topic resources. See Publisher auto closed resource details Example 27-1. See Subscriber auto closed resource details Example 27-2.

Assume subscriber group durableSubscriber is statically configured for topic aTopic as shown in Example 22-3.

Example 27-1 try-with-resource for Publisher

NamedTopic<String> topic = session.getTopic("aTopic",
        ValueTypeAssertion.withType(String.class));
try (Publisher<String> publisher = topic.createPublisher()) {
    // Observe each completion; this also bounds outstanding application work.
    for (int i = 0; i < 1000; i++) {
        publisher.publish("value" + i).join();
    }
    publisher.flush().join();
}

Example 27-2 Subscriber processing with try-with-resource

try (Subscriber<String> subscriber = topic.createSubscriber(
        Subscriber.inGroup("durableSubscriber"), Subscriber.CompleteOnEmpty.enabled())) {
    while (true) {
        Subscriber.Element<String> element = subscriber.receive().join();
        if (element == null) {
            break;
        }
        System.out.println(element.getValue()); // replace with successful application processing
        Subscriber.CommitResult committed = element.commit();
        System.out.println(committed);
    }
}

NamedTopic.destroy() releases all the topics resources and destroys the instance. All outstanding Publishers and Subscribers are closed. Outstanding operations on Publishers and Subscribers are completed exceptionally. Subscriber groups are destroyed and all topic storage is released, including persistent storage.

Example setup and lifecycle

These method fragments reuse a started Session and use the topic API imports. CompleteOnEmpty ends this drain when no element is immediately available; concurrent publication can arrive later. The durable group retains its identity after this subscriber closes. Inspect commit outcomes and apply the application’s retry/idempotence policy; logging an outcome is not a transaction with external processing.

Understanding Topics Flow Control

Given that topics Publisher.send and Subscriber.receive methods provide an asynchronous (non-blocking) way of submitting data exchange requests, there is a default flow control mechanism to efficiently manage these requests.

When the automated flow control detects too many outstanding send and/or receive operations, running out of storage on a topic for outstanding values to be consumed, the outstanding CompletableFuture(s) for the async send and/or receive may be throttled to allow the system to catch up with all the outstanding requests. Both Publisher and Subscriber class provide a way for the application to get a FlowControl instance that allows it to opt-out of automatic flow control and manually govern the rate of the request flow. See the com.tangosol.net.FlowControl Javadoc for more details.

Managing the Publisher Flow Control to Place Upper Bound on Topics Storage

Use publisher flow control and topic storage limits when publishers can outpace subscribers.

By default, there are no storage constraints on the storage size for the values being retained by a topic. If a storage limit is configured for the topic as illustrated Example 27-3, there are several options available to restrict the topic’s storage usage.

After you configure a storage limit for a topic, publishers to that topic are throttled (blocked) by default when the next send would exceed the topic’s storage limit. Blocking the publishers gives subscribers time to process pending messages while limiting the storage used by the topic. After all subscribers and subscriber groups consume those values, publishers can resume sending.

There are Publisher creation time options to change the default behavior. The Publisher.FailOnFull.enabled() option indicates that a publisher send should not block, but rather complete exceptionally. The default Publisher option Publisher.OnFailure.Stop results in the Publisher auto closing when this exception occurs. This default preserves the order of published values from the publisher since the publisher is closed on first exceptional case that the value could not be delivered.

Example 27-3 FailOnFull publishing

NamedTopic<String> topic = session.getTopic("aTopic",
        ValueTypeAssertion.withType(String.class));
try (Publisher<String> publisher = topic.createPublisher(Publisher.FailOnFull.enabled())) {
    for (int i = 0; i < 1000; i++) {
        // Observe completion and propagate failure, including a full topic.
        publisher.publish("value" + i).join();
    }
}

If there is no need to preserve this order, the Publisher can be created with option Publisher.OnFailure.Continue and only the sends that occur when the topic is full fail.

To override the default blocking on sending to a full topic, one can use the following pattern to exempt a thread from flow control pause. This technique disables flow control and eliminates the storage limit on the topic for this NonBlocking thread.

Example 27-4 Exempt thread from flow control pause when reaching topic storage limit

NamedTopic<String> topic = session.getTopic("aTopic",
        ValueTypeAssertion.withType(String.class));
try (Publisher<String> publisher = topic.createPublisher()) {
    var pending = new ArrayList<CompletableFuture<Publisher.Status>>();
    for (int base = 0; base < 1000; base += 100) {
        pending.clear();
        try (var nonBlocking = new NonBlocking()) {
            for (int i = base; i < base + 100; i++) {
                pending.add(publisher.publish("value" + i));
            }
        }
        // Await outside the nonblocking scope and bound application backlog.
        CompletableFuture.allOf(pending.toArray(CompletableFuture[]::new)).join();
    }
}

Example setup and completion

These method fragments use a started Session, com.tangosol.net.topic.NamedTopic, Publisher, com.tangosol.net.ValueTypeAssertion, java.util.ArrayList, java.util.concurrent.CompletableFuture and com.oracle.coherence.common.base.NonBlocking. In the first example a failed publish is propagated as an exceptional completion; inspect the cause before identifying capacity as the problem. The second example bounds application submissions to batches of 100. Bypassing publisher backpressure without an application bound can exhaust memory; the scope alone does not bound topic storage.