Aeron Client to 1.43.0

driver/Aeron.Driver.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>Aeron.Driver</id>
-    <version>1.40.0</version>
+    <version>1.43.0</version>
     <title>Aeron Driver</title>
     <authors>Adaptive Financial Consulting Ltd.</authors>
     <owners>Adaptive Financial Consulting Ltd.</owners>
@@ -11,7 +11,7 @@
     <iconUrl>https://raw.githubusercontent.com/AdaptiveConsulting/Aeron.NET/master/images/adaptive.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>Aeron Media Driver for use with Aeron.Client.</description>
-    <copyright>Copyright 2023</copyright>
+    <copyright>Copyright 2024</copyright>
     <tags>aeron driver messaging networking transport udp ipc</tags>
 
     <contentFiles>

driver/version.txt

@@ -1,2 +1,2 @@
 Driver source:
-https://repo1.maven.org/maven2/io/aeron/aeron-all/1.40.0/aeron-all-1.40.0.jar
+https://repo1.maven.org/maven2/io/aeron/aeron-all/1.43.0/aeron-all-1.43.0.jar

src/Adaptive.Aeron.Tests/Adaptive.Aeron.Tests.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
     <PropertyGroup>
-        <TargetFramework>netcoreapp3.1</TargetFramework>
+        <TargetFramework>net6.0</TargetFramework>
     </PropertyGroup>
 
     <ItemGroup>

src/Adaptive.Aeron.Tests/ClientConductorTest.cs

@@ -74,7 +74,7 @@ public class ClientConductorTest
 
         private readonly TestEpochClock EpochClock = new TestEpochClock();
         private readonly TestNanoClock NanoClock = new TestNanoClock();
-        private ErrorHandler MockClientErrorHandler;
+        private IErrorHandler MockClientErrorHandler;
 
         private ClientConductor Conductor;
         private DriverProxy DriverProxy;
@@ -84,20 +84,32 @@ public class ClientConductorTest
         private ILogBuffersFactory LogBuffersFactory;
         private ILock mockClientLock = A.Fake<ILock>();
         private Aeron MockAeron;
+
         private bool SuppressPrintError = false;
+
+        private class PrintingErrorHandler : IErrorHandler
+        {
+            private readonly ClientConductorTest _test;
 
+            public PrintingErrorHandler(ClientConductorTest test)
+            {
+                _test = test;
+            }
 
-        [SetUp]
-        public void SetUp()
-        {
-            MockClientErrorHandler = A.Fake<ErrorHandler>(options => options.Wrapping(throwable =>
+            public void OnError(Exception throwable)
             {
-                if (!SuppressPrintError)
+                if (!_test.SuppressPrintError)
                 {
                     Console.WriteLine(throwable.ToString());
                     Console.Write(throwable.StackTrace);
                 }
-            }));
+            }
+        }
+
+        [SetUp]
+        public void SetUp()
+        {
+            MockClientErrorHandler = A.Fake<IErrorHandler>(options => options.Wrapping(new PrintingErrorHandler(this)));
 
             PublicationReady = new PublicationBuffersReadyFlyweight();
             SubscriptionReady = new SubscriptionReadyFlyweight();
@@ -492,7 +504,7 @@ public void ShouldTimeoutInterServiceIfTooLongBetweenDoWorkCalls()
             NanoClock.AdvancedNanos(1);
             Conductor.DoWork();
 
-            A.CallTo(() => MockClientErrorHandler(A<ConductorServiceTimeoutException>._)).MustHaveHappened();
+            A.CallTo(() => MockClientErrorHandler.OnError(A<ConductorServiceTimeoutException>._)).MustHaveHappened();
 
             Assert.True(Conductor.IsTerminating());
         }
@@ -504,7 +516,7 @@ public void ShouldTerminateAndErrorOnClientTimeoutFromDriver()
 
             Conductor.OnClientTimeout();
 
-            A.CallTo(() => MockClientErrorHandler.Invoke(A<Exception>._)).MustHaveHappened();
+            A.CallTo(() => MockClientErrorHandler.OnError(A<Exception>._)).MustHaveHappened();
 
             bool threwException = false;
             try
@@ -537,7 +549,7 @@ public void ShouldNotCloseAndErrorOnClientTimeoutForAnotherClientIdFromDriver()
 
             Conductor.DoWork();
 
-            A.CallTo(() => MockClientErrorHandler.Invoke(A<Exception>._)).MustNotHaveHappened();
+            A.CallTo(() => MockClientErrorHandler.OnError(A<Exception>._)).MustNotHaveHappened();
 
             Assert.False(Conductor.IsClosed());
         }

src/Adaptive.Aeron.Tests/FlyweightTests.cs

