Port to 1.10.4

RELEASE.md

@@ -1,4 +1,4 @@
 #### Port
 Aeron.NET has been ported against Java version:
-- Agrona: 0.9.18-13-g0378ffa 
-- Aeron:  1.10.2
+- Agrona: 0.9.22
+- Aeron:  1.10.4

driver/Aeron.Driver.nuspec

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

driver/version.txt

@@ -1,2 +1,2 @@
 Driver source:
-http://repo1.maven.org/maven2/io/aeron/aeron-all/1.10.3/aeron-all-1.10.3.jar
+http://repo1.maven.org/maven2/io/aeron/aeron-all/1.10.4/aeron-all-1.10.4.jar

src/Adaptive.Aeron/Adaptive.Aeron.csproj

@@ -2,7 +2,7 @@
   <PropertyGroup>
     <TargetFrameworks>netstandard2.0;net45</TargetFrameworks>
     <PackageId>Aeron.Client</PackageId>
-    <VersionPrefix>1.10.3</VersionPrefix>
+    <VersionPrefix>1.10.4</VersionPrefix>
     <Authors>Adaptive Financial Consulting Ltd.</Authors>
     <Company>Adaptive Financial Consulting Ltd.</Company>
     <Product>Aeron Client</Product>

src/Adaptive.Aeron/Aeron.cs

@@ -526,6 +526,11 @@ public class Context : IDisposable
             /// Key for the tags for a channel
             /// </summary>
             public const string TAGS_PARAM_NAME = "tags";
+
+            /// <summary>
+            /// Parameter name for channel URI param to indicate if term buffers should be sparse. Value is boolean.
+            /// </summary>
+            public const string SPARSE_PARAM_NAME = "sparse";
 
             /// <summary>
             /// Get the default directory name to be used if <seealso cref="AeronDirectoryName(String)"/> is not set. This will take
@@ -1418,12 +1423,16 @@ public int SaveErrorLog(StreamWriter writer, MappedByteBuffer cncByteBuffer)
                         cncVersion);
                 }
 
+                int distinctErrorCount = 0;
                 UnsafeBuffer buffer = CncFileDescriptor.CreateErrorLogBuffer(cncByteBuffer, cncMetaDataBuffer);
 
-                void ErrorConsumer(int count, long firstTimestamp, long lastTimestamp, string ex)
-                    => FormatError(writer, count, firstTimestamp, lastTimestamp, ex);
+                if (ErrorLogReader.HasErrors(buffer))
+                {
+                    void ErrorConsumer(int count, long firstTimestamp, long lastTimestamp, string ex)
+                        => FormatError(writer, count, firstTimestamp, lastTimestamp, ex);
 
-                var distinctErrorCount = ErrorLogReader.Read(buffer, ErrorConsumer);
+                    distinctErrorCount = ErrorLogReader.Read(buffer, ErrorConsumer);
+                }
 
                 writer.WriteLine();
                 writer.WriteLine("{0} distinct errors observed.", distinctErrorCount);

src/Adaptive.Aeron/ChannelUriStringBuilder.cs

@@ -24,6 +24,7 @@ public class ChannelUriStringBuilder
         private string _controlMode;
         private string _tags;
         private bool? _reliable;
+        private bool? _sparse;
         private int? _ttl;
         private int? _mtu;
         private int? _termLength;
@@ -272,6 +273,30 @@ public ChannelUriStringBuilder Reliable(bool? isReliable)
             return _reliable;
         }
 
+        /// <summary>
+        /// Set to indicate if a term log buffer should be sparse on disk or not. Sparse saves space at the potential
+        /// expense of latency.
+        /// </summary>
+        /// <param name="isSparse"> true if the term buffer log is sparse on disk. </param>
+        /// <returns> this for a fluent API. </returns>
+        /// <see cref="Aeron.Context.SPARSE_PARAM_NAME"/>
+        public ChannelUriStringBuilder Sparse(bool? isSparse)
+        {
+            _sparse = isSparse;
+            return this;
+        }
+
+        /// <summary>
+        /// Get if a term log buffer should be sparse on disk or not. Sparse saves space at the potential expense of latency.
+        /// </summary>
+        /// <returns> true if the term buffer log is sparse on disk. </returns>
+        /// <see cref="Aeron.Context.SPARSE_PARAM_NAME"/>
+        public bool? Sparse()
+        {
+            return _sparse;
+        }
+
+
         /// <summary>
         /// Set the Time To Live (TTL) for a multicast datagram. Valid values are 0-255 for the number of hops the datagram
         /// can progress along.
