Check

real-logic · Jan 24, 2024 · 35060d0 · 35060d0
1 parent 6379ad3
commit 35060d0
Show file tree

Hide file tree

Showing 7 changed files with 521 additions and 8 deletions.
diff --git a/agrona/build.gradle.kts b/agrona/build.gradle.kts
@@ -1,17 +1,28 @@
-
 plugins {
     application
     checkstyle
 }
 
+
+@Suppress("DEPRECATION")
+val generatedDir = file("${buildDir}/generated/src/main/java")
+val codecGeneration = configurations.create("codecGeneration")
+
 dependencies {
+    "codecGeneration"(libs.sbe)
     checkstyle(libs.checkstyle)
     implementation(libs.agrona)
     implementation(libs.slf4j)
     implementation(libs.logback)
     testImplementation(libs.bundles.testing)
 }
 
+sourceSets {
+    main {
+        java.srcDir(generatedDir)
+    }
+}
+
 testing {
     suites {
         // Configure the built-in test suite
@@ -22,3 +33,25 @@ testing {
         }
     }
 }
+
+tasks {
+    task("generateCodecs", JavaExec::class) {
+        group = "sbe"
+        val codecsFile = "src/main/resources/messages.xml"
+        val sbeFile = "src/main/resources/sbe/sbe.xsd"
+        inputs.files(codecsFile, sbeFile)
+        outputs.dir(generatedDir)
+        classpath = codecGeneration
+        mainClass.set("uk.co.real_logic.sbe.SbeTool")
+        args = listOf(codecsFile)
+        systemProperties["sbe.output.dir"] = generatedDir
+        systemProperties["sbe.target.language"] = "Java"
+        systemProperties["sbe.validation.xsd"] = sbeFile
+        systemProperties["sbe.validation.stop.on.error"] = "true"
+        outputs.dir(generatedDir)
+    }
+
+    compileJava {
+        dependsOn("generateCodecs")
+    }
+}
diff --git a/agrona/src/main/java/com/aeroncookbook/agrona/ringbuffer/ReceiveAgent.java b/agrona/src/main/java/com/aeroncookbook/agrona/ringbuffer/ReceiveAgent.java
@@ -16,6 +16,8 @@
 
 package com.aeroncookbook.agrona.ringbuffer;
 
+import com.aeroncookbook.sbe.MessageHeaderDecoder;
+import com.aeroncookbook.sbe.SampleSimpleDecoder;
 import org.agrona.DirectBuffer;
 import org.agrona.concurrent.Agent;
 import org.agrona.concurrent.ShutdownSignalBarrier;
@@ -29,6 +31,8 @@ public class ReceiveAgent implements Agent
     private final ManyToOneRingBuffer ringBuffer;
     private final int sendCount;
     private final Logger logger = LoggerFactory.getLogger(ReceiveAgent.class);
