Port to 1.38.1

Author: JPWatson

driver/Aeron.Driver.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>Aeron.Driver</id>
-    <version>1.37.0</version>
+    <version>1.38.1</version>
     <title>Aeron Driver</title>
     <authors>Adaptive Financial Consulting Ltd.</authors>
     <owners>Adaptive Financial Consulting Ltd.</owners>

driver/media-driver.jar

@@ -86,7 +86,7 @@ public void SetUp()
 
             LogBufferDescriptor.InitialiseTailWithTermId(_logMetaDataBuffer, PartionIndex, TermID1);
 
-            A.CallTo(() => _conductor.ReleasePublication(_publication)).Invokes(() => _publication.InternalClose());
+            A.CallTo(() => _conductor.RemovePublication(_publication)).Invokes(() => _publication.InternalClose());
         }
 
         [Test]
@@ -95,7 +95,7 @@ public void ShouldEnsureThePublicationIsOpenBeforeReadingPosition()
             _publication.Dispose();
             Assert.AreEqual(Publication.CLOSED, _publication.Position);
 
-            A.CallTo(() => _conductor.ReleasePublication(_publication)).MustHaveHappened();
+            A.CallTo(() => _conductor.RemovePublication(_publication)).MustHaveHappened();
         }
 
         [Test]
@@ -147,7 +147,21 @@ public void ShouldReleasePublicationOnClose()
         {
             _publication.Dispose();
 
-            A.CallTo(() => _conductor.ReleasePublication(_publication)).MustHaveHappened();
+            A.CallTo(() => _conductor.RemovePublication(_publication)).MustHaveHappened();
+        }
+        
+        [Test]
+        public void ShouldReturnErrorMessages()
+        {
+            Assert.Equals("NOT_CONNECTED", Publication.ErrorString(-1L));
+            Assert.Equals("BACK_PRESSURED", Publication.ErrorString(-2L));
+            Assert.Equals("ADMIN_ACTION", Publication.ErrorString(-3L));
+            Assert.Equals("CLOSED", Publication.ErrorString(-4L));
+            Assert.Equals("MAX_POSITION_EXCEEDED", Publication.ErrorString(-5L));
+            Assert.Equals("NONE", Publication.ErrorString(0L));
+            Assert.Equals("NONE", Publication.ErrorString(1L));
+            Assert.Equals("UNKNOWN", Publication.ErrorString(-6L));
+            Assert.Equals("UNKNOWN", Publication.ErrorString(long.MinValue));
         }
     }
 }

src/Adaptive.Aeron.Tests/PublicationTest.cs

@@ -0,0 +1,36 @@
+using System;
+using Adaptive.Aeron.Security;
+using Adaptive.Agrona;
+using FakeItEasy;
+using NUnit.Framework;
+
+namespace Adaptive.Aeron.Tests.Security
+{
+    public class authorisation_service_test
+    {
+        [Test]
+        public void ShouldAllowAnyCommandIfAllowAllIsUsed()
+        {
+            byte[] encodedCredentials = {0x1, 0x2, 0x3};
+            var errorHandler = A.Fake<ErrorHandler>();
+            const int protocolId = 77;
+            int actionId = new Random().Next();
+
+            Assert.True(AllowAllAuthorisationService.INSTANCE.IsAuthorised(protocolId, actionId, null, encodedCredentials));
+            A.CallTo(errorHandler).MustNotHaveHappened();
+        }
+
+        [Test]
+        public void ShouldForbidAllCommandsIfDenyAllIsUsed()
+        {
+            byte[] encodedCredentials = {0x4, 0x5, 0x6};
+            var errorHandler = A.Fake<ErrorHandler>();
+            const int protocolId = 77;
+            int actionId = new Random().Next();
+
+            Assert.False(DenyAllAuthorisationService.INSTANCE.IsAuthorised(protocolId, actionId, null, encodedCredentials));
+            A.CallTo(errorHandler).MustNotHaveHappened();
+        }
+    }
+
+}

src/Adaptive.Aeron.Tests/Security/AuthorisationServiceTest.cs

@@ -78,7 +78,7 @@ public void Setup()
                 AvailableImageHandler,
                 UnavailableImageHandler);
 
-            A.CallTo(() => Conductor.ReleaseSubscription(Subscription)).Invokes(() => Subscription.InternalClose(Aeron.NULL_VALUE));
+            A.CallTo(() => Conductor.RemoveSubscription(Subscription)).Invokes(() => Subscription.InternalClose(Aeron.NULL_VALUE));
         }
 
         [Test]