@@ -15,6 +15,7 @@
  */
 
 using Adaptive.Aeron.Protocol;
+using Adaptive.Agrona;
 using Adaptive.Agrona.Concurrent;
 using NUnit.Framework;
 
@@ -38,7 +39,7 @@ private void InitializeInstanceFields()
             aBuff = new UnsafeBuffer(buffer);
         }
 
-        private readonly byte[] buffer = new byte[512];
+        private readonly ByteBuffer buffer = BufferUtil.Allocate(512);
 
         private UnsafeBuffer aBuff;
         private readonly HeaderFlyweight encodeHeader = new HeaderFlyweight();
@@ -57,14 +58,14 @@ public void ShouldWriteCorrectValuesForGenericHeaderFields()
             encodeHeader.FrameLength(8);
 
             // little endian
-            Assert.AreEqual((byte)0x08, buffer[0]);
-            Assert.AreEqual((byte)0x00, buffer[1]);
-            Assert.AreEqual((byte)0x00, buffer[2]);
-            Assert.AreEqual((byte)0x00, buffer[3]);
-            Assert.AreEqual((byte)0x01, buffer[4]);
-            Assert.AreEqual((byte)0xC0, buffer[5]);
-            Assert.AreEqual((byte)HeaderFlyweight.HDR_TYPE_DATA, buffer[6]);
-            Assert.AreEqual((byte)0x00, buffer[7]);
+            Assert.AreEqual((byte)0x08, buffer.Get(0));
+            Assert.AreEqual((byte)0x00, buffer.Get(1));
+            Assert.AreEqual((byte)0x00, buffer.Get(2));
+            Assert.AreEqual((byte)0x00, buffer.Get(3));
+            Assert.AreEqual((byte)0x01, buffer.Get(4));
+            Assert.AreEqual((byte)0xC0, buffer.Get(5));
+            Assert.AreEqual((byte)HeaderFlyweight.HDR_TYPE_DATA, buffer.Get(6));
+            Assert.AreEqual((byte)0x00, buffer.Get(7));
         }
 
         [Test]

src/Adaptive.Aeron.Tests/ImageTest.cs

@@ -58,7 +58,7 @@ static ImageTest()
         private IControlledFragmentHandler MockControlledFragmentHandler;
         private IPosition Position;
         private LogBuffers LogBuffers;
-        private ErrorHandler ErrorHandler;
+        private IErrorHandler ErrorHandler;
         private Subscription Subscription;
 
         private UnsafeBuffer[] TermBuffers;
@@ -72,7 +72,7 @@ public void SetUp()
             MockControlledFragmentHandler = A.Fake<IControlledFragmentHandler>();
             Position = A.Fake<IPosition>(options => options.Wrapping(new AtomicLongPosition()));
             LogBuffers = A.Fake<LogBuffers>();
-            ErrorHandler = A.Fake<ErrorHandler>();
+            ErrorHandler = A.Fake<IErrorHandler>();
             Subscription = A.Fake<Subscription>();
 
             TermBuffers = new UnsafeBuffer[LogBufferDescriptor.PARTITION_COUNT];

src/Adaptive.Aeron.Tests/LogBuffer/TermReaderTest.cs

@@ -35,7 +35,7 @@ public class TermReaderTest
 
         private Header header;
         private UnsafeBuffer termBuffer;
-        private ErrorHandler errorHandler;
+        private IErrorHandler errorHandler;
         private IFragmentHandler handler;
         private IPosition subscriberPosition;
 
