Port to 1.10.2

AdaptiveConsulting · Aug 1, 2018 · 773c007 · 773c007
1 parent d1ac7ac
commit 773c007
Show file tree

Hide file tree

Showing 21 changed files with 1,093 additions and 26 deletions.
diff --git a/RELEASE.md b/RELEASE.md
@@ -1,4 +1,4 @@
 #### Port
 Aeron.NET has been ported against Java version:
 - Agrona: 0.9.18-13-g0378ffa 
-- Aeron:  1.10.1
+- Aeron:  1.10.2
diff --git a/driver/Aeron.Driver.nuspec b/driver/Aeron.Driver.nuspec
@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>Aeron.Driver</id>
-    <version>1.10.1</version>
+    <version>1.10.2</version>
     <title>Aeron Driver</title>
     <authors>Adaptive Financial Consulting Ltd.</authors>
     <owners>Adaptive Financial Consulting Ltd.</owners>

diff --git a/driver/media-driver.jar b/driver/media-driver.jar
diff --git a/driver/version.txt b/driver/version.txt
@@ -1,2 +1,2 @@
 Driver source:
-http://repo1.maven.org/maven2/io/aeron/aeron-all/1.10.1/aeron-all-1.10.1.jar
+http://repo1.maven.org/maven2/io/aeron/aeron-all/1.10.2/aeron-all-1.10.2.jar
diff --git a/scripts/build-prerelease-nuget-packages - Copy.bat b/scripts/build-prerelease-nuget-packages - Copy.bat
@@ -0,0 +1,19 @@
+@echo off
+pushd %~dp0..
+SET nuget_source=https://www.myget.org/F/aeron/api/v2/package
+
+del nupkgs\*.nupkg
+
+call dotnet pack src\Adaptive.Aeron\Adaptive.Aeron.csproj                -c Release --output ..\..\nupkgs
+call dotnet pack src\Adaptive.Agrona\Adaptive.Agrona.csproj              -c Release --output ..\..\nupkgs 
+call dotnet pack src\Adaptive.Cluster\Adaptive.Cluster.csproj            -c Release --output ..\..\nupkgs
+call dotnet pack src\Adaptive.Archiver\Adaptive.Archiver.csproj          -c Release --output ..\..\nupkgs
+call .\scripts\nuget pack .\driver\Aeron.Driver.nuspec                   -OutputDirectory nupkgs
+
+call dotnet nuget push nupkgs\Agrona.*.nupkg -s %nuget_source%
+call dotnet nuget push nupkgs\Aeron.Client.*.nupkg -s %nuget_source%
+call dotnet nuget push nupkgs\Aeron.Driver.*.nupkg -s %nuget_source%
+call dotnet nuget push nupkgs\Aeron.Cluster.*.nupkg -s %nuget_source%
+call dotnet nuget push nupkgs\Aeron.Archiver.*.nupkg -s %nuget_source%
+
+popd
diff --git a/src/Adaptive.Aeron/Adaptive.Aeron.csproj b/src/Adaptive.Aeron/Adaptive.Aeron.csproj
@@ -2,7 +2,7 @@
   <PropertyGroup>
     <TargetFrameworks>netstandard2.0;net45</TargetFrameworks>
     <PackageId>Aeron.Client</PackageId>
-    <VersionPrefix>1.10.1</VersionPrefix>
+    <VersionPrefix>1.10.2</VersionPrefix>
     <Authors>Adaptive Financial Consulting Ltd.</Authors>
     <Company>Adaptive Financial Consulting Ltd.</Company>
     <Product>Aeron Client</Product>

diff --git a/src/Adaptive.Aeron/Aeron.cs b/src/Adaptive.Aeron/Aeron.cs
@@ -120,7 +120,6 @@ internal Aeron(Context ctx)
         /// 
         /// Threads required for interacting with the media driver are created and managed within the Aeron instance.
         /// 
-        /// </para>
         /// </summary>
         /// <returns> the new <seealso cref="Aeron"/> instance connected to the Media Driver. </returns>
         public static Aeron Connect()