@@ -87,7 +87,7 @@ public void ShouldEnsureTheSubscriptionIsOpenWhenPolling()
             Subscription.Dispose();
             Assert.True(Subscription.IsClosed);
 
-            A.CallTo(() => Conductor.ReleaseSubscription(Subscription)).MustHaveHappened();
+            A.CallTo(() => Conductor.RemoveSubscription(Subscription)).MustHaveHappened();
         }
 
         [Test]

src/Adaptive.Aeron.Tests/SubscriptionTest.cs

@@ -2,7 +2,7 @@
     <PropertyGroup>
         <TargetFramework>netstandard2.0</TargetFramework>
         <PackageId>Aeron.Client</PackageId>
-        <VersionPrefix>1.37.0</VersionPrefix>
+        <VersionPrefix>1.38.1</VersionPrefix>
         <Authors>Adaptive Financial Consulting Ltd.</Authors>
         <Company>Adaptive Financial Consulting Ltd.</Company>
         <Product>Aeron Client</Product>

src/Adaptive.Aeron/Adaptive.Aeron.csproj

@@ -257,6 +257,17 @@ public long AsyncAddPublication(string channel, int streamId)
         {
             return _conductor.AsyncAddPublication(channel, streamId);
         }
+        
+        /// <summary>
+        /// Asynchronously remove a <seealso cref="Publication"/>.
+        /// </summary>
+        /// <param name="registrationId"> to be of the publication removed. </param>
+        /// <seealso cref="AsyncAddPublication(String, int)"/>
+        /// <seealso cref="AsyncAddExclusivePublication(String, int)"/>
+        public void AsyncRemovePublication(long registrationId)
+        {
+            _conductor.RemovePublication(registrationId);
+        }
 
         /// <summary>
         /// Asynchronously add a <seealso cref="Publication"/> for publishing messages to subscribers from a single thread.
@@ -948,6 +959,30 @@ static Context()
             /// Placeholder value to use in URIs to specify that a timestamp should be stored in the reserved value field.
             /// </summary>
             public const string RESERVED_OFFSET = "reserved";
+            
+            /// <summary>
+            /// Property name for a fallback PrintStream based logger when it is not possible to use the error logging
+            /// callback.  Supported values are stdout, stderr, no_op (stderr is the default).
+            /// </summary>
+            public const string FALLBACK_LOGGER_PROP_NAME = "aeron.fallback.logger";
+
+            /// <summary>
+            /// Get the current fallback logger based on the supplied property.
+            /// </summary>
+            /// <returns> the configured PrintStream. </returns>
+            public static TextWriter FallbackLogger()
+            {
+                string fallbackLoggerName = Config.GetProperty(FALLBACK_LOGGER_PROP_NAME, "stderr");
+                switch (fallbackLoggerName)
+                {
+                    case "stdout":
+                        return Console.Out;
+
+                    case "stderr":
+                    default:
+                        return Console.Error;
+                }
+            }
 
             /// <summary>
             /// Get the default directory name to be used if <seealso cref="AeronDirectoryName(String)"/> is not set. This will take