@@ -603,6 +628,11 @@ public string Build()
                 _sb.Append(Aeron.Context.RELIABLE_STREAM_PARAM_NAME).Append('=').Append(_reliable).Append('|');
             }
 
+            if (null != _sparse)
+            {
+                _sb.Append(Aeron.Context.SPARSE_PARAM_NAME).Append('=').Append(_sparse).Append('|');
+            }
+
             if (null != _ttl)
             {
                 _sb.Append(Aeron.Context.TTL_PARAM_NAME).Append('=').Append(_ttl.Value).Append('|');

src/Adaptive.Aeron/Publication.cs

@@ -71,7 +71,7 @@ public abstract class Publication : IDisposable
         /// The offer failed due to reaching the maximum position of the stream given term buffer length times the total
         /// possible number of terms.
         /// <para>
-        /// If this happen then the publication should be closed and a new one added. To make it less likely to happen then
+        /// If this happens then the publication should be closed and a new one added. To make it less likely to happen then
         /// increase the term buffer length.
         /// </para>
         /// </summary>

src/Adaptive.Agrona/Adaptive.Agrona.csproj

@@ -3,7 +3,7 @@
     <TargetFrameworks>netstandard2.0;net45</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <PackageId>Agrona</PackageId>
-    <VersionPrefix>1.10.3</VersionPrefix>
+    <VersionPrefix>1.10.4</VersionPrefix>
     <Authors>Adaptive Financial Consulting Ltd.</Authors>
     <Company>Adaptive Financial Consulting Ltd.</Company>
     <Product>Agrona libraries initially included in Aeron Client</Product>

src/Adaptive.Agrona/Concurrent/Errors/ErrorReader.cs

@@ -23,6 +23,16 @@ namespace Adaptive.Agrona.Concurrent.Errors
     /// </summary>
     public class ErrorLogReader
     {
+        /// <summary>
+        /// Has the error buffer any recorded errors?
+        /// </summary>
+        /// <param name="buffer"> containing the <seealso cref="DistinctErrorLog"/>. </param>
+        /// <returns> true if there is at least one error. </returns>
+        public static bool HasErrors(IAtomicBuffer buffer)
+        {
+            return 0 != buffer.GetIntVolatile(DistinctErrorLog.LengthOffset);
+        }
+
         /// <summary>
         /// Read all the errors in a log since the creation of the log.
         /// </summary>

src/Adaptive.Archiver/Adaptive.Archiver.csproj

@@ -3,7 +3,7 @@
     <TargetFrameworks>netstandard2.0;net45</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <PackageId>Aeron.Archiver</PackageId>
-    <VersionPrefix>1.10.3</VersionPrefix>
+    <VersionPrefix>1.10.4</VersionPrefix>
     <Authors>Adaptive Financial Consulting Ltd.</Authors>
     <Company>Adaptive Financial Consulting Ltd.</Company>
     <Product>Archiving over the Aeron transport</Product>

src/Adaptive.Archiver/AeronArchive.cs

@@ -301,7 +301,8 @@ public virtual void CheckForErrorResponse()
                     if (controlResponsePoller.TemplateId() == ControlResponseDecoder.TEMPLATE_ID &&
                         controlResponsePoller.Code() == ControlResponseCode.ERROR)
                     {
-                        throw new ArchiveException(controlResponsePoller.ErrorMessage(), (int) controlResponsePoller.RelevantId());
+                        throw new ArchiveException(controlResponsePoller.ErrorMessage(),
+                            (int) controlResponsePoller.RelevantId());
                     }
                 }
             }
@@ -1066,6 +1067,16 @@ public class Configuration
             /// </summary>
             public const int RECORDING_EVENTS_STREAM_ID_DEFAULT = 30;
 
+            /// <summary>
+            /// Sparse term buffer indicator for control streams.
+            /// </summary>
+            private const string CONTROL_TERM_BUFFER_SPARSE_PARAM_NAME = "aeron.archive.control.term.buffer.sparse";
+
+            /// <summary>
+            /// Overrides driver's sparse term buffer indicator for control streams.
+            /// </summary>
+            private const bool CONTROL_TERM_BUFFER_SPARSE_DEFAULT = true;
+
             /// <summary>
             /// Term length for control streams.
             /// </summary>
@@ -1077,12 +1088,12 @@ public class Configuration
             internal const int CONTROL_TERM_BUFFER_LENGTH_DEFAULT = 64 * 1024;
 
             /// <summary>
