Industream.EventHub.Client 1.0.0-preview.9

Industream.EventHub.Client

.NET client for EventHub: publishes and subscribes to events stored in NATS JetStream, sends the commands that modify them to the EventHub backend, and reads their history from it. It builds on the event contract of Industream.EventHub.Models and on NATS.Net.

dotnet add package Industream.EventHub.Client

Publishing an event

using Industream.EventHub.Client.Publishing;
using Industream.EventHub.Models;
using NATS.Client.Core;

await using var connection = new NatsConnection(new NatsOpts { Url = "nats://localhost:4222" });
var publisher = new EventPublisher(connection);

var cloudEvent = EventFactory.Create(thermalAlarm);
var published = await publisher.PublishAsync(cloudEvent, "eventhub.raceway.process.alarm.thermalalarm");
// published.EventId, published.StreamSequence, published.Duplicate

PublishAsync takes a CloudEvent built with EventFactory, then:

• checks the subject (EventSubject.Validate);
• publishes the CloudEvent as structured JSON, with the Content-Type: application/cloudevents+json header and the Nats-Msg-Id header set to the event id: JetStream ignores the same event published again within its duplicate window, and Duplicate is then true;
• waits for the JetStream acknowledgement. When no stream stores the subject, or JetStream refuses the event, it throws an EventPublishException. Connection errors (a closed connection…) come as the NATS.Net exceptions.

To publish an event again after an error, publish the same CloudEvent: it keeps its id, so JetStream stores it only once. Building it again from the EventModel would give it a new id, and the event would be stored twice.

Subscribing

using Industream.EventHub.Client.Subscribing;

var subscriber = new EventSubscriber(connection);

await subscriber.ConsumeAsync(
    consumerName: "recipemaker-analyses",
    filter: "eventhub.materialcatalog.>",
    handler: async (receivedEvent, cancellationToken) =>
    {
        // react to receivedEvent (decoded by EventDecoder, see Industream.EventHub.Models)
    },
    options: new ConsumeOptions
    {
        OnHandlerFailure = failure => logger.LogError(failure.Exception, "Event {Id} failed", failure.Event.Id)
    },
    cancellationToken: stoppingToken);

The consumer name may not be empty nor contain whitespace, ., *, >, / or \, and MaxDeliveries must be at least 1: ConsumeAsync throws an ArgumentException otherwise. ConsumeAsync runs until its cancellation token is cancelled, even before it has started: it then ends quietly, hands no more event (those already received come back next time), and a handler that stops on the cancellation ends it quietly too. It uses a durable consumer: NATS remembers where the consumer named consumerName stopped, and the next ConsumeAsync with the same name resumes there, so an application that was stopped receives the events it missed.

• Every received event carries the NATS subject it was published on, NatsSubject, and its position in the stream, StreamSequence.
• An event is acknowledged once the handler succeeds. When the handler throws, the failure is passed to OnHandlerFailure and the event comes back a second later, up to MaxDeliveries deliveries (5 by default). After the last one (IsLastAttempt), the consumer gives the event up; it stays in the stream.
• Events are handed one at a time, and delivered at least once: an event comes back when the application stops before acknowledging it, or when its handler runs longer than the acknowledgement wait of NATS (30 s). A handler must accept an event it has already handled (the event id tells).
• A message that is not an EventHub event is passed to OnInvalidMessage (an InvalidMessage: its NATS subject, stream sequence, body and error) and set aside for good, so that it does not come back in a loop.
• Start only matters the first time the consumer is created: SubscriptionStart.NewEvents (default) receives the events published from then on, SubscriptionStart.AllStoredEvents also receives the events still stored. A later ConsumeAsync with the same name keeps the position of the consumer, and updates its filter and MaxDeliveries.

To read the past (replay), ReadFromAsync returns, in order, the events stored from a start, then stops once caught up. FollowFromAsync reads the same way, then goes on with the new events as they are published, until its cancellation token is cancelled:

await foreach (var receivedEvent in subscriber.ReadFromAsync(ReadStart.At(since), "eventhub.raceway.>"))
{
}

var start = ReadStart.After(lastSequence);
var events = subscriber.FollowFromAsync(start, "eventhub.raceway.>", cancellationToken: stoppingToken);
await foreach (var receivedEvent in events)
{
    lastSequence = receivedEvent.StreamSequence!.Value;
}

• The start is a date (ReadStart.At) or the stream sequence of an event already read (ReadStart.After): every received event carries its position in the stream, StreamSequence. NATS cannot look an event up by its id.
• Unlike ConsumeAsync, they leave nothing behind in NATS: to resume later, the application keeps the StreamSequence of the last event it handled.
• A read survives the loss of its NATS consumer: NATS deletes it when the application takes more than a minute on the events fetched ahead (up to 256), or on a restart. The read then goes on right after the last event read, each event once and in order.
• They skip messages that are not EventHub events, after passing them to ReadOptions.OnInvalidMessage.
• Every consumer they create has a name, sent in the subject of the request: the NATS rights of a deployment can then forbid a client to take over a consumer that is not its own (Deployment/nats/nats-secured.conf).
• When ReadStart.After asks for events already removed by the stream retention, ReadOptions.OnMissedEvents gets the requested and the oldest stored sequences before the read goes on from the oldest stored event. A start at a date cannot tell: NATS does not keep the time of the removed events.

All three check the filter with EventSubject.ValidateFilter: wildcards as whole tokens, > only last.

Modifying an event

Only the EventHub backend modifies an event: EventModifier sends it the commands (NATS Request-Reply, see the backend README) and returns the event with its full new state: a ModifiedEvent of Industream.EventHub.Models, which also holds EventState, Acknowledgement, EventEnd and Actor.

