[tipatterson-dev]

Publishes CS API Part 3 Batch Resource Events for the observations of each datastream (/req/resource-events/batch-events, /req/mqtt/batch-resource-events). A datastream can create many observations each second, so the service sends one event with the number of new observations in each time window, not one event for each observation.

Closes #258. Part of #249.

Stacked on #264 (Resource Events, #254) (base feature/cs-api-mqtt/254-resource-events). Merge #261, #263 and #264 first.

Changes

• TopicMapper accepts the Batch Resource Event topic (no :data): {nodeId}/{systemPath}/datastreams/{id}/observations.
• New BatchWindow: counts the observations of one datastream and publishes at most one batch in each window.
  • After a quiet window, the first batch is published at once. The next ones are published at the end of the window.
  • A window with no new observations publishes nothing.
• New BatchEventPublisher:
  • Listens to the data channel of each datastream in the hub database, and of each datastream created later (system registry events).
  • type is org.ogc.api.csapi.observation.create. source is the API root. subject is the canonical observations URL ({apiRoot}/datastreams/{id}/observations) and resourceparent is the canonical datastream URL. Each event has a new id.
  • data is {"timerange": [first, last], "count": n}, with datacontenttype application/json.
  • Stops listening, and drops pending counts, when a datastream or its system is removed, and when the service stops.
• New config setting batchEventWindowMs (default 1000). 0 publishes one event each time a datastream gets observations. A negative value is rejected.
• CsApiMqttConnector (no code change): the batch topic uses the event topic checks of csapi-mqtt: Resource Events for systems, datastreams and control streams #264. A subscriber must be able to read the datastream, and client publications are rejected as not authorized.
• AsyncAPI description: the observationBatchEvents channel, a batch CloudEvents schema with data, and the configured window in the channel description.
• README: Batch Resource Events section, the window setting, status, and known limits.

Not included

• Batch events for commands, command status, command results and system events. Part 3 allows a server to publish batch events for some collections, if the service description lists them.
• Events for one observation (.../observations/{id}) are rejected. They need osh-core changes first.
• Wildcard subscriptions (consys-mqtt: accept wildcard subscriptions #255).

Known limits

• Events are published for each datastream that gets observations, also when no client subscribes.
• Observations not yet in an event when the service stops are not published.

Testing

• TestBatchWindow: first event at once, events after a quiet window, one flush at the end of the window, second window cycle, quiet window, time range order, rejected schedule, close, zero window.
• TestBatchEventPublisher: topic and CloudEvent JSON, a new id for each event, unknown datastream, datastream and system removal, stop. testEventsFromEventBus uses a real osh-core EventBus, so that the subscription delivers events.
• TestTopicMapper: batch topics for top-level systems and nested subsystems, and rejected topics.
• TestAsyncApiHandler: batch channel, event type and the configured window.
• On a test node:
  • 1 Hz drivers (counter, fake weather): one event each second, count 1, with the correct subject, resourceparent and timerange
  • datastream created over HTTP after start: 50 observations in 0.4 s gave count 1 at once, then count 49 at the end of the window; after a quiet gap, 20 observations gave 1, then 19
  • a datastream with no new observations: no events
  • subscribe to an unknown datastream or to .../observations/{id}: rejected; client publish to the batch topic: not authorized

Stack

1. csapi-mqtt: Part 3 topic scheme (new module, 0.1.0) #261 Part 3 topic scheme
2. csapi-mqtt: AsyncAPI service description #263 AsyncAPI service description
3. csapi-mqtt: Resource Events for systems, datastreams and control streams #264 Resource Events
4. This PR