Data Subscriptions: Subscription Consumption Examples
In this page:
Worker with a specified batch size
Client with full exception handling and processing retries
Subscription that ends when no documents left
Worker that processes dynamic objects
Subscription that works with lowest level API
Two subscription workers that are waiting for each other
Worker with a specified batch size
Here we create a worker, specifying the maximum batch size we want to receive.
var workerWBatch = store.Subscriptions.GetSubscriptionWorker<Order>(
new SubscriptionWorkerOptions(subscriptionName)
{
MaxDocsPerBatch = 20
});
_ = workerWBatch.Run(x => { /* custom logic */ });
Client with full exception handling and processing retries
Here we implement a client that treats exceptions thrown by worker, and retries creating the worker if an exception is recoverable.
while (true)
{
var options = new SubscriptionWorkerOptions(subscriptionName);
// here we configure that we allow a down time of up to 2 hours, and will wait for 2 minutes for reconnecting
options.MaxErroneousPeriod = TimeSpan.FromHours(2);
options.TimeToWaitBeforeConnectionRetry = TimeSpan.FromMinutes(2);
subscriptionWorker = store.Subscriptions.GetSubscriptionWorker<Order>(options);
try
{
// here we are able to be informed of any exception that happens during processing
subscriptionWorker.OnSubscriptionConnectionRetry += exception =>
{
Logger.Error("Error during subscription processing: " + subscriptionName, exception);
};
await subscriptionWorker.Run(async batch =>
{
foreach (var item in batch.Items)
{
// we want to force close the subscription processing in that case
// and let the external code decide what to do with that
if (item.Result.Company == "companies/832-A")
throw new UnsupportedCompanyException("Company Id can't be 'companies/832-A', you must fix this");
await ProcessOrder(item.Result);
}
}, cancellationToken);
// Run will complete normally if you have disposed the subscription
return;
}
catch (Exception e)
{
Logger.Error("Failure in subscription: " + subscriptionName, e);
if (e is DatabaseDoesNotExistException ||
e is SubscriptionDoesNotExistException ||
e is SubscriptionInvalidStateException ||
e is AuthorizationException)
throw; // not recoverable
if (e is SubscriptionClosedException)
// closed explicitly by admin, probably
return;
if (e is SubscriberErrorException se)
{
// for UnsupportedCompanyException type, we want to throw an exception, otherwise
// we continue processing
if (se.InnerException != null && se.InnerException is UnsupportedCompanyException)
{
throw;
}
continue;
}
// handle this depending on subscription
// open strategy (discussed later)
if (e is SubscriptionInUseException)
continue;
return;
}
finally
{
subscriptionWorker.Dispose();
}
}
Subscription that ends when no documents left
Here we create a subscription client that runs only up to the point there are no more new documents left to process.
This is useful for an ad-hoc single use processing that the user wants to be sure is performed completely.
var highValueOrdersWorker = store.Subscriptions.GetSubscriptionWorker<OrderAndCompany>(
new SubscriptionWorkerOptions(subsId)
{
// Here we ask the worker to stop when there are no documents left to send.
// Will throw SubscriptionClosedException when it finishes it's job
CloseWhenNoDocsLeft = true
});
try
{
await highValueOrdersWorker.Run(async batch =>
{
foreach (var item in batch.Items)
{
await SendThankYouNoteToEmployee(item.Result);
}
});
}
catch (SubscriptionClosedException)
{
// that's expected
}
Worker that processes dynamic objects
Here we create a worker that processes received data as dynamic objects.
await store.Subscriptions.CreateAsync(new SubscriptionCreationOptions<Order>()
{
Projection = order => new { DynanamicField_1 = "Company: " + order.Company + " Employee: " + order.Employee }
});
var subscriptionWorker = store.Subscriptions.GetSubscriptionWorker(subscriptionId);
_ = subscriptionWorker.Run(async batch =>
{
foreach (var item in batch.Items)
{
await RaiseNotification(item.Result.DynanamicField_1);
}
});
Subscription that works with lowest level API
Here we create a subscription that works with blittable document representation that can be useful in very high performance scenarios, but it may be dangerous due to the direct usage of unmanaged memory.
await store.Subscriptions.CreateAsync(
new SubscriptionCreationOptions<Order> {
Projection = x=>new
{
x.Employee
}
});
var subscriptionWorker = store.Subscriptions.GetSubscriptionWorker<BlittableJsonReaderObject>(subscriptionId);
_ = subscriptionWorker.Run(async batch =>
{
foreach (var item in batch.Items)
{
await RaiseNotification(item.Result["Employee"].ToString());
}
});
Two subscription workers that are waiting for each other
Here we create two workers:
* The main worker with the TakeOver
strategy that will take over the other one and will take the lead
* The secondary worker that will wait for the first one fail (due to machine failure etc.)
The main worker:
var worker = store.Subscriptions.GetSubscriptionWorker<Order>(new SubscriptionWorkerOptions(subscriptionName)
{
Strategy = SubscriptionOpeningStrategy.TakeOver
});
while (true)
{
try
{
await worker.Run(x =>
{
// your logic
});
}
catch (Exception)
{
// retry
}
}
The secondary worker:
var worker2 = store.Subscriptions.GetSubscriptionWorker<Order>(new SubscriptionWorkerOptions(subscriptionName)
{
Strategy = SubscriptionOpeningStrategy.WaitForFree
});
while (true)
{
try
{
await worker.Run(x =>
{
// your logic
});
}
catch (Exception)
{
// retry
}
}