-            /// Term length for control streams.
+            /// MTU length for control streams.
             /// </summary>
             internal const string CONTROL_MTU_LENGTH_PARAM_NAME = "aeron.archive.control.mtu.length";
 
             /// <summary>
-            /// MTU to reflect default control term length.
+            ///  MTU to reflect default for the control streams.
             /// </summary>
             internal const int CONTROL_MTU_LENGTH_DEFAULT = 4 * 1024;
 
@@ -1096,11 +1107,22 @@ public static long MessageTimeoutNs()
                 return Config.GetDurationInNanos(MESSAGE_TIMEOUT_PROP_NAME, MESSAGE_TIMEOUT_DEFAULT_NS);
             }
 
+            /// <summary>
+            /// Should term buffer files be sparse for control request and response streams.
+            /// </summary>
+            /// <returns> true if term buffer files should be sparse for control request and response streams. </returns>
+            /// <seealso cref="CONTROL_TERM_BUFFER_SPARSE_PARAM_NAME"/>
+            public static bool ControlTermBufferSparse()
+            {
+                string propValue = Config.GetProperty(CONTROL_TERM_BUFFER_SPARSE_PARAM_NAME);
+                return null != propValue ? "true".Equals(propValue) : CONTROL_TERM_BUFFER_SPARSE_DEFAULT;
+            }
+
             /// <summary>
             /// Term buffer length to be used for control request and response streams.
             /// </summary>
             /// <returns> term buffer length to be used for control request and response streams. </returns>
-            /// <seealso cref= #CONTROL_TERM_BUFFER_LENGTH_PARAM_NAME </seealso>
+            /// <seealso cref="CONTROL_TERM_BUFFER_LENGTH_PARAM_NAME"></seealso>
             public static int ControlTermBufferLength()
             {
                 return Config.GetSizeAsInt(CONTROL_TERM_BUFFER_LENGTH_PARAM_NAME, CONTROL_TERM_BUFFER_LENGTH_DEFAULT);
@@ -1217,6 +1239,7 @@ public class Context : IDisposable
             internal int controlRequestStreamId = Configuration.ControlStreamId();
             internal string controlResponseChannel = Configuration.ControlResponseChannel();
             internal int controlResponseStreamId = Configuration.ControlResponseStreamId();
+            internal bool controlTermBufferSparse = Configuration.ControlTermBufferSparse();
             internal int controlTermBufferLength = Configuration.ControlTermBufferLength();
             internal int controlMtuLength = Configuration.ControlMtuLength();
 
@@ -1260,6 +1283,7 @@ public void Conclude()
                 ChannelUri uri = ChannelUri.Parse(controlRequestChannel);
                 uri.Put(Aeron.Aeron.Context.TERM_LENGTH_PARAM_NAME, Convert.ToString(controlTermBufferLength));
                 uri.Put(Aeron.Aeron.Context.MTU_LENGTH_PARAM_NAME, Convert.ToString(controlMtuLength));
+                uri.Put(Aeron.Aeron.Context.SPARSE_PARAM_NAME, Convert.ToString(controlTermBufferSparse));
                 controlRequestChannel = uri.ToString();
             }
 
@@ -1418,9 +1442,31 @@ public int ControlResponseStreamId()
             {
                 return controlResponseStreamId;
             }
+
+            /// <summary>
+            /// Should the control streams use sparse file term buffers.
+            /// </summary>
+            /// <param name="controlTermBufferSparse"> for the control stream. </param>
+            /// <returns> this for a fluent API. </returns>
+            /// <seealso cref="Configuration.CONTROL_TERM_BUFFER_SPARSE_PARAM_NAME"></seealso>
+            public Context ControlTermBufferSparse(bool controlTermBufferSparse)
+            {
+                this.controlTermBufferSparse = controlTermBufferSparse;
+                return this;
+            }
+
+            /// <summary>
+            /// Should the control streams use sparse file term buffers.
+            /// </summary>
+            /// <returns> true if the control stream should use sparse file term buffers. </returns>
+            /// <seealso cref="Configuration.CONTROL_TERM_BUFFER_SPARSE_PARAM_NAME"></seealso>
+            public bool ControlTermBufferSparse()
+            {
+                return controlTermBufferSparse;
+            }
 
             /// <summary>
-            /// Set the term buffer length for the control stream.
+            /// Set the term buffer length for the control streams.
             /// </summary>
             /// <param name="controlTermBufferLength"> for the control stream. </param>
             /// <returns> this for a fluent API. </returns>
