Data Subscriptions: Consumption API Overview

In this page:

Subscription worker generation
SubscriptionWorkerOptions
Running subscription worker
SubscriptionBatch<T>
SubscriptionWorker<T>

Subscription worker generation

Subscription worker generation is accessible through the DocumentStore's Subscriptions Property, of type DocumentSubscriptions:

SubscriptionWorker<dynamic> GetSubscriptionWorker(
  string subscriptionName, string database = null);

SubscriptionWorker<dynamic> GetSubscriptionWorker(
  SubscriptionWorkerOptions options, string database = null);

SubscriptionWorker<T> GetSubscriptionWorker<T>(
  string subscriptionName, string database = null) where T : class;

SubscriptionWorker<T> GetSubscriptionWorker<T>(
  SubscriptionWorkerOptions options, string database = null) where T : class;

Parameters
subscriptionName	string	The subscription's name. This parameter appears in more simple overloads allowing to start processing without creating a SubscriptionCreationOptions instance, relying on the default values
options	SubscriptionWorkerOptions	Contains subscription worker, affecting the interaction of the specific worker with the subscription, but does not affect the subscription's definition
database	string	Name of the database to look for the data subscription. If null, the default database configured in DocumentStore will be used.

Return value
SubscriptionWorker	A created data subscription worker. When returned, the worker is Idle and it will start working only when the Run function is called.

SubscriptionWorkerOptions

The only mandatory parameter for SubscriptionWorkerOptions creation is the subscription's name.

Member	Type	Description
SubscriptionName	string	Returns the subscription name passed to the constructor. This name will be used by the server side to identify the subscription in question.
TimeToWaitBeforeConnectionRetry	TimeSpan	Time to wait before reconnecting, in the case of non-aborting failure during the subscription processing. Default: 5 seconds.
IgnoreSubscriberErrors	bool	If true, will not abort subscription processing if client code, passed to the Run function, throws an unhandled exception. Default: false.
Strategy	SubscriptionOpeningStrategy(enum)	Sets the way the server will treat current and/or other clients when they will try to connect. See Workers interplay. Default: OpenIfFree.
MaxDocsPerBatch	int	Maximum amount of documents that the server will try sending in a batch. If the server will not find "enough" documents, it won't wait and send the amount it found. Default: 4096.
CloseWhenNoDocsLeft	bool	If true, it performs an "ad-hoc" operation that processes all possible documents, until the server can't find any new documents to send. At that moment, the task returned by the Run function will fail and throw a SubscriptionClosedException exception. Default: false.

Running subscription worker

After receiving a subscription worker, the subscription worker is still not processing any documents. SubscriptionWorker's Run function allows you to start processing worker operations.
The Run function receives the client-side code as a delegate that will process the received batches:

Task Run(Action<SubscriptionBatch<T>> processDocuments,
  CancellationToken ct = default(CancellationToken));

Task Run(Func<SubscriptionBatch<T>, Task> processDocuments,
  CancellationToken ct = default(CancellationToken));

Parameters
processDocuments	Action<SubscriptionBatch<T>>	Delegate for sync batches processing
processDocuments	Func<SubscriptionBatch<T>, Task>	Delegate for async batches processing
ct	CancellationToken	Cancellation token used in order to halt the worker operation

Return value
Task	Task that is alive as long as the subscription worker is processing or tries processing. If the processing is aborted, the task exits with an exception

SubscriptionBatch<T>

Member	Type	Description
Items	List<SubscriptionBatch<T>.Item>	Batch's items list.
NumberOfItemsInBatch	int	Amount of items in the batch.

Method Signature	Return value	Description
OpenSession()	IDocumentSession	New document session, that tracks all items and included items of the current batch.
OpenAsyncSession()	IDocumentSession	New asynchronous document session, that tracks all items and included items of the current batch.

Subscription Worker Connectivity

As long as there is no exception, the worker will continue addressing the same server that the first batch was received from.
If the worker fails to reach that node, it will try to failover to another node from the session's topology list.
The node that the worker succeeded connecting to, will inform the worker which node is currently responsible for data subscriptions.

SubscriptionBatch<T>.Item

if T is BlittableJsonReaderObject, no deserialization will take place

Member	Type	Description
Result	T	Current batch item.
ExceptionMessage	string	Message of the exception thrown during current document processing in the server side.
Id	string	Current batch item's underlying document ID.
ChangeVector	string	Current batch item's underlying document change vector of the current document.
RawResult	BlittableJsonReaderObject	Current batch item before serialization to T.
RawMetadata	BlittableJsonReaderObject	Current batch item's underlying document metadata.
Metadata	IMetadataDictionary	Current batch item's underlying metadata values.

Usage of RawResult, RawMetadata, and Metadata values outside of the document processing delegate are not supported

SubscriptionWorker<T>

Methods

Method Signature	Return Type	Description
Dispose()	void	Aborts subscription worker operation ungracefully by waiting for the task returned by the Run function to finish running.
DisposeAsync()	Task	Async version of Dispose().
Dispose(bool waitForSubscriptionTask)	void	Aborts the subscription worker, but allows deciding whether to wait for the Run function task or not.
DisposeAsync(bool waitForSubscriptionTask)	void	Async version of DisposeAsync(bool waitForSubscriptionTask).
Run (multiple overloads)	Task	Starts the subscription worker work of processing batches, receiving the batch processing delegates (see above).

Events

Event	Type\Return type	Description
AfterAcknowledgment	AfterAcknowledgmentAction (event)	Event that is risen after each the server acknowledges batch processing progress.
OnSubscriptionConnectionRetry	Action<Exception> (event)	Event that is fired when the subscription worker tries to reconnect to the server after a failure. The event receives as a parameter the exception that interrupted the processing.
OnDisposed	Action<SubscriptionWorker<T>> (event)	Event that is fired after the subscription worker was disposed.

AfterAcknowledgmentAction

Parameters
batch	SubscriptionBatch<T>	The batch process which was acknowledged

Return value
Task	Task for which the worker will wait for the event processing to be finished (for async functions etc.)

Properties

Member	Type\Return type	Description
CurrentNodeTag	string	Returns current processing RavenDB server's node tag.
SubscriptionName	string	Returns processed subscription's name.