@@ -1999,7 +2034,7 @@ public MappedByteBuffer MapExistingCncFile(Action<string> logProgress)
             {
                 FileInfo cncFile = new FileInfo(Path.Combine(_aeronDirectory.FullName, CncFileDescriptor.CNC_FILE));
 
-                if (cncFile.Exists && cncFile.Length > 0)
+                if (cncFile.Exists && cncFile.Length > CncFileDescriptor.END_OF_METADATA_OFFSET)
                 {
                     if (null != logProgress)
                     {
@@ -2023,7 +2058,7 @@ public static bool IsDriverActive(DirectoryInfo directory, long driverTimeoutMs,
             {
                 FileInfo cncFile = new FileInfo(Path.Combine(directory.FullName, CncFileDescriptor.CNC_FILE));
 
-                if (cncFile.Exists && cncFile.Length > 0)
+                if (cncFile.Exists && cncFile.Length > CncFileDescriptor.END_OF_METADATA_OFFSET)
                 {
                     logger("INFO: Aeron CnC file " + cncFile + " exists");
 
@@ -2120,23 +2155,26 @@ public static bool RequestDriverTermination(
             {
                 FileInfo cncFile = new FileInfo(Path.Combine(directory.FullName, CncFileDescriptor.CNC_FILE));
 
-                if (cncFile.Exists && cncFile.Length > 0)
+                if (cncFile.Exists && cncFile.Length > CncFileDescriptor.END_OF_METADATA_OFFSET)
                 {
                     var cncByteBuffer = IoUtil.MapExistingFile(cncFile, "CnC file");
                     try
                     {
                         UnsafeBuffer cncMetaDataBuffer = CncFileDescriptor.CreateMetaDataBuffer(cncByteBuffer);
                         int cncVersion = cncMetaDataBuffer.GetIntVolatile(CncFileDescriptor.CncVersionOffset(0));
 
-                        CncFileDescriptor.CheckVersion(cncVersion);
+                        if (cncVersion > 0)
+                        {
+                            CncFileDescriptor.CheckVersion(cncVersion);
 
-                        ManyToOneRingBuffer toDriverBuffer =
-                            new ManyToOneRingBuffer(
-                                CncFileDescriptor.CreateToDriverBuffer(cncByteBuffer, cncMetaDataBuffer));
-                        long clientId = toDriverBuffer.NextCorrelationId();
-                        DriverProxy driverProxy = new DriverProxy(toDriverBuffer, clientId);
+                            ManyToOneRingBuffer toDriverBuffer =
+                                new ManyToOneRingBuffer(
+                                    CncFileDescriptor.CreateToDriverBuffer(cncByteBuffer, cncMetaDataBuffer));
+                            long clientId = toDriverBuffer.NextCorrelationId();
+                            DriverProxy driverProxy = new DriverProxy(toDriverBuffer, clientId);
 
-                        return driverProxy.TerminateDriver(tokenBuffer, tokenOffset, tokenLength);
+                            return driverProxy.TerminateDriver(tokenBuffer, tokenOffset, tokenLength);
+                        }
                     }
                     finally
                     {
@@ -2191,7 +2229,7 @@ public int SaveErrorLog(StreamWriter writer, MappedByteBuffer cncByteBuffer)
             public static int PrintErrorLog(IAtomicBuffer errorBuffer, TextWriter @out)
             {
                 int distinctErrorCount = 0;
-                if (ErrorLogReader.HasErrors(errorBuffer))
+                if (errorBuffer.Capacity > 0 && ErrorLogReader.HasErrors(errorBuffer))
                 {
                     void ErrorConsumer(int count, long firstTimestamp, long lastTimestamp, string ex)
                         => @out.WriteLine(

src/Adaptive.Aeron/Aeron.cs

@@ -113,13 +113,13 @@ public ChannelUri Media(string media)
         /// Is the channel <seealso cref="Media()"/> equal to <seealso cref="Aeron.Context.UDP_MEDIA"/>.
         /// </summary>
         /// <returns> true the channel <seealso cref="Media()"/> equals <seealso cref="Aeron.Context.UDP_MEDIA"/>. </returns>
-        public bool Udp => Aeron.Context.UDP_MEDIA.Equals(_media);
+        public bool IsUdp => Aeron.Context.UDP_MEDIA.Equals(_media);
 
         /// <summary>
         /// Is the channel <seealso cref="Media()"/> equal to <seealso cref="Aeron.Context.IPC_MEDIA"/>.
         /// </summary>
         /// <returns> true the channel <seealso cref="Media()"/> equals <seealso cref="Aeron.Context.IPC_MEDIA"/>. </returns>
-        public bool Ipc => Aeron.Context.IPC_MEDIA.Equals(_media);
+        public bool IsIpc => Aeron.Context.IPC_MEDIA.Equals(_media);
 
         /// <summary>
         /// The scheme for the URI. Must be "aeron".
@@ -473,6 +473,30 @@ public static long GetTag(string paramValue)
         {
             return IsTagged(paramValue) ? long.Parse(paramValue.Substring(4, paramValue.Length - 4)) : INVALID_TAG;
         }
+        
+        /// <summary>
+        /// Create a channel URI for a destination, i.e. a channel that uses {@code media} and {@code interface} parameters
+        /// of the original channel and adds specified {@code endpoint} to it. For example given the input channel is
+        /// {@code aeron:udp?mtu=1440|ttl=0|endpoint=localhost:8090|term-length=128k|interface=eth0} and the endpoint is
+        /// {@code 192.168.0.14} the output of this method will be {@code aeron:udp?endpoint=192.168.0.14|interface=eth0}.
+        /// </summary>
+        /// <param name="channel">  for which the destination is being added. </param>
+        /// <param name="endpoint"> for the target destination. </param>
+        /// <returns> new channel URI for a destination. </returns>
+        public static string CreateDestinationUri(string channel, string endpoint)
+        {
+            ChannelUri channelUri = ChannelUri.Parse(channel);
+            string uri = AERON_PREFIX + channelUri.Media() + "?" + Aeron.Context.ENDPOINT_PARAM_NAME + "=" + endpoint;
+            string networkInterface = channelUri.Get(Aeron.Context.INTERFACE_PARAM_NAME);
+
+            if (null != networkInterface)
+            {
+                return uri + "|" + Aeron.Context.INTERFACE_PARAM_NAME + "=" + networkInterface;
+            }
+
+            return uri;
+        }
+
 
         private static void ValidateMedia(string media)
         {

src/Adaptive.Aeron/ChannelUri.cs

@@ -1458,9 +1458,13 @@ public ChannelUriStringBuilder SpiesSimulateConnection(ChannelUri channelUri)
         /// <returns> this for a fluent API. </returns>
         public ChannelUriStringBuilder InitialPosition(long position, int initialTermId, int termLength)
         {
-            if (position < 0 || 0 != (position & (FRAME_ALIGNMENT - 1)))
+            if (position < 0)
             {
-                throw new ArgumentException("invalid position: " + position);
+                throw new ArgumentException("invalid position=" + position + " < 0");
+            }
+            if (0 != (position & (FRAME_ALIGNMENT - 1)))
+            {
+                throw new ArgumentException("invalid position=" + position + " does not have frame alignment=" + FRAME_ALIGNMENT);
             }
 
             int bitsToShift = LogBufferDescriptor.PositionBitsToShift(termLength);

src/Adaptive.Aeron/ChannelUriStringBuilder.cs

@@ -310,7 +310,6 @@ internal void OnNewPublication(
             );
 
             _resourceByRegIdMap.Put(correlationId, publication);
-            _asyncCommandIdSet.Remove(correlationId);
         }
 
         internal void OnNewExclusivePublication(
@@ -342,7 +341,6 @@ internal void OnNewExclusivePublication(
             );
 
             _resourceByRegIdMap.Put(correlationId, publication);
-            _asyncCommandIdSet.Remove(correlationId);
         }
 
         internal void OnNewSubscription(long correlationId, int statusIndicatorId)
@@ -575,7 +573,7 @@ internal ExclusivePublication GetExclusivePublication(long registrationId)
             }
         }
 
-        internal void ReleasePublication(Publication publication)
+        internal void RemovePublication(Publication publication)
         {
             _clientLock.Lock();
             try
@@ -604,6 +602,44 @@ internal void ReleasePublication(Publication publication)
                 _clientLock.Unlock();
             }
         }
+        
+        internal void RemovePublication(long publicationRegistrationId)
+        {
+            _clientLock.Lock();
+            try
+            {
+                if (Aeron.NULL_VALUE == publicationRegistrationId || _isTerminating || _isClosed)
+                {
+                    return;
+                }
+
+                EnsureNotReentrant();
+
+                object resource = _resourceByRegIdMap.Get(publicationRegistrationId);
+                if (null != resource && !(resource is Publication))
+                {
+                    throw new AeronException("registration id is not a Publication");
+                }
+
+                Publication publication = (Publication)resource;
+                if (null != publication)
+                {
+                    _resourceByRegIdMap.Remove(publicationRegistrationId);
+                    publication.InternalClose();
+                    ReleaseLogBuffers(publication.LogBuffers, publication.OriginalRegistrationId, EXPLICIT_CLOSE_LINGER_NS);
+                }
+
+                if (_asyncCommandIdSet.Remove(publicationRegistrationId) || null != publication)
+                {
+                    _driverProxy.RemovePublication(publicationRegistrationId);
+                    _stashedChannelByRegistrationId.Remove(publicationRegistrationId);
+                }
+            }
+            finally
+            {
+                _clientLock.Unlock();
+            }
+        }
 
         internal Subscription AddSubscription(string channel, int streamId)
         {
@@ -634,7 +670,7 @@ internal Subscription AddSubscription(string channel, int streamId, AvailableIma
             }
         }
 
-        internal void ReleaseSubscription(Subscription subscription)
+        internal void RemoveSubscription(Subscription subscription)
         {
             _clientLock.Lock();
             try