diff --git a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/BooleanValueMapper.java b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/BooleanValueMapper.java index 4abbafe60e..f172b9c04a 100644 --- a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/BooleanValueMapper.java +++ b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/BooleanValueMapper.java @@ -1,10 +1,12 @@ package gov.nasa.jpl.aerie.contrib.serialization.mappers; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.framework.Result; import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; import java.util.function.Function; public final class BooleanValueMapper implements ValueMapper { @@ -25,4 +27,9 @@ public Result deserializeValue(final SerializedValue serialized public SerializedValue serializeValue(final Boolean value) { return SerializedValue.of(value); } + + @Override + public void writeJson(final Boolean value, final JsonGenerator gen) throws IOException { + gen.writeBoolean(value); + } } diff --git a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/DoubleValueMapper.java b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/DoubleValueMapper.java index 3a3795feb9..5a4815c033 100644 --- a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/DoubleValueMapper.java +++ b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/DoubleValueMapper.java @@ -1,10 +1,13 @@ package gov.nasa.jpl.aerie.contrib.serialization.mappers; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.framework.Result; import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; +import java.math.BigDecimal; import java.util.function.Function; public final class DoubleValueMapper implements ValueMapper { @@ -25,4 +28,9 @@ public Result deserializeValue(final SerializedValue serializedV public SerializedValue serializeValue(final Double value) { return SerializedValue.of(value); } + + @Override + public void writeJson(final Double value, final JsonGenerator gen) throws IOException { + gen.writeNumber(BigDecimal.valueOf(value)); + } } diff --git a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/IntegerValueMapper.java b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/IntegerValueMapper.java index 4201ae93ae..0bb5ad52c8 100644 --- a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/IntegerValueMapper.java +++ b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/IntegerValueMapper.java @@ -1,10 +1,12 @@ package gov.nasa.jpl.aerie.contrib.serialization.mappers; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.framework.Result; import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; import java.util.function.Function; public final class IntegerValueMapper implements ValueMapper { @@ -36,4 +38,9 @@ public Result deserializeValue(final SerializedValue serialized public SerializedValue serializeValue(final Integer value) { return SerializedValue.of(value); } + + @Override + public void writeJson(final Integer value, final JsonGenerator gen) throws IOException { + gen.writeNumber(value); + } } diff --git a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/ListValueMapper.java b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/ListValueMapper.java index 5129cd01cc..3acc97f640 100644 --- a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/ListValueMapper.java +++ b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/ListValueMapper.java @@ -1,10 +1,12 @@ package gov.nasa.jpl.aerie.contrib.serialization.mappers; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.framework.Result; import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.function.Function; @@ -51,4 +53,13 @@ public SerializedValue serializeValue(final List elements) { } return SerializedValue.of(serializedElements); } + + @Override + public void writeJson(final List elements, final JsonGenerator gen) throws IOException { + gen.writeStartArray(); + for (final var element : elements) { + this.elementMapper.writeJson(element, gen); + } + gen.writeEndArray(); + } } diff --git a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/LongValueMapper.java b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/LongValueMapper.java index caecf276bb..2aded52966 100644 --- a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/LongValueMapper.java +++ b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/LongValueMapper.java @@ -1,10 +1,12 @@ package gov.nasa.jpl.aerie.contrib.serialization.mappers; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.framework.Result; import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; import java.util.function.Function; public final class LongValueMapper implements ValueMapper { @@ -25,4 +27,9 @@ public Result deserializeValue(final SerializedValue serializedVal public SerializedValue serializeValue(final Long value) { return SerializedValue.of(value); } + + @Override + public void writeJson(final Long value, final JsonGenerator gen) throws IOException { + gen.writeNumber(value); + } } diff --git a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/MapValueMapper.java b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/MapValueMapper.java index 64ddf58bf7..8bfba960e6 100644 --- a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/MapValueMapper.java +++ b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/MapValueMapper.java @@ -1,10 +1,12 @@ package gov.nasa.jpl.aerie.contrib.serialization.mappers; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.framework.Result; import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -75,4 +77,18 @@ public SerializedValue serializeValue(final Map fields) { } return SerializedValue.of(elementSpecs); } + + @Override + public void writeJson(final Map fields, final JsonGenerator gen) throws IOException { + gen.writeStartArray(); + for (final var entry : fields.entrySet()) { + gen.writeStartObject(); + gen.writeFieldName("key"); + keyMapper.writeJson(entry.getKey(), gen); + gen.writeFieldName("value"); + elementMapper.writeJson(entry.getValue(), gen); + gen.writeEndObject(); + } + gen.writeEndArray(); + } } diff --git a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/RecordValueMapper.java b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/RecordValueMapper.java index c529300f2d..956412d13b 100644 --- a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/RecordValueMapper.java +++ b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/RecordValueMapper.java @@ -1,10 +1,12 @@ package gov.nasa.jpl.aerie.contrib.serialization.mappers; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.framework.Result; import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.RecordComponent; @@ -86,6 +88,20 @@ public SerializedValue serializeValue(final R value) { return SerializedValue.of(map); } + @Override + public void writeJson(final R value, final JsonGenerator gen) throws IOException { + gen.writeStartObject(); + for (final var component : components) { + gen.writeFieldName(component.name); + writeJsonHelper(value, component, gen); + } + gen.writeEndObject(); + } + + private void writeJsonHelper(final R value, final Component component, final JsonGenerator gen) throws IOException { + component.mapper.writeJson(component.projection.apply(value), gen); + } + private SerializedValue serializeHelper(final R value, final Component component) { return component.mapper.serializeValue(component.projection.apply(value)); } diff --git a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/StringValueMapper.java b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/StringValueMapper.java index f895cc62f0..d48cb9e54c 100644 --- a/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/StringValueMapper.java +++ b/contrib/src/main/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/StringValueMapper.java @@ -1,10 +1,12 @@ package gov.nasa.jpl.aerie.contrib.serialization.mappers; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.framework.Result; import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; import java.util.function.Function; public final class StringValueMapper implements ValueMapper { @@ -25,4 +27,9 @@ public Result deserializeValue(final SerializedValue serializedV public SerializedValue serializeValue(final String value) { return SerializedValue.of(value); } + + @Override + public void writeJson(final String value, final JsonGenerator gen) throws IOException { + gen.writeString(value); + } } diff --git a/contrib/src/test/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/WriteJsonTest.java b/contrib/src/test/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/WriteJsonTest.java new file mode 100644 index 0000000000..8822a46559 --- /dev/null +++ b/contrib/src/test/java/gov/nasa/jpl/aerie/contrib/serialization/mappers/WriteJsonTest.java @@ -0,0 +1,299 @@ +package gov.nasa.jpl.aerie.contrib.serialization.mappers; + +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; +import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.StringWriter; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * For each mapper, verifies that writeJson output matches serializeValue -> writeTo output. + * This is the key correctness invariant: the optimized streaming path must produce + * identical JSON to the legacy serialize-then-stream path. + */ +class WriteJsonTest { + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + + private static String viaWriteJson(ValueMapper mapper, T value) throws IOException { + final var sw = new StringWriter(); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + mapper.writeJson(value, gen); + } + return sw.toString(); + } + + private static String viaSerializeValue(ValueMapper mapper, T value) throws IOException { + return mapper.serializeValue(value).toJsonString(); + } + + private static void assertWriteJsonMatchesSerialize(ValueMapper mapper, T value) throws IOException { + final var fromWriteJson = viaWriteJson(mapper, value); + final var fromSerialize = viaSerializeValue(mapper, value); + // Compare semantically: parse both JSON strings back to SerializedValue and check equality. + // This handles legitimate field ordering differences (HashMap vs component order). + final var parsedWriteJson = parseJson(fromWriteJson); + final var parsedSerialize = parseJson(fromSerialize); + assertEquals(parsedSerialize, parsedWriteJson, + "writeJson output must be semantically equal to serializeValue->toJsonString for " + value + + "\n writeJson: " + fromWriteJson + + "\n serialize: " + fromSerialize); + } + + private static SerializedValue parseJson(String json) throws IOException { + try (final var parser = JSON_FACTORY.createParser(json)) { + return parseValue(parser); + } + } + + private static SerializedValue parseValue(com.fasterxml.jackson.core.JsonParser parser) throws IOException { + final var token = parser.nextToken(); + return switch (token) { + case VALUE_NULL -> SerializedValue.NULL; + case VALUE_TRUE -> SerializedValue.of(true); + case VALUE_FALSE -> SerializedValue.of(false); + case VALUE_NUMBER_INT -> SerializedValue.of(parser.getBigIntegerValue().longValueExact()); + case VALUE_NUMBER_FLOAT -> SerializedValue.of(parser.getDecimalValue()); + case VALUE_STRING -> SerializedValue.of(parser.getText()); + case START_ARRAY -> { + final var list = new java.util.ArrayList(); + while (parser.nextToken() != com.fasterxml.jackson.core.JsonToken.END_ARRAY) { + list.add(parseCurrentValue(parser)); + } + yield SerializedValue.of(list); + } + case START_OBJECT -> { + final var map = new java.util.HashMap(); + while (parser.nextToken() != com.fasterxml.jackson.core.JsonToken.END_OBJECT) { + final var key = parser.getCurrentName(); + parser.nextToken(); + map.put(key, parseCurrentValue(parser)); + } + yield SerializedValue.of(map); + } + default -> throw new IOException("Unexpected token: " + token); + }; + } + + private static SerializedValue parseCurrentValue(com.fasterxml.jackson.core.JsonParser parser) throws IOException { + final var token = parser.currentToken(); + return switch (token) { + case VALUE_NULL -> SerializedValue.NULL; + case VALUE_TRUE -> SerializedValue.of(true); + case VALUE_FALSE -> SerializedValue.of(false); + case VALUE_NUMBER_INT -> SerializedValue.of(parser.getBigIntegerValue().longValueExact()); + case VALUE_NUMBER_FLOAT -> SerializedValue.of(parser.getDecimalValue()); + case VALUE_STRING -> SerializedValue.of(parser.getText()); + case START_ARRAY -> { + final var list = new java.util.ArrayList(); + while (parser.nextToken() != com.fasterxml.jackson.core.JsonToken.END_ARRAY) { + list.add(parseCurrentValue(parser)); + } + yield SerializedValue.of(list); + } + case START_OBJECT -> { + final var map = new java.util.HashMap(); + while (parser.nextToken() != com.fasterxml.jackson.core.JsonToken.END_OBJECT) { + final var key = parser.getCurrentName(); + parser.nextToken(); + map.put(key, parseCurrentValue(parser)); + } + yield SerializedValue.of(map); + } + default -> throw new IOException("Unexpected token: " + token); + }; + } + + // --- Primitive mappers --- + + @Test + void integerMapper() throws IOException { + final var mapper = new IntegerValueMapper(); + assertWriteJsonMatchesSerialize(mapper, 0); + assertWriteJsonMatchesSerialize(mapper, 42); + assertWriteJsonMatchesSerialize(mapper, -100); + assertWriteJsonMatchesSerialize(mapper, Integer.MAX_VALUE); + assertWriteJsonMatchesSerialize(mapper, Integer.MIN_VALUE); + } + + @Test + void longMapper() throws IOException { + final var mapper = new LongValueMapper(); + assertWriteJsonMatchesSerialize(mapper, 0L); + assertWriteJsonMatchesSerialize(mapper, 999999999999L); + assertWriteJsonMatchesSerialize(mapper, -1L); + assertWriteJsonMatchesSerialize(mapper, Long.MAX_VALUE); + } + + @Test + void doubleMapper() throws IOException { + final var mapper = new DoubleValueMapper(); + assertWriteJsonMatchesSerialize(mapper, 0.0); + assertWriteJsonMatchesSerialize(mapper, 3.14159265358979); + assertWriteJsonMatchesSerialize(mapper, -1.0e10); + assertWriteJsonMatchesSerialize(mapper, Double.MAX_VALUE); + } + + @Test + void booleanMapper() throws IOException { + final var mapper = new BooleanValueMapper(); + assertWriteJsonMatchesSerialize(mapper, true); + assertWriteJsonMatchesSerialize(mapper, false); + } + + @Test + void stringMapper() throws IOException { + final var mapper = new StringValueMapper(); + assertWriteJsonMatchesSerialize(mapper, ""); + assertWriteJsonMatchesSerialize(mapper, "hello"); + assertWriteJsonMatchesSerialize(mapper, "line1\nline2\ttab\"quote\\backslash"); + assertWriteJsonMatchesSerialize(mapper, "unicode: \u00e9\u00e8\u00ea"); + } + + // --- Collection mappers --- + + @Test + void listMapper_empty() throws IOException { + final var mapper = new ListValueMapper<>(new IntegerValueMapper()); + assertWriteJsonMatchesSerialize(mapper, List.of()); + } + + @Test + void listMapper_integers() throws IOException { + final var mapper = new ListValueMapper<>(new IntegerValueMapper()); + assertWriteJsonMatchesSerialize(mapper, List.of(1, 2, 3, 4, 5)); + } + + @Test + void listMapper_strings() throws IOException { + final var mapper = new ListValueMapper<>(new StringValueMapper()); + assertWriteJsonMatchesSerialize(mapper, List.of("a", "b", "c")); + } + + @Test + void listMapper_nested() throws IOException { + final var mapper = new ListValueMapper<>(new ListValueMapper<>(new BooleanValueMapper())); + assertWriteJsonMatchesSerialize(mapper, List.of( + List.of(true, false), + List.of(false, true, true) + )); + } + + // --- Record mappers --- + + record Point(double x, double y) {} + record Labeled(String label, Point point) {} + record Config(String name, int count, boolean enabled, List weights) {} + + private static RecordValueMapper pointMapper() { + return new RecordValueMapper<>(Point.class, List.of( + new RecordValueMapper.Component<>("x", Point::x, new DoubleValueMapper()), + new RecordValueMapper.Component<>("y", Point::y, new DoubleValueMapper()) + )); + } + + private static RecordValueMapper labeledMapper() { + return new RecordValueMapper<>(Labeled.class, List.of( + new RecordValueMapper.Component<>("label", Labeled::label, new StringValueMapper()), + new RecordValueMapper.Component<>("point", Labeled::point, pointMapper()) + )); + } + + private static RecordValueMapper configMapper() { + return new RecordValueMapper<>(Config.class, List.of( + new RecordValueMapper.Component<>("name", Config::name, new StringValueMapper()), + new RecordValueMapper.Component<>("count", Config::count, new IntegerValueMapper()), + new RecordValueMapper.Component<>("enabled", Config::enabled, new BooleanValueMapper()), + new RecordValueMapper.Component<>("weights", Config::weights, new ListValueMapper<>(new DoubleValueMapper())) + )); + } + + @Test + void recordMapper_simplePoint() throws IOException { + assertWriteJsonMatchesSerialize(pointMapper(), new Point(1.0, 2.5)); + } + + @Test + void recordMapper_nestedRecord() throws IOException { + assertWriteJsonMatchesSerialize( + labeledMapper(), + new Labeled("origin", new Point(0.0, 0.0))); + } + + @Test + void recordMapper_withCollectionField() throws IOException { + assertWriteJsonMatchesSerialize( + configMapper(), + new Config("test", 5, true, List.of(0.1, 0.2, 0.7))); + } + + // --- Map mappers --- + + @Test + void mapMapper_stringToInt() throws IOException { + final var mapper = new MapValueMapper<>(new StringValueMapper(), new IntegerValueMapper()); + // Single entry for deterministic ordering + assertWriteJsonMatchesSerialize(mapper, Map.of("one", 1)); + } + + @Test + void mapMapper_empty() throws IOException { + final var mapper = new MapValueMapper<>(new StringValueMapper(), new IntegerValueMapper()); + assertWriteJsonMatchesSerialize(mapper, Map.of()); + } + + // --- Stress: large data volumes --- + + @Test + void listMapper_largeList() throws IOException { + final var mapper = new ListValueMapper<>(new DoubleValueMapper()); + // 10,000 element list + final var list = new java.util.ArrayList(10_000); + for (int i = 0; i < 10_000; i++) list.add(i * 0.001); + assertWriteJsonMatchesSerialize(mapper, list); + } + + @Test + void recordMapper_repeated() throws IOException { + // Serialize same record 1000 times — simulates per-step overhead + final var mapper = configMapper(); + final var value = new Config("sensor_data", 42, true, List.of(1.0, 2.0, 3.0, 4.0, 5.0)); + + for (int i = 0; i < 1000; i++) { + assertWriteJsonMatchesSerialize(mapper, value); + } + } + + // --- Default fallback behavior --- + + @Test + void defaultWriteJson_matchesSerializeValue() throws IOException { + // A mapper that does NOT override writeJson should still produce correct output via the default + final var fallbackMapper = new ValueMapper() { + @Override + public gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema getValueSchema() { + return gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema.STRING; + } + @Override + public gov.nasa.jpl.aerie.merlin.framework.Result deserializeValue(SerializedValue serializedValue) { + return serializedValue.asString() + .map(gov.nasa.jpl.aerie.merlin.framework.Result::success) + .orElseGet(() -> gov.nasa.jpl.aerie.merlin.framework.Result.failure("not a string")); + } + @Override + public SerializedValue serializeValue(String value) { + return SerializedValue.of(value); + } + // NOTE: No writeJson override — uses default from ValueMapper interface + }; + + assertWriteJsonMatchesSerialize(fallbackMapper, "testing default path"); + } +} diff --git a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/engine/SimulationEngine.java b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/engine/SimulationEngine.java index 44b575f8c1..95aa1b4320 100644 --- a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/engine/SimulationEngine.java +++ b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/engine/SimulationEngine.java @@ -1,6 +1,7 @@ package gov.nasa.jpl.aerie.merlin.driver.engine; import gov.nasa.jpl.aerie.merlin.driver.MissionModel.SerializableTopic; +import gov.nasa.jpl.aerie.merlin.driver.resources.DeferredSerializedValue; import gov.nasa.jpl.aerie.types.ActivityInstance; import gov.nasa.jpl.aerie.types.ActivityInstanceId; import gov.nasa.jpl.aerie.merlin.driver.resources.SimulationResourceManager; @@ -41,6 +42,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.IdentityHashMap; import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; @@ -181,7 +183,7 @@ record AtDuration() implements Status{} record Nominal( Duration elapsedTime, Map> realResourceUpdates, - Map> dynamicResourceUpdates + Map> dynamicResourceUpdates ) implements Status {} } @@ -217,9 +219,9 @@ public Status step(Duration simulationDuration) throws Throwable { throw results.error.get(); } - // Serialize the resources updated in this batch + // Collect the resources updated in this batch — discrete values are deferred (not serialized yet) final var realResourceUpdates = new HashMap>(); - final var dynamicResourceUpdates = new HashMap>(); + final var dynamicResourceUpdates = new HashMap>(); for (final var update : results.resourceUpdates.updates()) { final var name = update.resourceId().id(); @@ -231,7 +233,7 @@ public Status step(Duration simulationDuration) throws Throwable { name, Pair.of( schema, - SimulationEngine.extractDiscreteDynamics(update))); + SimulationEngine.extractDeferredDiscreteDynamics(update))); } } @@ -249,8 +251,8 @@ private static RealDynamics extractRealDynamics(final ResourceUpdates return RealDynamics.linear(initial, rate); } - private static SerializedValue extractDiscreteDynamics(final ResourceUpdates.ResourceUpdate update) { - return update.resource.getOutputType().serialize(update.update.dynamics()); + private static DeferredSerializedValue extractDeferredDiscreteDynamics(final ResourceUpdates.ResourceUpdate update) { + return new DeferredSerializedValue(update.update.dynamics(), update.resource.getOutputType()); } /** Schedule a new task to be performed at the given time. */ @@ -605,9 +607,23 @@ public ActivityDirectiveId getDirective(SpanId id) { return this.spanToPlannedDirective.get(id); } - public record Trait(Iterable> topics, Topic activityTopic) - implements EffectTrait> + public record Trait( + Map, SerializableTopic> inputTopics, + Map, SerializableTopic> outputTopics, + Topic activityTopic + ) implements EffectTrait> { + public Trait(final Iterable> topics, final Topic activityTopic) { + this(new IdentityHashMap<>(), new IdentityHashMap<>(), activityTopic); + for (final var topic : topics) { + if (topic.name().startsWith("ActivityType.Input.")) { + this.inputTopics.put(topic.topic(), topic); + } else if (topic.name().startsWith("ActivityType.Output.")) { + this.outputTopics.put(topic.topic(), topic); + } + } + } + @Override public Consumer empty() { return spanInfo -> {}; @@ -637,41 +653,41 @@ public Consumer concurrently(final Consumer left, final Cons public Consumer atom(final Event ev) { return spanInfo -> { + final var topic = ev.topic(); + // Identify activities. - ev.extract(this.activityTopic) - .ifPresent(directiveId -> spanInfo.spanToPlannedDirective.put(ev.provenance(), directiveId)); + if (topic == this.activityTopic) { + ev.extract(this.activityTopic) + .ifPresent(directiveId -> spanInfo.spanToPlannedDirective.put(ev.provenance(), directiveId)); + } - for (final var topic : this.topics) { - // Identify activity inputs. - extractInput(topic, ev, spanInfo); + // Identify activity inputs and outputs. + final var inputTopic = this.inputTopics.get(topic); + if (inputTopic != null) { + extractHelper(inputTopic, ev, spanInfo); + } - // Identify activity outputs. - extractOutput(topic, ev, spanInfo); + final var outputTopic = this.outputTopics.get(topic); + if (outputTopic != null) { + extractOutputHelper(outputTopic, ev, spanInfo); } }; } - private static - void extractInput(final SerializableTopic topic, final Event ev, final SpanInfo spanInfo) { - if (!topic.name().startsWith("ActivityType.Input.")) return; - - ev.extract(topic.topic()).ifPresent(input -> { - final var activityType = topic.name().substring("ActivityType.Input.".length()); - + private static void extractHelper(SerializableTopic inputTopic, Event ev, SpanInfo spanInfo) { + ev.extract(inputTopic.topic()).ifPresent(input -> { + final var activityType = inputTopic.name().substring("ActivityType.Input.".length()); spanInfo.input.put( ev.provenance(), - new SerializedActivity(activityType, topic.outputType().serialize(input).asMap().orElseThrow())); + new SerializedActivity(activityType, inputTopic.outputType().serialize(input).asMap().orElseThrow())); }); } - private static - void extractOutput(final SerializableTopic topic, final Event ev, final SpanInfo spanInfo) { - if (!topic.name().startsWith("ActivityType.Output.")) return; - - ev.extract(topic.topic()).ifPresent(output -> { + private static void extractOutputHelper(SerializableTopic outputTopic, Event ev, SpanInfo spanInfo) { + ev.extract(outputTopic.topic()).ifPresent(output -> { spanInfo.output.put( ev.provenance(), - topic.outputType().serialize(output)); + outputTopic.outputType().serialize(output)); }); } } diff --git a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/json/JsonEncoding.java b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/json/JsonEncoding.java index 4aee81f4a8..a06f2f3348 100644 --- a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/json/JsonEncoding.java +++ b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/json/JsonEncoding.java @@ -1,12 +1,18 @@ package gov.nasa.jpl.aerie.merlin.driver.json; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import javax.json.JsonValue; +import java.io.IOException; +import java.io.StringWriter; import static gov.nasa.jpl.aerie.merlin.driver.json.SerializedValueJsonParser.serializedValueP; public final class JsonEncoding { + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + public static JsonValue encode(final SerializedValue value) { return serializedValueP.unparse(value); } @@ -16,4 +22,20 @@ public static SerializedValue decode(final JsonValue value) { .parse(value) .getSuccessOrThrow($ -> new Error("Unable to parse JSON as SerializedValue: " + $)); } + + public static void writeToGenerator(final SerializedValue value, final JsonGenerator gen) throws IOException { + value.writeTo(gen); + } + + public static String writeToString(final SerializedValue value) throws IOException { + return value.toJsonString(); + } + + public static JsonGenerator createGenerator(final StringWriter writer) throws IOException { + return JSON_FACTORY.createGenerator(writer); + } + + public static JsonFactory getJsonFactory() { + return JSON_FACTORY; + } } diff --git a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/json/SerializedValueJsonParser.java b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/json/SerializedValueJsonParser.java index 82e3a70e2f..9bb818ac17 100644 --- a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/json/SerializedValueJsonParser.java +++ b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/json/SerializedValueJsonParser.java @@ -70,6 +70,11 @@ public JsonValue onNumeric(final BigDecimal value) { return Json.createValue(value); } + @Override + public JsonValue onDouble(final double value) { + return Json.createValue(value); + } + @Override public JsonValue onString(final String value) { return Json.createValue(value); diff --git a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/DeferredSerializedValue.java b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/DeferredSerializedValue.java new file mode 100644 index 0000000000..9d7a56e8f0 --- /dev/null +++ b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/DeferredSerializedValue.java @@ -0,0 +1,37 @@ +package gov.nasa.jpl.aerie.merlin.driver.resources; + +import com.fasterxml.jackson.core.JsonGenerator; +import gov.nasa.jpl.aerie.merlin.protocol.model.OutputType; +import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; + +import java.io.IOException; + +/** + * Wraps a raw Java value and its OutputType, deferring serialization until needed. + * When streaming to JSON, calls OutputType.writeJson() directly, bypassing SerializedValue entirely. + * When SerializedValue is needed (e.g., for constraints/scheduler), lazily computes and caches it. + */ +public final class DeferredSerializedValue { + private final Object rawValue; + private final OutputType outputType; + private volatile SerializedValue cached; + + @SuppressWarnings("unchecked") + public DeferredSerializedValue(final T rawValue, final OutputType outputType) { + this.rawValue = rawValue; + this.outputType = (OutputType) outputType; + } + + public SerializedValue toSerializedValue() { + var result = cached; + if (result == null) { + result = outputType.serialize(rawValue); + cached = result; + } + return result; + } + + public void writeJson(final JsonGenerator gen) throws IOException { + outputType.writeJson(rawValue, gen); + } +} diff --git a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/InMemorySimulationResourceManager.java b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/InMemorySimulationResourceManager.java index 5ca174e16c..8637190206 100644 --- a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/InMemorySimulationResourceManager.java +++ b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/InMemorySimulationResourceManager.java @@ -18,7 +18,7 @@ */ public class InMemorySimulationResourceManager implements SimulationResourceManager { private final HashMap> realResourceSegments; - private final HashMap> discreteResourceSegments; + private final HashMap> discreteResourceSegments; private Duration lastReceivedTime; @@ -95,7 +95,7 @@ public ResourceProfiles computeProfiles(final Duration elapsedDuration, Set(elapsedDuration.minus(finalSegment.startOffset()), finalSegment.dynamics())); } - // Compute Discrete Profiles + // Compute Discrete Profiles — materialize DeferredSerializedValue to SerializedValue for(final var resource : discreteResourceSegments.entrySet()) { final var name = resource.getKey(); final var schema = resource.getValue().valueSchema(); @@ -109,12 +109,16 @@ public ResourceProfiles computeProfiles(final Duration elapsedDuration, Set(nextSegment.startOffset().minus(segment.startOffset()), segment.dynamics())); + profile.add(new ProfileSegment<>( + nextSegment.startOffset().minus(segment.startOffset()), + segment.dynamics().toSerializedValue())); } // Process final segment final var finalSegment = segments.getLast(); - profile.add(new ProfileSegment<>(elapsedDuration.minus(finalSegment.startOffset()), finalSegment.dynamics())); + profile.add(new ProfileSegment<>( + elapsedDuration.minus(finalSegment.startOffset()), + finalSegment.dynamics().toSerializedValue())); } return profiles; @@ -130,7 +134,7 @@ public ResourceProfiles computeProfiles(final Duration elapsedDuration, Set> realResourceUpdates, - final Map> discreteResourceUpdates + final Map> discreteResourceUpdates ) { if(elapsedTime.shorterThan(lastReceivedTime)) { throw new IllegalArgumentException(("elapsedTime must be monotonically increasing between calls.\n" diff --git a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/SimulationResourceManager.java b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/SimulationResourceManager.java index 50097d295d..f68261b7d4 100644 --- a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/SimulationResourceManager.java +++ b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/SimulationResourceManager.java @@ -2,7 +2,6 @@ import gov.nasa.jpl.aerie.merlin.protocol.types.Duration; import gov.nasa.jpl.aerie.merlin.protocol.types.RealDynamics; -import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; import org.apache.commons.lang3.tuple.Pair; @@ -30,11 +29,11 @@ public interface SimulationResourceManager { * Process resource updates for a given time. * @param elapsedTime the amount of time elapsed since the start of simulation. Must be monotonically increasing on subsequent calls. * @param realResourceUpdates the set of updates to real resources. Up to one update per resource is permitted. - * @param discreteResourceUpdates the set of updates to discrete resources. Up to one update per resource is permitted. + * @param discreteResourceUpdates the set of updates to discrete resources, with deferred serialization. Up to one update per resource is permitted. */ void acceptUpdates( final Duration elapsedTime, final Map> realResourceUpdates, - final Map> discreteResourceUpdates + final Map> discreteResourceUpdates ); } diff --git a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/StreamingSimulationResourceManager.java b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/StreamingSimulationResourceManager.java index 1dc8aa333b..c9c04fb3c5 100644 --- a/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/StreamingSimulationResourceManager.java +++ b/merlin-driver/src/main/java/gov/nasa/jpl/aerie/merlin/driver/resources/StreamingSimulationResourceManager.java @@ -18,7 +18,7 @@ */ public class StreamingSimulationResourceManager implements SimulationResourceManager { private final HashMap> realResourceSegments; - private final HashMap> discreteResourceSegments; + private final HashMap> discreteResourceSegments; private final AsyncConsumer streamer; @@ -74,7 +74,9 @@ public ResourceProfiles computeProfiles(final Duration elapsedDuration) { profiles.discreteProfiles() .get(name) .segments() - .add(new ProfileSegment<>(elapsedDuration.minus(finalSegment.startOffset()), finalSegment.dynamics())); + .add(new ProfileSegment<>( + elapsedDuration.minus(finalSegment.startOffset()), + finalSegment.dynamics().toSerializedValue())); // Remove final segment segments.clear(); @@ -123,7 +125,7 @@ private ResourceProfiles computeProfiles() { segments.add(finalSegment); } - // Compute Discrete Profiles + // Compute Discrete Profiles — materialize DeferredSerializedValue to SerializedValue for(final var resource : discreteResourceSegments.entrySet()) { final var name = resource.getKey(); final var schema = resource.getValue().valueSchema(); @@ -135,7 +137,9 @@ private ResourceProfiles computeProfiles() { for(int i = 0; i < segments.size()-1; i++) { final var segment = segments.get(i); final var nextSegment = segments.get(i+1); - profile.add(new ProfileSegment<>(nextSegment.startOffset().minus(segment.startOffset()), segment.dynamics())); + profile.add(new ProfileSegment<>( + nextSegment.startOffset().minus(segment.startOffset()), + segment.dynamics().toSerializedValue())); } // Remove the completed segments, leaving only the final (incomplete) segment in the current set @@ -159,7 +163,7 @@ private ResourceProfiles computeProfiles() { public void acceptUpdates( final Duration elapsedTime, final Map> realResourceUpdates, - final Map> discreteResourceUpdates + final Map> discreteResourceUpdates ) { if(elapsedTime.shorterThan(lastReceivedTime)) { throw new IllegalArgumentException(("elapsedTime must be monotonically increasing between calls.\n" diff --git a/merlin-driver/src/test/java/gov/nasa/jpl/aerie/merlin/driver/JacksonStreamingTortureTest.java b/merlin-driver/src/test/java/gov/nasa/jpl/aerie/merlin/driver/JacksonStreamingTortureTest.java new file mode 100644 index 0000000000..c582c21cee --- /dev/null +++ b/merlin-driver/src/test/java/gov/nasa/jpl/aerie/merlin/driver/JacksonStreamingTortureTest.java @@ -0,0 +1,358 @@ +package gov.nasa.jpl.aerie.merlin.driver; + +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.BooleanValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.DoubleValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.IntegerValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.ListValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.LongValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.RecordValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.StringValueMapper; +import gov.nasa.jpl.aerie.merlin.driver.json.JsonEncoding; +import gov.nasa.jpl.aerie.merlin.driver.resources.DeferredSerializedValue; +import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; +import gov.nasa.jpl.aerie.merlin.protocol.model.OutputType; +import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; +import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.StringWriter; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Torture test exercising the Jackson streaming serialization pipeline at scale. + * + * Verifies: + * 1. Correctness: writeJson output matches legacy serializeValue path for all mapper types + * 2. Scale: handles 200+ resources × 500 segments without issues + * 3. DeferredSerializedValue: lazy caching + direct streaming + * 4. Large composite types: deeply nested records with collections + */ +class JacksonStreamingTortureTest { + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + + // --- Record types simulating realistic mission model resources --- + + record ThermalState(double temperature, double heaterPower, boolean heaterOn) {} + record AttitudeState(double roll, double pitch, double yaw, double rate) {} + record TelemetryPacket(String name, int sequenceNumber, long timestamp, List channels) {} + record SubsystemStatus(String subsystem, boolean active, int errorCount, ThermalState thermal) {} + + // --- Mapper factories --- + + private static RecordValueMapper thermalMapper() { + return new RecordValueMapper<>(ThermalState.class, List.of( + new RecordValueMapper.Component<>("temperature", ThermalState::temperature, new DoubleValueMapper()), + new RecordValueMapper.Component<>("heaterPower", ThermalState::heaterPower, new DoubleValueMapper()), + new RecordValueMapper.Component<>("heaterOn", ThermalState::heaterOn, new BooleanValueMapper()) + )); + } + + private static RecordValueMapper attitudeMapper() { + return new RecordValueMapper<>(AttitudeState.class, List.of( + new RecordValueMapper.Component<>("roll", AttitudeState::roll, new DoubleValueMapper()), + new RecordValueMapper.Component<>("pitch", AttitudeState::pitch, new DoubleValueMapper()), + new RecordValueMapper.Component<>("yaw", AttitudeState::yaw, new DoubleValueMapper()), + new RecordValueMapper.Component<>("rate", AttitudeState::rate, new DoubleValueMapper()) + )); + } + + private static RecordValueMapper telemetryMapper() { + return new RecordValueMapper<>(TelemetryPacket.class, List.of( + new RecordValueMapper.Component<>("name", TelemetryPacket::name, new StringValueMapper()), + new RecordValueMapper.Component<>("sequenceNumber", TelemetryPacket::sequenceNumber, new IntegerValueMapper()), + new RecordValueMapper.Component<>("timestamp", TelemetryPacket::timestamp, new LongValueMapper()), + new RecordValueMapper.Component<>("channels", TelemetryPacket::channels, new ListValueMapper<>(new DoubleValueMapper())) + )); + } + + private static RecordValueMapper subsystemMapper() { + return new RecordValueMapper<>(SubsystemStatus.class, List.of( + new RecordValueMapper.Component<>("subsystem", SubsystemStatus::subsystem, new StringValueMapper()), + new RecordValueMapper.Component<>("active", SubsystemStatus::active, new BooleanValueMapper()), + new RecordValueMapper.Component<>("errorCount", SubsystemStatus::errorCount, new IntegerValueMapper()), + new RecordValueMapper.Component<>("thermal", SubsystemStatus::thermal, thermalMapper()) + )); + } + + // --- Helper to create an OutputType from a ValueMapper --- + + private static OutputType outputTypeFrom(ValueMapper mapper) { + return new OutputType<>() { + @Override + public ValueSchema getSchema() { return mapper.getValueSchema(); } + @Override + public SerializedValue serialize(T value) { return mapper.serializeValue(value); } + @Override + public void writeJson(T value, JsonGenerator gen) throws IOException { mapper.writeJson(value, gen); } + }; + } + + // --- Correctness: writeJson vs serializeValue for every mapper --- + + @Test + void allMappers_writeJsonMatchesSerialize_1000iterations() throws IOException { + // Simulate 1000 steps × multiple resource types + final var thermal = thermalMapper(); + final var attitude = attitudeMapper(); + final var telemetry = telemetryMapper(); + final var subsystem = subsystemMapper(); + final var intMapper = new IntegerValueMapper(); + final var dblMapper = new DoubleValueMapper(); + final var boolMapper = new BooleanValueMapper(); + final var strMapper = new StringValueMapper(); + + for (int step = 0; step < 1000; step++) { + final double t = step * 0.1; + + // Primitive resources + assertJsonEquals(intMapper, step); + assertJsonEquals(dblMapper, Math.sin(t) * 100); + assertJsonEquals(boolMapper, step % 2 == 0); + assertJsonEquals(strMapper, "mode_" + (step % 5)); + + // Complex resources + assertJsonEquals(thermal, new ThermalState(20.0 + t * 0.5, step % 10, step % 3 == 0)); + assertJsonEquals(attitude, new AttitudeState(t, -t * 0.5, t * 2, 0.01 * step)); + assertJsonEquals(telemetry, new TelemetryPacket( + "pkt_" + step, step, System.nanoTime(), + List.of(1.0 + t, 2.0 - t, 3.0 * t, 4.0 / (t + 1), 5.0))); + assertJsonEquals(subsystem, new SubsystemStatus( + "THERMAL", true, step % 100, + new ThermalState(-40.0 + step * 0.1, 50.0, step % 7 != 0))); + } + } + + // --- Scale: simulate 200+ resources × 500 segments --- + + @Test + void scaleTest_200Resources_500Segments() throws IOException { + final int NUM_RESOURCES = 200; + final int NUM_SEGMENTS = 500; + + // Create a mix of mappers representing the resource fleet + @SuppressWarnings("unchecked") + final ValueMapper[] mappers = new ValueMapper[NUM_RESOURCES]; + for (int i = 0; i < NUM_RESOURCES; i++) { + mappers[i] = switch (i % 5) { + case 0 -> (ValueMapper) (ValueMapper) new IntegerValueMapper(); + case 1 -> (ValueMapper) (ValueMapper) new DoubleValueMapper(); + case 2 -> (ValueMapper) (ValueMapper) new BooleanValueMapper(); + case 3 -> (ValueMapper) (ValueMapper) thermalMapper(); + case 4 -> (ValueMapper) (ValueMapper) attitudeMapper(); + default -> throw new IllegalStateException(); + }; + } + + // Simulate writing all segments to a single large JSON stream + final var sw = new StringWriter(1024 * 1024); // 1MB initial buffer + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + gen.writeStartArray(); // array of resources + + for (int r = 0; r < NUM_RESOURCES; r++) { + gen.writeStartObject(); + gen.writeStringField("name", "resource_" + r); + gen.writeArrayFieldStart("segments"); + + for (int s = 0; s < NUM_SEGMENTS; s++) { + gen.writeStartObject(); + gen.writeStringField("extent", "00:00:01.000000"); + gen.writeFieldName("dynamics"); + + final Object value = switch (r % 5) { + case 0 -> s; + case 1 -> s * 0.1; + case 2 -> s % 2 == 0; + case 3 -> new ThermalState(20.0 + s, s * 0.5, s % 3 == 0); + case 4 -> new AttitudeState(s * 0.01, -s * 0.01, s * 0.02, 0.001); + default -> throw new IllegalStateException(); + }; + + mappers[r].writeJson(value, gen); + gen.writeEndObject(); + } + + gen.writeEndArray(); // end segments + gen.writeEndObject(); // end resource + } + + gen.writeEndArray(); // end array of resources + } + + final var json = sw.toString(); + // Verify it's valid JSON and has substantial content + assertTrue(json.startsWith("["), "Output should start with array"); + assertTrue(json.endsWith("]"), "Output should end with array"); + assertTrue(json.length() > 100_000, "200×500 segments should produce substantial output, got " + json.length()); + } + + // --- DeferredSerializedValue: lazy caching + streaming --- + + @Test + void deferredSerializedValue_writeJsonMatchesLazySerialization() throws IOException { + final var mapper = subsystemMapper(); + final var outputType = outputTypeFrom(mapper); + final var value = new SubsystemStatus("POWER", true, 5, + new ThermalState(25.0, 10.0, true)); + + // Create deferred value — no serialization yet + final var deferred = new DeferredSerializedValue(value, outputType); + + // Stream directly via writeJson (bypasses SerializedValue entirely) + final var directJson = writeToString(gen -> deferred.writeJson(gen)); + + // Lazy serialize, then stream + final var lazySerialized = deferred.toSerializedValue(); + final var lazyJson = writeToString(gen -> lazySerialized.writeTo(gen)); + + // Both paths must produce semantically identical output (field order may differ) + assertEquals(parseJson(lazyJson), parseJson(directJson)); + + // Verify caching: second call returns same instance + final var lazySerialized2 = deferred.toSerializedValue(); + assertTrue(lazySerialized == lazySerialized2, "toSerializedValue should return cached instance"); + } + + @Test + void deferredSerializedValue_primitiveTypes() throws IOException { + // Verify deferred works for each primitive type + assertDeferredCorrect(new IntegerValueMapper(), 42); + assertDeferredCorrect(new DoubleValueMapper(), 3.14); + assertDeferredCorrect(new BooleanValueMapper(), true); + assertDeferredCorrect(new StringValueMapper(), "hello"); + assertDeferredCorrect(new LongValueMapper(), 999999999999L); + } + + @Test + void deferredSerializedValue_stressWithManyInstances() throws IOException { + // Simulate what happens during simulation: one DeferredSerializedValue per resource per step + final var mapper = thermalMapper(); + final var outputType = outputTypeFrom(mapper); + final int NUM_STEPS = 10_000; + + final var deferredValues = new ArrayList(NUM_STEPS); + + // Phase 1: Create deferred values (simulating simulation loop — no serialization) + for (int i = 0; i < NUM_STEPS; i++) { + deferredValues.add(new DeferredSerializedValue( + new ThermalState(20.0 + i * 0.01, i * 0.5, i % 3 == 0), + outputType)); + } + + // Phase 2: Stream all values (simulating post-simulation output) + final var sw = new StringWriter(NUM_STEPS * 100); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + gen.writeStartArray(); + for (final var deferred : deferredValues) { + deferred.writeJson(gen); + } + gen.writeEndArray(); + } + + final var json = sw.toString(); + assertTrue(json.startsWith("[")); + assertTrue(json.length() > NUM_STEPS * 10, + "10K thermal states should produce substantial output, got " + json.length()); + } + + // --- JsonEncoding convenience methods --- + + @Test + void jsonEncoding_writeToString_matchesWriteTo() throws IOException { + final var complex = SerializedValue.of(Map.of( + "data", SerializedValue.of(List.of( + SerializedValue.of(1L), + SerializedValue.of(2.5), + SerializedValue.of("three"))), + "meta", SerializedValue.of(Map.of( + "count", SerializedValue.of(3L))))); + + final var viaHelper = JsonEncoding.writeToString(complex); + final var viaMethod = complex.toJsonString(); + assertEquals(viaMethod, viaHelper); + } + + // --- Helpers --- + + @FunctionalInterface + private interface JsonWriterAction { + void write(JsonGenerator gen) throws IOException; + } + + private static String writeToString(JsonWriterAction action) throws IOException { + final var sw = new StringWriter(); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + action.write(gen); + } + return sw.toString(); + } + + private static SerializedValue parseJson(String json) throws IOException { + try (final var parser = JSON_FACTORY.createParser(json)) { + return parseValue(parser); + } + } + + private static SerializedValue parseValue(com.fasterxml.jackson.core.JsonParser parser) throws IOException { + final var token = parser.nextToken(); + return parseFromToken(parser, token); + } + + private static SerializedValue parseFromToken(com.fasterxml.jackson.core.JsonParser parser, com.fasterxml.jackson.core.JsonToken token) throws IOException { + return switch (token) { + case VALUE_NULL -> SerializedValue.NULL; + case VALUE_TRUE -> SerializedValue.of(true); + case VALUE_FALSE -> SerializedValue.of(false); + case VALUE_NUMBER_INT -> SerializedValue.of(parser.getBigIntegerValue().longValueExact()); + case VALUE_NUMBER_FLOAT -> SerializedValue.of(parser.getDecimalValue()); + case VALUE_STRING -> SerializedValue.of(parser.getText()); + case START_ARRAY -> { + final var list = new java.util.ArrayList(); + com.fasterxml.jackson.core.JsonToken t; + while ((t = parser.nextToken()) != com.fasterxml.jackson.core.JsonToken.END_ARRAY) { + list.add(parseFromToken(parser, t)); + } + yield SerializedValue.of(list); + } + case START_OBJECT -> { + final var map = new java.util.HashMap(); + com.fasterxml.jackson.core.JsonToken t; + while ((t = parser.nextToken()) != com.fasterxml.jackson.core.JsonToken.END_OBJECT) { + final var key = parser.getCurrentName(); + final var valToken = parser.nextToken(); + map.put(key, parseFromToken(parser, valToken)); + } + yield SerializedValue.of(map); + } + default -> throw new IOException("Unexpected token: " + token); + }; + } + + @SuppressWarnings("unchecked") + private static void assertJsonEquals(ValueMapper mapper, T value) throws IOException { + final var viaWriteJson = writeToString(gen -> mapper.writeJson(value, gen)); + final var viaSerialized = mapper.serializeValue(value).toJsonString(); + assertEquals(parseJson(viaSerialized), parseJson(viaWriteJson), + "Mismatch for " + mapper.getClass().getSimpleName() + " with value " + value); + } + + private static void assertDeferredCorrect(ValueMapper mapper, T value) throws IOException { + final var outputType = outputTypeFrom(mapper); + final var deferred = new DeferredSerializedValue(value, outputType); + + // Direct streaming + final var directJson = writeToString(gen -> deferred.writeJson(gen)); + // Via lazy serialization + final var lazyJson = deferred.toSerializedValue().toJsonString(); + + assertEquals(parseJson(lazyJson), parseJson(directJson), + "Deferred mismatch for " + mapper.getClass().getSimpleName()); + } +} diff --git a/merlin-driver/src/test/java/gov/nasa/jpl/aerie/merlin/driver/SerializationBenchmark.java b/merlin-driver/src/test/java/gov/nasa/jpl/aerie/merlin/driver/SerializationBenchmark.java new file mode 100644 index 0000000000..2602f20c75 --- /dev/null +++ b/merlin-driver/src/test/java/gov/nasa/jpl/aerie/merlin/driver/SerializationBenchmark.java @@ -0,0 +1,149 @@ +package gov.nasa.jpl.aerie.merlin.driver; + +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.BooleanValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.DoubleValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.IntegerValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.ListValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.RecordValueMapper; +import gov.nasa.jpl.aerie.contrib.serialization.mappers.StringValueMapper; +import gov.nasa.jpl.aerie.merlin.framework.ValueMapper; +import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.StringWriter; +import java.util.List; + +/** + * Benchmark comparing the legacy serialization path (serializeValue -> writeTo) + * against the optimized writeJson path for representative resource types. + * + * Run with: ./gradlew merlin-driver:test --tests '*SerializationBenchmark' + */ +class SerializationBenchmark { + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + private static final int WARMUP = 2_000; + private static final int ITERATIONS = 50_000; + + // --- Representative resource types --- + + record ThermalState(double temperature, double heaterPower, boolean heaterOn) {} + record TelemetryPacket(String name, int sequenceNumber, List channels) {} + + private static final RecordValueMapper THERMAL_MAPPER = new RecordValueMapper<>( + ThermalState.class, List.of( + new RecordValueMapper.Component<>("temperature", ThermalState::temperature, new DoubleValueMapper()), + new RecordValueMapper.Component<>("heaterPower", ThermalState::heaterPower, new DoubleValueMapper()), + new RecordValueMapper.Component<>("heaterOn", ThermalState::heaterOn, new BooleanValueMapper()) + )); + + private static final RecordValueMapper TELEMETRY_MAPPER = new RecordValueMapper<>( + TelemetryPacket.class, List.of( + new RecordValueMapper.Component<>("name", TelemetryPacket::name, new StringValueMapper()), + new RecordValueMapper.Component<>("sequenceNumber", TelemetryPacket::sequenceNumber, new IntegerValueMapper()), + new RecordValueMapper.Component<>("channels", TelemetryPacket::channels, new ListValueMapper<>(new DoubleValueMapper())) + )); + + // --- Benchmark: legacy path (serializeValue -> toJsonString) --- + + private static long benchmarkLegacyPath(ValueMapper mapper, T value, int iterations) throws IOException { + // Warmup + for (int i = 0; i < WARMUP; i++) { + final SerializedValue sv = mapper.serializeValue(value); + sv.toJsonString(); + } + + final long start = System.nanoTime(); + for (int i = 0; i < iterations; i++) { + final SerializedValue sv = mapper.serializeValue(value); + sv.toJsonString(); + } + return System.nanoTime() - start; + } + + // --- Benchmark: optimized writeJson path --- + + private static long benchmarkWriteJsonPath(ValueMapper mapper, T value, int iterations) throws IOException { + // Warmup + for (int i = 0; i < WARMUP; i++) { + final var sw = new StringWriter(256); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + mapper.writeJson(value, gen); + } + } + + final long start = System.nanoTime(); + for (int i = 0; i < iterations; i++) { + final var sw = new StringWriter(256); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + mapper.writeJson(value, gen); + } + } + return System.nanoTime() - start; + } + + // --- Benchmark: legacy path writing to shared generator --- + + private static long benchmarkLegacyToGenerator(ValueMapper mapper, T value, int iterations) throws IOException { + // Warmup + for (int i = 0; i < WARMUP; i++) { + final var sw = new StringWriter(256); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + mapper.serializeValue(value).writeTo(gen); + } + } + + final long start = System.nanoTime(); + for (int i = 0; i < iterations; i++) { + final var sw = new StringWriter(256); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + mapper.serializeValue(value).writeTo(gen); + } + } + return System.nanoTime() - start; + } + + @Test + void benchmark() throws IOException { + System.out.println("\n=== Serialization Benchmark (" + ITERATIONS + " iterations, " + WARMUP + " warmup) ===\n"); + + // Integer primitive + benchmarkMapper("IntegerValueMapper", new IntegerValueMapper(), 42); + + // Double primitive + benchmarkMapper("DoubleValueMapper", new DoubleValueMapper(), 3.14159); + + // Boolean primitive + benchmarkMapper("BooleanValueMapper", new BooleanValueMapper(), true); + + // String primitive + benchmarkMapper("StringValueMapper", new StringValueMapper(), "sensor_active"); + + // 3-field record (thermal state) + benchmarkMapper("ThermalState (3 fields)", + THERMAL_MAPPER, new ThermalState(25.3, 10.0, true)); + + // Record with list (telemetry packet, 5-element channel list) + benchmarkMapper("TelemetryPacket (3 fields + 5-elem list)", + TELEMETRY_MAPPER, new TelemetryPacket("pkt_001", 42, List.of(1.0, 2.5, 3.7, 4.2, 5.9))); + + System.out.println("=== End Benchmark ===\n"); + } + + private static void benchmarkMapper(String name, ValueMapper mapper, T value) throws IOException { + final long legacyNs = benchmarkLegacyPath(mapper, value, ITERATIONS); + final long serializeThenStreamNs = benchmarkLegacyToGenerator(mapper, value, ITERATIONS); + final long writeJsonNs = benchmarkWriteJsonPath(mapper, value, ITERATIONS); + + final double legacyMs = legacyNs / 1_000_000.0; + final double serStreamMs = serializeThenStreamNs / 1_000_000.0; + final double writeJsonMs = writeJsonNs / 1_000_000.0; + final double speedupVsLegacy = legacyMs / writeJsonMs; + final double speedupVsSerStream = serStreamMs / writeJsonMs; + + System.out.printf("%-42s legacy=%.1fms ser+stream=%.1fms writeJson=%.1fms speedup=%.2fx (vs legacy) %.2fx (vs ser+stream)%n", + name, legacyMs, serStreamMs, writeJsonMs, speedupVsLegacy, speedupVsSerStream); + } +} diff --git a/merlin-framework-processor/src/main/java/gov/nasa/jpl/aerie/merlin/processor/generator/MissionModelGenerator.java b/merlin-framework-processor/src/main/java/gov/nasa/jpl/aerie/merlin/processor/generator/MissionModelGenerator.java index d3e480e024..299b33701c 100644 --- a/merlin-framework-processor/src/main/java/gov/nasa/jpl/aerie/merlin/processor/generator/MissionModelGenerator.java +++ b/merlin-framework-processor/src/main/java/gov/nasa/jpl/aerie/merlin/processor/generator/MissionModelGenerator.java @@ -557,22 +557,8 @@ public JavaFile generateActivityTypes(final MissionModelRecord missionModel) { WildcardTypeName.subtypeOf(Object.class))), "directiveTypes", Modifier.PUBLIC, Modifier.STATIC, Modifier.FINAL) - .initializer( - "$T.ofEntries($>$>$L$<$<)", - ClassName.get(Map.class), - missionModel.activityTypes() - .stream() - .map(activityType -> CodeBlock - .builder() - .add( - "\n$T.entry($S, $L)", - ClassName.get(Map.class), - activityType.name(), - activityType.inputType().mapper().name.canonicalName().replace(".", "_"))) - .reduce((x, y) -> x.add(",").add(y.build())) - .orElse(CodeBlock.builder()) - .build()) .build()) + .addStaticBlock(buildDirectiveTypesStaticInit(missionModel)) .addMethod( MethodSpec .methodBuilder("registerTopics") @@ -621,6 +607,34 @@ public JavaFile generateActivityTypes(final MissionModelRecord missionModel) { .build(); } + /** + * Build a static initializer block for the directiveTypes field that uses HashMap.put() calls + * instead of Map.ofEntries(). This avoids a Java compiler type inference limitation that + * causes compilation failures when there are many (700+) activity types. + */ + private static CodeBlock buildDirectiveTypesStaticInit(final MissionModelRecord missionModel) { + final var builder = CodeBlock.builder() + .addStatement( + "final var _map = new $T<$T, $T<$T, ?, ?>>()", + ClassName.get(java.util.HashMap.class), + ClassName.get(String.class), + ClassName.get(gov.nasa.jpl.aerie.merlin.framework.ActivityMapper.class), + ClassName.get(missionModel.topLevelModel())); + + for (final var activityType : missionModel.activityTypes()) { + builder.addStatement( + "_map.put($S, $L)", + activityType.name(), + activityType.inputType().mapper().name.canonicalName().replace(".", "_")); + } + + builder.addStatement( + "directiveTypes = $T.unmodifiableMap(_map)", + ClassName.get(java.util.Collections.class)); + + return builder.build(); + } + private record ComputedAttributesCodeBlocks(TypeName typeName, FieldSpec fieldDef) {} /** Generate an `InputType` implementation. */ diff --git a/merlin-framework/src/main/java/gov/nasa/jpl/aerie/merlin/framework/Registrar.java b/merlin-framework/src/main/java/gov/nasa/jpl/aerie/merlin/framework/Registrar.java index d9866999fd..3e0a245b6f 100644 --- a/merlin-framework/src/main/java/gov/nasa/jpl/aerie/merlin/framework/Registrar.java +++ b/merlin-framework/src/main/java/gov/nasa/jpl/aerie/merlin/framework/Registrar.java @@ -1,5 +1,6 @@ package gov.nasa.jpl.aerie.merlin.framework; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.protocol.driver.Initializer; import gov.nasa.jpl.aerie.merlin.protocol.driver.Querier; import gov.nasa.jpl.aerie.merlin.protocol.model.OutputType; @@ -7,6 +8,7 @@ import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; import java.util.Map; import java.util.Objects; import java.util.function.Function; @@ -24,11 +26,11 @@ public boolean isInitializationComplete() { } public void discrete(final String name, final Resource resource, final ValueMapper mapper) { - this.builder.resource(name, makeResource("discrete", resource, mapper.getValueSchema(), mapper::serializeValue, null)); + this.builder.resource(name, makeResource("discrete", resource, mapper.getValueSchema(), mapper::serializeValue, mapper::writeJson, null)); } public void discrete(final String name, final Resource resource, final ValueMapper mapper, final String description) { - this.builder.resource(name, makeResource("discrete", resource, mapper.getValueSchema(), mapper::serializeValue, description)); + this.builder.resource(name, makeResource("discrete", resource, mapper.getValueSchema(), mapper::serializeValue, mapper::writeJson, description)); } public void real(final String name, final Resource resource) { @@ -47,6 +49,13 @@ public void realWithMetadata(final String name, final Resource real(name, resource, $ -> ValueSchema.withMeta(key, metadataValueMapper.serializeValue(metadata), $), description); } + private static final JsonWriter REAL_DYNAMICS_JSON_WRITER = (dynamics, gen) -> { + gen.writeStartObject(); + gen.writeNumberField("initial", dynamics.initial); + gen.writeNumberField("rate", dynamics.rate); + gen.writeEndObject(); + }; + private void real(final String name, final Resource resource, UnaryOperator schemaModifier) { this.builder.resource( name, @@ -59,6 +68,7 @@ private void real(final String name, final Resource resource, Unar dynamics -> SerializedValue.of(Map.of( "initial", SerializedValue.of(dynamics.initial), "rate", SerializedValue.of(dynamics.rate))), + REAL_DYNAMICS_JSON_WRITER, null )); } @@ -75,15 +85,22 @@ private void real(final String name, final Resource resource, Unar dynamics -> SerializedValue.of(Map.of( "initial", SerializedValue.of(dynamics.initial), "rate", SerializedValue.of(dynamics.rate))), + REAL_DYNAMICS_JSON_WRITER, description )); } + @FunctionalInterface + private interface JsonWriter { + void write(T value, JsonGenerator gen) throws IOException; + } + private static gov.nasa.jpl.aerie.merlin.protocol.model.Resource makeResource( final String type, final Resource resource, final ValueSchema valueSchema, final Function serializer, + final JsonWriter jsonWriter, final String description ) { return new gov.nasa.jpl.aerie.merlin.protocol.model.Resource<>() { @@ -107,6 +124,11 @@ public ValueSchema getSchema() { public SerializedValue serialize(final Value value) { return serializer.apply(value); } + + @Override + public void writeJson(final Value value, final JsonGenerator gen) throws IOException { + jsonWriter.write(value, gen); + } }; } @@ -132,6 +154,11 @@ public ValueSchema getSchema() { public SerializedValue serialize(final Event value) { return mapper.serializeValue(value); } + + @Override + public void writeJson(final Event value, final JsonGenerator gen) throws IOException { + mapper.writeJson(value, gen); + } }); } } diff --git a/merlin-framework/src/main/java/gov/nasa/jpl/aerie/merlin/framework/ValueMapper.java b/merlin-framework/src/main/java/gov/nasa/jpl/aerie/merlin/framework/ValueMapper.java index 3ae025d5bf..5b92b12b05 100644 --- a/merlin-framework/src/main/java/gov/nasa/jpl/aerie/merlin/framework/ValueMapper.java +++ b/merlin-framework/src/main/java/gov/nasa/jpl/aerie/merlin/framework/ValueMapper.java @@ -1,8 +1,11 @@ package gov.nasa.jpl.aerie.merlin.framework; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; + /** * A mapping between (a) the mission-specific representation of a data type defined by a mission model (b) to a * mission-agnostic representation of that data type. @@ -30,4 +33,12 @@ public interface ValueMapper { * @return A mission-agnostic representation of {@code value}. */ SerializedValue serializeValue(T value); + + /** + * Writes a value directly to a Jackson {@link JsonGenerator}, bypassing intermediate SerializedValue allocations. + * The default implementation falls back to {@link #serializeValue(T)} followed by streaming write. + */ + default void writeJson(T value, JsonGenerator gen) throws IOException { + serializeValue(value).writeTo(gen); + } } diff --git a/merlin-sdk/build.gradle b/merlin-sdk/build.gradle index 3ed5781f42..8b67928271 100644 --- a/merlin-sdk/build.gradle +++ b/merlin-sdk/build.gradle @@ -30,6 +30,8 @@ jacocoTestReport { javadoc.options.addStringOption('Xdoclint:none', '-quiet') dependencies { + api 'com.fasterxml.jackson.core:jackson-core:2.15.3' + testImplementation 'org.junit.jupiter:junit-jupiter-engine:6.0.1' testRuntimeOnly 'org.junit.platform:junit-platform-launcher' diff --git a/merlin-sdk/src/main/java/gov/nasa/jpl/aerie/merlin/protocol/model/OutputType.java b/merlin-sdk/src/main/java/gov/nasa/jpl/aerie/merlin/protocol/model/OutputType.java index 99ed8094d2..63f63d189c 100644 --- a/merlin-sdk/src/main/java/gov/nasa/jpl/aerie/merlin/protocol/model/OutputType.java +++ b/merlin-sdk/src/main/java/gov/nasa/jpl/aerie/merlin/protocol/model/OutputType.java @@ -1,8 +1,11 @@ package gov.nasa.jpl.aerie.merlin.protocol.model; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import java.io.IOException; + /** * A type of data produced as output by a Merlin model. * @@ -24,4 +27,12 @@ public interface OutputType { /** Extracts a value conforming to this type's {@linkplain #getSchema() schema} from an opaque value of type {@code T}. */ SerializedValue serialize(T value); + + /** + * Writes a value directly to a Jackson {@link JsonGenerator}, bypassing intermediate SerializedValue allocations. + * The default implementation falls back to {@link #serialize(T)} followed by streaming write. + */ + default void writeJson(T value, JsonGenerator gen) throws IOException { + serialize(value).writeTo(gen); + } } diff --git a/merlin-sdk/src/main/java/gov/nasa/jpl/aerie/merlin/protocol/types/SerializedValue.java b/merlin-sdk/src/main/java/gov/nasa/jpl/aerie/merlin/protocol/types/SerializedValue.java index 9c8dba185e..167502e570 100644 --- a/merlin-sdk/src/main/java/gov/nasa/jpl/aerie/merlin/protocol/types/SerializedValue.java +++ b/merlin-sdk/src/main/java/gov/nasa/jpl/aerie/merlin/protocol/types/SerializedValue.java @@ -1,5 +1,9 @@ package gov.nasa.jpl.aerie.merlin.protocol.types; +import com.fasterxml.jackson.core.JsonGenerator; + +import java.io.IOException; +import java.io.StringWriter; import java.math.BigDecimal; import java.util.ArrayList; import java.util.List; @@ -57,6 +61,7 @@ public sealed interface SerializedValue extends Comparable { interface Visitor { T onNull(); T onNumeric(BigDecimal value); + T onDouble(double value); T onBoolean(boolean value); T onString(String value); T onMap(Map value); @@ -111,6 +116,29 @@ public int hashCode() { } } + record DoubleValue(double value) implements SerializedValue { + @Override + public T match(final Visitor visitor) { + return visitor.onDouble(value); + } + + @Override + public Double getValue() { + return value; + } + + @Override + public boolean equals(final Object obj) { + if (!(obj instanceof DoubleValue other)) return false; + return Double.compare(this.value, other.value) == 0; + } + + @Override + public int hashCode() { + return Double.hashCode(value); + } + } + record BooleanValue(boolean value) implements SerializedValue { @Override public T match(final Visitor visitor) { @@ -181,7 +209,7 @@ static SerializedValue of(final BigDecimal value) { * @return A new {@link SerializedValue} containing a real number. */ static SerializedValue of(final double value) { - return new NumericValue(BigDecimal.valueOf(value)); + return new DoubleValue(value); } /** @@ -261,6 +289,11 @@ public T onNumeric(final BigDecimal value) { return this.onDefault(); } + @Override + public T onDouble(final double value) { + return this.onDefault(); + } + @Override public T onBoolean(final boolean value) { return this.onDefault(); @@ -325,6 +358,11 @@ default Optional asNumeric() { public Optional onNumeric(final BigDecimal value) { return Optional.of(value); } + + @Override + public Optional onDouble(final double value) { + return Optional.of(BigDecimal.valueOf(value)); + } }); } @@ -340,6 +378,11 @@ default Optional asReal() { public Optional onNumeric(final BigDecimal value) { return Optional.of(value.doubleValue()); } + + @Override + public Optional onDouble(final double value) { + return Optional.of(value); + } }); } @@ -359,6 +402,12 @@ public Optional onNumeric(final BigDecimal value) { return Optional.empty(); } } + + @Override + public Optional onDouble(final double value) { + if (value % 1 == 0) return Optional.of((long) value); + return Optional.empty(); + } }); } @@ -421,4 +470,84 @@ public Optional> onList(final List value) } }); } + + /** + * Writes this value directly to a Jackson {@link JsonGenerator}, bypassing intermediate JsonNode allocations. + */ + default void writeTo(final JsonGenerator gen) throws IOException { + try { + this.match(new Visitor() { + @Override + public Void onNull() { + try { gen.writeNull(); } catch (final IOException e) { throw new UncheckedIOException(e); } + return null; + } + @Override + public Void onNumeric(final BigDecimal value) { + try { gen.writeNumber(value); } catch (final IOException e) { throw new UncheckedIOException(e); } + return null; + } + @Override + public Void onDouble(final double value) { + try { gen.writeNumber(value); } catch (final IOException e) { throw new UncheckedIOException(e); } + return null; + } + @Override + public Void onBoolean(final boolean value) { + try { gen.writeBoolean(value); } catch (final IOException e) { throw new UncheckedIOException(e); } + return null; + } + @Override + public Void onString(final String value) { + try { gen.writeString(value); } catch (final IOException e) { throw new UncheckedIOException(e); } + return null; + } + @Override + public Void onMap(final Map value) { + try { + gen.writeStartObject(); + for (final var entry : value.entrySet()) { + gen.writeFieldName(entry.getKey()); + entry.getValue().writeTo(gen); + } + gen.writeEndObject(); + } catch (final IOException e) { throw new UncheckedIOException(e); } + return null; + } + @Override + public Void onList(final List value) { + try { + gen.writeStartArray(); + for (final var element : value) { + element.writeTo(gen); + } + gen.writeEndArray(); + } catch (final IOException e) { throw new UncheckedIOException(e); } + return null; + } + }); + } catch (final UncheckedIOException e) { + throw e.getCause(); + } + } + + final class UncheckedIOException extends RuntimeException { + UncheckedIOException(final IOException cause) { super(cause); } + @Override public IOException getCause() { return (IOException) super.getCause(); } + } + + /** + * Serializes this value to a JSON string using Jackson streaming. + */ + default String toJsonString() throws IOException { + try { + final var writer = new StringWriter(); + try (final var gen = new com.fasterxml.jackson.core.JsonFactory().createGenerator(writer)) { + this.writeTo(gen); + } + return writer.toString(); + } catch (final UncheckedIOException e) { + throw e.getCause(); + } + } } diff --git a/merlin-sdk/src/test/java/gov/nasa/jpl/aerie/merlin/protocol/types/SerializedValueStreamingTest.java b/merlin-sdk/src/test/java/gov/nasa/jpl/aerie/merlin/protocol/types/SerializedValueStreamingTest.java new file mode 100644 index 0000000000..6507642e40 --- /dev/null +++ b/merlin-sdk/src/test/java/gov/nasa/jpl/aerie/merlin/protocol/types/SerializedValueStreamingTest.java @@ -0,0 +1,158 @@ +package gov.nasa.jpl.aerie.merlin.protocol.types; + +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.StringWriter; +import java.math.BigDecimal; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Tests that SerializedValue.writeTo(JsonGenerator) produces identical output + * to the legacy javax.json-based serialization path. + */ +class SerializedValueStreamingTest { + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + + private static String writeToString(SerializedValue value) throws IOException { + final var sw = new StringWriter(); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + value.writeTo(gen); + } + return sw.toString(); + } + + // --- Primitive types --- + + @Test + void writeTo_null() throws IOException { + assertEquals("null", writeToString(SerializedValue.NULL)); + } + + @Test + void writeTo_boolean_true() throws IOException { + assertEquals("true", writeToString(SerializedValue.of(true))); + } + + @Test + void writeTo_boolean_false() throws IOException { + assertEquals("false", writeToString(SerializedValue.of(false))); + } + + @Test + void writeTo_integer() throws IOException { + assertEquals("42", writeToString(SerializedValue.of(42L))); + } + + @Test + void writeTo_negative_integer() throws IOException { + assertEquals("-7", writeToString(SerializedValue.of(-7L))); + } + + @Test + void writeTo_zero() throws IOException { + assertEquals("0", writeToString(SerializedValue.of(0L))); + } + + @Test + void writeTo_double() throws IOException { + assertEquals("3.14", writeToString(SerializedValue.of(3.14))); + } + + @Test + void writeTo_bigDecimal() throws IOException { + assertEquals( + "123456789.987654321", + writeToString(SerializedValue.of(new BigDecimal("123456789.987654321")))); + } + + @Test + void writeTo_string() throws IOException { + assertEquals("\"hello world\"", writeToString(SerializedValue.of("hello world"))); + } + + @Test + void writeTo_string_with_escapes() throws IOException { + final var json = writeToString(SerializedValue.of("line1\nline2\ttab\"quote")); + assertEquals("\"line1\\nline2\\ttab\\\"quote\"", json); + } + + @Test + void writeTo_empty_string() throws IOException { + assertEquals("\"\"", writeToString(SerializedValue.of(""))); + } + + // --- Composite types --- + + @Test + void writeTo_empty_list() throws IOException { + assertEquals("[]", writeToString(SerializedValue.of(List.of()))); + } + + @Test + void writeTo_list_of_primitives() throws IOException { + final var value = SerializedValue.of(List.of( + SerializedValue.of(1L), + SerializedValue.of("two"), + SerializedValue.of(true), + SerializedValue.NULL + )); + assertEquals("[1,\"two\",true,null]", writeToString(value)); + } + + @Test + void writeTo_empty_map() throws IOException { + assertEquals("{}", writeToString(SerializedValue.of(Map.of()))); + } + + @Test + void writeTo_simple_map() throws IOException { + // Use a single-entry map for deterministic key order + final var json = writeToString(SerializedValue.of(Map.of("key", SerializedValue.of("value")))); + assertEquals("{\"key\":\"value\"}", json); + } + + // --- Nested structures --- + + @Test + void writeTo_nested_list_of_maps() throws IOException { + final var value = SerializedValue.of(List.of( + SerializedValue.of(Map.of("x", SerializedValue.of(1L))), + SerializedValue.of(Map.of("y", SerializedValue.of(2L))) + )); + assertEquals("[{\"x\":1},{\"y\":2}]", writeToString(value)); + } + + @Test + void writeTo_deeply_nested() throws IOException { + // {"a":{"b":{"c":[1,2,3]}}} + final var innerList = SerializedValue.of(List.of( + SerializedValue.of(1L), SerializedValue.of(2L), SerializedValue.of(3L))); + final var innerMap = SerializedValue.of(Map.of("c", innerList)); + final var middleMap = SerializedValue.of(Map.of("b", innerMap)); + final var outerMap = SerializedValue.of(Map.of("a", middleMap)); + + assertEquals("{\"a\":{\"b\":{\"c\":[1,2,3]}}}", writeToString(outerMap)); + } + + // --- Round-trip consistency: toJsonString() should match writeTo() --- + + @Test + void toJsonString_matches_writeTo() throws IOException { + final var complex = SerializedValue.of(Map.of( + "name", SerializedValue.of("test"), + "values", SerializedValue.of(List.of( + SerializedValue.of(1L), + SerializedValue.of(2.5), + SerializedValue.NULL + )) + )); + + assertEquals(writeToString(complex), complex.toJsonString()); + } +} diff --git a/merlin-worker/src/main/java/gov/nasa/jpl/aerie/merlin/worker/postgres/PostgresProfileQueryHandler.java b/merlin-worker/src/main/java/gov/nasa/jpl/aerie/merlin/worker/postgres/PostgresProfileQueryHandler.java index 8f00cb4872..7d215133d2 100644 --- a/merlin-worker/src/main/java/gov/nasa/jpl/aerie/merlin/worker/postgres/PostgresProfileQueryHandler.java +++ b/merlin-worker/src/main/java/gov/nasa/jpl/aerie/merlin/worker/postgres/PostgresProfileQueryHandler.java @@ -1,6 +1,7 @@ package gov.nasa.jpl.aerie.merlin.worker.postgres; -import gov.nasa.jpl.aerie.json.JsonParser; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; import gov.nasa.jpl.aerie.merlin.driver.resources.ResourceProfile; import gov.nasa.jpl.aerie.merlin.driver.resources.ResourceProfiles; import gov.nasa.jpl.aerie.merlin.protocol.types.Duration; @@ -13,14 +14,14 @@ import org.apache.commons.lang3.tuple.Pair; import javax.sql.DataSource; +import java.io.IOException; +import java.io.StringWriter; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.SQLException; import java.sql.Statement; import java.util.HashMap; -import static gov.nasa.jpl.aerie.merlin.driver.json.SerializedValueJsonParser.serializedValueP; -import static gov.nasa.jpl.aerie.merlin.server.http.ProfileParsers.realDynamicsP; import static gov.nasa.jpl.aerie.merlin.server.remotes.postgres.PostgresParsers.discreteProfileTypeP; import static gov.nasa.jpl.aerie.merlin.server.remotes.postgres.PostgresParsers.realProfileTypeP; @@ -28,6 +29,8 @@ * Utility class to handle upload of resource profiles to the database. * */ public class PostgresProfileQueryHandler implements AutoCloseable { + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + private final Connection connection; private final HashMap profileIds; private final HashMap profileDurations; @@ -87,10 +90,10 @@ public void uploadResourceProfiles(final ResourceProfiles resourceProfiles) { // Post Segments for (final var realEntry : resourceProfiles.realProfiles().entrySet()) { - addProfileSegmentsToBatch(realEntry.getKey(), realEntry.getValue(), realDynamicsP); + addRealProfileSegmentsToBatch(realEntry.getKey(), realEntry.getValue()); } for (final var discreteEntry : resourceProfiles.discreteProfiles().entrySet()) { - addProfileSegmentsToBatch(discreteEntry.getKey(), discreteEntry.getValue(), serializedValueP); + addDiscreteProfileSegmentsToBatch(discreteEntry.getKey(), discreteEntry.getValue()); } postProfileSegments(); @@ -152,15 +155,42 @@ private void updateProfileDurations() throws SQLException { } } - private void addProfileSegmentsToBatch(final String name, ResourceProfile profile, JsonParser dynamicsP) throws SQLException { + private void addRealProfileSegmentsToBatch(final String name, ResourceProfile profile) throws SQLException { + final var id = profileIds.get(name); + this.postSegmentsStatement.setLong(1, id); + + var newDuration = profileDurations.get(name); + for (final var segment : profile.segments()) { + PreparedStatements.setDuration(this.postSegmentsStatement, 2, newDuration); + try { + this.postSegmentsStatement.setString(3, realDynamicsToJsonString(segment.dynamics())); + } catch (final IOException e) { + throw new SQLException("Failed to serialize real dynamics", e); + } + this.postSegmentsStatement.addBatch(); + + newDuration = newDuration.plus(segment.extent()); + } + + this.updateDurationStatement.setLong(2, id); + PreparedStatements.setDuration(this.updateDurationStatement, 1, newDuration); + this.updateDurationStatement.addBatch(); + + profileDurations.put(name, newDuration); + } + + private void addDiscreteProfileSegmentsToBatch(final String name, ResourceProfile profile) throws SQLException { final var id = profileIds.get(name); this.postSegmentsStatement.setLong(1, id); var newDuration = profileDurations.get(name); for (final var segment : profile.segments()) { PreparedStatements.setDuration(this.postSegmentsStatement, 2, newDuration); - final var dynamics = dynamicsP.unparse(segment.dynamics()).toString(); - this.postSegmentsStatement.setString(3, dynamics); + try { + this.postSegmentsStatement.setString(3, segment.dynamics().toJsonString()); + } catch (final IOException e) { + throw new SQLException("Failed to serialize discrete dynamics", e); + } this.postSegmentsStatement.addBatch(); newDuration = newDuration.plus(segment.extent()); @@ -173,6 +203,17 @@ private void addProfileSegmentsToBatch(final String name, ResourceProfile profileDurations.put(name, newDuration); } + private static String realDynamicsToJsonString(final RealDynamics dynamics) throws IOException { + final var sw = new StringWriter(); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + gen.writeStartObject(); + gen.writeNumberField("initial", dynamics.initial); + gen.writeNumberField("rate", dynamics.rate); + gen.writeEndObject(); + } + return sw.toString(); + } + @Override public void close() throws SQLException { this.postProfileStatement.close(); diff --git a/orchestration-utils/src/main/java/gov/nasa/jpl/aerie/orchestration/simulation/ResourceFileStreamer.java b/orchestration-utils/src/main/java/gov/nasa/jpl/aerie/orchestration/simulation/ResourceFileStreamer.java index ea3ac9cdf4..5796f5394f 100644 --- a/orchestration-utils/src/main/java/gov/nasa/jpl/aerie/orchestration/simulation/ResourceFileStreamer.java +++ b/orchestration-utils/src/main/java/gov/nasa/jpl/aerie/orchestration/simulation/ResourceFileStreamer.java @@ -1,22 +1,26 @@ package gov.nasa.jpl.aerie.orchestration.simulation; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import gov.nasa.jpl.aerie.merlin.driver.engine.ProfileSegment; import gov.nasa.jpl.aerie.merlin.driver.resources.AsyncConsumer; import gov.nasa.jpl.aerie.merlin.driver.resources.ResourceProfiles; +import gov.nasa.jpl.aerie.merlin.protocol.types.RealDynamics; +import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; -import javax.json.Json; import java.io.FileWriter; import java.io.IOException; +import java.io.StringWriter; import java.util.Arrays; import java.util.HashMap; import java.util.UUID; -import static gov.nasa.jpl.aerie.merlin.driver.json.SerializedValueJsonParser.serializedValueP; -import static gov.nasa.jpl.aerie.merlin.server.http.ProfileParsers.realDynamicsP; - /** * A consumer that writes resource segments to the file system. */ public class ResourceFileStreamer implements AsyncConsumer { + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + private final UUID uuid; private final HashMap fileNames; @@ -56,11 +60,8 @@ public void accept(final ResourceProfiles resourceProfile) { final var name = getFileName(r.getKey()); try (final var fileWriter = new FileWriter(name, true)) { for(final var segment : r.getValue().segments()) { - final var s = Json.createObjectBuilder() - .add("extent", segment.extent().toString()) - .add("dynamics", realDynamicsP.unparse(segment.dynamics())) - .build(); - fileWriter.write(s.toString()+"\n"); + fileWriter.write(segmentToJsonString(segment, true)); + fileWriter.write('\n'); } fileWriter.flush(); } catch (IOException e) { @@ -72,11 +73,8 @@ public void accept(final ResourceProfiles resourceProfile) { final var name = getFileName(d.getKey()); try (final var fileWriter = new FileWriter(name, true)) { for(final var segment : d.getValue().segments()) { - final var s = Json.createObjectBuilder() - .add("extent", segment.extent().toString()) - .add("dynamics", serializedValueP.unparse(segment.dynamics())) - .build(); - fileWriter.write(s.toString()+"\n"); + fileWriter.write(segmentToJsonString(segment, false)); + fileWriter.write('\n'); } fileWriter.flush(); } catch (IOException e) { @@ -85,6 +83,26 @@ public void accept(final ResourceProfiles resourceProfile) { } } + private static String segmentToJsonString(final ProfileSegment segment, final boolean isReal) throws IOException { + final var sw = new StringWriter(); + try (final var gen = JSON_FACTORY.createGenerator(sw)) { + gen.writeStartObject(); + gen.writeStringField("extent", segment.extent().toString()); + gen.writeFieldName("dynamics"); + if (isReal) { + final var dynamics = (RealDynamics) segment.dynamics(); + gen.writeStartObject(); + gen.writeNumberField("initial", dynamics.initial); + gen.writeNumberField("rate", dynamics.rate); + gen.writeEndObject(); + } else { + ((SerializedValue) segment.dynamics()).writeTo(gen); + } + gen.writeEndObject(); + } + return sw.toString(); + } + /** * Converts a resource's name into a legal file name and saves it in its cache of filenames. */ diff --git a/orchestration-utils/src/main/java/gov/nasa/jpl/aerie/orchestration/simulation/SimulationResultsWriter.java b/orchestration-utils/src/main/java/gov/nasa/jpl/aerie/orchestration/simulation/SimulationResultsWriter.java index 9ab944e33f..50e1d30a67 100644 --- a/orchestration-utils/src/main/java/gov/nasa/jpl/aerie/orchestration/simulation/SimulationResultsWriter.java +++ b/orchestration-utils/src/main/java/gov/nasa/jpl/aerie/orchestration/simulation/SimulationResultsWriter.java @@ -1,39 +1,37 @@ package gov.nasa.jpl.aerie.orchestration.simulation; -import gov.nasa.jpl.aerie.json.JsonParser; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.util.DefaultIndenter; +import com.fasterxml.jackson.core.util.DefaultPrettyPrinter; import gov.nasa.jpl.aerie.merlin.driver.SimulationResults; import gov.nasa.jpl.aerie.merlin.driver.resources.ResourceProfile; +import gov.nasa.jpl.aerie.merlin.protocol.types.Duration; +import gov.nasa.jpl.aerie.merlin.protocol.types.RealDynamics; +import gov.nasa.jpl.aerie.merlin.protocol.types.SerializedValue; +import gov.nasa.jpl.aerie.merlin.protocol.types.ValueSchema; +import gov.nasa.jpl.aerie.merlin.server.remotes.postgres.EventGraphFlattener; +import gov.nasa.jpl.aerie.types.Plan; +import gov.nasa.jpl.aerie.types.Timestamp; -import javax.json.Json; -import javax.json.JsonReader; -import javax.json.stream.JsonGenerator; +import java.io.BufferedReader; import java.io.FileWriter; import java.io.IOException; import java.io.OutputStreamWriter; -import java.io.StringReader; import java.io.Writer; import java.nio.file.Files; import java.nio.file.Path; import java.time.temporal.ChronoUnit; -import java.util.Map; - -import gov.nasa.jpl.aerie.merlin.protocol.types.Duration; -import gov.nasa.jpl.aerie.merlin.server.remotes.postgres.EventGraphFlattener; -import gov.nasa.jpl.aerie.types.Plan; -import gov.nasa.jpl.aerie.types.Timestamp; import static gov.nasa.jpl.aerie.merlin.driver.json.SerializedValueJsonParser.serializedValueP; import static gov.nasa.jpl.aerie.merlin.driver.json.ValueSchemaJsonParser.valueSchemaP; -import static gov.nasa.jpl.aerie.merlin.server.http.ProfileParsers.realDynamicsP; import static gov.nasa.jpl.aerie.merlin.server.remotes.postgres.PostgresParsers.activityArgumentsP; import static gov.nasa.jpl.aerie.merlin.server.remotes.postgres.PostgresParsers.simulationArgumentsP; public class SimulationResultsWriter { - private final static double SCHEMA_VERSION = 1; - - // Write JSONs with Pretty Printing - private final static Map config = Map.of(JsonGenerator.PRETTY_PRINTING, ""); + private static final double SCHEMA_VERSION = 1; + private static final JsonFactory JSON_FACTORY = new JsonFactory(); private final SimulationResults results; private final Plan plan; @@ -91,221 +89,266 @@ public void writeResults(CanceledListener canceledListener, Path outputFilePath) } public void writeResults(CanceledListener canceledListener, Writer outputWriter) { - try (final var resultsJsonGenerator = Json.createGeneratorFactory(config).createGenerator(outputWriter)) { + final var printer = new DefaultPrettyPrinter(); + printer.indentArraysWith(DefaultIndenter.SYSTEM_LINEFEED_INSTANCE); + + try (final var gen = JSON_FACTORY.createGenerator(outputWriter)) { + gen.setPrettyPrinter(printer); + // Start the top-level object - resultsJsonGenerator.writeStartObject(); + gen.writeStartObject(); // Output the starting information, a set of top-level fields - writeOpening(resultsJsonGenerator, canceledListener.get()); + writeOpening(gen, canceledListener.get()); // Write each of the main subsections + gen.writeFieldName("simulationConfiguration"); + writeSimConfig(gen); - resultsJsonGenerator.writeKey("simulationConfiguration"); - writeSimConfig(resultsJsonGenerator); - - resultsJsonGenerator.writeKey("profiles"); - writeProfiles(resultsJsonGenerator); + gen.writeFieldName("profiles"); + writeProfiles(gen); - resultsJsonGenerator.writeKey("spans"); - writeSpans(resultsJsonGenerator); + gen.writeFieldName("spans"); + writeSpans(gen); - resultsJsonGenerator.writeKey("topics"); - writeTopics(resultsJsonGenerator); + gen.writeFieldName("topics"); + writeTopics(gen); - resultsJsonGenerator.writeKey("events"); - writeEvents(resultsJsonGenerator); + gen.writeFieldName("events"); + writeEvents(gen); // End the top-level object - resultsJsonGenerator.writeEnd(); + gen.writeEndObject(); + } catch (IOException e) { + throw new RuntimeException(e); } } /** Write the top-level fields of the results JSON */ - private void writeOpening(JsonGenerator resultsGenerator, boolean canceled) { + private void writeOpening(JsonGenerator gen, boolean canceled) throws IOException { final var simEndTime = plan.simulationStartTimestamp.plusMicros(results.duration.in(Duration.MICROSECOND)); - resultsGenerator - .write("version", SCHEMA_VERSION) - .write("simulationStartTime", plan.simulationStartTimestamp.toString()) - .write("simulationEndTime", simEndTime.toString()) - .write("canceled", canceled); + gen.writeNumberField("version", SCHEMA_VERSION); + gen.writeStringField("simulationStartTime", plan.simulationStartTimestamp.toString()); + gen.writeStringField("simulationEndTime", simEndTime.toString()); + gen.writeBooleanField("canceled", canceled); } /** Write the simulation configuration section of the results */ - private void writeSimConfig(JsonGenerator resultsGenerator) { - resultsGenerator.writeStartObject() - .write("startTime", plan.simulationStartTimestamp.toString()) - .write("endTime", plan.simulationEndTimestamp.toString()) - .write("arguments", simulationArgumentsP.unparse(plan.simulationConfiguration())) - .writeEnd(); + private void writeSimConfig(JsonGenerator gen) throws IOException { + gen.writeStartObject(); + gen.writeStringField("startTime", plan.simulationStartTimestamp.toString()); + gen.writeStringField("endTime", plan.simulationEndTimestamp.toString()); + gen.writeFieldName("arguments"); + writeJsonValue(gen, simulationArgumentsP.unparse(plan.simulationConfiguration())); + gen.writeEndObject(); } /** * Write the profiles section of the results. * Will get profile segments from resourceFileStreamer if it's non-null, or from results' maps otherwise. */ - private void writeProfiles(JsonGenerator resultsGenerator) { - resultsGenerator.writeStartObject(); + private void writeProfiles(JsonGenerator gen) throws IOException { + gen.writeStartObject(); // Each real profile is an object in the array realProfiles - resultsGenerator.writeStartArray("realProfiles"); + gen.writeArrayFieldStart("realProfiles"); for (var e : results.realProfiles.entrySet()) { - writeProfile(resultsGenerator, e.getKey(), e.getValue(), realDynamicsP); + writeRealProfile(gen, e.getKey(), e.getValue()); } - resultsGenerator.writeEnd(); + gen.writeEndArray(); // Each discrete profile is an object in the array discreteProfiles - resultsGenerator.writeStartArray("discreteProfiles"); + gen.writeArrayFieldStart("discreteProfiles"); for (var e : results.discreteProfiles.entrySet()) { - writeProfile(resultsGenerator, e.getKey(), e.getValue(), serializedValueP); + writeDiscreteProfile(gen, e.getKey(), e.getValue()); } - resultsGenerator.writeEnd(); + gen.writeEndArray(); - resultsGenerator.writeEnd(); // end of profiles object + gen.writeEndObject(); // end of profiles object } - /** Write a single resource profile object */ - private void writeProfile( - JsonGenerator resultsGenerator, + /** Write a single real resource profile object */ + private void writeRealProfile( + JsonGenerator gen, String profileName, - ResourceProfile profile, - JsonParser dynamicsParser - ) { - resultsGenerator.writeStartObject() - .write("name", profileName) - .write("schema", valueSchemaP.unparse(profile.schema())) - .writeStartArray("segments"); + ResourceProfile profile + ) throws IOException { + gen.writeStartObject(); + gen.writeStringField("name", profileName); + gen.writeFieldName("schema"); + writeValueSchema(gen, profile.schema()); + gen.writeArrayFieldStart("segments"); if (resourceFileStreamer != null) { - // We expect RFS made a temp file where each line is a profile segment - var resourceTempFile = Path.of(resourceFileStreamer.getFileName(profileName)); - try (final var stream = Files.lines(resourceTempFile)) { - stream.forEach(s -> { - if (!s.isBlank()) { - // s is a JSON object for a single segment, write it as a value to the results generator - // Sadly, this requires reading the object into a JsonValue, just to write it back out (!) - try (final JsonReader jr = Json.createReader(new StringReader(s))) { - resultsGenerator.write(jr.readValue()); - } - } - }); - } catch (IOException ex) { - throw new RuntimeException(ex); + writeSegmentsFromFile(gen, profileName); + } else { + for (var s : profile.segments()) { + gen.writeStartObject(); + gen.writeStringField("extent", s.extent().toString()); + gen.writeFieldName("dynamics"); + gen.writeStartObject(); + gen.writeNumberField("initial", s.dynamics().initial); + gen.writeNumberField("rate", s.dynamics().rate); + gen.writeEndObject(); + gen.writeEndObject(); } - resourceTempFile.toFile().delete(); + } + + gen.writeEndArray(); + gen.writeEndObject(); + } + + /** Write a single discrete resource profile object */ + private void writeDiscreteProfile( + JsonGenerator gen, + String profileName, + ResourceProfile profile + ) throws IOException { + gen.writeStartObject(); + gen.writeStringField("name", profileName); + gen.writeFieldName("schema"); + writeValueSchema(gen, profile.schema()); + gen.writeArrayFieldStart("segments"); + + if (resourceFileStreamer != null) { + writeSegmentsFromFile(gen, profileName); } else { for (var s : profile.segments()) { - resultsGenerator.writeStartObject() - .write("extent", s.extent().toString()) - .write("dynamics", dynamicsParser.unparse(s.dynamics())) - .writeEnd(); + gen.writeStartObject(); + gen.writeStringField("extent", s.extent().toString()); + gen.writeFieldName("dynamics"); + s.dynamics().writeTo(gen); + gen.writeEndObject(); } } - resultsGenerator.writeEnd().writeEnd(); + gen.writeEndArray(); + gen.writeEndObject(); + } + + /** Write segments from a resource file streamer temp file */ + private void writeSegmentsFromFile(JsonGenerator gen, String profileName) throws IOException { + var resourceTempFile = Path.of(resourceFileStreamer.getFileName(profileName)); + try (final BufferedReader reader = Files.newBufferedReader(resourceTempFile)) { + String line; + while ((line = reader.readLine()) != null) { + if (!line.isBlank()) { + gen.writeRawValue(line); + } + } + } + resourceTempFile.toFile().delete(); } /** Write the spans section of the results file, containing all activity spans */ - private void writeSpans(JsonGenerator resultsGenerator) { - resultsGenerator.writeStartObject(); + private void writeSpans(JsonGenerator gen) throws IOException { + gen.writeStartObject(); // Each simulated activity is an object in the array simulatedActivities - resultsGenerator.writeStartArray("simulatedActivities"); + gen.writeArrayFieldStart("simulatedActivities"); for (var e : results.simulatedActivities.entrySet()) { final var id = e.getKey(); final var act = e.getValue(); - resultsGenerator.writeStartObject(); + gen.writeStartObject(); final var startOffset = Duration.of(plan.simulationStartTimestamp.microsUntil(new Timestamp(act.start())), Duration.MICROSECOND).toString(); final var endTime = act.start().plus(act.duration().in(Duration.MICROSECOND), ChronoUnit.MICROS).toString(); - resultsGenerator.write("id", id.id()); + gen.writeNumberField("id", id.id()); - resultsGenerator.writeKey("directiveId"); - act.directiveId().ifPresentOrElse( - did -> resultsGenerator.write(did.id()), - resultsGenerator::writeNull); + if (act.directiveId().isPresent()) { + gen.writeNumberField("directiveId", act.directiveId().get().id()); + } else { + gen.writeNullField("directiveId"); + } - resultsGenerator.writeKey("parentId"); if (act.parentId() != null) { - resultsGenerator.write(act.parentId().id()); + gen.writeNumberField("parentId", act.parentId().id()); } else { - resultsGenerator.writeNull(); + gen.writeNullField("parentId"); } - resultsGenerator.writeStartArray("childIds"); - for (var ci : act.childIds()) resultsGenerator.write(ci.id()); - resultsGenerator.writeEnd(); + gen.writeArrayFieldStart("childIds"); + for (var ci : act.childIds()) gen.writeNumber(ci.id()); + gen.writeEndArray(); + + gen.writeStringField("type", act.type()); + gen.writeStringField("startOffset", startOffset); + gen.writeStringField("duration", act.duration().toString()); - resultsGenerator - .write("type", act.type()) - .write("startOffset", startOffset) - .write("duration", act.duration().toString()) - .write("attributes", serializedValueP.unparse(act.computedAttributes())) - .write("arguments", activityArgumentsP.unparse(act.arguments())) - .write("startTime", act.start().toString()) - .write("endTime", endTime); + gen.writeFieldName("attributes"); + act.computedAttributes().writeTo(gen); - resultsGenerator.writeEnd(); + gen.writeFieldName("arguments"); + writeJsonValue(gen, activityArgumentsP.unparse(act.arguments())); + + gen.writeStringField("startTime", act.start().toString()); + gen.writeStringField("endTime", endTime); + + gen.writeEndObject(); } - resultsGenerator.writeEnd(); + gen.writeEndArray(); // Each unfinished activity is an object in the array unfinishedActivities - resultsGenerator.writeStartArray("unfinishedActivities"); + gen.writeArrayFieldStart("unfinishedActivities"); for (var e : results.unfinishedActivities.entrySet()) { final var id = e.getKey(); final var act = e.getValue(); - resultsGenerator.writeStartObject(); + gen.writeStartObject(); final var startOffset = Duration.of(plan.simulationStartTimestamp.microsUntil(new Timestamp(act.start())), Duration.MICROSECOND).toString(); - resultsGenerator.write("id", id.id()); + gen.writeNumberField("id", id.id()); - resultsGenerator.writeKey("directiveId"); - act.directiveId().ifPresentOrElse( - did -> resultsGenerator.write(did.id()), - resultsGenerator::writeNull); + if (act.directiveId().isPresent()) { + gen.writeNumberField("directiveId", act.directiveId().get().id()); + } else { + gen.writeNullField("directiveId"); + } - resultsGenerator.writeKey("parentId"); if (act.parentId() != null) { - resultsGenerator.write(act.parentId().id()); + gen.writeNumberField("parentId", act.parentId().id()); } else { - resultsGenerator.writeNull(); + gen.writeNullField("parentId"); } - resultsGenerator.writeStartArray("childIds"); - for (var ci : act.childIds()) resultsGenerator.write(ci.id()); - resultsGenerator.writeEnd(); + gen.writeArrayFieldStart("childIds"); + for (var ci : act.childIds()) gen.writeNumber(ci.id()); + gen.writeEndArray(); + + gen.writeStringField("type", act.type()); + gen.writeStringField("startOffset", startOffset); + + gen.writeFieldName("arguments"); + writeJsonValue(gen, activityArgumentsP.unparse(act.arguments())); - resultsGenerator - .write("type", act.type()) - .write("startOffset", startOffset) - .write("arguments", activityArgumentsP.unparse(act.arguments())) - .write("startTime", act.start().toString()); + gen.writeStringField("startTime", act.start().toString()); - resultsGenerator.writeEnd(); + gen.writeEndObject(); } - resultsGenerator.writeEnd(); + gen.writeEndArray(); - resultsGenerator.writeEnd(); // end of spans object + gen.writeEndObject(); // end of spans object } /** Write the topics section of the results */ - private void writeTopics(JsonGenerator resultsGenerator) { - resultsGenerator.writeStartObject(); + private void writeTopics(JsonGenerator gen) throws IOException { + gen.writeStartObject(); for (var t : results.topics) { - resultsGenerator.writeStartObject(t.getMiddle()) - .write("schema", valueSchemaP.unparse(t.getRight())) - .writeEnd(); + gen.writeObjectFieldStart(t.getMiddle()); + gen.writeFieldName("schema"); + writeValueSchema(gen, t.getRight()); + gen.writeEndObject(); } - resultsGenerator.writeEnd(); // end of topics object + gen.writeEndObject(); // end of topics object } /** Write the events section of the results */ - private void writeEvents(JsonGenerator resultsGenerator) { - resultsGenerator.writeStartArray(); + private void writeEvents(JsonGenerator gen) throws IOException { + gen.writeStartArray(); for (var e : results.events.entrySet()) { var realTime = e.getKey(); @@ -318,31 +361,51 @@ private void writeEvents(JsonGenerator resultsGenerator) { for (var entry : flattenedEventGraph) { var event = entry.getRight(); - resultsGenerator.writeStartObject() - .write("causalTime", entry.getLeft()) - .write("realTime", realTime.toString()) - .write("transactionIndex", transactionIndex) - .write("value", serializedValueP.unparse(event.value())); + gen.writeStartObject(); + gen.writeStringField("causalTime", entry.getLeft()); + gen.writeStringField("realTime", realTime.toString()); + gen.writeNumberField("transactionIndex", transactionIndex); + gen.writeFieldName("value"); + event.value().writeTo(gen); //grab the topic from the event's topic id results.topics .stream() .filter(topic -> topic.getLeft() == event.topicId()) .findFirst() - .ifPresent(topic -> resultsGenerator.write("topic", topic.getMiddle())); + .ifPresent(topic -> { + try { + gen.writeStringField("topic", topic.getMiddle()); + } catch (IOException ex) { + throw new RuntimeException(ex); + } + }); // optional span id - resultsGenerator.writeKey("spanId"); - event.spanId().ifPresentOrElse(resultsGenerator::write, resultsGenerator::writeNull); + if (event.spanId().isPresent()) { + gen.writeNumberField("spanId", event.spanId().get()); + } else { + gen.writeNullField("spanId"); + } - resultsGenerator.writeEnd(); // end of event object + gen.writeEndObject(); // end of event object } ++transactionIndex; } } - resultsGenerator.writeEnd(); // end of events array + gen.writeEndArray(); // end of events array + } + + /** Write a javax.json JsonValue into a Jackson generator by converting to string */ + private static void writeJsonValue(JsonGenerator gen, javax.json.JsonValue jsonValue) throws IOException { + gen.writeRawValue(jsonValue.toString()); + } + + /** Write a ValueSchema to a Jackson generator */ + private static void writeValueSchema(JsonGenerator gen, ValueSchema schema) throws IOException { + writeJsonValue(gen, valueSchemaP.unparse(schema)); } } diff --git a/scheduler-driver/src/main/java/gov/nasa/jpl/aerie/scheduler/constraints/activities/ActivityExpression.java b/scheduler-driver/src/main/java/gov/nasa/jpl/aerie/scheduler/constraints/activities/ActivityExpression.java index 573e58aadb..58ea3769b2 100644 --- a/scheduler-driver/src/main/java/gov/nasa/jpl/aerie/scheduler/constraints/activities/ActivityExpression.java +++ b/scheduler-driver/src/main/java/gov/nasa/jpl/aerie/scheduler/constraints/activities/ActivityExpression.java @@ -538,6 +538,12 @@ public Boolean onNumeric(final BigDecimal value) { return argumentsAsNumeric.map(bigDecimal -> bigDecimal.equals(value)).orElse(false); } + @Override + public Boolean onDouble(final double value) { + final var argumentsAsReal = superset.asReal(); + return argumentsAsReal.map(realValue -> realValue.equals(value)).orElse(false); + } + @Override public Boolean onBoolean(final boolean value) { final var argumentsAsBoolean = superset.asBoolean(); diff --git a/scheduler-driver/src/test/java/gov/nasa/jpl/aerie/scheduler/simulation/SimulationResultsComparisonUtils.java b/scheduler-driver/src/test/java/gov/nasa/jpl/aerie/scheduler/simulation/SimulationResultsComparisonUtils.java index 3e4a983042..872a00e82f 100644 --- a/scheduler-driver/src/test/java/gov/nasa/jpl/aerie/scheduler/simulation/SimulationResultsComparisonUtils.java +++ b/scheduler-driver/src/test/java/gov/nasa/jpl/aerie/scheduler/simulation/SimulationResultsComparisonUtils.java @@ -87,12 +87,17 @@ public SerializedValue onNull() { } @Override - public SerializedValue onNumeric(final BigDecimal value) { + public SerializedValue onNumeric(BigDecimal value) { return SerializedValue.of(value); } @Override - public SerializedValue onBoolean(final boolean value) { + public SerializedValue onDouble(double value) { + return SerializedValue.of(value); + } + + @Override + public SerializedValue onBoolean(boolean value) { return SerializedValue.of(value); } diff --git a/scheduler-server/src/main/java/gov/nasa/jpl/aerie/scheduler/server/services/TypescriptCodeGenerationService.java b/scheduler-server/src/main/java/gov/nasa/jpl/aerie/scheduler/server/services/TypescriptCodeGenerationService.java index b18e0bf61c..e3284108c6 100644 --- a/scheduler-server/src/main/java/gov/nasa/jpl/aerie/scheduler/server/services/TypescriptCodeGenerationService.java +++ b/scheduler-server/src/main/java/gov/nasa/jpl/aerie/scheduler/server/services/TypescriptCodeGenerationService.java @@ -321,6 +321,11 @@ public String onNull() { @Override public String onNumeric(final BigDecimal value) { return "number"; + @Override + public String onDouble(final double value) { + return "number"; + } + } @Override