Skip to content

Update command status serialization for osh-core changes and add inli… #125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mdhsl merged 1 commit into from
Oct 10, 2025
Merged

Update command status serialization for osh-core changes and add inli… #125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 38 additions & 7 deletions ...src/main/java/org/sensorhub/impl/datastore/postgis/command/PostgisCommandStatusStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,10 @@
import com.zaxxer.hikari.HikariDataSource;
import org.postgresql.util.PGobject;
import org.sensorhub.api.command.ICommandStatus;
import org.sensorhub.api.command.ICommandStreamInfo;
import org.sensorhub.api.common.BigId;
import org.sensorhub.api.datastore.DataStoreException;
import org.sensorhub.api.datastore.command.CommandStatusFilter;
import org.sensorhub.api.datastore.command.ICommandStatusStore;
import org.sensorhub.api.datastore.command.ICommandStore;
import org.sensorhub.api.datastore.command.ICommandStreamStore;
import org.sensorhub.api.datastore.command.*;
import org.sensorhub.api.datastore.system.ISystemDescStore;
import org.sensorhub.impl.datastore.postgis.PostgisStore;
import org.sensorhub.impl.datastore.postgis.utils.PostgisUtils;
Expand All @@ -40,6 +38,8 @@
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import static org.sensorhub.api.datastore.command.ICommandStatusStore.CommandStatusField.COMMAND_ID;


