Industream.EventHub.Models 1.0.0-preview.9

Industream.EventHub.Models

The EventHub event contract in .NET: builds CloudEvents that follow the EventHub event model (Docs/Event-Model.md in the EventHub repository). It has no NATS dependency: the EventHub clients and the backend both use it.

dotnet add package Industream.EventHub.Models

Building an event

using Industream.EventHub.Models;

var thermalAlarm = new EventModel
{
    Application = "raceway",            // type industream.raceway.thermalalarm
    Name = "thermalalarm",
    Source = "/industream/flowmaker/raceway-flow",   // optional, default: /industream/raceway
    Severity = EventSeverity.Alarm,
    Category = EventCategory.Process,
    Subject = "tuyere/12",
    Attributes = { ["tuyere"] = 12 },
    Data = new { code = "THERMAL_OUT_OF_RANGE", flameTemperature = 2412 }
};

var cloudEvent = EventFactory.Create(thermalAlarm);

EventFactory.Create fills in:

• id (UUID v7), specversion 1.0, datacontenttype application/json;
• type (industream.<Application>.<Name>) and source (Source, or /industream/<Application> by default);
• time in UTC, truncated to the millisecond (now when the model has no Time);
• the Severity extension (severitytext and severitynumber), category, and project (Application unless the model sets Project);
• rootevent when the model sets RootEventId (secondary events);
• keephistory: false when the model sets KeepInHistory = false: the event goes through EventHub as usual, but the backend leaves it out of the database history (an ordinary event does not carry the extension);
• lasting: true when the model sets Lasting = true: the event lasts until it is ended, like an alarm (an instantaneous event does not carry the extension);
• the publisher attributes, keeping their type (string, int, bool, DateTimeOffset, Uri).

EventJson.Encode(cloudEvent) writes the CloudEvent as structured JSON, as it is published; data given as a .NET object is written in camelCase (FlameTemperature becomes flameTemperature), like the rest of the contract.

Invalid models

An invalid model is a programming error: EventFactory.Create throws an ArgumentException whose message says what is wrong: empty application, event or project name, unknown severity or category, empty subject, source that is not a URI reference, invalid or reserved attribute name, unsupported attribute value.

Decoding a received event

var receivedEvent = EventDecoder.Decode(messageBody);   // throws EventDecodeException if invalid

ReceivedEvent gives the CloudEvents fields, the severity, the category, the project, RootEventId, KeepInHistory (true when keephistory is absent), Lasting (false when lasting is absent), the publisher attributes and data (as JSON), plus the received CloudEvent and, when a client read it from the stream, its StreamSequence and NatsSubject. The severity is read by OpenTelemetry range (severitynumber 13 to 16 is Warn, TRACE 1 to 4 is read as Debug), or from severitytext when there is no number: an event published without the EventHub client may use any number of a level. Publisher attributes keep numbers and booleans typed; timestamps and URIs, plain strings in JSON, come back as strings.

A message is rejected when it is not a valid CloudEvent in structured JSON (a blank id, type or source included), has no time, severity, category or project, has a severitynumber that is not an integer, a keephistory or a lasting that is not a boolean, a rootevent that is not an event id in its standard form, or an extension value outside the CloudEvents type system (an object, a list, a number that is not a 32-bit integer). Data comes back as JSON; data of another content type (text…) comes back as a JSON string.

Secondary events

When the backend acknowledges or ends an event, it publishes a secondary event: its RootEventId is the id of the main event, its type industream.eventhub.event.<change>. A subscriber receives it like any other event, with Modification set: what changed and the full state of the main event, read from its data. Modification is null for every other event.

if (receivedEvent.Modification is { } modification)
{
    if (modification.Change == EventChange.Acknowledged) { … }
    var end = modification.State.End;   // the full, cumulative state
}

Change is a string: EventChange.Acknowledged, EventChange.Ended, or a change added later, kept as received. A secondary event whose data is not { change, event, state } with the change of its type and a valid state is rejected like any other invalid event. The main event stays in Data, as it was published.

NATS subject

The publisher chooses the NATS subject of its events; it only has to start with eventhub., the root the EventHub stream captures (why: Docs/Event-Model.md §7.2).

EventSubject.Validate("eventhub.raceway.process.alarm.thermalalarm");   // throws ArgumentException if invalid

A subject is rejected when it does not start with eventhub., has an empty token, contains whitespace or contains a wildcard (*, >: subscriptions only). EventSubject.All (eventhub.>) covers every EventHub subject.

A subscription filter accepts wildcards as whole tokens, > only last:

EventSubject.ValidateFilter("eventhub.*.*.alarm.>");

State of a modified event

Only the EventHub backend modifies an event, through its commands (sent with EventModifier of Industream.EventHub.Client). The state it gives the event (Docs/Event-Model.md §8.3) is part of the contract, shared by the backend and the client:

• ModifiedEvent: the id of the main event and its EventState, as a command replies it;
• EventState: the Acknowledgement (Actor, Time, optional Comment) and the EventEnd (Actor, Time), each null until it happens; EventState.Initial for an event never modified;
• Actor: its ActorType (User, Application, Flow), its Id and its ActorProof (HubToken, NatsPermission, Declared).

EventStateDecoder.Decode(JsonElement) reads a state as the backend writes it (the state of a bucket value, of a secondary event or of a command reply): kebab-case values ("hub-token"), RFC 3339 times, a field it does not know ignored. An invalid state throws an EventDecodeException that names the wrong field ('acknowledgement.actor.type' is unknown: 'robot').

LatestEventVersionDecoder.Decode(JsonElement) reads a value of the EventContract.LatestVersionBucket bucket (eventhub-latest): Event, the main event decoded as EventDecoder does with its NatsSubject, NatsSubject and State. The pending secondary event the backend may keep there is ignored.

ModificationJsonOptions.Value writes and reads them as the contract does: camelCase fields, kebab-case values ("hub-token"), times in UTC to the millisecond. EventContract.CommandSubjectRoot (eventhubcmd) and EventContract.DeclaredActorHeader (Eventhub-Actor) name the subjects and the header of the commands.

Build and test

From the repository root:

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

The unit tests replay the contract cases shared by every client and check the models against the shared schemas (see Shared/README.md).