diff --git a/src/Adaptive.Aeron/ClientConductor.cs b/src/Adaptive.Aeron/ClientConductor.cs
@@ -47,6 +47,7 @@ internal class ClientConductor : IAgent, IDriverEventsListener
         private long _timeOfLastResourcesCheckNs;
         private long _timeOfLastServiceNs;
         private bool _isClosed;
+        private bool _isInCallback;
         private string _stashedChannel;
         private RegistrationException _driverException;
 
@@ -270,6 +271,8 @@ public void OnAvailableImage(
                 AvailableImageHandler handler = subscription.AvailableImageHandler();
                 if (null != handler)
                 {
+                    _isInCallback = true;
+
                     try
                     {
                         handler(image);
@@ -278,6 +281,10 @@ public void OnAvailableImage(
                     {
                         HandleError(ex);
                     }
+                    finally
+                    {
+                        _isInCallback = false;
+                    }
                 }
 
                 subscription.AddImage(image);
@@ -295,6 +302,8 @@ public void OnUnavailableImage(long correlationId, long subscriptionRegistration
                     UnavailableImageHandler handler = subscription.UnavailableImageHandler();
                     if (null != handler)
                     {
+                        _isInCallback = true;
+
                         try
                         {
                             handler(image);
@@ -303,6 +312,10 @@ public void OnUnavailableImage(long correlationId, long subscriptionRegistration
                         {
                             HandleError(ex);
                         }
+                        finally
+                        {
+                            _isInCallback = false;
+                        }
                     }
                 }
             }
@@ -318,6 +331,8 @@ public void OnAvailableCounter(long registrationId, int counterId)
         {
             if (null != _availableCounterHandler)
             {
+                _isInCallback = true;
+
                 try
                 {
                     _availableCounterHandler(_countersReader, registrationId, counterId);
@@ -326,13 +341,19 @@ public void OnAvailableCounter(long registrationId, int counterId)
                 {
                     HandleError(ex);
                 }
+                finally
+                {
+                    _isInCallback = false;
+                }
             }
         }
 
         public void OnUnavailableCounter(long registrationId, int counterId)
         {
             if (null != _unavailableCounterHandler)
             {
+                _isInCallback = true;
+
                 try
                 {
                     _unavailableCounterHandler(_countersReader, registrationId, counterId);
@@ -341,6 +362,10 @@ public void OnUnavailableCounter(long registrationId, int counterId)
                 {
                     HandleError(ex);
                 }
+                finally
+                {
+                    _isInCallback = false;
+                }
             }
         }
 
@@ -360,6 +385,7 @@ internal ConcurrentPublication AddPublication(string channel, int streamId)
             try
             {
                 EnsureOpen();
+                EnsureNotReentrant();
 
                 _stashedChannel = channel;
                 long registrationId = _driverProxy.AddPublication(channel, streamId);
@@ -379,6 +405,7 @@ internal virtual ExclusivePublication AddExclusivePublication(string channel, in
             try
             {
                 EnsureOpen();
+                EnsureNotReentrant();
 
                 _stashedChannel = channel;
                 long registrationId = _driverProxy.AddExclusivePublication(channel, streamId);
@@ -402,6 +429,7 @@ internal virtual void ReleasePublication(Publication publication)
                     publication.InternalClose();
 
                     EnsureOpen();
+                    EnsureNotReentrant();
 
                     var removedPublication = _resourceByRegIdMap[publication.RegistrationId];
 
@@ -429,6 +457,7 @@ internal virtual Subscription AddSubscription(string channel, int streamId, Avai
             try
             {
                 EnsureOpen();
+                EnsureNotReentrant();
 
                 long correlationId = _driverProxy.AddSubscription(channel, streamId);
                 Subscription subscription = new Subscription(this, channel, streamId, correlationId, availableImageHandler, unavailableImageHandler);
@@ -455,6 +484,7 @@ internal virtual void ReleaseSubscription(Subscription subscription)
                     subscription.InternalClose();
 
                     EnsureOpen();
+                    EnsureNotReentrant();
 
                     long registrationId = subscription.RegistrationId;
                     AwaitResponse(_driverProxy.RemoveSubscription(registrationId));
@@ -473,6 +503,7 @@ internal virtual void AddDestination(long registrationId, string endpointChannel
             try
             {
                 EnsureOpen();
+                EnsureNotReentrant();
 
                 AwaitResponse(_driverProxy.AddDestination(registrationId, endpointChannel));
             }
@@ -488,6 +519,7 @@ internal virtual void RemoveDestination(long registrationId, string endpointChan
             try
             {
                 EnsureOpen();
+                EnsureNotReentrant();
 
                 AwaitResponse(_driverProxy.RemoveDestination(registrationId, endpointChannel));
             }
@@ -503,6 +535,7 @@ internal void AddRcvDestination(long registrationId, string endpointChannel)
             try
             {
                 EnsureOpen();
+                EnsureNotReentrant();
 
                 AwaitResponse(_driverProxy.AddRcvDestination(registrationId, endpointChannel));
             }
@@ -518,7 +551,8 @@ internal void RemoveRcvDestination(long registrationId, string endpointChannel)
             try
             {
                 EnsureOpen();
-
+                EnsureNotReentrant();
+
                 AwaitResponse(_driverProxy.RemoveRcvDestination(registrationId, endpointChannel));
             }
             finally
@@ -534,6 +568,7 @@ internal virtual Counter AddCounter(int typeId, IDirectBuffer keyBuffer, int key
             try
             {
                 EnsureOpen();
+                EnsureNotReentrant();
 
                 if (keyLength < 0 || keyLength > CountersManager.MAX_KEY_LENGTH)
                 {
@@ -563,6 +598,7 @@ internal Counter AddCounter(int typeId, string label)
             try
             {
                 EnsureOpen();
+                EnsureNotReentrant();
 
                 if (label.Length > CountersManager.MAX_LABEL_LENGTH)
                 {
@@ -591,6 +627,7 @@ internal virtual void ReleaseCounter(Counter counter)
                     counter.InternalClose();
 
                     EnsureOpen();
+                    EnsureNotReentrant();
 
                     long registrationId = counter.RegistrationId();
                     AwaitResponse(_driverProxy.RemoveCounter(registrationId));
@@ -646,6 +683,14 @@ private void EnsureOpen()
             }
         }
 
+        private void EnsureNotReentrant()
+        {
+            if (_isInCallback)
+            {
+                throw new AeronException("Reentrant calls not permitted during callbacks");
+            }
+        }
+
         private LogBuffers LogBuffers(long registrationId, string logFileName)
         {
             LogBuffers logBuffers = _logBuffersByIdMap[registrationId];

diff --git a/src/Adaptive.Agrona/Adaptive.Agrona.csproj b/src/Adaptive.Agrona/Adaptive.Agrona.csproj
@@ -3,7 +3,7 @@
     <TargetFrameworks>netstandard2.0;net45</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <PackageId>Agrona</PackageId>
-    <VersionPrefix>1.10.1</VersionPrefix>
+    <VersionPrefix>1.10.2</VersionPrefix>
     <Authors>Adaptive Financial Consulting Ltd.</Authors>
     <Company>Adaptive Financial Consulting Ltd.</Company>
     <Product>Agrona libraries initially included in Aeron Client</Product>

diff --git a/src/Adaptive.Archiver/Adaptive.Archiver.csproj b/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.1</VersionPrefix>
+    <VersionPrefix>1.10.2</VersionPrefix>
     <Authors>Adaptive Financial Consulting Ltd.</Authors>
     <Company>Adaptive Financial Consulting Ltd.</Company>
     <Product>Archiving over the Aeron transport</Product>

diff --git a/src/Adaptive.Archiver/AeronArchive.cs b/src/Adaptive.Archiver/AeronArchive.cs
@@ -398,7 +398,7 @@ public ExclusivePublication AddRecordedExclusivePublication(string channel, int
         /// <param name="channel">        to be recorded. </param>
         /// <param name="streamId">       to be recorded. </param>
         /// <param name="sourceLocation"> of the publication to be recorded. </param>
-        /// <returns> the correlationId used to identify the request. </returns>
+        /// <returns> the subscriptionId of the recording. </returns>
         public long StartRecording(string channel, int streamId, SourceLocation sourceLocation)
         {
             _lock.Lock();
@@ -411,9 +411,7 @@ public long StartRecording(string channel, int streamId, SourceLocation sourceLo
                     throw new ArchiveException("failed to send start recording request");
                 }
 
-                PollForResponse(correlationId);
-
-                return correlationId;
+                return PollForResponse(correlationId);
             }
             finally
             {
@@ -432,7 +430,7 @@ public long StartRecording(string channel, int streamId, SourceLocation sourceLo
         /// <param name="channel">        to be recorded. </param>
         /// <param name="streamId">       to be recorded. </param>
         /// <param name="sourceLocation"> of the publication to be recorded. </param>
-        /// <returns> the correlationId used to identify the request. </returns>
+        /// <returns> the subscriptionId of the recording. </returns>
         public long ExtendRecording(long recordingId, string channel, int streamId,
             SourceLocation sourceLocation)
         {
@@ -447,9 +445,7 @@ public long ExtendRecording(long recordingId, string channel, int streamId,
                     throw new ArchiveException("failed to send extend recording request");
                 }
 
-                PollForResponse(correlationId);
-
-                return correlationId;
+                return PollForResponse(correlationId);
             }
             finally
             {
@@ -500,16 +496,48 @@ public void StopRecording(Publication publication)
         }
 
         /// <summary>
-        /// Start a replay for a length in bytes of a recording from a position. If the position is <seealso cref="#NULL_POSITION"/>
+        /// Stop recording for a subscriptionId that has been returned from
+        /// <seealso cref="#startRecording(String, int, SourceLocation)"/> or
+        /// <seealso cref="#extendRecording(long, String, int, SourceLocation)"/>.
+        /// </summary>
+        /// <param name="subscriptionId"> the subscription was registered with for the recording. </param>
+        public void StopRecording(long subscriptionId)
+        {
+            _lock.Lock();
+            try
+            {
+                long correlationId = aeron.NextCorrelationId();
+
+                if (!archiveProxy.StopRecording(subscriptionId, correlationId, controlSessionId))
+                {
+                    throw new ArchiveException("failed to send stop recording request");
+                }
+
+                PollForResponse(correlationId);
+            }
+            finally
+            {
+                _lock.Unlock();
+            }
+        }
+
+
+        /// <summary>
+        /// Start a replay for a length in bytes of a recording from a position. If the position is <seealso cref="NULL_POSITION"/>
         /// then the stream will be replayed from the start.
+        ///
+        /// The lower 32-bits of the returned value contain the <see cref="Image.SessionId"/> of the received replay. All
+        /// 64-bits are required to uniquely identify the replay when calling <see cref="StopReplay(long)"/>. The lower 32-bits
+        /// can be obtained by casting the <see cref="long"/> value to an <see cref="int"/>.
+        /// 
         /// </summary>
         /// <param name="recordingId">    to be replayed. </param>
-        /// <param name="position">       from which the replay should begin or <seealso cref="#NULL_POSITION"/> if from the start. </param>
-        /// <param name="length">         of the stream to be replayed. Use <seealso cref="Long#MAX_VALUE"/> to follow a live recording or <see cref="NULL_LENGTH"/> to replay the whole stream of unknown length. </param>
+        /// <param name="position">       from which the replay should begin or <seealso cref="NULL_POSITION"/> if from the start. </param>
+        /// <param name="length">         of the stream to be replayed. Use <seealso cref="long.MaxValue"/> to follow a live recording or <see cref="NULL_LENGTH"/> to replay the whole stream of unknown length. </param>
         /// <param name="replayChannel">  to which the replay should be sent. </param>
         /// <param name="replayStreamId"> to which the replay should be sent. </param>
-        /// <returns> the id of the replay session which will be the same as the <seealso cref="Image#sessionId()"/> of the received
-        /// replay for correlation with the matching channel and stream id. </returns>
+        /// <returns> the id of the replay session which will be the same as the <seealso cref="Image.SessionId"/> of the received
+        /// replay for correlation with the matching channel and stream id in the lower 32 bits. </returns>
         public long StartReplay(long recordingId, long position, long length, string replayChannel,
             int replayStreamId)
         {