From 3a7a458f69824029fab458f1fdbe4a5a4da71394 Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 12 Oct 2026 16:26:27 -0500 Subject: [PATCH 01/13] [CSAPI] Commit the write database before acknowledging writes Handlers holding the read-only federated store skipped the commit. Fixes #18 --- .../consys/resource/BaseResourceHandler.java | 24 ++++++++++++++----- 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/resource/BaseResourceHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/resource/BaseResourceHandler.java index fdc5de940..d26810d4d 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/resource/BaseResourceHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/resource/BaseResourceHandler.java @@ -63,6 +63,7 @@ public abstract class BaseResourceHandler Date: Mon, 12 Oct 2026 16:29:09 -0500 Subject: [PATCH 02/13] [H2] Don't reuse series IDs after removing records by key An emptied series stayed registered, so a new series could get its ID and two datastreams shared records. Applies to the obs and command stores. Refs opensensorhub/osh-core#340 --- .../datastore/AbstractTestCommandStore.java | 33 +++++++++++++++++++ .../impl/datastore/AbstractTestObsStore.java | 33 +++++++++++++++++++ .../impl/datastore/h2/MVCommandStoreImpl.java | 9 ++++- .../impl/datastore/h2/MVObsStoreImpl.java | 9 ++++- 4 files changed, 82 insertions(+), 2 deletions(-) diff --git a/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestCommandStore.java b/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestCommandStore.java index 407e75c65..8d087d567 100644 --- a/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestCommandStore.java +++ b/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestCommandStore.java @@ -483,6 +483,39 @@ public void testSelectCommandsByMultipleIDs() throws Exception } + @Test + public void testRemoveByKeyDoesNotMixCommandStreams() throws Exception + { + var cs1ID = addSimpleCommandStream(bigId(10), "out1").getInternalID(); + var cs2ID = addSimpleCommandStream(bigId(20), "out2").getInternalID(); + + // empty the newest command series by removing its only command + var removed = addCommands(cs1ID, Instant.parse("2020-01-01T00:00:00Z"), 1, 1000); + for (var k: removed.keySet()) + { + cmdStore.remove(k); + allCommands.remove(k); + } + forceReadBackFromStorage(); + + // new series for command stream 2, then more commands in the existing series of stream 1 + var cs2Cmds = addCommands(cs2ID, Instant.parse("2021-01-01T00:00:00Z"), 10, 1000); + forceReadBackFromStorage(); + var cs1Cmds = addCommands(cs1ID, Instant.parse("2022-01-01T00:00:00Z"), 3, 1000); + forceReadBackFromStorage(); + + var filter = new CommandFilter.Builder() + .withCommandStreams(cs1ID) + .build(); + checkSelectedEntries(cmdStore.selectEntries(filter), cs1Cmds, filter); + + filter = new CommandFilter.Builder() + .withCommandStreams(cs2ID) + .build(); + checkSelectedEntries(cmdStore.selectEntries(filter), cs2Cmds, filter); + } + + @Test public void testSelectCommandsByDataStreamIDAndTime() throws Exception { diff --git a/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestObsStore.java b/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestObsStore.java index 8cca79fc4..92bc8b1e1 100644 --- a/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestObsStore.java +++ b/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestObsStore.java @@ -383,6 +383,39 @@ public void testAddAndRemoveByKey() throws Exception } + @Test + public void testRemoveByKeyDoesNotMixDataStreams() throws Exception + { + var ds1ID = addSimpleDataStream(bigId(10), "out1"); + var ds2ID = addSimpleDataStream(bigId(20), "out2"); + + // empty the newest obs series by removing its only obs + var removed = addSimpleObsWithoutResultTime(ds1ID, BigId.NONE, Instant.parse("2020-01-01T00:00:00Z"), 1); + for (var k: removed.keySet()) + { + obsStore.remove(k); + allObs.remove(k); + } + forceReadBackFromStorage(); + + // new series for datastream 2, then more obs in the existing series of datastream 1 + var ds2Obs = addSimpleObsWithoutResultTime(ds2ID, BigId.NONE, Instant.parse("2021-01-01T00:00:00Z"), 10); + forceReadBackFromStorage(); + var ds1Obs = addSimpleObsWithoutResultTime(ds1ID, BigId.NONE, Instant.parse("2022-01-01T00:00:00Z"), 3); + forceReadBackFromStorage(); + + var filter = new ObsFilter.Builder() + .withDataStreams(ds1ID) + .build(); + checkSelectedEntries(obsStore.selectEntries(filter), ds1Obs, filter); + + filter = new ObsFilter.Builder() + .withDataStreams(ds2ID) + .build(); + checkSelectedEntries(obsStore.selectEntries(filter), ds2Obs, filter); + } + + @Test public void testAddAndRemoveEntireDatastream() throws Exception { diff --git a/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVCommandStoreImpl.java b/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVCommandStoreImpl.java index 1435fe8c7..2d7c38345 100644 --- a/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVCommandStoreImpl.java +++ b/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVCommandStoreImpl.java @@ -66,6 +66,7 @@ public class MVCommandStoreImpl implements ICommandStore protected MVCommandStatusStoreImpl cmdStatusStore; protected MVBTreeMap cmdRecordsIndex; protected MVBTreeMap cmdSeriesMainIndex; + protected long lastSeriesID; protected int idScope; protected IFoiStore foiStore; @@ -143,6 +144,12 @@ private MVCommandStoreImpl init(MVStore mvStore, int idScope, IdProviderType dsI .keyType(new MVCommandSeriesKeyByCommandStreamDataType()) .valueType(new MVTimeSeriesInfoDataType())); + // next series ID must follow every series still registered, including empty ones, + // otherwise a new series can reuse the ID of an existing one + this.lastSeriesID = Math.max( + cmdSeriesMainIndex.values().stream().mapToLong(s -> s.id).max().orElse(0), + cmdRecordsIndex.isEmpty() ? 0 : cmdRecordsIndex.lastKey().seriesID); + return this; } @@ -621,7 +628,7 @@ public synchronized BigId add(ICommandData cmd) MVTimeSeriesInfo series = cmdSeriesMainIndex.computeIfAbsent(seriesKey, k -> { return new MVTimeSeriesInfo( - cmdRecordsIndex.isEmpty() ? 1 : cmdRecordsIndex.lastKey().seriesID + 1); + ++lastSeriesID); }); // add to main command index diff --git a/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVObsStoreImpl.java b/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVObsStoreImpl.java index 6b9b2aa8e..557b65d06 100644 --- a/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVObsStoreImpl.java +++ b/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVObsStoreImpl.java @@ -74,6 +74,7 @@ public class MVObsStoreImpl implements IObsStore protected MVDataStreamStoreImpl dataStreamStore; protected MVBTreeMap obsRecordsIndex; protected MVBTreeMap obsSeriesMainIndex; + protected long lastSeriesID; protected MVBTreeMap obsSeriesByFoiIndex; protected int idScope; @@ -165,6 +166,12 @@ private MVObsStoreImpl init(MVStore mvStore, int idScope, IdProviderType dsIdPro .keyType(new MVObsSeriesKeyByFoiDataType()) .valueType(new MVVoidDataType())); + // next series ID must follow every series still registered, including empty ones, + // otherwise a new series can reuse the ID of an existing one + this.lastSeriesID = Math.max( + obsSeriesMainIndex.values().stream().mapToLong(s -> s.id).max().orElse(0), + obsRecordsIndex.isEmpty() ? 0 : obsRecordsIndex.lastKey().seriesID); + return this; } @@ -908,7 +915,7 @@ public synchronized BigId add(IObsData obs) obsSeriesByFoiIndex.putIfAbsent(seriesKey, Boolean.TRUE); return new MVTimeSeriesInfo( - obsRecordsIndex.isEmpty() ? 1 : obsRecordsIndex.lastKey().seriesID + 1); + ++lastSeriesID); }); // add to main obs index From 834f98e05ead23a1cbc3698da6b45cb4f62e336c Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 12 Oct 2026 16:31:56 -0500 Subject: [PATCH 03/13] [CSAPI] AND the parent datastream with system, dataStream and observedProperty These params replaced the parent filter. Observations under /samplingFeatures/{id} are now filtered by FOI. Refs #21, opensensorhub/osh-core#340 --- .../impl/service/consys/obs/ObsHandler.java | 102 +++++++++++++++--- .../impl/service/consys/TestObservations.java | 35 +++++- 2 files changed, 119 insertions(+), 18 deletions(-) diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java index a696407e1..6d54bbd67 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java @@ -16,7 +16,9 @@ import java.io.IOException; import java.time.Instant; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import java.util.Map.Entry; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -35,6 +37,7 @@ import org.sensorhub.api.datastore.DataStoreException; import org.sensorhub.api.datastore.SpatialFilter; import org.sensorhub.api.datastore.TemporalFilter; +import org.sensorhub.api.datastore.obs.DataStreamFilter; import org.sensorhub.api.datastore.obs.DataStreamKey; import org.sensorhub.api.datastore.obs.IObsStore; import org.sensorhub.api.datastore.obs.ObsFilter; @@ -99,7 +102,7 @@ protected ResourceBinding getBinding(RequestContext ctx, boolea ctx.setData(contextData); // try to fetch datastream since it's needed to configure binding - var dsID = ctx.getParentID(); + var dsID = getParentDataStreamID(ctx); if (dsID != null) contextData.dsInfo = db.getDataStreamStore().get(new DataStreamKey(dsID)); @@ -208,6 +211,17 @@ protected boolean isHttpRequestFromBrowser(RequestContext ctx) } + /* + * Observations are also mounted under sampling features, so the parent + * is only a datastream when it was reached through the datastream handler + */ + protected BigId getParentDataStreamID(RequestContext ctx) + { + var parent = ctx.getParentRef(); + return parent.type instanceof DataStreamHandler ? parent.internalID : null; + } + + @Override public void doPost(RequestContext ctx) throws IOException { @@ -224,7 +238,7 @@ protected void subscribe(final RequestContext ctx) throws InvalidRequestExceptio ctx.getSecurityHandler().checkPermission(permissions.stream); Asserts.checkNotNull(ctx.getStreamHandler(), StreamHandler.class); - var dsID = ctx.getParentID(); + var dsID = getParentDataStreamID(ctx); if (dsID == null) throw ServiceErrors.badRequest("Streaming is only supported on a specific datastream"); @@ -476,8 +490,18 @@ protected ObsFilter getFilter(ResourceRef parent, Map queryPar var builder = new ObsFilter.Builder(); // filter on parent if needed + BigId parentDsID = null; + boolean foiParent = false; if (parent.internalID != null) - builder.withDataStreams(parent.internalID); + { + if (parent.type instanceof DataStreamHandler) + parentDsID = parent.internalID; + else + { + builder.withFois(parent.internalID); + foiParent = true; + } + } var phenomenonTimeFilterBuilder = new TemporalFilter.Builder(); @@ -500,7 +524,7 @@ protected ObsFilter getFilter(ResourceRef parent, Map queryPar // foi param var foiIDs = parseResourceIdsOrUids("foi", queryParams, idEncoders.getFoiIdEncoder()); - if (foiIDs != null && !foiIDs.isEmpty()) + if (foiIDs != null && !foiIDs.isEmpty() && !foiParent) { if (foiIDs.isUids()) builder.withFois().withUniqueIDs(foiIDs.getUids()).done(); @@ -508,25 +532,69 @@ protected ObsFilter getFilter(ResourceRef parent, Map queryPar builder.withFois(foiIDs.getBigIds()); } - // system param + // the parent datastream and the system, dataStream and observedProperty params + // all restrict the same datastream filter, so they are combined with AND var sysIDs = parseResourceIdsOrUids("system", queryParams, idEncoders.getSystemIdEncoder()); - if (sysIDs != null && !sysIDs.isEmpty()) + var dsIDs = parseResourceIdsOrUids("dataStream", queryParams, idEncoders.getDataStreamIdEncoder()); + var obsProps = parseMultiValuesArg("observedProperty", queryParams); + boolean hasSysIDs = sysIDs != null && !sysIDs.isEmpty(); + boolean hasObsProps = obsProps != null && !obsProps.isEmpty(); + + // explicit datastream IDs: the parent, the dataStream param, or their intersection + // (null means no datastream ID was given) + Set selectedDsIDs = null; + if (parentDsID != null) + selectedDsIDs = new HashSet<>(Set.of(parentDsID)); + if (dsIDs != null && !dsIDs.isEmpty()) + { + if (selectedDsIDs == null) + selectedDsIDs = new HashSet<>(dsIDs.getBigIds()); + else // an empty intersection selects no observations + selectedDsIDs.retainAll(dsIDs.getBigIds()); + } + + // other datastream criteria, by system and observed property + var dsFilter = new DataStreamFilter.Builder(); + if (hasSysIDs) { if (sysIDs.isUids()) - builder.withSystems().withUniqueIDs(sysIDs.getUids()).done(); + dsFilter.withSystems().withUniqueIDs(sysIDs.getUids()).done(); else - builder.withSystems(sysIDs.getBigIds()); + dsFilter.withSystems(sysIDs.getBigIds()); } + if (hasObsProps) + dsFilter.withObservedProperties(obsProps); - // datastream param - var dsIDs = parseResourceIdsOrUids("dataStream", queryParams, idEncoders.getDataStreamIdEncoder()); - if (dsIDs != null && !dsIDs.isEmpty()) - builder.withDataStreams(dsIDs.getBigIds()); - - // observedProperty param - var obsProps = parseMultiValuesArg("observedProperty", queryParams); - if (obsProps != null && !obsProps.isEmpty()) - builder.withDataStreams().withObservedProperties(obsProps).done(); + if (selectedDsIDs != null) + { + // stores use datastream IDs alone when they are given, + // so check the other criteria against each selected datastream here + if (hasSysIDs || hasObsProps) + { + var propFilter = dsFilter.build(); + selectedDsIDs.removeIf(id -> { + // drop unknown datastreams + var dsInfo = db.getDataStreamStore().get(new DataStreamKey(id)); + if (dsInfo == null) + return true; + + // drop datastreams of other systems (matched by UID or ID) + if (hasSysIDs && !(sysIDs.isUids() ? + sysIDs.getUids().contains(dsInfo.getSystemID().getUniqueID()) : + sysIDs.getBigIds().contains(dsInfo.getSystemID().getInternalID()))) + return true; + + // drop datastreams without one of the observed properties + return !propFilter.testObservedProperty(dsInfo); + }); + } + + builder.withDataStreams(selectedDsIDs); + } + + // no datastream IDs given, so the store can select datastreams by the other criteria + else if (hasSysIDs || hasObsProps) + builder.withDataStreams(dsFilter.build()); // use opensearch bbox param to filter spatially var bbox = parseBboxArg("bbox", queryParams); diff --git a/sensorhub-service-consys/src/test/java/org/sensorhub/impl/service/consys/TestObservations.java b/sensorhub-service-consys/src/test/java/org/sensorhub/impl/service/consys/TestObservations.java index c29e0cdd6..7dee271e2 100644 --- a/sensorhub-service-consys/src/test/java/org/sensorhub/impl/service/consys/TestObservations.java +++ b/sensorhub-service-consys/src/test/java/org/sensorhub/impl/service/consys/TestObservations.java @@ -141,8 +141,41 @@ public void testAddObsAndDeleteDatastream() throws Exception } + @Test + public void testDatastreamFiltersCombineWithParent() throws Exception + { + // add 2 systems with 3 datastreams + var sysUrl1 = systemTests.addFeature(7); + var sysUrl2 = systemTests.addFeature(8); + var dsUrl1 = datastreamTests.addDatastreamOmJson(sysUrl1, 1); + var dsUrl2 = datastreamTests.addDatastreamOmJson(sysUrl1, 2); + var dsUrl3 = datastreamTests.addDatastreamOmJson(sysUrl2, 3); + var sysId1 = getResourceId(sysUrl1); + var sysId2 = getResourceId(sysUrl2); + var dsId1 = getResourceId(dsUrl1); + var dsId2 = getResourceId(dsUrl2); + + addObservationBatch(dsUrl1, Instant.now(), 100, 10, false); + addObservationBatch(dsUrl2, Instant.now(), 100, 5, false); + addObservationBatch(dsUrl3, Instant.now(), 100, 3, false); + + // params narrow the parent datastream, they don't replace it + var dsObs = concat(dsUrl1, OBS_COLLECTION); + sendGetRequestAndGetItems(dsObs + "?dataStream=" + dsId1, 10); + sendGetRequestAndGetItems(dsObs + "?dataStream=" + dsId2, 0); + sendGetRequestAndGetItems(dsObs + "?system=" + sysId1, 10); + sendGetRequestAndGetItems(dsObs + "?system=" + sysId2, 0); + sendGetRequestAndGetItems(dsObs + "?observedProperty=" + TestDataStreams.URI_DECIMAL_PROP, 10); + sendGetRequestAndGetItems(dsObs + "?observedProperty=http://blabla/nothing", 0); + + // params combine with each other at the top level + sendGetRequestAndGetItems(OBS_COLLECTION + "?system=" + sysId1 + "&observedProperty=" + TestDataStreams.URI_DECIMAL_PROP, 15); + sendGetRequestAndGetItems(OBS_COLLECTION + "?dataStream=" + dsId2 + "&system=" + sysId2, 0); + } + + // Non-Test helper methods - + protected String addObservation(String dsUrl, Instant startTime, int num) throws Exception { return addObservation(dsUrl, startTime, 1000L, num); From f05af517171cf39b2a7f63a76030e1bdb1763abf Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 12 Oct 2026 16:33:45 -0500 Subject: [PATCH 04/13] [Core] Pick the latest datastream or command stream version by valid time Query results aren't sorted, so the first one isn't always the latest. Refs opensensorhub/osh-core#343 --- .../command/ICommandStreamStore.java | 12 ++++- .../api/datastore/obs/IDataStreamStore.java | 12 ++++- .../mem/TestInMemCommandStreamStore.java | 47 +++++++++++++++++++ .../mem/TestInMemDataStreamStore.java | 47 +++++++++++++++++++ 4 files changed, 116 insertions(+), 2 deletions(-) diff --git a/sensorhub-core/src/main/java/org/sensorhub/api/datastore/command/ICommandStreamStore.java b/sensorhub-core/src/main/java/org/sensorhub/api/datastore/command/ICommandStreamStore.java index 8b085d1ed..97b75e1bd 100644 --- a/sensorhub-core/src/main/java/org/sensorhub/api/datastore/command/ICommandStreamStore.java +++ b/sensorhub-core/src/main/java/org/sensorhub/api/datastore/command/ICommandStreamStore.java @@ -14,6 +14,8 @@ package org.sensorhub.api.datastore.command; +import java.time.Instant; +import java.util.Comparator; import org.sensorhub.api.command.ICommandStreamInfo; import org.sensorhub.api.datastore.DataStoreException; import org.sensorhub.api.datastore.command.ICommandStreamStore.CommandStreamInfoField; @@ -111,12 +113,20 @@ public default Entry getLatestVersionEntry .done() .withControlInputNames(controlInput) .build()) - .findFirst(); + // results are not guaranteed to be sorted, so select the version with the latest valid time + .max(Comparator.comparing(e -> getValidTimeBegin(e.getValue()))); return entryOpt.isPresent() ? entryOpt.get() : null; } + private static Instant getValidTimeBegin(ICommandStreamInfo info) + { + var validTime = info.getValidTime(); + return validTime != null ? validTime.begin() : Instant.MIN; + } + + /** * Remove all command streams that are associated to the given system command input * @param sysUID Unique ID of system receiving the command stream diff --git a/sensorhub-core/src/main/java/org/sensorhub/api/datastore/obs/IDataStreamStore.java b/sensorhub-core/src/main/java/org/sensorhub/api/datastore/obs/IDataStreamStore.java index 210e9583e..72bbff0d8 100644 --- a/sensorhub-core/src/main/java/org/sensorhub/api/datastore/obs/IDataStreamStore.java +++ b/sensorhub-core/src/main/java/org/sensorhub/api/datastore/obs/IDataStreamStore.java @@ -14,6 +14,8 @@ package org.sensorhub.api.datastore.obs; +import java.time.Instant; +import java.util.Comparator; import org.sensorhub.api.data.IDataStreamInfo; import org.sensorhub.api.datastore.DataStoreException; import org.sensorhub.api.datastore.obs.IDataStreamStore.DataStreamInfoField; @@ -111,12 +113,20 @@ public default Entry getLatestVersionEntry(Strin .done() .withOutputNames(outputName) .build()) - .findFirst(); + // results are not guaranteed to be sorted, so select the version with the latest valid time + .max(Comparator.comparing(e -> getValidTimeBegin(e.getValue()))); return entryOpt.isPresent() ? entryOpt.get() : null; } + private static Instant getValidTimeBegin(IDataStreamInfo info) + { + var validTime = info.getValidTime(); + return validTime != null ? validTime.begin() : Instant.MIN; + } + + /** * Remove all datastreams that are associated to the given system output * @param sysUID diff --git a/sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemCommandStreamStore.java b/sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemCommandStreamStore.java index 6d1602b86..12ceb41f3 100644 --- a/sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemCommandStreamStore.java +++ b/sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemCommandStreamStore.java @@ -14,7 +14,24 @@ package org.sensorhub.impl.datastore.mem; +import static org.junit.Assert.assertEquals; +import java.time.Instant; +import java.util.AbstractMap.SimpleEntry; +import java.util.Map.Entry; +import java.util.Set; +import java.util.stream.Stream; +import org.junit.Test; +import org.sensorhub.api.command.CommandStreamInfo; +import org.sensorhub.api.command.ICommandStreamInfo; +import org.sensorhub.api.common.BigId; +import org.sensorhub.api.datastore.command.CommandStreamFilter; +import org.sensorhub.api.datastore.command.CommandStreamKey; +import org.sensorhub.api.datastore.command.ICommandStreamStore.CommandStreamInfoField; +import org.sensorhub.api.feature.FeatureId; import org.sensorhub.impl.datastore.AbstractTestCommandStreamStore; +import org.vast.data.TextEncodingImpl; +import org.vast.swe.SWEHelper; +import org.vast.util.TimeExtent; public class TestInMemCommandStreamStore extends AbstractTestCommandStreamStore @@ -73,4 +90,34 @@ protected void testAddAndSelectByKeywords_ExpectedResults(int testCaseIdx) case 4: addToExpectedResults(4); break; } } + + + @Test + public void testLatestVersionEntryWithUnsortedResults() throws Exception + { + var sysId = new FeatureId(BigId.fromLong(DATABASE_NUM, 1), "urn:osh:test:sys1"); + var rec = new SWEHelper().createRecord().name("in1").addField("f1", new SWEHelper().createQuantity()).build(); + var oldInfo = new CommandStreamInfo.Builder() + .withName("v1").withSystem(sysId).withRecordDescription(rec).withRecordEncoding(new TextEncodingImpl()) + .withValidTime(TimeExtent.beginAt(Instant.parse("2020-01-01T00:00:00Z"))) + .build(); + var newInfo = new CommandStreamInfo.Builder() + .withName("v2").withSystem(sysId).withRecordDescription(rec).withRecordEncoding(new TextEncodingImpl()) + .withValidTime(TimeExtent.beginAt(Instant.parse("2024-01-01T00:00:00Z"))) + .build(); + + // store that returns versions oldest first + var store = new InMemoryCommandStreamStore(new InMemoryCommandStore(DATABASE_NUM)) { + @Override + public Stream> selectEntries(CommandStreamFilter filter, Set fields) + { + return Stream.of( + new SimpleEntry<>(new CommandStreamKey(DATABASE_NUM, 1), oldInfo), + new SimpleEntry<>(new CommandStreamKey(DATABASE_NUM, 2), newInfo)); + } + }; + + var e = store.getLatestVersionEntry(sysId.getUniqueID(), "in1"); + assertEquals("v2", e.getValue().getName()); + } } diff --git a/sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemDataStreamStore.java b/sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemDataStreamStore.java index 28adb71c4..0c04a9525 100644 --- a/sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemDataStreamStore.java +++ b/sensorhub-core/src/test/java/org/sensorhub/impl/datastore/mem/TestInMemDataStreamStore.java @@ -14,7 +14,24 @@ package org.sensorhub.impl.datastore.mem; +import static org.junit.Assert.assertEquals; +import java.time.Instant; +import java.util.AbstractMap.SimpleEntry; +import java.util.Map.Entry; +import java.util.Set; +import java.util.stream.Stream; +import org.junit.Test; +import org.sensorhub.api.common.BigId; +import org.sensorhub.api.data.DataStreamInfo; +import org.sensorhub.api.data.IDataStreamInfo; +import org.sensorhub.api.datastore.obs.DataStreamFilter; +import org.sensorhub.api.datastore.obs.DataStreamKey; +import org.sensorhub.api.datastore.obs.IDataStreamStore.DataStreamInfoField; +import org.sensorhub.api.feature.FeatureId; import org.sensorhub.impl.datastore.AbstractTestDataStreamStore; +import org.vast.data.TextEncodingImpl; +import org.vast.swe.SWEHelper; +import org.vast.util.TimeExtent; public class TestInMemDataStreamStore extends AbstractTestDataStreamStore @@ -73,4 +90,34 @@ protected void testAddAndSelectByKeywords_ExpectedResults(int testCaseIdx) case 4: addToExpectedResults(4); break; } } + + + @Test + public void testLatestVersionEntryWithUnsortedResults() throws Exception + { + var sysId = new FeatureId(BigId.fromLong(DATABASE_NUM, 1), "urn:osh:test:sys1"); + var rec = new SWEHelper().createRecord().name("out1").addField("f1", new SWEHelper().createQuantity()).build(); + var oldInfo = new DataStreamInfo.Builder() + .withName("v1").withSystem(sysId).withRecordDescription(rec).withRecordEncoding(new TextEncodingImpl()) + .withValidTime(TimeExtent.beginAt(Instant.parse("2020-01-01T00:00:00Z"))) + .build(); + var newInfo = new DataStreamInfo.Builder() + .withName("v2").withSystem(sysId).withRecordDescription(rec).withRecordEncoding(new TextEncodingImpl()) + .withValidTime(TimeExtent.beginAt(Instant.parse("2024-01-01T00:00:00Z"))) + .build(); + + // store that returns versions oldest first + var store = new InMemoryDataStreamStore(new InMemoryObsStore(DATABASE_NUM)) { + @Override + public Stream> selectEntries(DataStreamFilter filter, Set fields) + { + return Stream.of( + new SimpleEntry<>(new DataStreamKey(DATABASE_NUM, 1), oldInfo), + new SimpleEntry<>(new DataStreamKey(DATABASE_NUM, 2), newInfo)); + } + }; + + var e = store.getLatestVersionEntry(sysId.getUniqueID(), "out1"); + assertEquals("v2", e.getValue().getName()); + } } From e38bfc2592b33ffa6b873b3716d865468e4a9da5 Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 12 Oct 2026 16:36:26 -0500 Subject: [PATCH 05/13] [H2] Respect the phenomenonTime range with resultTime=latest Only for series where result time equals phenomenon time. Refs opensensorhub/osh-core#331 --- .../impl/datastore/AbstractTestObsStore.java | 27 +++++++++++++++++++ .../impl/datastore/h2/MVObsStoreImpl.java | 5 ++-- 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestObsStore.java b/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestObsStore.java index 92bc8b1e1..fca341922 100644 --- a/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestObsStore.java +++ b/sensorhub-core/src/testFixtures/java/org/sensorhub/impl/datastore/AbstractTestObsStore.java @@ -416,6 +416,33 @@ public void testRemoveByKeyDoesNotMixDataStreams() throws Exception } + @Test + public void testSelectLatestResultWithinPhenomenonTimeRange() throws Exception + { + var dsID = addSimpleDataStream(bigId(10), "out1"); + var startTime = Instant.parse("2020-01-01T00:00:00Z"); + var obs = new ArrayList<>(addSimpleObsWithoutResultTime(dsID, BigId.NONE, startTime, 5, 60000).entrySet()); + forceReadBackFromStorage(); + + // range covers the 2nd to 3rd obs, so only the 3rd is expected + var filter = new ObsFilter.Builder() + .withDataStreams(dsID) + .withPhenomenonTimeDuring(startTime.plusSeconds(60), startTime.plusSeconds(150)) + .withLatestResult() + .build(); + var expected = Map.of(obs.get(2).getKey(), obs.get(2).getValue()); + checkSelectedEntries(obsStore.selectEntries(filter), expected, filter); + + // range after all obs + filter = new ObsFilter.Builder() + .withDataStreams(dsID) + .withPhenomenonTimeDuring(startTime.plusSeconds(3600), startTime.plusSeconds(7200)) + .withLatestResult() + .build(); + checkSelectedEntries(obsStore.selectEntries(filter), Collections.emptyMap(), filter); + } + + @Test public void testAddAndRemoveEntireDatastream() throws Exception { diff --git a/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVObsStoreImpl.java b/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVObsStoreImpl.java index 557b65d06..d617e2be6 100644 --- a/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVObsStoreImpl.java +++ b/sensorhub-datastore-h2/src/main/java/org/sensorhub/impl/datastore/h2/MVObsStoreImpl.java @@ -310,11 +310,12 @@ Stream> getObsStream(MVTimeSeriesInfo ser } // if request if for latest result only, get only the latest obs in series + // within the requested phenomenon time range if (timeParams.latestResultOnly) { - MVTimeSeriesRecordKey maxKey = new MVTimeSeriesRecordKey(series.id, Instant.MAX); + MVTimeSeriesRecordKey maxKey = new MVTimeSeriesRecordKey(series.id, phenomenonTimeRange.upperEndpoint()); Entry e = obsRecordsIndex.floorEntry(maxKey); - if (e != null && e.getKey().seriesID == series.id) + if (e != null && e.getKey().seriesID == series.id && phenomenonTimeRange.contains(e.getKey().timeStamp)) return Stream.of(e); else return Stream.empty(); From f5297cb333b6b114094770ad6675ef1963562f1f Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 12 Oct 2026 16:42:28 -0500 Subject: [PATCH 06/13] [CSAPI] Test write commits and sampling feature observations Refs #18, #21 --- .../impl/service/consys/TestObservations.java | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/sensorhub-service-consys/src/test/java/org/sensorhub/impl/service/consys/TestObservations.java b/sensorhub-service-consys/src/test/java/org/sensorhub/impl/service/consys/TestObservations.java index 7dee271e2..69f26aa86 100644 --- a/sensorhub-service-consys/src/test/java/org/sensorhub/impl/service/consys/TestObservations.java +++ b/sensorhub-service-consys/src/test/java/org/sensorhub/impl/service/consys/TestObservations.java @@ -15,6 +15,7 @@ package org.sensorhub.impl.service.consys; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import java.io.IOException; import java.io.StringWriter; import java.time.Instant; @@ -23,6 +24,7 @@ import org.junit.Before; import org.junit.Test; import org.sensorhub.api.common.SensorHubException; +import org.sensorhub.impl.datastore.h2.MVObsSystemDatabase; import com.google.gson.JsonArray; import com.google.gson.JsonObject; import com.google.gson.JsonParser; @@ -33,6 +35,7 @@ public class TestObservations extends AbstractTestApiBase { TestSystems systemTests = new TestSystems(); TestDataStreams datastreamTests = new TestDataStreams(); + TestFois foiTests = new TestFois(); @Before @@ -41,6 +44,7 @@ public void setup() throws IOException, SensorHubException super.setup(); systemTests.apiRootUrl = apiRootUrl; datastreamTests.apiRootUrl = apiRootUrl; + foiTests.apiRootUrl = apiRootUrl; } @@ -174,6 +178,43 @@ public void testDatastreamFiltersCombineWithParent() throws Exception } + @Test + public void testWritesAreCommittedBeforeResponse() throws Exception + { + var mvStore = ((MVObsSystemDatabase)db).getMVStore(); + + var sysUrl = systemTests.addFeature(9); + assertFalse("System not committed", mvStore.hasUnsavedChanges()); + + var dsUrl = datastreamTests.addDatastreamOmJson(sysUrl, 1); + assertFalse("Datastream not committed", mvStore.hasUnsavedChanges()); + + var urls = addObservationBatch(dsUrl, Instant.now(), 100, 5, false); + assertFalse("Observations not committed", mvStore.hasUnsavedChanges()); + + sendDeleteRequestAndCheckStatus(urls.get(0), 204); + assertFalse("Observation delete not committed", mvStore.hasUnsavedChanges()); + } + + + @Test + public void testSamplingFeatureObservations() throws Exception + { + var sysUrl = systemTests.addFeature(10); + var dsUrl = datastreamTests.addDatastreamOmJson(sysUrl, 1); + var foiId = getResourceId(foiTests.addFoi(null, 1)); + + // obs with and without the sampling feature + var obsUrl = concat(dsUrl, OBS_COLLECTION); + for (int i = 0; i < 4; i++) + sendPostRequest(obsUrl + "?foi=" + foiId, createObservationNoFoi(Instant.now(), 100, i)); + addObservationBatch(dsUrl, Instant.now(), 100, 3, false); + + sendGetRequestAndGetItems(concat(concat(FOI_COLLECTION, foiId), OBS_COLLECTION), 4); + sendGetRequestAndGetItems(obsUrl, 7); + } + + // Non-Test helper methods protected String addObservation(String dsUrl, Instant startTime, int num) throws Exception From 232c2591cb54793f867c029e2c0084bb9a6eb25e Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Wed, 17 Jun 2026 14:56:03 -0500 Subject: [PATCH 07/13] feat(consys): add custom obs/command format hooks for data and served schemas Extend the CustomObsFormat SPI with command-side hooks (getCommandBinding, getCommandSchemaBinding) and an isAutoSelectable flag, and thread the configured customFormats map through the obs, command, command-status, datastream-schema and command-stream-schema handlers so a registered format (e.g. application/swe+proto) can serve its own data and schema views. --- .../impl/service/consys/ConSysApiService.java | 8 +-- .../service/consys/obs/CustomObsFormat.java | 54 +++++++++++++++++-- .../consys/obs/DataStreamSchemaHandler.java | 23 +++++++- .../impl/service/consys/obs/ObsHandler.java | 10 ++-- .../service/consys/task/CommandHandler.java | 36 ++++++++++--- .../consys/task/CommandStatusHandler.java | 6 ++- .../task/CommandStreamSchemaHandler.java | 35 +++++++++--- 7 files changed, 145 insertions(+), 27 deletions(-) diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java index 2671639ed..33ff879e9 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java @@ -262,7 +262,7 @@ protected void doStart() throws SensorHubException rootHandler.addSubResource(dataStreamHandler); systemsHandler.addSubResource(dataStreamHandler); sysMembersHandler.addSubResource(dataStreamHandler); - var dataSchemaHandler = new DataStreamSchemaHandler(handlerCtx, security.datastream_permissions); + var dataSchemaHandler = new DataStreamSchemaHandler(handlerCtx, security.datastream_permissions, customFormats); dataStreamHandler.addSubResource(dataSchemaHandler); // observations @@ -282,11 +282,11 @@ protected void doStart() throws SensorHubException rootHandler.addSubResource(cmdStreamHandler); systemsHandler.addSubResource(cmdStreamHandler); sysMembersHandler.addSubResource(cmdStreamHandler); - var cmdSchemaHandler = new CommandStreamSchemaHandler(handlerCtx, security.commandstream_permissions); + var cmdSchemaHandler = new CommandStreamSchemaHandler(handlerCtx, security.commandstream_permissions, customFormats); cmdStreamHandler.addSubResource(cmdSchemaHandler); - + // commands - var cmdHandler = new CommandHandler(handlerCtx, threadPool, security.command_permissions); + var cmdHandler = new CommandHandler(handlerCtx, threadPool, security.command_permissions, customFormats); cmdStreamHandler.addSubResource(cmdHandler); // command status diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/CustomObsFormat.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/CustomObsFormat.java index 8423a3b28..ae5fcd376 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/CustomObsFormat.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/CustomObsFormat.java @@ -15,10 +15,13 @@ package org.sensorhub.impl.service.consys.obs; import java.io.IOException; +import org.sensorhub.api.command.ICommandData; +import org.sensorhub.api.command.ICommandStreamInfo; import org.sensorhub.api.common.BigId; import org.sensorhub.api.common.IdEncoders; import org.sensorhub.api.data.IDataStreamInfo; import org.sensorhub.api.data.IObsData; +import org.sensorhub.api.datastore.command.CommandStreamKey; import org.sensorhub.api.datastore.obs.DataStreamKey; import org.sensorhub.impl.service.consys.resource.RequestContext; import org.sensorhub.impl.service.consys.resource.ResourceBinding; @@ -26,11 +29,54 @@ public interface CustomObsFormat { + /** + * @return true if this format can encode observations of the given + * datastream. Drives the "formats" list advertised on the datastream + * resource and explicit format selection. + */ boolean isCompatible(IDataStreamInfo dsInfo); - - + + + /** + * @return true if this format should be auto-selected for browser requests + * with no explicit format (e.g. video formats playable in a browser). + * Defaults to {@link #isCompatible}; override to return false for formats + * that should be advertised but only served when explicitly requested. + */ + default boolean isAutoSelectable(IDataStreamInfo dsInfo) + { + return isCompatible(dsInfo); + } + + ResourceBinding getSchemaBinding(RequestContext ctx, IdEncoders idEncoders, IDataStreamInfo dsInfo) throws IOException; - - + + ResourceBinding getObsBinding(RequestContext ctx, IdEncoders idEncoders, IDataStreamInfo dsInfo) throws IOException; + + + /** + * @return a binding serving this format's view of a command stream's + * parameter schema, or null if this format does not support commands. + * Consulted by CommandStreamSchemaHandler when the commandFormat query + * param matches this format's mime type. + */ + default ResourceBinding getCommandSchemaBinding(RequestContext ctx, IdEncoders idEncoders, ICommandStreamInfo csInfo) throws IOException + { + return null; + } + + + /** + * @param forReading true when the binding will deserialize commands from + * the request body (POST/publish ingestion), false when it will + * serialize commands to the response (GET/stream) + * @return a binding encoding/decoding commands in this format, or null if + * this format does not support commands. Consulted by CommandHandler for + * both reads (GET/stream) and ingestion (POST with this content type). + */ + default ResourceBinding getCommandBinding(RequestContext ctx, IdEncoders idEncoders, ICommandStreamInfo csInfo, boolean forReading) throws IOException + { + return null; + } } diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/DataStreamSchemaHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/DataStreamSchemaHandler.java index e648673a1..f243e8c3e 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/DataStreamSchemaHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/DataStreamSchemaHandler.java @@ -36,11 +36,20 @@ public class DataStreamSchemaHandler extends ResourceHandler { public static final String[] NAMES = { "schema" }; - - + + final Map customFormats; + + public DataStreamSchemaHandler(HandlerContext ctx, ResourcePermissions permissions) + { + this(ctx, permissions, java.util.Collections.emptyMap()); + } + + + public DataStreamSchemaHandler(HandlerContext ctx, ResourcePermissions permissions, Map customFormats) { super(ctx.getReadDb().getDataStreamStore(), ctx.getDataStreamIdEncoder(), ctx, permissions); + this.customFormats = customFormats != null ? customFormats : java.util.Collections.emptyMap(); } @@ -62,6 +71,16 @@ else if (format.equals(ResourceFormat.AUTO) && ctx.isBrowserHtmlRequest()) return new DataStreamBindingHtml(ctx, idEncoders, obsFormat); else if (obsFormat.isOneOf(ResourceFormat.JSON, ResourceFormat.OM_JSON)) return new DataStreamSchemaBindingOmJson(ctx, idEncoders, forReading); + // custom obs formats (e.g. application/swe+proto) provide their own schema + // binding — checked before the SWE_FORMAT_PREFIX branch since swe+proto + // also starts with "application/swe+" + else if (customFormats.containsKey(obsFormat.getMimeType())) + { + var dsInfo = dataStore.get(new DataStreamKey(ctx.getParentRef().internalID)); + if (dsInfo == null) + throw ServiceErrors.notFound(); + return customFormats.get(obsFormat.getMimeType()).getSchemaBinding(ctx, idEncoders, dsInfo); + } else if (obsFormat.getMimeType().startsWith(ResourceFormat.SWE_FORMAT_PREFIX)) return new DataStreamSchemaBindingSweCommon(obsFormat, ctx, idEncoders, forReading); else diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java index 6d54bbd67..bd4d2d38e 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java @@ -172,7 +172,7 @@ protected ResourceBinding getCustomFormatBinding(RequestContext var mimeType = entry.getKey(); var formatImpl = entry.getValue(); - if (formatImpl.isCompatible(dsInfo)) + if (formatImpl.isAutoSelectable(dsInfo)) { obsFormat = formatImpl; ctx.getLogger().info("Auto-selecting format {}", mimeType); @@ -244,9 +244,13 @@ protected void subscribe(final RequestContext ctx) throws InvalidRequestExceptio var queryParams = ctx.getParameterMap(); var filter = getFilter(ctx.getParentRef(), queryParams, 0, Long.MAX_VALUE); - var responseFormat = parseFormat(queryParams); + // default to the format already set by the transport layer, if any + // (e.g. MQTT ":data/" subtopics); an explicit f= query + // param still takes precedence + var defaultFormat = ctx.getFormat() != null ? ctx.getFormat() : ResourceFormat.AUTO; + var responseFormat = parseFormat(queryParams, defaultFormat); ctx.setFormatOptions(responseFormat, parseSelectArg(queryParams)); - + // detect if real-time request boolean isRealTime = (filter.getPhenomenonTime() == null && filter.getResultTime() == null) || (filter.getResultTime() != null && filter.getResultTime().beginsNow()) || diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java index 0af2e1241..e14c5153e 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java @@ -46,6 +46,7 @@ import org.sensorhub.impl.service.consys.HandlerContext; import org.sensorhub.impl.service.consys.ServiceErrors; import org.sensorhub.impl.service.consys.RestApiServlet.ResourcePermissions; +import org.sensorhub.impl.service.consys.obs.CustomObsFormat; import org.sensorhub.impl.service.consys.resource.BaseResourceHandler; import org.sensorhub.impl.service.consys.resource.IResourceHandler; import org.sensorhub.impl.service.consys.resource.RequestContext; @@ -68,9 +69,10 @@ public class CommandHandler extends BaseResourceHandler customFormats; final Random random = new SecureRandom(); - - + + public static class CommandHandlerContextData { public BigId streamID; @@ -78,16 +80,23 @@ public static class CommandHandlerContextData public BigId foiId; public CommandStreamTransactionHandler dsHandler; } - - + + public CommandHandler(HandlerContext ctx, ScheduledExecutorService threadPool, ResourcePermissions permissions) + { + this(ctx, threadPool, permissions, java.util.Collections.emptyMap()); + } + + + public CommandHandler(HandlerContext ctx, ScheduledExecutorService threadPool, ResourcePermissions permissions, Map customFormats) { super(ctx.getReadDb().getCommandStore(), ctx.getCommandIdEncoder(), ctx, permissions); - + this.eventBus = ctx.getEventBus(); this.db = ctx.getReadDb(); this.threadPool = threadPool; - + this.customFormats = customFormats != null ? customFormats : java.util.Collections.emptyMap(); + // I know the doc says otherwise but we need to use the federated DB for command transactions here // because we don't write to DB directly but rather send commands to systems that can be in other databases this.readOnlyTxHandler = new SystemDatabaseTransactionHandler(eventBus, ctx.getReadDb()); @@ -151,6 +160,15 @@ protected ResourceBinding getBinding(RequestContext ctx, bo } // select binding depending on format + // custom formats (e.g. application/swe+proto) take precedence when the + // requested mime type matches and the format supports commands + if (contextData.csInfo != null && customFormats.containsKey(format.getMimeType())) + { + var binding = customFormats.get(format.getMimeType()).getCommandBinding(ctx, idEncoders, contextData.csInfo, forReading); + if (binding != null) + return binding; + } + if (format.isOneOf(ResourceFormat.AUTO, ResourceFormat.JSON)) return new CommandBindingJson(ctx, idEncoders, forReading, dataStore); else @@ -181,7 +199,11 @@ protected void subscribe(final RequestContext ctx) throws InvalidRequestExceptio var queryParams = ctx.getParameterMap(); var filter = getFilter(ctx.getParentRef(), queryParams, 0, Long.MAX_VALUE); - var responseFormat = parseFormat(queryParams); + // default to the format already set by the transport layer, if any + // (e.g. MQTT ":data/" subtopics); an explicit f= query + // param still takes precedence + var defaultFormat = ctx.getFormat() != null ? ctx.getFormat() : ResourceFormat.AUTO; + var responseFormat = parseFormat(queryParams, defaultFormat); ctx.setFormatOptions(responseFormat, parseSelectArg(queryParams)); // continue when streaming actually starts diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStatusHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStatusHandler.java index 19b0f780d..d55e81cf7 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStatusHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStatusHandler.java @@ -146,7 +146,11 @@ protected void subscribe(final RequestContext ctx) throws InvalidRequestExceptio var queryParams = ctx.getParameterMap(); var filter = getFilter(ctx.getParentRef(), queryParams, 0, Long.MAX_VALUE); - var responseFormat = parseFormat(queryParams); + // default to the format already set by the transport layer, if any + // (e.g. MQTT ":data/" subtopics); an explicit f= query + // param still takes precedence + var defaultFormat = ctx.getFormat() != null ? ctx.getFormat() : ResourceFormat.AUTO; + var responseFormat = parseFormat(queryParams, defaultFormat); ctx.setFormatOptions(responseFormat, parseSelectArg(queryParams)); // continue when streaming actually starts diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamSchemaHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamSchemaHandler.java index 86c4f533d..13cbdb7d4 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamSchemaHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamSchemaHandler.java @@ -25,6 +25,7 @@ import org.sensorhub.impl.service.consys.HandlerContext; import org.sensorhub.impl.service.consys.ServiceErrors; import org.sensorhub.impl.service.consys.RestApiServlet.ResourcePermissions; +import org.sensorhub.impl.service.consys.obs.CustomObsFormat; import org.sensorhub.impl.service.consys.resource.RequestContext; import org.sensorhub.impl.service.consys.resource.ResourceBinding; import org.sensorhub.impl.service.consys.resource.ResourceFormat; @@ -36,30 +37,52 @@ public class CommandStreamSchemaHandler extends ResourceHandler { public static final String[] NAMES = { "schema" }; - - + + final Map customFormats; + + public CommandStreamSchemaHandler(HandlerContext ctx, ResourcePermissions permissions) + { + this(ctx, permissions, java.util.Collections.emptyMap()); + } + + + public CommandStreamSchemaHandler(HandlerContext ctx, ResourcePermissions permissions, Map customFormats) { super(ctx.getReadDb().getCommandStreamStore(), ctx.getCommandStreamIdEncoder(), ctx, permissions); + this.customFormats = customFormats != null ? customFormats : java.util.Collections.emptyMap(); } - - + + @Override protected ResourceBinding getBinding(RequestContext ctx, boolean forReading) throws IOException { var format = ctx.getFormat(); if (!format.isOneOf(ResourceFormat.AUTO, ResourceFormat.JSON)) throw ServiceErrors.unsupportedFormat(format); - + // generate proper schema depending on command format var cmdFormat = parseFormat("commandFormat", ctx.getParameterMap()); if (cmdFormat == null) cmdFormat = ResourceFormat.JSON; - + if (format.equals(ResourceFormat.AUTO) && ctx.isBrowserHtmlRequest()) return new CommandStreamBindingHtml(ctx, idEncoders); else if (cmdFormat.equals(ResourceFormat.JSON)) return new CommandStreamSchemaBindingJson(ctx, idEncoders, forReading); + // custom command formats (e.g. application/swe+proto) provide their own + // schema binding — checked before the SWE_FORMAT_PREFIX branch since + // swe+proto also starts with "application/swe+" + else if (customFormats.containsKey(cmdFormat.getMimeType())) + { + var csInfo = dataStore.get(new CommandStreamKey(ctx.getParentRef().internalID)); + if (csInfo == null) + throw ServiceErrors.notFound(); + var binding = customFormats.get(cmdFormat.getMimeType()).getCommandSchemaBinding(ctx, idEncoders, csInfo); + if (binding == null) + throw ServiceErrors.unsupportedFormat(cmdFormat); + return binding; + } else if (cmdFormat.getMimeType().startsWith(ResourceFormat.SWE_FORMAT_PREFIX)) return new CommandStreamSchemaBindingSweCommon(cmdFormat, ctx, idEncoders, forReading); else From bb6979897869cfdab9315137d2aab7508b354368 Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Wed, 17 Jun 2026 14:56:34 -0500 Subject: [PATCH 08/13] feat(consys): route custom schema formats on datastream/controlstream POST Dispatch datastream and control-stream creation to a registered custom format's binding when the embedded schema names it. DataStreamBindingJson gains a custom-format branch ahead of the application/swe+ prefix match; CommandStreamBindingJson buffers the schema object and dispatches on commandFormat, staying backward-compatible (a missing commandFormat still uses the default JSON binding); CommandStreamHandler threads the customFormats map through to the binding. --- .../impl/service/consys/ConSysApiService.java | 2 +- .../consys/obs/DataStreamBindingJson.java | 9 +++ .../consys/task/CommandStreamBindingJson.java | 66 +++++++++++++------ .../consys/task/CommandStreamHandler.java | 17 +++-- 4 files changed, 70 insertions(+), 24 deletions(-) diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java index 33ff879e9..93ff7cbaa 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java @@ -278,7 +278,7 @@ protected void doStart() throws SensorHubException //foiHandler.addSubResource(obsStatsHandler); // command streams - var cmdStreamHandler = new CommandStreamHandler(handlerCtx, security.commandstream_permissions); + var cmdStreamHandler = new CommandStreamHandler(handlerCtx, security.commandstream_permissions, customFormats); rootHandler.addSubResource(cmdStreamHandler); systemsHandler.addSubResource(cmdStreamHandler); sysMembersHandler.addSubResource(cmdStreamHandler); diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/DataStreamBindingJson.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/DataStreamBindingJson.java index 859110e11..5e3fb7318 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/DataStreamBindingJson.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/DataStreamBindingJson.java @@ -151,6 +151,15 @@ else if ("schema".equals(prop)) ResourceBindingJson schemaBinding = null; if (ResourceFormat.OM_JSON.getMimeType().equals(obsFormat)) schemaBinding = new DataStreamSchemaBindingOmJson(ctx, idEncoders, reader); + // custom obs formats (e.g. application/swe+proto) bring their own + // schema binding — checked BEFORE the SWE_FORMAT_PREFIX branch since + // swe+proto also matches that prefix (mirrors DataStreamSchemaHandler.getBinding) + else if (customFormats.containsKey(obsFormat)) + { + var custom = customFormats.get(obsFormat).getSchemaBinding(ctx, idEncoders, null); + if (custom instanceof ResourceBindingJson) + schemaBinding = (ResourceBindingJson) custom; + } else if (obsFormat.startsWith(ResourceFormat.SWE_FORMAT_PREFIX)) schemaBinding = new DataStreamSchemaBindingSweCommon(ResourceFormat.fromMimeType(obsFormat), ctx, idEncoders, reader); if (schemaBinding == null) diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java index b57b19d18..c1f494476 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java @@ -15,8 +15,11 @@ package org.sensorhub.impl.service.consys.task; import java.io.IOException; +import java.io.StringReader; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; +import java.util.Map; import net.opengis.swe.v20.BinaryEncoding; import org.sensorhub.api.command.CommandStreamInfo; import org.sensorhub.api.command.ICommandStreamInfo; @@ -27,6 +30,8 @@ import org.sensorhub.impl.service.consys.InvalidRequestException; import org.sensorhub.impl.service.consys.ResourceParseException; import org.sensorhub.impl.service.consys.SWECommonUtils; +import org.sensorhub.impl.service.consys.ServiceErrors; +import org.sensorhub.impl.service.consys.obs.CustomObsFormat; import org.sensorhub.impl.service.consys.resource.RequestContext; import org.sensorhub.impl.service.consys.resource.ResourceBindingJson; import org.sensorhub.impl.service.consys.resource.ResourceFormat; @@ -37,6 +42,7 @@ import org.vast.swe.SWEStaxBindings; import org.vast.swe.json.SWEJsonStreamReader; import org.vast.swe.json.SWEJsonStreamWriter; +import com.google.gson.JsonParser; import com.google.gson.stream.JsonReader; import com.google.gson.stream.JsonToken; import com.google.gson.stream.JsonWriter; @@ -47,18 +53,26 @@ public class CommandStreamBindingJson extends ResourceBindingJson customFormats; SWEJsonStreamReader sweReader; SWEJsonStreamWriter sweWriter; - - + + public CommandStreamBindingJson(RequestContext ctx, IdEncoders idEncoders, IObsSystemDatabase db, boolean forReading) throws IOException + { + this(ctx, idEncoders, db, forReading, Collections.emptyMap()); + } + + + public CommandStreamBindingJson(RequestContext ctx, IdEncoders idEncoders, IObsSystemDatabase db, boolean forReading, Map customFormats) throws IOException { super(ctx, idEncoders, forReading); - + this.assocs = new CommandStreamAssocs(db, idEncoders); this.rootURL = ctx.getApiRootURL(); this.sweBindings = new SWEStaxBindings(); - + this.customFormats = customFormats != null ? customFormats : Collections.emptyMap(); + if (forReading) this.sweReader = new SWEJsonStreamReader(reader); else @@ -103,21 +117,35 @@ else if ("system@link".equals(prop)) sysRef = readFeatureRef(reader); else if ("schema".equals(prop)) { - reader.beginObject(); - - /*// obsFormat must come first! - if (!reader.nextName().equals("commandFormat")) - throw new ResourceParseException(MISSING_PROP_ERROR_MSG + "schema/obsFormat"); - var obsFormat = reader.nextString(); - - ResourceBindingJson schemaBinding = null; - if (ResourceFormat.JSON.getMimeType().equals(obsFormat)) - schemaBinding = new CommandStreamSchemaBindingJson(ctx, idEncoders, reader); - - if (schemaBinding == null) - throw ServiceErrors.unsupportedFormat(obsFormat);*/ - var schemaBinding = new CommandStreamSchemaBindingJson(ctx, idEncoders, reader); - csInfo = schemaBinding.deserialize(reader); + // Buffer the schema object so commandFormat can be inspected + // without destructively consuming the stream. This preserves + // backward compatibility with bodies that omit commandFormat or + // don't list it first (e.g. legacy "parametersSchema"-first), while + // still letting a custom format (application/swe+proto) opt in. + boolean wasLenient = reader.isLenient(); + reader.setLenient(true); + var schemaObj = JsonParser.parseReader(reader).getAsJsonObject(); + reader.setLenient(wasLenient); + + var commandFormat = schemaObj.has("commandFormat") + ? schemaObj.get("commandFormat").getAsString() : null; + var bufReader = new JsonReader(new StringReader(schemaObj.toString())); + + ResourceBindingJson schemaBinding; + if (commandFormat != null && customFormats.containsKey(commandFormat)) + { + var custom = customFormats.get(commandFormat).getCommandSchemaBinding(ctx, idEncoders, null); + if (!(custom instanceof ResourceBindingJson)) + throw ServiceErrors.unsupportedFormat(commandFormat); + schemaBinding = (ResourceBindingJson) custom; + } + else + { + // unchanged default: SWE-Common JSON via the JSON binding + schemaBinding = new CommandStreamSchemaBindingJson(ctx, idEncoders, bufReader); + } + + csInfo = schemaBinding.deserialize(bufReader); } else reader.skipValue(); diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamHandler.java index 5d7ed8bd2..888f2d410 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamHandler.java @@ -30,6 +30,7 @@ import org.sensorhub.impl.service.consys.ConSysApiSecurity; import org.sensorhub.impl.service.consys.ServiceErrors; import org.sensorhub.impl.service.consys.RestApiServlet.ResourcePermissions; +import org.sensorhub.impl.service.consys.obs.CustomObsFormat; import org.sensorhub.impl.service.consys.resource.RequestContext; import org.sensorhub.impl.service.consys.resource.ResourceBinding; import org.sensorhub.impl.service.consys.resource.ResourceFormat; @@ -46,14 +47,22 @@ public class CommandStreamHandler extends ResourceHandler customFormats; + + public CommandStreamHandler(HandlerContext ctx, ResourcePermissions permissions) + { + this(ctx, permissions, java.util.Collections.emptyMap()); + } + + + public CommandStreamHandler(HandlerContext ctx, ResourcePermissions permissions, Map customFormats) { super(ctx.getReadDb().getCommandStreamStore(), ctx.getCommandStreamIdEncoder(), ctx, permissions); this.db = ctx.getReadDb(); this.transactionHandler = new SystemDatabaseTransactionHandler(ctx.getEventBus(), ctx.getWriteDb()); - + this.customFormats = customFormats != null ? customFormats : java.util.Collections.emptyMap(); + this.eventsHandler = new CommandStreamEventsHandler(ctx, permissions); addSubResource(eventsHandler); } @@ -70,7 +79,7 @@ protected ResourceBinding getBinding(Reque return new CommandStreamBindingHtml(ctx, idEncoders, db, true, title); } else if (format.isOneOf(ResourceFormat.AUTO, ResourceFormat.JSON)) - return new CommandStreamBindingJson(ctx, idEncoders, db, forReading); + return new CommandStreamBindingJson(ctx, idEncoders, db, forReading, customFormats); else throw ServiceErrors.unsupportedFormat(format); } From 291c96f2ddd897bb74387a4709a7042a920e6508 Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 6 Jul 2026 01:08:42 -0500 Subject: [PATCH 09/13] feat(consys): add RequestContext constructor for one-shot GET responses from alternate transports --- .../consys/resource/RequestContext.java | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/resource/RequestContext.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/resource/RequestContext.java index 318e7c180..b8a4c8304 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/resource/RequestContext.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/resource/RequestContext.java @@ -171,6 +171,25 @@ public RequestContext(RestApiServlet servlet, URI resourceURI, InputStream is) } + /* + * Constructor for other protocols providing their own path/params and output stream, + * for one-shot (non-streaming) GET responses (e.g. NATS request-reply read) + */ + public RequestContext(RestApiServlet servlet, URI resourceURI, OutputStream os) + { + this.clientSide = false; + this.servlet = Asserts.checkNotNull(servlet, Servlet.class); + this.req = null; + this.resp = null; + this.requestPathInfo = null; + this.streamHandler = null; + this.inputStream = null; + this.outputStream = Asserts.checkNotNull(os, OutputStream.class); + this.path = splitPath(resourceURI.getPath()); + this.queryParams = parseQueryParams(resourceURI.getQuery()); + } + + /* * Constructor for other protocols providing their own path/params and stream handler (e.g. MQTT subscribe) */ From 6da18afb3c4974333cdd9624fec91b4b9308cfad Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Tue, 13 Oct 2026 11:23:54 -0500 Subject: [PATCH 10/13] [CSAPI] Split custom command formats into CustomCommandFormat --- .../impl/service/consys/ConSysApiService.java | 19 +++++-- .../service/consys/obs/CustomObsFormat.java | 29 ----------- .../service/consys/task/CommandHandler.java | 5 +- .../consys/task/CommandStreamBindingJson.java | 7 ++- .../consys/task/CommandStreamHandler.java | 5 +- .../task/CommandStreamSchemaHandler.java | 7 ++- .../consys/task/CustomCommandFormat.java | 50 +++++++++++++++++++ 7 files changed, 74 insertions(+), 48 deletions(-) create mode 100644 sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CustomCommandFormat.java diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java index 93ff7cbaa..1fda242e1 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/ConSysApiService.java @@ -51,6 +51,7 @@ import org.sensorhub.impl.service.consys.task.CommandStatusHandler; import org.sensorhub.impl.service.consys.task.CommandStreamHandler; import org.sensorhub.impl.service.consys.task.CommandStreamSchemaHandler; +import org.sensorhub.impl.service.consys.task.CustomCommandFormat; import org.sensorhub.utils.NamedThreadFactory; import com.google.common.base.Strings; import com.google.common.collect.ImmutableSet; @@ -166,15 +167,23 @@ protected void doStart() throws SensorHubException // load custom formats Map customFormats = new HashMap<>(); + Map customCmdFormats = new HashMap<>(); for (var formatConfig: config.customFormats) { try { // find impl for this mime type + // a single class can handle observations, commands or both ModuleRegistry moduleReg = getParentHub().getModuleRegistry(); - var clazz = moduleReg.findClass(formatConfig.className); + var clazz = moduleReg.findClass(formatConfig.className); var formatImpl = clazz.getDeclaredConstructor().newInstance(); - customFormats.put(formatConfig.mimeType, formatImpl); + if (!(formatImpl instanceof CustomObsFormat) && !(formatImpl instanceof CustomCommandFormat)) + throw new IllegalArgumentException(formatConfig.className + " must implement " + + CustomObsFormat.class.getSimpleName() + " or " + CustomCommandFormat.class.getSimpleName()); + if (formatImpl instanceof CustomObsFormat) + customFormats.put(formatConfig.mimeType, (CustomObsFormat)formatImpl); + if (formatImpl instanceof CustomCommandFormat) + customCmdFormats.put(formatConfig.mimeType, (CustomCommandFormat)formatImpl); getLogger().info("Loaded custom {} format implementation: {}", formatConfig.mimeType, formatConfig.className); } catch (Exception e) @@ -278,15 +287,15 @@ protected void doStart() throws SensorHubException //foiHandler.addSubResource(obsStatsHandler); // command streams - var cmdStreamHandler = new CommandStreamHandler(handlerCtx, security.commandstream_permissions, customFormats); + var cmdStreamHandler = new CommandStreamHandler(handlerCtx, security.commandstream_permissions, customCmdFormats); rootHandler.addSubResource(cmdStreamHandler); systemsHandler.addSubResource(cmdStreamHandler); sysMembersHandler.addSubResource(cmdStreamHandler); - var cmdSchemaHandler = new CommandStreamSchemaHandler(handlerCtx, security.commandstream_permissions, customFormats); + var cmdSchemaHandler = new CommandStreamSchemaHandler(handlerCtx, security.commandstream_permissions, customCmdFormats); cmdStreamHandler.addSubResource(cmdSchemaHandler); // commands - var cmdHandler = new CommandHandler(handlerCtx, threadPool, security.command_permissions, customFormats); + var cmdHandler = new CommandHandler(handlerCtx, threadPool, security.command_permissions, customCmdFormats); cmdStreamHandler.addSubResource(cmdHandler); // command status diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/CustomObsFormat.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/CustomObsFormat.java index ae5fcd376..fdf2f93f6 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/CustomObsFormat.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/CustomObsFormat.java @@ -15,13 +15,10 @@ package org.sensorhub.impl.service.consys.obs; import java.io.IOException; -import org.sensorhub.api.command.ICommandData; -import org.sensorhub.api.command.ICommandStreamInfo; import org.sensorhub.api.common.BigId; import org.sensorhub.api.common.IdEncoders; import org.sensorhub.api.data.IDataStreamInfo; import org.sensorhub.api.data.IObsData; -import org.sensorhub.api.datastore.command.CommandStreamKey; import org.sensorhub.api.datastore.obs.DataStreamKey; import org.sensorhub.impl.service.consys.resource.RequestContext; import org.sensorhub.impl.service.consys.resource.ResourceBinding; @@ -53,30 +50,4 @@ default boolean isAutoSelectable(IDataStreamInfo dsInfo) ResourceBinding getObsBinding(RequestContext ctx, IdEncoders idEncoders, IDataStreamInfo dsInfo) throws IOException; - - - /** - * @return a binding serving this format's view of a command stream's - * parameter schema, or null if this format does not support commands. - * Consulted by CommandStreamSchemaHandler when the commandFormat query - * param matches this format's mime type. - */ - default ResourceBinding getCommandSchemaBinding(RequestContext ctx, IdEncoders idEncoders, ICommandStreamInfo csInfo) throws IOException - { - return null; - } - - - /** - * @param forReading true when the binding will deserialize commands from - * the request body (POST/publish ingestion), false when it will - * serialize commands to the response (GET/stream) - * @return a binding encoding/decoding commands in this format, or null if - * this format does not support commands. Consulted by CommandHandler for - * both reads (GET/stream) and ingestion (POST with this content type). - */ - default ResourceBinding getCommandBinding(RequestContext ctx, IdEncoders idEncoders, ICommandStreamInfo csInfo, boolean forReading) throws IOException - { - return null; - } } diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java index e14c5153e..3c624b580 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java @@ -46,7 +46,6 @@ import org.sensorhub.impl.service.consys.HandlerContext; import org.sensorhub.impl.service.consys.ServiceErrors; import org.sensorhub.impl.service.consys.RestApiServlet.ResourcePermissions; -import org.sensorhub.impl.service.consys.obs.CustomObsFormat; import org.sensorhub.impl.service.consys.resource.BaseResourceHandler; import org.sensorhub.impl.service.consys.resource.IResourceHandler; import org.sensorhub.impl.service.consys.resource.RequestContext; @@ -69,7 +68,7 @@ public class CommandHandler extends BaseResourceHandler customFormats; + final Map customFormats; final Random random = new SecureRandom(); @@ -88,7 +87,7 @@ public CommandHandler(HandlerContext ctx, ScheduledExecutorService threadPool, R } - public CommandHandler(HandlerContext ctx, ScheduledExecutorService threadPool, ResourcePermissions permissions, Map customFormats) + public CommandHandler(HandlerContext ctx, ScheduledExecutorService threadPool, ResourcePermissions permissions, Map customFormats) { super(ctx.getReadDb().getCommandStore(), ctx.getCommandIdEncoder(), ctx, permissions); diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java index c1f494476..b14a41e72 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java @@ -31,7 +31,6 @@ import org.sensorhub.impl.service.consys.ResourceParseException; import org.sensorhub.impl.service.consys.SWECommonUtils; import org.sensorhub.impl.service.consys.ServiceErrors; -import org.sensorhub.impl.service.consys.obs.CustomObsFormat; import org.sensorhub.impl.service.consys.resource.RequestContext; import org.sensorhub.impl.service.consys.resource.ResourceBindingJson; import org.sensorhub.impl.service.consys.resource.ResourceFormat; @@ -53,7 +52,7 @@ public class CommandStreamBindingJson extends ResourceBindingJson customFormats; + final Map customFormats; SWEJsonStreamReader sweReader; SWEJsonStreamWriter sweWriter; @@ -64,7 +63,7 @@ public CommandStreamBindingJson(RequestContext ctx, IdEncoders idEncoders, IObsS } - public CommandStreamBindingJson(RequestContext ctx, IdEncoders idEncoders, IObsSystemDatabase db, boolean forReading, Map customFormats) throws IOException + public CommandStreamBindingJson(RequestContext ctx, IdEncoders idEncoders, IObsSystemDatabase db, boolean forReading, Map customFormats) throws IOException { super(ctx, idEncoders, forReading); @@ -134,7 +133,7 @@ else if ("schema".equals(prop)) ResourceBindingJson schemaBinding; if (commandFormat != null && customFormats.containsKey(commandFormat)) { - var custom = customFormats.get(commandFormat).getCommandSchemaBinding(ctx, idEncoders, null); + var custom = customFormats.get(commandFormat).getSchemaBinding(ctx, idEncoders, null); if (!(custom instanceof ResourceBindingJson)) throw ServiceErrors.unsupportedFormat(commandFormat); schemaBinding = (ResourceBindingJson) custom; diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamHandler.java index 888f2d410..5ee89fa6f 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamHandler.java @@ -30,7 +30,6 @@ import org.sensorhub.impl.service.consys.ConSysApiSecurity; import org.sensorhub.impl.service.consys.ServiceErrors; import org.sensorhub.impl.service.consys.RestApiServlet.ResourcePermissions; -import org.sensorhub.impl.service.consys.obs.CustomObsFormat; import org.sensorhub.impl.service.consys.resource.RequestContext; import org.sensorhub.impl.service.consys.resource.ResourceBinding; import org.sensorhub.impl.service.consys.resource.ResourceFormat; @@ -47,7 +46,7 @@ public class CommandStreamHandler extends ResourceHandler customFormats; + final Map customFormats; public CommandStreamHandler(HandlerContext ctx, ResourcePermissions permissions) @@ -56,7 +55,7 @@ public CommandStreamHandler(HandlerContext ctx, ResourcePermissions permissions) } - public CommandStreamHandler(HandlerContext ctx, ResourcePermissions permissions, Map customFormats) + public CommandStreamHandler(HandlerContext ctx, ResourcePermissions permissions, Map customFormats) { super(ctx.getReadDb().getCommandStreamStore(), ctx.getCommandStreamIdEncoder(), ctx, permissions); this.db = ctx.getReadDb(); diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamSchemaHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamSchemaHandler.java index 13cbdb7d4..da985c704 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamSchemaHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamSchemaHandler.java @@ -25,7 +25,6 @@ import org.sensorhub.impl.service.consys.HandlerContext; import org.sensorhub.impl.service.consys.ServiceErrors; import org.sensorhub.impl.service.consys.RestApiServlet.ResourcePermissions; -import org.sensorhub.impl.service.consys.obs.CustomObsFormat; import org.sensorhub.impl.service.consys.resource.RequestContext; import org.sensorhub.impl.service.consys.resource.ResourceBinding; import org.sensorhub.impl.service.consys.resource.ResourceFormat; @@ -38,7 +37,7 @@ public class CommandStreamSchemaHandler extends ResourceHandler customFormats; + final Map customFormats; public CommandStreamSchemaHandler(HandlerContext ctx, ResourcePermissions permissions) @@ -47,7 +46,7 @@ public CommandStreamSchemaHandler(HandlerContext ctx, ResourcePermissions permis } - public CommandStreamSchemaHandler(HandlerContext ctx, ResourcePermissions permissions, Map customFormats) + public CommandStreamSchemaHandler(HandlerContext ctx, ResourcePermissions permissions, Map customFormats) { super(ctx.getReadDb().getCommandStreamStore(), ctx.getCommandStreamIdEncoder(), ctx, permissions); this.customFormats = customFormats != null ? customFormats : java.util.Collections.emptyMap(); @@ -78,7 +77,7 @@ else if (customFormats.containsKey(cmdFormat.getMimeType())) var csInfo = dataStore.get(new CommandStreamKey(ctx.getParentRef().internalID)); if (csInfo == null) throw ServiceErrors.notFound(); - var binding = customFormats.get(cmdFormat.getMimeType()).getCommandSchemaBinding(ctx, idEncoders, csInfo); + var binding = customFormats.get(cmdFormat.getMimeType()).getSchemaBinding(ctx, idEncoders, csInfo); if (binding == null) throw ServiceErrors.unsupportedFormat(cmdFormat); return binding; diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CustomCommandFormat.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CustomCommandFormat.java new file mode 100644 index 000000000..d5ed0b2fe --- /dev/null +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CustomCommandFormat.java @@ -0,0 +1,50 @@ +/***************************** BEGIN LICENSE BLOCK *************************** + +The contents of this file are subject to the Mozilla Public License, v. 2.0. +If a copy of the MPL was not distributed with this file, You can obtain one +at http://mozilla.org/MPL/2.0/. + +Software distributed under the License is distributed on an "AS IS" basis, +WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License +for the specific language governing rights and limitations under the License. + +Copyright (C) 2026 Botts Innovative Research Inc. All Rights Reserved. + +******************************* END LICENSE BLOCK ***************************/ + +package org.sensorhub.impl.service.consys.task; + +import java.io.IOException; +import org.sensorhub.api.command.ICommandData; +import org.sensorhub.api.command.ICommandStreamInfo; +import org.sensorhub.api.common.BigId; +import org.sensorhub.api.common.IdEncoders; +import org.sensorhub.api.datastore.command.CommandStreamKey; +import org.sensorhub.impl.service.consys.resource.RequestContext; +import org.sensorhub.impl.service.consys.resource.ResourceBinding; + + +public interface CustomCommandFormat +{ + /** + * @return true if this format can encode commands of the given + * command stream. Drives the "formats" list advertised on the + * command stream resource. + */ + boolean isCompatible(ICommandStreamInfo csInfo); + + + /** + * @return a binding serving this format's view of the command stream + * schema, or null if not supported. Used when the commandFormat query + * param matches this format's mime type. + */ + ResourceBinding getSchemaBinding(RequestContext ctx, IdEncoders idEncoders, ICommandStreamInfo csInfo) throws IOException; + + + /** + * @param forReading true when the binding will read commands from the + * request body, false when it will write commands to the response + */ + ResourceBinding getCommandBinding(RequestContext ctx, IdEncoders idEncoders, ICommandStreamInfo csInfo, boolean forReading) throws IOException; +} From 353740385ebe0398363d713a60e119efca9a3e4d Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Tue, 13 Oct 2026 11:24:05 -0500 Subject: [PATCH 11/13] [CSAPI] Select custom command bindings like obs bindings --- .../service/consys/task/CommandHandler.java | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java index 3c624b580..da0dc7595 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java @@ -159,20 +159,29 @@ protected ResourceBinding getBinding(RequestContext ctx, bo } // select binding depending on format - // custom formats (e.g. application/swe+proto) take precedence when the - // requested mime type matches and the format supports commands - if (contextData.csInfo != null && customFormats.containsKey(format.getMimeType())) + // use a custom format if available for selected mime type + // do this only if commands from single command stream are requested + if (contextData.csInfo != null) { - var binding = customFormats.get(format.getMimeType()).getCommandBinding(ctx, idEncoders, contextData.csInfo, forReading); + var binding = getCustomFormatBinding(ctx, contextData.csInfo, forReading); if (binding != null) return binding; } + // otherwise use standard formats if (format.isOneOf(ResourceFormat.AUTO, ResourceFormat.JSON)) return new CommandBindingJson(ctx, idEncoders, forReading, dataStore); else return new CommandBindingSweCommon(ctx, idEncoders, forReading, dataStore); } + + + protected ResourceBinding getCustomFormatBinding(RequestContext ctx, ICommandStreamInfo csInfo, boolean forReading) throws IOException + { + var format = ctx.getFormat(); + var cmdFormat = customFormats.get(format.getMimeType()); + return cmdFormat != null ? cmdFormat.getCommandBinding(ctx, idEncoders, csInfo, forReading) : null; + } @Override From 2f559fac84bc31380f0babe32d4eb60d07acbc2a Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Tue, 13 Oct 2026 11:24:18 -0500 Subject: [PATCH 12/13] [CSAPI] List custom command formats on command streams --- .../impl/service/consys/SWECommonUtils.java | 31 +++++++++++++++++++ .../consys/task/CommandStreamBindingJson.java | 10 ++---- 2 files changed, 33 insertions(+), 8 deletions(-) diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/SWECommonUtils.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/SWECommonUtils.java index e5d2f4c8f..65c629733 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/SWECommonUtils.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/SWECommonUtils.java @@ -18,9 +18,11 @@ import java.util.Collection; import java.util.Map; import java.util.Set; +import org.sensorhub.api.command.ICommandStreamInfo; import org.sensorhub.api.data.IDataStreamInfo; import org.sensorhub.impl.service.consys.obs.CustomObsFormat; import org.sensorhub.impl.service.consys.resource.ResourceFormat; +import org.sensorhub.impl.service.consys.task.CustomCommandFormat; import org.vast.cdm.common.DataStreamParser; import org.vast.cdm.common.DataStreamWriter; import org.vast.data.DataIterator; @@ -231,6 +233,35 @@ public static Collection getAvailableFormats(IDataStreamInfo dsInfo, Map } + /* + * Get list of format available for a given command stream + */ + public static Collection getAvailableFormats(ICommandStreamInfo csInfo, Map customFormats) + { + var formatList = new ArrayList(); + + formatList.add(ResourceFormat.JSON.getMimeType()); + + if (ResourceFormat.allowNonBinaryFormat(csInfo.getRecordEncoding())) + { + formatList.add(ResourceFormat.SWE_JSON.getMimeType()); + formatList.add(ResourceFormat.SWE_TEXT.getMimeType()); + formatList.add(ResourceFormat.SWE_XML.getMimeType()); + } + + formatList.add(ResourceFormat.SWE_BINARY.getMimeType()); + + // also list compatible custom formats + for (var formatEntry: customFormats.entrySet()) + { + if (formatEntry.getValue().isCompatible(csInfo)) + formatList.add(formatEntry.getKey()); + } + + return formatList; + } + + public static Iterable getProperties(DataComponent dataStruct) { return Iterables.filter(new DataIterator(dataStruct), comp -> { diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java index b14a41e72..5479f9d2f 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStreamBindingJson.java @@ -308,14 +308,8 @@ public void serialize(CommandStreamKey key, ICommandStreamInfo csInfo, boolean s // available formats writer.name("formats").beginArray(); - writer.value(ResourceFormat.JSON.getMimeType()); - if (ResourceFormat.allowNonBinaryFormat(csInfo.getRecordEncoding())) - { - writer.value(ResourceFormat.SWE_JSON.getMimeType()); - writer.value(ResourceFormat.SWE_TEXT.getMimeType()); - writer.value(ResourceFormat.SWE_XML.getMimeType()); - } - writer.value(ResourceFormat.SWE_BINARY.getMimeType()); + for (var f: SWECommonUtils.getAvailableFormats(csInfo, customFormats)) + writer.value(f); writer.endArray(); } From c931a9c74230efe98d4d956b9a3d04ddb7995765 Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Tue, 13 Oct 2026 11:25:02 -0500 Subject: [PATCH 13/13] [CSAPI] Take the stream format from the query params only --- .../org/sensorhub/impl/service/consys/obs/ObsHandler.java | 8 ++------ .../impl/service/consys/task/CommandHandler.java | 6 +----- .../impl/service/consys/task/CommandStatusHandler.java | 6 +----- 3 files changed, 4 insertions(+), 16 deletions(-) diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java index bd4d2d38e..093bb7abd 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/ObsHandler.java @@ -244,13 +244,9 @@ protected void subscribe(final RequestContext ctx) throws InvalidRequestExceptio var queryParams = ctx.getParameterMap(); var filter = getFilter(ctx.getParentRef(), queryParams, 0, Long.MAX_VALUE); - // default to the format already set by the transport layer, if any - // (e.g. MQTT ":data/" subtopics); an explicit f= query - // param still takes precedence - var defaultFormat = ctx.getFormat() != null ? ctx.getFormat() : ResourceFormat.AUTO; - var responseFormat = parseFormat(queryParams, defaultFormat); + var responseFormat = parseFormat(queryParams); ctx.setFormatOptions(responseFormat, parseSelectArg(queryParams)); - + // detect if real-time request boolean isRealTime = (filter.getPhenomenonTime() == null && filter.getResultTime() == null) || (filter.getResultTime() != null && filter.getResultTime().beginsNow()) || diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java index da0dc7595..4b75a31f0 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandHandler.java @@ -207,11 +207,7 @@ protected void subscribe(final RequestContext ctx) throws InvalidRequestExceptio var queryParams = ctx.getParameterMap(); var filter = getFilter(ctx.getParentRef(), queryParams, 0, Long.MAX_VALUE); - // default to the format already set by the transport layer, if any - // (e.g. MQTT ":data/" subtopics); an explicit f= query - // param still takes precedence - var defaultFormat = ctx.getFormat() != null ? ctx.getFormat() : ResourceFormat.AUTO; - var responseFormat = parseFormat(queryParams, defaultFormat); + var responseFormat = parseFormat(queryParams); ctx.setFormatOptions(responseFormat, parseSelectArg(queryParams)); // continue when streaming actually starts diff --git a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStatusHandler.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStatusHandler.java index d55e81cf7..19b0f780d 100644 --- a/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStatusHandler.java +++ b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/task/CommandStatusHandler.java @@ -146,11 +146,7 @@ protected void subscribe(final RequestContext ctx) throws InvalidRequestExceptio var queryParams = ctx.getParameterMap(); var filter = getFilter(ctx.getParentRef(), queryParams, 0, Long.MAX_VALUE); - // default to the format already set by the transport layer, if any - // (e.g. MQTT ":data/" subtopics); an explicit f= query - // param still takes precedence - var defaultFormat = ctx.getFormat() != null ? ctx.getFormat() : ResourceFormat.AUTO; - var responseFormat = parseFormat(queryParams, defaultFormat); + var responseFormat = parseFormat(queryParams); ctx.setFormatOptions(responseFormat, parseSelectArg(queryParams)); // continue when streaming actually starts