Repository navigation
Node federation service, CS API MQTT Part 3 and proto/flatbuffers formats - #100
Closed
tipatterson-dev wants to merge 57 commits into
Closed
tipatterson-dev wants to merge 57 commits into
tipatterson-dev wants to merge 57 commits into
Conversation
- add cloud event types - add a resource event publisher
- Replace permission checks in ConSysApiMqttConnector with validator - Add unit tests for ConSysTopicValidator
- fix a parsing bug with # wildcards - simplify test readability for validator
Recognize an optional format token appended to a Resource Data Topic (e.g. .../observations:data/swe-json) per the OGC CS API Part 3 content negotiation rules, and apply it as the request/response format. ConSysTopicValidator gains DATA_SUFFIX, FORMAT_SUBTOPICS (json, swe-json, swe-binary, swe-csv, swe-proto, om-json, sml-json), isDataTopic, parseDataTopicFormat and stripDataSuffix; ConSysApiMqttConnector consumes them for both publish (request content type) and subscribe (response format), replacing the bare :data suffix handling.
Add the sensorhub-service-consys-proto module skeleton: the Connected Systems and SWE Common .proto definitions (including the swe_options field annotations carrying SWE metadata), the Gradle build wiring the protobuf plugin, and the IModuleProvider service registration. No Java yet — the codec implementation follows in the next commit.
Implement the swe+proto codec end to end. Outbound: SWE→proto schema generation (ProtoSchemaWriter, GeneratedSchemaCache, DataStreamSchemaCache) and observation/command encoding (ProtoObsEncoder, ProtoFormat bindings). Inbound: observation/command decode for ingestion (ProtoRecordDecoder, ObsBindingProto, CommandBindingProto) and client schema ingestion that rebuilds a SWE Common record from a wire FileDescriptorSet (ProtoSchemaReader, DataStream/CommandStream schema bindings). Includes the module service classes (Activator, ConSysApiProtoService, config, descriptor) and the test suite.
…print cache GeneratedSchemaCache now rebuilds the FileDescriptor/FileDescriptorSet on every get() instead of memoizing per stream behind a structural fingerprint. The hashing (fingerprint()/mix(), the entries map, Entry.fingerprint) is removed; the class name and get(BigId, DataComponent) signature are kept (streamId unused) so the cache can be reinstated without touching ProtoFormat or the bindings. The prior fingerprint-cache implementation is preserved on branch parked/schema-fingerprint-cache. - TestGeneratedSchemaCache: assert per-call rebuild + deterministic output instead of build-count memoization. - ObsBindingProto / CommandBindingProto / ProtoFormat: refresh now-stale "memoized/fingerprinted" comments.
Move bindings/codecs/schema machinery out of the flat proto package into subpackages by concern: schema/ (writer, reader, both caches), codec/ (encoder, decoder), and datastreams/ observations/ controlstreams/ commands/ for the resource bindings. Service/format/infra classes (ProtoFormat, ConSysApiProtoService, *Config, *Descriptor, Activator) stay at the root so the deployment-config FQN, the IModuleProvider SPI entry, and the OSGi Bundle-Activator keep resolving. Add the now-required cross-package imports and make the four binding name constants (OBS_PACKAGE/OBS_MESSAGE, CMD_PACKAGE/CMD_MESSAGE) public so ProtoFormat reads them across the boundary. Tests mirror the layout. No behavior change; full module suite green.
Increment 1 of DataArray support. Fixed-size arrays of scalars and of records/vectors now round-trip on the same-node encode->decode path: - ProtoObsEncoder: DataArray -> repeated field; K = getComponentCount() elements walked from the flat DataBlock (contiguous for fixed arrays). - ProtoRecordDecoder: repeated field -> flat block, with a wire-length vs schema-size check. - ProtoArrays.hasNonFlatLayout: shared guard so a fixed array hiding a DataChoice or variable-size sub-array (DataBlockList-backed, not flat- addressable) throws loudly in both stages instead of mis-reading atoms. - Extracted scalarValue()/setAtom() so the array path reuses the scalar type switch (no fifth copy). Not yet: variable-size arrays, Matrix, foreign ingest (ProtoSchemaReader still rejects repeated fields) - follow-up increments. Tests: TestProtoArrayRoundTrip. Full module suite green (37).
Increment 2. Variable-size arrays now round-trip; the element count travels as the repeated field's wire length (no separate size field needed). - ProtoObsEncoder.encode/encodeCommand: bind struct.setData(data) so a variable array's getComponentCount() reflects the observation's data; drop the variable-size rejection. - ProtoRecordDecoder: the choice pre-pass becomes prepass(), which also updateSize()s each variable array from msg.getRepeatedFieldCount() before createDataBlock() so the flat block is allocated correctly; drop the variable-size rejection. - Non-flat (DataBlockList) elements — a fixed array hiding a DataChoice or a variable-size sub-array — still throw via ProtoArrays.hasNonFlatLayout. Tests: variable-size round-trip (trailing scalar catches drift) + a direct unit test of the non-flat guard. Full module suite green.
Increment 3. A DataArray whose element is itself a DataArray (a Matrix row; proto has no repeated-of-repeated) now round-trips via a generated wrapper message holding the inner array as its single repeated field. - ProtoSchemaWriter: route a DataArray element to the nested-message path; buildMessage already wraps a non-structured component as a one-field message, so the wrapper falls out for free (recurses for N-D). - ProtoObsEncoder: treat a DataArray element as nested (encodeMessage wraps the inner array as the wrapper's field 1). - ProtoRecordDecoder: new branch decodes each wrapper's field-1 inner array. Fixed-size matrices only (their flat DataBlock is contiguous); variable-size inner arrays remain non-flat and are rejected by ProtoArrays.hasNonFlatLayout. Test: 2x3 array-of-array round-trip with a trailing scalar. Suite green.
Increment 4. ProtoSchemaReader (the foreign-schema ingest path) no longer rejects repeated fields — it rebuilds a variable-size DataArray: - buildArray(): element from the field's scalar/record type, or — for a Matrix-row wrapper message (one repeated field) — a nested DataArray (recursive, so N-D matrices ingest too). - The array is implicit variable-size (an elementCount carrying a Count whose value is unset → isVariableSize()), so ProtoRecordDecoder.prepass() sizes it from the repeated field's wire length. A foreign descriptor carries no count, so ingested arrays are necessarily variable-size. Test: encode with the original struct, decode with the reader-rebuilt one (the foreign-ingest property). Suite green.
Increment 5. ProtoFormat.isCompatible no longer returns true unconditionally (its standing TODO): it returns canEncode(recordStructure), so a datastream whose structure the codec can't handle won't list swe+proto in its formats and then 500 on encode. canEncode rejects: - structures with a non-flat array (a DataArray whose element hides a DataChoice or variable-size sub-array — DataBlockList-backed; the writer emits these but the flat-index codec can't walk them), via ProtoArrays.hasNonFlatLayout (made public); - anything the schema build throws on (Geometry, unmapped scalar types, invalid descriptor) — caught by attempting write()+resolve(). Test: TestProtoFormat. Suite green.
…rows)
Proves a capability the increment-3 message understated ("fixed matrices
only"). The codec's only constraint is that every array ELEMENT has a fixed
atom count, so a variable-size array is fine as the OUTER matrix dimension:
array[var n] of (array[3] double) — a variable count of fixed-width rows —
round-trips on a flat block. Only a variable INNER array (ragged rows) or an
array-of-choice is DataBlockList-backed and rejected.
Increment 6. A nested DataArray with BOTH dimensions variable (M x N, sized by count components, rectangular / "square per message") now round-trips — M and N may differ per message. Verified empirically that SWE lays these out as a flat DataBlockMixed (NOT a DataBlockList), so the flat-index codec handles them; the earlier guard rejected them on a wrong premise. - ProtoArrays: hasNonFlatLayout -> elementHasChoice. Rectangular arrays (fixed or variable, nested) stay flat and are no longer flagged; only a DataChoice in an array element is still rejected (the pre-pass doesn't apply per-element selections inside an array yet). - ProtoRecordDecoder.prepass: size the (shared) inner array dimension from row 0's wrapper message, so createDataBlock allocates the full M x N flat block. A ragged row then mismatches the per-row wireCount check and throws loudly. - ProtoFormat: hasNonFlatArray -> hasArrayWithChoiceElement. Encoder already handled it via setData + the nested-array path. Tests: M,N varying per message (2x3, 3x2, 1x5) on one schema; updated guard unit test.
…e names of ProtoEncoder and ProtoDecoder
…s opaque bytes fields
Adoption fell through to a name-only match whenever either schema signature was null, and ran before the schema-null guard, so a stream whose schema fetch failed could bind onto an unrelated commander datastream. Match on outputName/inputName (what the CS API actually keys streams on) and require both signatures present and equal.
The dedup lookup listed the commander system's datastreams (one request plus a schema fetch each) on every iteration of the mirror loop, making startup quadratic in stream count. Build the outputName/inputName index once per commander system and reuse it.
HashMap accepts a null key, so every system without a UID collapsed onto one entry and all their streams were mirrored into a single commander system. Skip them on both the index and lookup sides.
"procedureLink@link"/"deploymentLink@link" match no wire property, so the remote procedure and deployment links were POSTed to the commander unstripped. Use procedure@link/deployment@link, also clear system@link, and drop the "id":"default" placeholder in favour of removing the key.
…collisions allow setting root topic, minor bug fixes
…lizers The module had no tests at all. Wire JUnit in the same shape as consys-nats and cover APIHelper's root normalization, which until now was only checked by a throwaway class compiled against the build output.
…rvations Identifies an observation fleet-wide by (systemUid, outputName, phenomenonTime) so an echo is recognized regardless of which node's datastream id it comes back under. Reads the time from CS API JSON, and from the leading 8-byte double of a SWE binary frame so the video relay can be deduped too. Fails open throughout. Not yet wired in.
The observation pump holds only a Datastream, but identifying an observation across nodes needs the system UID — datastream ids are per-node. Set it wherever the owning system is known.
Listing the commander among the nodes is legal and useful — a commander carrying its own sensors then appears in the federated set like any other node. Relaying its data back is not: the observation is already stored where it would be sent, so the copy is republished and returns. Mirror such nodes as before, but skip their observation subscriptions, their pumps, and their command forwarding. Node identity compares the resolved address so the same host spelled two ways still matches.
Consult ObsEcho on both the JSON and binary relay paths, so a re-delivered observation is relayed at most once — covering re-mirroring after a restart, an MQTT reconnect replaying an overlap, and a cycle of brokers relaying to each other. Fingerprints are marked before the insert and rolled back if it fails, so a dropped insert stays retryable.
# Conflicts: # services/sensorhub-service-federation/src/main/java/org/sensorhub/impl/service/federation/MirroringMixin.java
# Conflicts: # services/sensorhub-service-federation/src/main/java/org/sensorhub/impl/service/federation/BrokerContext.java # services/sensorhub-service-federation/src/main/java/org/sensorhub/impl/service/federation/OSHDataBroker.java
tipatterson-dev
marked this pull request as ready for review
October 13, 2026 20:02
Author
|
I tested with fakeweather, ffmpeg, and the counter sim for commands. It all worked fine for me. |
|
added backfill on mirror to query remote history. Fixed dropped obs for post on startup sensors. |
rmw-workbotts
approved these changes
Oct 14, 2026
rmw-workbotts
left a comment
rmw-workbotts
left a comment
There was a problem hiding this comment.
Commits are stable for federation
drewbotts
approved these changes
Oct 15, 2026
|
This should be going to OpenSensorHub - Master rather than Botts Innovative Research - Master, correct? |
Author
|
Closing to direct merge into osh's addons repo |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Brings in the node federation service and the CS API pub/sub pieces it relies on. Cleaned-up version of
fed-service, brought current withmaster.Depends on Botts-Innovative-Research/osh-core#45 (custom obs/command formats). The proto and flatbuffers modules don't compile without it.
Changes
sensorhub-service-federation(new): mirrors systems, datastreams and control streams from remote nodes into a commander node over the CS API + MQTT, and routes commands from the commander to the remotes. Includes the loop guard (no self-federation, no re-forwarding already-forwarded observations).sensorhub-service-consys-mqtt: CS API Part 3 draft updates. Adds CloudEvents resource events, wildcard subscriptions with a topic validator, and:data/<format>topics. The format token is passed to the API asf=. Also drops an unused dependency on the video service, and an orphaned test for batching code that was never committed.sensorhub-service-mqtt-hivemq: wildcard subscriptions enabled in the default broker config (needed by the above).sensorhub-service-consys-proto/-flatbuffers(new):application/swe+protoandapplication/swe+flatbuffersencodings for observations, commands and their schemas, through core'sCustomObsFormat/CustomCommandFormat.The NATS binding (
sensorhub-service-consys-nats) is split out of this PR to be tested separately.Testing
swe-proto; commands sent to the commander over HTTP and MQTT ran on the remote counter sim driver.CI note
CI builds against the published osh-core package, so the proto/flatbuffers modules will fail there until a core build with osh-core#45 is published.