@@ -1432,19 +1478,19 @@ public Context ControlTermBufferLength(int controlTermBufferLength)
             }
 
             /// <summary>
-            /// Get the term buffer length for the control steam.
+            /// Get the term buffer length for the control streams.
             /// </summary>
-            /// <returns> the term buffer length for the control steam. </returns>
+            /// <returns> the term buffer length for the control streams. </returns>
             /// <seealso cref= Configuration#CONTROL_TERM_BUFFER_LENGTH_PARAM_NAME </seealso>
             public int ControlTermBufferLength()
             {
                 return controlTermBufferLength;
             }
 
             /// <summary>
-            /// Set the MTU length for the control stream.
+            /// Set the MTU length for the control streams.
             /// </summary>
-            /// <param name="controlMtuLength"> for the control stream. </param>
+            /// <param name="controlMtuLength"> for the control streams. </param>
             /// <returns> this for a fluent API. </returns>
             /// <seealso cref= Configuration#CONTROL_MTU_LENGTH_PARAM_NAME </seealso>
             public Context ControlMtuLength(int controlMtuLength)
@@ -1454,9 +1500,9 @@ public Context ControlMtuLength(int controlMtuLength)
             }
 
             /// <summary>
-            /// Get the MTU length for the control steam.
+            /// Get the MTU length for the control steams.
             /// </summary>
-            /// <returns> the MTU length for the control steam. </returns>
+            /// <returns> the MTU length for the control steams. </returns>
             /// <seealso cref= Configuration#CONTROL_MTU_LENGTH_PARAM_NAME </seealso>
             public int ControlMtuLength()
             {
@@ -1648,7 +1694,8 @@ public AeronArchive Poll()
 
                 if (2 == step)
                 {
-                    if (!archiveProxy.TryConnect(ctx.ControlResponseChannel(), ctx.ControlResponseStreamId(), connectCorrelationId))
+                    if (!archiveProxy.TryConnect(ctx.ControlResponseChannel(), ctx.ControlResponseStreamId(),
+                        connectCorrelationId))
                     {
                         return null;
                     }
@@ -1667,22 +1714,27 @@ public AeronArchive Poll()
                 }
 
                 controlResponsePoller.Poll();
-                if (controlResponsePoller.IsPollComplete() && controlResponsePoller.CorrelationId() == connectCorrelationId && controlResponsePoller.TemplateId() == ControlResponseDecoder.TEMPLATE_ID)
+                if (controlResponsePoller.IsPollComplete() &&
+                    controlResponsePoller.CorrelationId() == connectCorrelationId &&
+                    controlResponsePoller.TemplateId() == ControlResponseDecoder.TEMPLATE_ID)
                 {
                     ControlResponseCode code = controlResponsePoller.Code();
                     if (code != ControlResponseCode.OK)
                     {
                         if (code == ControlResponseCode.ERROR)
                         {
-                            throw new ArchiveException("error: " + controlResponsePoller.ErrorMessage(), (int) controlResponsePoller.RelevantId());
+                            throw new ArchiveException("error: " + controlResponsePoller.ErrorMessage(),
+                                (int) controlResponsePoller.RelevantId());
                         }
 
                         throw new ArchiveException("unexpected response: code=" + code);
                     }
 
                     long controlSessionId = controlResponsePoller.ControlSessionId();
                     Subscription subscription = controlResponsePoller.Subscription();
-                    return new AeronArchive(ctx, controlResponsePoller, archiveProxy, new RecordingDescriptorPoller(subscription, FRAGMENT_LIMIT, controlSessionId), controlSessionId);
+                    return new AeronArchive(ctx, controlResponsePoller, archiveProxy,
+                        new RecordingDescriptorPoller(subscription, FRAGMENT_LIMIT, controlSessionId),
+                        controlSessionId);
                 }
 
                 return null;

src/Adaptive.Cluster/Adaptive.Cluster.csproj

@@ -3,7 +3,7 @@
     <TargetFrameworks>netstandard2.0;net45</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <PackageId>Aeron.Cluster</PackageId>
-    <VersionPrefix>1.10.3</VersionPrefix>
+    <VersionPrefix>1.10.4</VersionPrefix>
     <Authors>Adaptive Financial Consulting Ltd.</Authors>
     <Company>Adaptive Financial Consulting Ltd.</Company>
     <Product>Clustering libraries over the Aeron transport</Product>