@@ -44,7 +44,7 @@ public void SetUp()
         {
             header = new Header(INITIAL_TERM_ID, TERM_BUFFER_CAPACITY);
             termBuffer = A.Fake<UnsafeBuffer>();
-            errorHandler = A.Fake<ErrorHandler>();
+            errorHandler = A.Fake<IErrorHandler>();
             handler = A.Fake<IFragmentHandler>();
             subscriberPosition = A.Fake<IPosition>();
 

src/Adaptive.Aeron.Tests/LogBuffer/TermRebuilderTest.cs

@@ -40,7 +40,7 @@ public void SetUp()
         [Test]
         public void ShouldInsertIntoEmptyBuffer()
         {
-            UnsafeBuffer packet = new UnsafeBuffer(new byte[256]);
+            UnsafeBuffer packet = new UnsafeBuffer(BufferUtil.Allocate(256));
             const int termOffset = 0;
             const int srcOffset = 0;
             const int length = 256;
@@ -63,7 +63,7 @@ public void ShouldInsertLastFrameIntoBuffer()
             const int srcOffset = 0;
             int tail = TERM_BUFFER_CAPACITY - frameLength;
             int termOffset = tail;
-            UnsafeBuffer packet = new UnsafeBuffer(new byte[frameLength]);
+            UnsafeBuffer packet = new UnsafeBuffer(BufferUtil.Allocate(frameLength));
             packet.PutShort(FrameDescriptor.TypeOffset(srcOffset), (short)FrameDescriptor.PADDING_FRAME_TYPE);
             packet.PutInt(srcOffset, frameLength);
 
@@ -80,7 +80,7 @@ public void ShouldFillSingleGap()
             const int srcOffset = 0;
             int tail = alignedFrameLength;
             int termOffset = tail;
-            UnsafeBuffer packet = new UnsafeBuffer(new byte[alignedFrameLength]);
+            UnsafeBuffer packet = new UnsafeBuffer(BufferUtil.Allocate(alignedFrameLength));
 
             TermRebuilder.Insert(_termBuffer, termOffset, packet, alignedFrameLength);
 
@@ -93,7 +93,7 @@ public void ShouldFillAfterAGap()
             const int frameLength = 50;
             int alignedFrameLength = BitUtil.Align(frameLength, FrameDescriptor.FRAME_ALIGNMENT);
             const int srcOffset = 0;
-            UnsafeBuffer packet = new UnsafeBuffer(new byte[alignedFrameLength]);
+            UnsafeBuffer packet = new UnsafeBuffer(BufferUtil.Allocate(alignedFrameLength));
             int termOffset = alignedFrameLength * 2;
 
             TermRebuilder.Insert(_termBuffer, termOffset, packet, alignedFrameLength);
@@ -107,7 +107,7 @@ public void ShouldFillGapButNotMoveTailOrHwm()
             const int frameLength = 50;
             int alignedFrameLength = BitUtil.Align(frameLength, FrameDescriptor.FRAME_ALIGNMENT);
             const int srcOffset = 0;
-            UnsafeBuffer packet = new UnsafeBuffer(new byte[alignedFrameLength]);
+            UnsafeBuffer packet = new UnsafeBuffer(BufferUtil.Allocate(alignedFrameLength));
             int termOffset = alignedFrameLength * 2;
 
             TermRebuilder.Insert(_termBuffer, termOffset, packet, alignedFrameLength);

src/Adaptive.Aeron.Tests/PublicationTest.cs

@@ -92,7 +92,7 @@ public void SetUp()
 
             for (var i = 0; i < PARTITION_COUNT; i++)
             {
-                _termBuffers[i] = new UnsafeBuffer(BufferUtil.AllocateDirectAligned(TERM_LENGTH, FRAME_ALIGNMENT));
+                _termBuffers[i] = new UnsafeBuffer(BufferUtil.AllocateDirect(TERM_LENGTH));
             }
 
             _publication = new ConcurrentPublication(

src/Adaptive.Aeron.Tests/SubscriptionTest.cs

@@ -44,18 +44,14 @@ public class SubscriptionTest
         private AvailableImageHandler AvailableImageHandler;
         private UnavailableImageHandler UnavailableImageHandler;
 
-        private readonly UnsafeBuffer valuesBuffer = new UnsafeBuffer(new byte[16 * 1024]);
-        private readonly UnsafeBuffer metaDataBuffer = new UnsafeBuffer(new byte[64 * 1024]);
         private readonly UnsafeBuffer tempBuffer = new UnsafeBuffer(new byte[1024]);
-        private CountersManager countersManager;
+        private CountersManager countersManager = Tests.NewCountersManager(16 * 1024);
 
         private Subscription Subscription;
 
         [SetUp]
         public void Setup()
         {
-            countersManager  = new CountersManager(metaDataBuffer, valuesBuffer, Encoding.ASCII);
-
             ImageOneMock = A.Fake<Image>();
             ImageTwoMock = A.Fake<Image>();
 

src/Adaptive.Aeron.Tests/Tests.cs

@@ -0,0 +1,20 @@
+using Adaptive.Agrona;
+using Adaptive.Agrona.Concurrent;
+using Adaptive.Agrona.Concurrent.Status;
+
+namespace Adaptive.Aeron.Tests;
+
+public class Tests
+{
+    public static CountersManager NewCountersManager(int dataLength)
+    {
+        return new CountersManager(
+            new UnsafeBuffer(BufferUtil.AllocateDirect(countersMetadataBufferLength(dataLength))),
+            new UnsafeBuffer(BufferUtil.AllocateDirect(dataLength)));
+    }
+
+    public static int countersMetadataBufferLength(int counterValuesBufferLength)
+    {
+        return counterValuesBufferLength * (CountersReader.METADATA_LENGTH / CountersReader.COUNTER_LENGTH);
+    }
+}