using Industream.EventHub.Client.Modifying;

var modifier = new EventModifier(connection);
var actor = CommandActor.Application("raceway");

var acknowledged = await modifier.AcknowledgeAsync(receivedEvent.Id, actor, comment: "Coal injection reduced.");
var ended = await modifier.EndAsync(receivedEvent.Id, actor);
// ended.State.Acknowledgement, ended.State.End: Actor (Type, Id, Proof), Time, Comment

• AcknowledgeAsync records that the actor has taken the event into account: any event, once, with an optional comment. EndAsync records that an event that lasts (Lasting, an alarm) is over: once. They are independent: an acknowledged alarm stays active until it is ended.

• The actor decides the subject and the headers of the command, and the backend records it with the modification:

  Actor	Who	Subject
  CommandActor.User(hubToken)	A user, proven by their Industream Hub access token (Authorization header)	eventhubcmd.<command>.user
  CommandActor.DeclaredUser(userName)	A user named without proof (Eventhub-Actor header), for a backend whose authentication is off	eventhubcmd.<command>.user
  CommandActor.Application(applicationName)	An application, proven by the NATS permissions of its NATS user	eventhubcmd.<command>.application.<name>
  CommandActor.Flow(flowName)	A FlowMaker flow (Eventhub-Actor header), sent by the FlowMaker box	eventhubcmd.<command>.flow

  An empty token, a blank name, a name with a control character, or an application name that is not a single subject token (whitespace, ., *, >) throws an ArgumentException, as does an empty event id.

• When the backend refuses the command, an EventModificationException carries its Status (the HTTP meaning: 400 instantaneous event or invalid request, 401 Hub token missing or invalid, 404 unknown event, 409 already acknowledged or ended), Title and Detail, for the Reason BackendErrorReason.Refused. When no backend answers (NoBackend), or not within 10 seconds (Timeout), or its reply cannot be read (UnreadableReply), Status is null. After a timeout, the modification may have been recorded: sent again, it gets 409.

Reading the history

The stream keeps the events for its retention only (30 days by default), and NATS filters them by subject and date only. The backend keeps every event in its database: EventHistoryReader reads them through its read queries (NATS Request-Reply, see the backend README), with the state of each event.

using Industream.EventHub.Client.Querying;

var reader = new EventHistoryReader(connection);
var filter = new EventHistoryFilter
{
    From = DateTimeOffset.UtcNow.AddDays(-7),
    Projects = ["raceway"],
    Severities = [EventSeverity.Alarm],
    InProgress = true,
    Attributes = new Dictionary<string, object> { ["tuyere"] = 12 },
};

var page = await reader.ReadHistoryAsync(filter, offset: 0, limit: 50);
// page.Total; page.Events: LatestEventVersion (Event, NatsSubject, State), most recent first

• Every filter is optional: the period [From, To[, projects, severities, categories, Acknowledged, InProgress (events that last and are not ended), publisher attributes (equality as text: 12 finds 12 and "12") and Text (contained in the type, the CloudEvents subject or an attribute value, whatever the case).
• The caller chooses the page: offset and limit (50 by default, 500 at most). Total counts the events of the filter, all pages included.
• Only the main events are listed, each with the full state its modifications gave it (EventState, no acknowledgement and no end when never modified).
• When the backend refuses the query (400: an invalid filter or page), an EventQueryException carries its Status, Title and Detail, for the Reason BackendErrorReason.Refused; when no backend answers (NoBackend), or not within 10 seconds (Timeout), or its reply cannot be read (UnreadableReply), Status is null. The queries are open: every NATS user that reads the stream may send them, without Hub token.

Statistics

ReadStatisticsAsync counts the main events of a history filter per group: per project, per severity, per time bucket, per publisher attribute… One call gives one grouping; a screen that shows several (the Overview of the UI) makes several calls.

var to = DateTimeOffset.UtcNow;
var filter = new EventHistoryFilter { From = to.AddHours(-24), To = to, Projects = ["raceway"] };

var perHour = await reader.ReadStatisticsAsync(
    filter,
    [StatisticsDimension.Time, StatisticsDimension.Severity],
    bucketSize: TimeSpan.FromHours(1),
    orderBy: StatisticsOrder.Key);
// perHour.Groups: Time, Severity and Count of each group; Total, GroupCount

var openAlarmsPerTuyere = await reader.ReadStatisticsAsync(
    new EventHistoryFilter { Severities = [EventSeverity.Alarm], InProgress = true },
    [StatisticsDimension.Attribute("tuyere")]);
// openAlarmsPerTuyere.Groups[0].Attributes["tuyere"]: 12, as published

• The dimensions: Project, Severity, Category, Type, Acknowledged, InProgress, Time and Attribute(name), each at most once. A group gives the keys of its dimensions only, the others are null.
• Time needs bucketSize, in whole seconds, and the From of the filter: the buckets start at From, every bucketSize, and Time is the start of a bucket. The buckets without events are not given: the caller fills them with 0.
• Attributes keeps the published values: 12 and "12" are two groups, a null value groups the events without the attribute.
• orderBy: Count (by default) gives the most numerous groups first, Key orders them by their keys, in the order of the dimensions. limit (1000 by default and at most) cuts the groups; Total and GroupCount are computed before it.
• Same errors as the history: an EventQueryException.

Build and test

From the repository root:

dotnet build Industream.EventHub.slnx
dotnet test Industream.EventHub.slnx

The end-to-end tests (Industream.EventHub.Client.E2ETests) start a throwaway NATS server with Testcontainers: Docker must be running.