Help decoding multiple messages using rust by resolving #987 (#991)

* Add new arguments for generated MessageDecoder#header method When decoding multiple messages from a single byte buffer, the second and following message decoder should offset to the new header offset plus header length. * Update code to reflect method signature change in decoder's header method * Add test for introduced changes --------- Co-authored-by: Elvis Wang <[email protected]>
aeron-io · Oct 25, 2024 · e34fc03 · e34fc03
1 parent 956a46b
commit e34fc03
Show file tree

Hide file tree

Showing 16 changed files with 112 additions and 14 deletions.
diff --git a/build.gradle b/build.gradle
@@ -652,6 +652,7 @@ tasks.register('generateRustTestCodecs', JavaExec) {
             'sbe-tool/src/test/resources/issue895.xml',
             'sbe-tool/src/test/resources/issue972.xml',
             'sbe-tool/src/test/resources/issue984.xml',
+            'sbe-tool/src/test/resources/issue987.xml',
             'sbe-tool/src/test/resources/example-bigendian-test-schema.xml',
             'sbe-tool/src/test/resources/nested-composite-name.xml',
     ]

diff --git a/rust/Cargo.toml b/rust/Cargo.toml
@@ -15,6 +15,7 @@ issue_435 = { path = "../generated/rust/issue435" }
 issue_895 = { path = "../generated/rust/issue895" }
 issue_972 = { path = "../generated/rust/issue972" }
 issue_984 = { path = "../generated/rust/issue984" }
+issue_987 = { path = "../generated/rust/issue987" }
 baseline_bigendian = { path = "../generated/rust/baseline-bigendian" }
 nested_composite_name = { path = "../generated/rust/nested-composite-name" }
 
@@ -27,4 +28,4 @@ harness = false
 
 [[bench]]
 name = "md_benchmark"
