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()); + } } 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..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 @@ -383,6 +383,66 @@ 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 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/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..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 @@ -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; } @@ -303,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(); @@ -908,7 +916,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 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..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) @@ -262,7 +271,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 @@ -278,15 +287,15 @@ 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, customCmdFormats); 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, customCmdFormats); cmdStreamHandler.addSubResource(cmdSchemaHandler); - + // commands - var cmdHandler = new CommandHandler(handlerCtx, threadPool, security.command_permissions); + 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/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/obs/CustomObsFormat.java b/sensorhub-service-consys/src/main/java/org/sensorhub/impl/service/consys/obs/CustomObsFormat.java index 8423a3b28..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 @@ -26,11 +26,28 @@ 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; } 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/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 a696407e1..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 @@ -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)); @@ -169,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); @@ -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/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 customFormats; final Random random = new SecureRandom(); - - + + public static class CommandHandlerContextData { public BigId streamID; @@ -78,16 +79,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,11 +159,29 @@ protected ResourceBinding getBinding(RequestContext ctx, bo } // select binding depending on format + // 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 = 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 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..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 @@ -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,7 @@ 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.resource.RequestContext; import org.sensorhub.impl.service.consys.resource.ResourceBindingJson; import org.sensorhub.impl.service.consys.resource.ResourceFormat; @@ -37,6 +41,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 +52,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 +116,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).getSchemaBinding(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(); @@ -281,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(); } 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..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 @@ -46,14 +46,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 +78,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); } 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..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 @@ -36,30 +36,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()).getSchemaBinding(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 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; +} 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..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; } @@ -141,8 +145,78 @@ public void testAddObsAndDeleteDatastream() throws Exception } - // Non-Test helper methods + @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); + } + + + @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 { return addObservation(dsUrl, startTime, 1000L, num);