public class PostgisCommandStatusStore extends PostgisStore<QueryBuilderCommandStatusStore> implements ICommandStatusStore {
private static final Logger logger = LoggerFactory.getLogger(PostgisCommandStoreImpl.class);
Expand Down Expand Up @@ -93,7 +93,23 @@ public Stream<Entry<BigId, ICommandStatus>> selectEntries(CommandStatusFilter fi
private Entry<BigId, ICommandStatus> resultSetToEntry(ResultSet resultSet, Set<ICommandStatusStore.CommandStatusField> fields) {
try {
BigId id = BigId.fromLong(idScope, resultSet.getLong("id"));
ICommandStatus cmdStatus = SerializerUtils.readICommandStatusFromJson(resultSet.getString("data"));
boolean noFields = fields != null && !fields.isEmpty();
BigId commandStreamId = null;
if (!noFields || fields.contains(COMMAND_ID)) {
long commandIdAsLong = resultSet.getLong(String.valueOf(COMMAND_ID));
if (!resultSet.wasNull()) {
var commandId = BigId.fromLong(idScope, commandIdAsLong);
commandStreamId = commandStore.get(commandId).getCommandStreamID();
}
}

ICommandStatus cmdStatus = null;
if (commandStreamId != null) {
ICommandStreamInfo commandStreamInfo = commandStore.getCommandStreams().get(new CommandStreamKey(commandStreamId));
cmdStatus = SerializerUtils.readICommandStatusFromJson(resultSet.getString("data"), commandStreamInfo);
} else {
cmdStatus = SerializerUtils.readICommandStatusFromJson(resultSet.getString("data"));
}
return Map.entry(id, cmdStatus);
} catch (Exception ex) {
throw new RuntimeException("Cannot parse resultSet to CommandStatus",ex);
Expand All @@ -105,13 +121,28 @@ public void commit() throws DataStoreException {

}

protected ICommandStreamInfo getContext(ICommandStatus status) {
var cmdId = status.getCommandID();
var cmdStreamId = commandStore.get(cmdId).getCommandStreamID();
var cmdStreamKey = new CommandStreamKey(cmdStreamId);
return commandStore.getCommandStreams().get(cmdStreamKey);
}

@Override
public BigId add(ICommandStatus rec) {
try (Connection connection1 = hikariDataSource.getConnection()) {
try (PreparedStatement preparedStatement = connection1.prepareStatement(queryBuilder.insertCommandQuery(), Statement.RETURN_GENERATED_KEYS)) {

// insert Object
String objectAsJson = SerializerUtils.writeICommandStatusToJson(rec);
String objectAsJson;
// Use context to serialize inline records
if (rec.getResult() != null &&
rec.getResult().getInlineRecords() != null &&
!rec.getResult().getInlineRecords().isEmpty()) {
var context = getContext(rec);
objectAsJson = SerializerUtils.writeICommandStatusToJson(rec, context);
} else {
objectAsJson = SerializerUtils.writeICommandStatusToJson(rec);
}

PGobject jsonObject = new PGobject();
jsonObject.setType("json");
Expand Down
106 changes: 88 additions & 18 deletions ...ore-postgis/src/main/java/org/sensorhub/impl/datastore/postgis/utils/SerializerUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,7 +310,7 @@ public static String writeICommandStreamInfoToJson(ICommandStreamInfo commandStr
}

// CommandStatus
public static ICommandStatus readICommandStatusFromJson(String json) throws IOException {
public static ICommandStatus readICommandStatusFromJson(String json, ICommandStreamInfo csInfo) throws IOException {
StringReader stringReader = new StringReader(json);
JsonReader jsonReader = new JsonReader(stringReader);

Expand Down Expand Up @@ -339,7 +339,7 @@ public static ICommandStatus readICommandStatusFromJson(String json) throws IOEx
} else if ("reportTime".equals(prop)) {
commandStatusBuilder = commandStatusBuilder.withReportTime(PostgisUtils.readInstantFromString(jsonReader.nextString(), false));
} else if ("commandResult".equals(prop)) {
commandStatusBuilder = commandStatusBuilder.withResult(readICommandResult(jsonReader));
commandStatusBuilder = commandStatusBuilder.withResult(readICommandResult(jsonReader, csInfo));
} else {
jsonReader.skipValue();
}
Expand All @@ -352,7 +352,15 @@ public static ICommandStatus readICommandStatusFromJson(String json) throws IOEx
return commandStatusBuilder.build();
}

public static ICommandStatus readICommandStatusFromJson(String json) throws IOException {
return readICommandStatusFromJson(json, null);
}

public static String writeICommandStatusToJson(ICommandStatus commandStatus) throws IOException{
return writeICommandStatusToJson(commandStatus, null);
}

public static String writeICommandStatusToJson(ICommandStatus commandStatus, ICommandStreamInfo commandStreamInfo) throws IOException {
try (ByteArrayOutputStream os = new ByteArrayOutputStream()) {
try (var osw = new OutputStreamWriter(os, StandardCharsets.UTF_8)) {
try (JsonWriter jsonWriter = new JsonWriter(osw)) {
Expand Down Expand Up @@ -382,7 +390,7 @@ public static String writeICommandStatusToJson(ICommandStatus commandStatus) thr
}

if(commandStatus.getResult() != null) {
writeICommandResult(jsonWriter, "commandResult",commandStatus.getResult());
writeICommandResult(jsonWriter, "commandResult",commandStatus.getResult(), commandStreamInfo);
}

jsonWriter.endObject();
Expand All @@ -396,41 +404,103 @@ public static String writeICommandStatusToJson(ICommandStatus commandStatus) thr
}

protected static void writeICommandResult(JsonWriter jsonWriter, String nodeName, ICommandResult commandResult) throws IOException {
jsonWriter.beginObject();
writeICommandResult(jsonWriter, nodeName, commandResult, null);
}

protected static void writeICommandResult(JsonWriter jsonWriter, String nodeName, ICommandResult commandResult, ICommandStreamInfo commandStreamInfo) throws IOException {
jsonWriter.name(nodeName);
writeBigId(jsonWriter, "datastream@id",commandResult.getDataStreamID());
jsonWriter.beginObject();

jsonWriter.name("observationRefs");
jsonWriter.beginArray();
for(BigId obsId: commandResult.getObservationRefs()) {
writeBigId(jsonWriter, null, obsId);
if (commandResult.getDataStreamIDs() != null) {
jsonWriter.name("datastreamRefs");
jsonWriter.beginArray();
for (var dsId: commandResult.getDataStreamIDs()) {
jsonWriter.beginObject();
writeBigId(jsonWriter, "datastream@id", dsId);
jsonWriter.endObject();
}
jsonWriter.endArray();
}
jsonWriter.endArray();

if (commandResult.getObservationIDs() != null) {
jsonWriter.name("observationRefs");
jsonWriter.beginArray();
for (var obsId: commandResult.getObservationIDs()) {
jsonWriter.beginObject();
writeBigId(jsonWriter, "observation@id", obsId);
jsonWriter.endObject();
}
jsonWriter.endArray();
}

if (commandResult.getInlineRecords() != null && commandStreamInfo != null) {
jsonWriter.name("inlineRecords");
jsonWriter.beginArray();
for (var inlineRecord: commandResult.getInlineRecords()) {
jsonWriter.jsonValue(writeDataBlockToJson(commandStreamInfo.getResultStructure(), commandStreamInfo.getResultEncoding(), inlineRecord));

}
jsonWriter.endArray();
}

jsonWriter.endObject();
}

protected static ICommandResult readICommandResult(JsonReader jsonReader) throws IOException {
BigId dataStreamId = null;
List<BigId> obsRef = null;
protected static ICommandResult readICommandResult(JsonReader jsonReader, ICommandStreamInfo csInfo) throws IOException {
List<BigId> dataStreamIds = null;
List<BigId> obsIds = null;
List<DataBlock> inlineRecords = null;

jsonReader.beginObject();
while (jsonReader.hasNext()) {
switch (jsonReader.nextName()) {
case "datastream@id":
dataStreamId = readBigId(jsonReader);
case "datastreamRefs":
dataStreamIds = new ArrayList<>();
jsonReader.beginArray();
while(jsonReader.hasNext()) {
jsonReader.beginObject();
jsonReader.nextName();
dataStreamIds.add(readBigId(jsonReader));
jsonReader.endObject();
}
jsonReader.endArray();
break;
case "observationRefs":
obsRef = new ArrayList<>();
obsIds = new ArrayList<>();
jsonReader.beginArray();
while (jsonReader.hasNext()) {
obsRef.add(readBigId(jsonReader));
jsonReader.beginObject();
jsonReader.nextName();
obsIds.add(readBigId(jsonReader));
jsonReader.endObject();
}
jsonReader.endArray();
break;
case "inlineRecords":
if (csInfo == null) {
jsonReader.skipValue();
break;
}
JsonDataParserGson parser = new JsonDataParserGson(jsonReader);
inlineRecords = new ArrayList<>();
jsonReader.beginArray();
while (jsonReader.hasNext()) {
parser.setDataComponents(csInfo.getResultStructure());
inlineRecords.add(parser.parseNextBlock());
}
jsonReader.endArray();
}
}
jsonReader.endObject();
return CommandResult.withObsInDatastream(dataStreamId, obsRef);

if (dataStreamIds != null && !dataStreamIds.isEmpty()) {
return CommandResult.withDatastreams(dataStreamIds);
} else if (obsIds != null && !obsIds.isEmpty()) {
return CommandResult.withObservations(obsIds);
} else if (inlineRecords != null && !inlineRecords.isEmpty()) {
return CommandResult.withData(inlineRecords);
}
return null;
}
//
protected static void writeBigId(JsonWriter jsonWriter, String nodeName, BigId bigId) throws IOException {
Expand Down
103 changes: 101 additions & 2 deletions ...-postgis/src/test/java/org/sensorhub/impl/datastore/postgis/TestSerializeDeserialize.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,18 +23,20 @@
import net.opengis.gml.v32.impl.GMLFactory;
import net.opengis.sensorml.v20.AbstractProcess;
import net.opengis.swe.v20.DataBlock;
import net.opengis.swe.v20.DataComponent;
import net.opengis.swe.v20.DataType;
import org.junit.Before;
import org.junit.Test;
import org.sensorhub.api.command.CommandStreamInfo;
import org.sensorhub.api.command.ICommandStreamInfo;
import org.sensorhub.api.command.*;
import org.sensorhub.api.common.BigId;
import org.sensorhub.api.common.BigIdLong;
import org.sensorhub.api.data.DataStreamInfo;
import org.sensorhub.api.data.IDataStreamInfo;
import org.sensorhub.api.data.ObsData;
import org.sensorhub.api.feature.FeatureId;
import org.sensorhub.api.system.ISystemWithDesc;
import org.sensorhub.impl.datastore.DataStoreFiltersTypeAdapterFactory;
import org.sensorhub.impl.datastore.DataStoreUtils;
import org.sensorhub.impl.datastore.postgis.utils.BigIdTypeAdapter;
import org.sensorhub.impl.datastore.postgis.utils.DataStreamExclusionStrategy;
import org.sensorhub.impl.datastore.postgis.utils.SerializerUtils;
Expand All @@ -49,6 +51,7 @@
import org.vast.ogc.gml.IFeature;
import org.vast.sensorML.SMLHelper;
import org.vast.sensorML.SMLJsonBindings;
import org.vast.swe.SWEBuilders;
import org.vast.swe.SWEHelper;

import static org.junit.Assert.*;
Expand All @@ -63,6 +66,8 @@
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalUnit;
import java.util.Collection;
import java.util.List;

public class TestSerializeDeserialize {

Expand Down Expand Up @@ -585,4 +590,98 @@ public void testDeserializeIDataStream() throws IOException {

IDataStreamInfo dataStreamInfo = SerializerUtils.readIDataStreamInfoFromJson(data);
}

private Collection<BigId> createExpectedBigIds() {
return List.of(
new BigIdLong(2, 4),
new BigIdLong(1, 5),
new BigIdLong(3, 10),
new BigIdLong(5, 100)
);
}

@Test
public void testSerializeCommandResultWithDataStreamIds() throws IOException {
var expectedIds = createExpectedBigIds();
var cmdRes = CommandResult.withDatastreams(expectedIds);
var cmdStatus = new CommandStatus.Builder()
.withCommand(new BigIdLong(1, 1))
.withResult(cmdRes)
.build();
var json = SerializerUtils.writeICommandStatusToJson(cmdStatus);
var fromJson = SerializerUtils.readICommandStatusFromJson(json);
assertTrue(fromJson.getResult().getDataStreamIDs().containsAll(expectedIds));
}

@Test
public void testSerializeCommandResultWithObservationIds() throws IOException {
var expectedIds = createExpectedBigIds();
var cmdRes = CommandResult.withObservations(expectedIds);
var cmdStatus = new CommandStatus.Builder()
.withCommand(new BigIdLong(1, 2))
.withResult(cmdRes)
.build();
var json = SerializerUtils.writeICommandStatusToJson(cmdStatus);
var fromJson = SerializerUtils.readICommandStatusFromJson(json);
assertTrue(fromJson.getResult().getObservationIDs().containsAll(expectedIds));

var one = expectedIds.stream().toList().get(0);
cmdRes = CommandResult.withObservation(one);
cmdStatus = new CommandStatus.Builder()
.withCommand(new BigIdLong(1, 3))
.withResult(cmdRes)
.build();
json = SerializerUtils.writeICommandStatusToJson(cmdStatus);
fromJson = SerializerUtils.readICommandStatusFromJson(json);
assertTrue(fromJson.getResult().getObservationIDs().size() == 1 && fromJson.getResult().getObservationIDs().contains(one));
}

private DataComponent createTestStructure() {
SWEHelper fac = new SWEHelper();
return fac.createRecord()
.name("testIO")
.addField("test1", fac.createText())
.addField("test2", fac.createCount())
.build();
}

@Test
public void testSerializeCommandResultWithInlineRecords() throws IOException {
var struct = createTestStructure();
var cmdStream = new CommandStreamInfo.Builder()
.withName("test")
.withRecordDescription(struct.copy())
.withRecordEncoding(new TextEncodingImpl())
.withResultDescription(struct.copy())
.withResultEncoding(new TextEncodingImpl())
.withSystem(new FeatureId(new BigIdLong(1, 5), "urn:osh:test:1"))
.build();

var data = struct.createDataBlock();
var d1 = "I am a string";
int d2 = 555;
data.setStringValue(0, d1);
data.setIntValue(1, d2);

var cmdRes = CommandResult.withData(data);
var cmdStatus = new CommandStatus.Builder()
.withCommand(new BigIdLong(1, 5))
.withResult(cmdRes)
.build();
var json = SerializerUtils.writeICommandStatusToJson(cmdStatus, cmdStream);
var fromJson = SerializerUtils.readICommandStatusFromJson(json, cmdStream);
var resRecords = fromJson.getResult().getInlineRecords().stream().toList();
assertFalse(resRecords.isEmpty());
assertEquals(d1, resRecords.get(0).getStringValue(0));
assertEquals(d2, resRecords.get(0).getIntValue(1));
}

@Test
public void testSerializeCommandStatus() throws IOException {
var cmdStatus = CommandStatus.accepted(new BigIdLong(1, 12));
var json = SerializerUtils.writeICommandStatusToJson(cmdStatus);
var fromJson = SerializerUtils.readICommandStatusFromJson(json);
assertEquals(ICommandStatus.CommandStatusCode.ACCEPTED, fromJson.getStatusCode());
}

}