-harness = false
+harness = false
diff --git a/rust/benches/car_benchmark.rs b/rust/benches/car_benchmark.rs
@@ -131,7 +131,7 @@ fn decode(state: &State) -> SbeResult<usize> {
 
     let buf = ReadBuf::new(state.buffer.as_slice());
     let header = MessageHeaderDecoder::default().wrap(buf, 0);
-    car = car.header(header);
+    car = car.header(header, 0);
 
     // Car...
     car.serial_number();

diff --git a/rust/benches/md_benchmark.rs b/rust/benches/md_benchmark.rs
@@ -90,7 +90,7 @@ fn decode_md(state: &State) -> SbeResult<usize> {
 
     let buf = ReadBuf::new(state.buffer.as_slice());
     let header = MessageHeaderDecoder::default().wrap(buf, 0);
-    market_data = market_data.header(header);
+    market_data = market_data.header(header, 0);
 
     market_data.transact_time();
     market_data.event_time_delta();

diff --git a/rust/car_example_baseline_data.sbe b/rust/car_example_baseline_data.sbe
diff --git a/rust/car_example_extension_data.sbe b/rust/car_example_extension_data.sbe
diff --git a/rust/tests/baseline_test.rs b/rust/tests/baseline_test.rs
@@ -44,7 +44,7 @@ fn decode_car_and_assert_expected_content(buffer: &[u8]) -> SbeResult<()> {
     let buf = ReadBuf::new(buffer);
     let header = MessageHeaderDecoder::default().wrap(buf, 0);
     assert_eq!(SBE_TEMPLATE_ID, header.template_id());
-    car = car.header(header);
+    car = car.header(header, 0);
 
     // Car...
     assert_eq!(1234, car.serial_number());

diff --git a/rust/tests/big_endian_test.rs b/rust/tests/big_endian_test.rs
@@ -43,7 +43,7 @@ fn decode_car_and_assert_expected_content(buffer: &[u8]) -> SbeResult<()> {
     let buf = ReadBuf::new(buffer);
     let header = MessageHeaderDecoder::default().wrap(buf, 0);
     assert_eq!(SBE_TEMPLATE_ID, header.template_id());
-    car = car.header(header);
+    car = car.header(header, 0);
 
     // Car...
     assert_eq!(1234, car.serial_number());

diff --git a/rust/tests/extension_test.rs b/rust/tests/extension_test.rs
@@ -43,7 +43,7 @@ fn decode_car_and_assert_expected_content(buffer: &[u8]) -> SbeResult<()> {
     let buf = ReadBuf::new(buffer);
     let header = MessageHeaderDecoder::default().wrap(buf, 0);
     assert_eq!(SBE_TEMPLATE_ID, header.template_id());
-    car = car.header(header);
+    car = car.header(header, 0);
 
     // Car...
     assert_eq!(1234, car.serial_number());

diff --git a/rust/tests/issue_435_test.rs b/rust/tests/issue_435_test.rs
@@ -33,7 +33,7 @@ fn issue_435_ref_test() -> SbeResult<()> {
     assert_eq!(SBE_SCHEMA_ID, header.schema_id());
     assert_eq!(*SetRef::default().set_one(true), header.s());
 
-    let decoder = Issue435Decoder::default().header(header);
+    let decoder = Issue435Decoder::default().header(header, 0);
     assert_eq!(EnumRef::Two, decoder.example_decoder().e());
 
     Ok(())

diff --git a/rust/tests/issue_895_test.rs b/rust/tests/issue_895_test.rs
@@ -27,7 +27,7 @@ fn issue_895_both_some() -> SbeResult<()> {
     assert_eq!(SBE_TEMPLATE_ID, header.template_id());
     assert_eq!(SBE_SCHEMA_ID, header.schema_id());
 
-    let decoder = Issue895Decoder::default().header(header);
+    let decoder = Issue895Decoder::default().header(header, 0);
     assert_eq!(Some(2.07), decoder.optional_float());
     assert_eq!(Some(4.12), decoder.optional_double());
 
@@ -46,7 +46,7 @@ fn issue_895_float_none() -> SbeResult<()> {
     let buf = ReadBuf::new(buf.as_slice());
     let header = MessageHeaderDecoder::default().wrap(buf, 0);
 
-    let decoder = Issue895Decoder::default().header(header);
+    let decoder = Issue895Decoder::default().header(header, 0);
     assert_eq!(None, decoder.optional_float());
     assert_eq!(Some(4.12), decoder.optional_double());
 
@@ -65,7 +65,7 @@ fn issue_895_double_none() -> SbeResult<()> {
     let buf = ReadBuf::new(buffer.as_slice());
     let header = MessageHeaderDecoder::default().wrap(buf, 0);
 
-    let decoder = Issue895Decoder::default().header(header);
+    let decoder = Issue895Decoder::default().header(header, 0);
     assert_eq!(Some(2.07), decoder.optional_float());
     assert_eq!(None, decoder.optional_double());
 

diff --git a/rust/tests/issue_972_test.rs b/rust/tests/issue_972_test.rs
@@ -26,7 +26,7 @@ fn round_trip() -> SbeResult<()> {
     assert_eq!(SBE_TEMPLATE_ID, header.template_id());
     assert_eq!(SBE_SCHEMA_ID, header.schema_id());
 
-    let decoder = Issue972Decoder::default().header(header);
+    let decoder = Issue972Decoder::default().header(header, 0);
     if let Either::Right(composite) = decoder.new_field_decoder() {
         assert_eq!(2007, composite.f1().unwrap());
         assert_eq!(2012, composite.f2().unwrap());

diff --git a/rust/tests/issue_984_test.rs b/rust/tests/issue_984_test.rs
@@ -34,7 +34,7 @@ fn round_trip() -> SbeResult<()> {
     assert_eq!(SBE_TEMPLATE_ID, header.template_id());
     assert_eq!(SBE_SCHEMA_ID, header.schema_id());
 
-    let simple_msg_decoder = SimpleMessageDecoder::default().header(header);
+    let simple_msg_decoder = SimpleMessageDecoder::default().header(header, 0);
     assert_eq!(1985, simple_msg_decoder.id());
     let mut grp_decoder = simple_msg_decoder.my_group_decoder();
     assert_eq!(1, grp_decoder.count());

diff --git a/rust/tests/issue_987_test.rs b/rust/tests/issue_987_test.rs
@@ -0,0 +1,71 @@
+use issue_987::{
+    issue_987_codec::{Issue987Decoder, Issue987Encoder, SBE_BLOCK_LENGTH, SBE_SCHEMA_ID, SBE_SCHEMA_VERSION, SBE_TEMPLATE_ID}, message_header_codec::MessageHeaderDecoder, *
+};
+
+fn create_encoder(buffer: &mut Vec<u8>, off: usize) -> Issue987Encoder {
+    let encoder = Issue987Encoder::default().wrap(
+        WriteBuf::new(buffer.as_mut_slice()),
+        off + message_header_codec::ENCODED_LENGTH,
+    );
+    let mut header = encoder.header(off);
+    header.parent().unwrap()
+}
+
+#[test]
+fn encode_2_messages_and_then_decode() -> SbeResult<()> {
+    let mut buffer = vec![0u8; 256];
+
+    let mut off = 0;
+
+    // encode message 1
+    let mut encoder = create_encoder(&mut buffer, off);
+    encoder.old_field(10);
+    let mut other_encoder = encoder.new_field_encoder();
+    other_encoder.f1(11);
+    other_encoder.f2(12);
+    let encoder = other_encoder.parent().unwrap();
+
+    // Update offset
+    off += message_header_codec::ENCODED_LENGTH + encoder.encoded_length();
+
+    // encode message 2
+    let mut encoder = create_encoder(&mut buffer, off);
+    encoder.old_field(20);
+    let mut other_encoder = encoder.new_field_encoder();
+    other_encoder.f1(21);
+    other_encoder.f2(22);
+
+    // decoding ...
+    off = 0;
+
+    // decode message 1
+    let buf = ReadBuf::new(buffer.as_slice());
+    let header = MessageHeaderDecoder::default().wrap(buf, off);
+    assert_eq!(SBE_BLOCK_LENGTH, header.block_length());
+    assert_eq!(SBE_SCHEMA_VERSION, header.version());
+    assert_eq!(SBE_TEMPLATE_ID, header.template_id());
+    assert_eq!(SBE_SCHEMA_ID, header.schema_id());
+    let decoder = Issue987Decoder::default().header(header, off);
+    assert_eq!(10, decoder.old_field());
+    let other_decoder = decoder.new_field_decoder();
+    assert_eq!(11, other_decoder.f1());
+    assert_eq!(12, other_decoder.f2());
+
+    // Update offset
+    off += message_header_codec::ENCODED_LENGTH + decoder.encoded_length();
+
+    // decode message 2
+    let buf = ReadBuf::new(buffer.as_slice());
+    let header = MessageHeaderDecoder::default().wrap(buf, off);
+    assert_eq!(SBE_BLOCK_LENGTH, header.block_length());
+    assert_eq!(SBE_SCHEMA_VERSION, header.version());
+    assert_eq!(SBE_TEMPLATE_ID, header.template_id());
+    assert_eq!(SBE_SCHEMA_ID, header.schema_id());
+    let decoder = Issue987Decoder::default().header(header, off);
+    assert_eq!(20, decoder.old_field());
+    let other_decoder = decoder.new_field_decoder();
+    assert_eq!(21, other_decoder.f1());
+    assert_eq!(22, other_decoder.f2());
+
+    Ok(())
+}
diff --git a/sbe-tool/src/main/java/uk/co/real_logic/sbe/generation/rust/MessageCoderDef.java b/sbe-tool/src/main/java/uk/co/real_logic/sbe/generation/rust/MessageCoderDef.java
@@ -141,13 +141,13 @@ void appendMessageHeaderEncoderFn(final Appendable out) throws IOException
 
     void appendMessageHeaderDecoderFn(final Appendable out) throws IOException
     {
-        indent(out, 2, "pub fn header(self, mut header: MessageHeaderDecoder<ReadBuf<'a>>) -> Self {\n");
+        indent(out, 2, "pub fn header(self, mut header: MessageHeaderDecoder<ReadBuf<'a>>, offset: usize) -> Self {\n");
         indent(out, 3, "debug_assert_eq!(SBE_TEMPLATE_ID, header.template_id());\n");
         indent(out, 3, "let acting_block_length = header.block_length();\n");
         indent(out, 3, "let acting_version = header.version();\n\n");
         indent(out, 3, "self.wrap(\n");
         indent(out, 4, "header.parent().unwrap(),\n");
-        indent(out, 4, "message_header_codec::ENCODED_LENGTH,\n");
+        indent(out, 4, "offset + message_header_codec::ENCODED_LENGTH,\n");
         indent(out, 4, "acting_block_length,\n");
         indent(out, 4, "acting_version,\n");
         indent(out, 3, ")\n");

diff --git a/sbe-tool/src/test/resources/issue987.xml b/sbe-tool/src/test/resources/issue987.xml
@@ -0,0 +1,25 @@
+<sbe:messageSchema
+    xmlns:sbe="http://fixprotocol.io/2016/sbe"
+    package="issue987"
+    id="987"
+    version="1"
+    semanticVersion="1.0.0"
+    byteOrder="littleEndian">
+  <types>
+    <composite name="messageHeader">
+      <type name="blockLength" primitiveType="uint16" description="Length of the root of the FIX message contained before repeating groups or variable/conditions fields."/>
+      <type name="templateId" primitiveType="uint16" description="Template ID used to encode the message."/>
+      <type name="schemaId" primitiveType="uint16" description="ID of the system publishing the message."/>
+      <type name="version" primitiveType="uint16" description="Schema version."/>
+    </composite>
+
+    <composite name="NewComposite">
+      <type name="f1" primitiveType="uint16" semanticType="Int"/>
+      <type name="f2" primitiveType="uint32" semanticType="Int" offset="4"/>
+    </composite>
+  </types>
+  <sbe:message name="issue987" id="1">
+    <field name="oldField" id="1" type="uint16" semanticType="Int"/>
+    <field name="newField" id="2" type="NewComposite" />
+  </sbe:message>
+</sbe:messageSchema>