+    private final MessageHeaderDecoder headerDecoder = new MessageHeaderDecoder();
+    private final SampleSimpleDecoder sampleSimpleDecoder = new SampleSimpleDecoder();
 
     public ReceiveAgent(final ManyToOneRingBuffer ringBuffer, final ShutdownSignalBarrier barrier, final int sendCount)
     {
@@ -46,11 +50,12 @@ public int doWork() throws Exception
 
     private void handler(final int messageType, final DirectBuffer buffer, final int offset, final int length)
     {
-        final int lastValue = buffer.getInt(offset);
+        sampleSimpleDecoder.wrapAndApplyHeader(buffer, offset, headerDecoder);
+        final long count = sampleSimpleDecoder.sequence();
 
-        if (lastValue == sendCount)
+        if (count == sendCount)
         {
-            logger.info("received: {}", lastValue);
+            logger.info("received: {}", count);
             barrier.signal();
         }
     }

diff --git a/agrona/src/main/java/com/aeroncookbook/agrona/ringbuffer/SendAgent1.java b/agrona/src/main/java/com/aeroncookbook/agrona/ringbuffer/SendAgent1.java
@@ -16,6 +16,9 @@
 
 package com.aeroncookbook.agrona.ringbuffer;
 
+import com.aeroncookbook.sbe.MessageHeaderEncoder;
+import com.aeroncookbook.sbe.SampleEnum;
+import com.aeroncookbook.sbe.SampleSimpleEncoder;
 import org.agrona.concurrent.Agent;
 import org.agrona.concurrent.AtomicBuffer;
 import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
@@ -24,6 +27,8 @@ public class SendAgent1 implements Agent
 {
     private final int sendCount;
     private final ManyToOneRingBuffer ringBuffer;
+    private final MessageHeaderEncoder messageHeaderEncoder = new MessageHeaderEncoder();
+    private final SampleSimpleEncoder sampleSimpleEncoder = new SampleSimpleEncoder();
     private int currentCountItem = 1;
 
     public SendAgent1(final ManyToOneRingBuffer ringBuffer, final int sendCount)
@@ -40,12 +45,15 @@ public int doWork()
             return 0;
         }
 
-        final int claimIndex = ringBuffer.tryClaim(1, Integer.BYTES);
+        final int claimIndex = ringBuffer.tryClaim(1, sampleSimpleEncoder.encodedLength() +
+            MessageHeaderEncoder.ENCODED_LENGTH);
         if (claimIndex > 0)
         {
             currentCountItem += 1;
             final AtomicBuffer buffer = ringBuffer.buffer();
-            buffer.putInt(claimIndex, currentCountItem);
+            sampleSimpleEncoder.wrapAndApplyHeader(buffer, claimIndex, messageHeaderEncoder);
+            sampleSimpleEncoder.sequence(currentCountItem);
+            sampleSimpleEncoder.enumField(SampleEnum.VALUE_1);
             ringBuffer.commit(claimIndex);
         }
         return 0;

diff --git a/agrona/src/main/java/com/aeroncookbook/agrona/ringbuffer/SendAgent2.java b/agrona/src/main/java/com/aeroncookbook/agrona/ringbuffer/SendAgent2.java
@@ -16,6 +16,9 @@
 
 package com.aeroncookbook.agrona.ringbuffer;
 
+import com.aeroncookbook.sbe.MessageHeaderEncoder;
+import com.aeroncookbook.sbe.SampleEnum;
+import com.aeroncookbook.sbe.SampleSimpleEncoder;
 import org.agrona.concurrent.Agent;
 import org.agrona.concurrent.AtomicBuffer;
 import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
@@ -25,6 +28,8 @@ public class SendAgent2 implements Agent
     private final int sendCount;
     private final ManyToOneRingBuffer ringBuffer;
     private int currentCountItem;
+    private final MessageHeaderEncoder messageHeaderEncoder = new MessageHeaderEncoder();
+    private final SampleSimpleEncoder sampleSimpleEncoder = new SampleSimpleEncoder();
 
     public SendAgent2(final ManyToOneRingBuffer ringBuffer, final int sendCount)
     {
@@ -46,7 +51,9 @@ public int doWork()
         {
             currentCountItem -= 1;
             final AtomicBuffer buffer = ringBuffer.buffer();
-            buffer.putInt(claimIndex, currentCountItem);
+            sampleSimpleEncoder.wrapAndApplyHeader(buffer, claimIndex, messageHeaderEncoder);
+            sampleSimpleEncoder.sequence(currentCountItem);
+            sampleSimpleEncoder.enumField(SampleEnum.VALUE_1);
             ringBuffer.commit(claimIndex);
         }
         return 0;

diff --git a/agrona/src/main/java/com/aeroncookbook/agrona/ringbuffer/StartHere.java b/agrona/src/main/java/com/aeroncookbook/agrona/ringbuffer/StartHere.java
@@ -34,7 +34,7 @@ public class StartHere
 
     public static void main(final String[] args)
     {
-        final int sendCount = 10_000_000;
+        final int sendCount = 20_000_000;
         final int bufferLength = 16384 + RingBufferDescriptor.TRAILER_LENGTH;
         final UnsafeBuffer unsafeBuffer = new UnsafeBuffer(ByteBuffer.allocateDirect(bufferLength));
         final IdleStrategy idleStrategySend1 = new BusySpinIdleStrategy();

diff --git a/agrona/src/main/resources/messages.xml b/agrona/src/main/resources/messages.xml
@@ -0,0 +1,56 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<!--
+  ~ Copyright 2019-2023 Adaptive Financial Consulting Ltd.
+  ~
+  ~ Licensed under the Apache License, Version 2.0 (the "License");
+  ~ you may not use this file except in compliance with the License.
+  ~ You may obtain a copy of the License at
+  ~
+  ~ https://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  -->
+
+<sbe:messageSchema xmlns:sbe="http://fixprotocol.io/2016/sbe"
+                   package="com.aeroncookbook.sbe"
+                   id="688"
+                   version="1"
+                   semanticVersion="0.1"
+                   description="Sample SBE Messages"
+                   byteOrder="littleEndian">
+    <types>
+        <composite name="messageHeader" description="Message identifiers and length of message root">
+            <type name="blockLength" primitiveType="uint16"/>
+            <type name="templateId" primitiveType="uint16"/>
+            <type name="schemaId" primitiveType="uint16"/>
+            <type name="version" primitiveType="uint16"/>
+        </composite>
+        <composite name="varStringEncoding">
+            <type name="length" primitiveType="uint32" maxValue="1073741824"/>
+            <type name="varData" primitiveType="uint8" length="0" characterEncoding="UTF-8"/>
+        </composite>
+        <composite name="groupSizeEncoding" description="Repeating group dimensions.">
+            <type name="blockLength" primitiveType="uint16"/>
+            <type name="numInGroup" primitiveType="uint16"/>
+        </composite>
+        <enum name="SampleEnum" encodingType="int32">
+            <validValue name="VALUE_1">1</validValue>
+            <validValue name="VALUE_2">2</validValue>
+            <validValue name="VALUE_3">3</validValue>
+        </enum>
+    </types>
+    <types>
+        <type name="Sequence" primitiveType="int64"/>
+    </types>
+
+    <sbe:message name="SampleSimple" id="1" description="Simple sample">
+        <field name="sequence" id="1" type="Sequence"/>
+        <field name="enumField" id="2" type="SampleEnum"/>
+        <data name="message" id="3" type="varStringEncoding"/>
+    </sbe:message>
+
+</sbe:messageSchema>