How to open data subscription?

Having a subscription identifier allows you to open it. Together with id you need to specify some details about the subscription connection:

Subscription<RavenJObject> Open(long id, SubscriptionConnectionOptions options, string database = null);

Subscription<T> Open<T>(long id, SubscriptionConnectionOptions options, string database = null) 

We have two method to open subscription. The first one is to deal with documents belonging to different collections - results are returned as RavenJObject objects then. The second one returns strongly typed subscription where retrieved documents are converted to a given type.

Documents are sent to a client in batches. SubscriptionConnectionOptions has BatchOptions property where you can specify:

• MaxDocCount - max number of docs that can be sent in a single batch (default: 4096),
• MaxSize - max total batch size in bytes (default: null - no size limit),
• AcknowledgmentTimeout - max time within the subscription needs to confirm that the batch has been successfully processed (default: 1 minute).

Additionally connection options have the following settings:

• IgnoreSubscribersErrors - determines if subscription should ignore errors thrown by subscription handlers (default: false),
• ClientAliveNotificationInterval - specifies how often the subscription sends heart beats to the server (server keeps the subscription open until a connected client sends these alive notifications - two undelivered notifications would let an another client to connect, default: 2 minutes),

• Strategy - the enum that represents subscription opening strategy. There are four strategies available:

  • OpenIfFree - the client will successfully open a subscription only if there isn't any other currently connected client. Otherwise it will end up with SubscriptionInUseException,
  • TakeOver - the connecting client will successfully open a subscription even if there is another active subscription's consumer. If the new client takes over the subscription then the existing one will get rejected. The subscription will always be processed by the last connected client.
  • ForceAndKeep - the client opening a subscription with forced strategy set will always get it and keep it open until another client with the same strategy gets connected.
  • WaitForFree - if the client currently cannot open the subscription because it is used by another client then it will subscribe Changes API to be notified about subscription status changes. Every time SubscriptionReleased notification arrives, it will repeat an attempt to open the subscription. After it succeeds in opening, it will process docs as usual.

Example I

var orders = store.Subscriptions.Open<Order>(id, new SubscriptionConnectionOptions()
{
	BatchOptions = new SubscriptionBatchOptions()
	{
		MaxDocCount = 16*1024,
		MaxSize = 4*1024*1024,
		AcknowledgmentTimeout = TimeSpan.FromMinutes(3)
	},
	IgnoreSubscribersErrors = false,
	ClientAliveNotificationInterval = TimeSpan.FromSeconds(30)
});

Processing documents

The result of opening subscription is Subscription<T> or Subscription<RavenJObject> instance. It implements IObservable interface, so it means that you can just utilize Reactive Extensions and subscribe to the incoming documents in order to process them. Also you will continue to get them, even for items that were added after you opened the subscription, because under the hood the Changes API is used to get notifications about any document updates.

Example II

orders.Subscribe(x =>
{
	GenerateInvoice(x);
});

orders.Subscribe(x =>
{
	if(x.RequireAt > DateTime.Now)
		SendReminder(x.Employee, x.Id);
});

Example III

You may want to dynamically manage subscription handlers. The returned subscriber object is type of IDisposable, in order to detach it from subscription just call Dispose on it:

var subscriber = orders.Subscribe(x => { });

subscriber.Dispose();