Removing unused ConnectionOpen method.

stebet · lukebakken · commit 5e0f118efd63 · 2023-04-12T15:42:39.000-07:00
Adding Timeout extension for ValueTask for connection handshake.
Don't target .NET Framework 4.7.2 in Benchmarks, Test Apps and Unit Tests when not building in Windows.
diff --git a/projects/Benchmarks/Benchmarks.csproj b/projects/Benchmarks/Benchmarks.csproj
@@ -1,7 +1,14 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
-  <PropertyGroup>
+  <PropertyGroup Condition="$([MSBuild]::IsOSPlatform('Windows'))">
     <TargetFrameworks>net6.0;net472</TargetFrameworks>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="!$([MSBuild]::IsOSPlatform('Windows'))">
+    <TargetFramework>net6.0</TargetFramework>
+  </PropertyGroup>
+  
+  <PropertyGroup>
     <OutputType>Exe</OutputType>
     <AssemblyOriginatorKeyFile>../rabbit.snk</AssemblyOriginatorKeyFile>
     <SignAssembly>true</SignAssembly>
diff --git a/projects/RabbitMQ.Client/client/api/IChannelExtensions.cs b/projects/RabbitMQ.Client/client/api/IChannelExtensions.cs
@@ -94,7 +94,7 @@ public static void BasicPublish(this IChannel channel, string exchange, string r
 
         public static void BasicPublish(this IChannel channel, CachedString exchange, CachedString routingKey, ReadOnlyMemory<byte> body = default, bool mandatory = false)
             => channel.BasicPublish(exchange, routingKey, in EmptyBasicProperty.Empty, body, mandatory);
-        
+
         public static ValueTask BasicPublishAsync(this IChannel channel, string exchange, string routingKey, ReadOnlyMemory<byte> body = default, bool mandatory = false)
             => channel.BasicPublishAsync(exchange, routingKey, in EmptyBasicProperty.Empty, body, mandatory);
 
diff --git a/projects/RabbitMQ.Client/client/framing/Channel.cs b/projects/RabbitMQ.Client/client/framing/Channel.cs
@@ -109,7 +109,7 @@ public override void _Private_ConnectionOpen(string virtualHost)
         {
             ChannelSend(new ConnectionOpen(virtualHost));
         }
-        
+
         public override ValueTask _Private_ConnectionOpenAsync(string virtualHost)
         {
             return ModelSendAsync(new ConnectionOpen(virtualHost));
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs
@@ -317,7 +317,7 @@ public void BasicPublish<TProperties>(string exchange, string routingKey, in TPr
         public void BasicPublish<TProperties>(CachedString exchange, CachedString routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader
             => InnerChannel.BasicPublish(exchange, routingKey, in basicProperties, body, mandatory);
-        
+
         public ValueTask BasicPublishAsync<TProperties>(string exchange, string routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader
             => InnerChannel.BasicPublishAsync(exchange, routingKey, in basicProperties, body, mandatory);
diff --git a/projects/RabbitMQ.Client/client/impl/ChannelBase.cs b/projects/RabbitMQ.Client/client/impl/ChannelBase.cs
@@ -241,29 +241,9 @@ private async Task CloseAsync(ShutdownEventArgs reason, bool abort)
             }
         }
 
-        internal void ConnectionOpen(string virtualHost)
+        internal async ValueTask ConnectionOpenAsync(string virtualHost)
         {
-            var k = new SimpleBlockingRpcContinuation();
-            lock (_rpcLock)
-            {
-                Enqueue(k);
-                try
-                {
-                    _Private_ConnectionOpen(virtualHost);
-                }
-                catch (AlreadyClosedException)
-                {
-                    // let continuation throw OperationInterruptedException,
-                    // which is a much more suitable exception before connection
-                    // negotiation finishes
-                }
-                k.GetReply(HandshakeContinuationTimeout);
-            }
-        }
-        
-        internal ValueTask ConnectionOpenAsync(string virtualHost)
-        {
-            return _Private_ConnectionOpenAsync(virtualHost);
+            await _Private_ConnectionOpenAsync(virtualHost).TimeoutAfter(HandshakeContinuationTimeout);
         }
 
         internal async ValueTask<ConnectionSecureOrTune> ConnectionSecureOkAsync(byte[] response)
@@ -401,7 +381,7 @@ protected void ChannelSend<T>(in T method) where T : struct, IOutgoingAmqpMethod
         {
             Session.Transmit(in method);
         }
-        
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         protected ValueTask ModelSendAsync<T>(in T method) where T : struct, IOutgoingAmqpMethod
         {
@@ -419,7 +399,7 @@ protected void ChannelSend<TMethod, THeader>(in TMethod method, in THeader heade
             }
             Session.Transmit(in method, in header, body);
         }
-        
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         protected ValueTask ModelSendAsync<TMethod, THeader>(in TMethod method, in THeader header, ReadOnlyMemory<byte> body)
             where TMethod : struct, IOutgoingAmqpMethod
@@ -429,7 +409,7 @@ protected ValueTask ModelSendAsync<TMethod, THeader>(in TMethod method, in THead
             {
                 _flowControlBlock.Wait();
             }
-            
+
             return Session.TransmitAsync(in method, in header, body);
         }
 
@@ -950,7 +930,7 @@ public void BasicPublish<TProperties>(CachedString exchange, CachedString routin
             var cmd = new BasicPublishMemory(exchange.Bytes, routingKey.Bytes, mandatory, default);
             ChannelSend(in cmd, in basicProperties, body);
         }
-        
+
         public ValueTask BasicPublishAsync<TProperties>(string exchange, string routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader
         {
@@ -965,7 +945,7 @@ public ValueTask BasicPublishAsync<TProperties>(string exchange, string routingK
             var cmd = new BasicPublish(exchange, routingKey, mandatory, default);
             return ModelSendAsync(in cmd, in basicProperties, body);
         }
-        
+
         public ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)
             where TProperties : IReadOnlyBasicProperties, IAmqpHeader
         {
diff --git a/projects/RabbitMQ.Client/client/impl/Connection.cs b/projects/RabbitMQ.Client/client/impl/Connection.cs
@@ -408,7 +408,7 @@ internal void Write(ReadOnlyMemory<byte> memory)
                 task.AsTask().GetAwaiter().GetResult();
             }
         }
-        
+
         internal ValueTask WriteAsync(ReadOnlyMemory<byte> memory)
         {
             return _frameHandler.WriteAsync(memory);
diff --git a/projects/RabbitMQ.Client/client/impl/ISession.cs b/projects/RabbitMQ.Client/client/impl/ISession.cs
@@ -78,7 +78,7 @@ internal interface ISession
         void Transmit<TMethod, THeader>(in TMethod cmd, in THeader header, ReadOnlyMemory<byte> body)
             where TMethod : struct, IOutgoingAmqpMethod
             where THeader : IAmqpHeader;
-        
+
         ValueTask TransmitAsync<TMethod, THeader>(in TMethod cmd, in THeader header, ReadOnlyMemory<byte> body)
             where TMethod : struct, IOutgoingAmqpMethod
             where THeader : IAmqpHeader;
diff --git a/projects/RabbitMQ.Client/client/impl/SessionBase.cs b/projects/RabbitMQ.Client/client/impl/SessionBase.cs
@@ -162,7 +162,7 @@ public void Transmit<TMethod, THeader>(in TMethod cmd, in THeader header, ReadOn
 
             Connection.Write(Framing.SerializeToFrames(ref Unsafe.AsRef(cmd), ref Unsafe.AsRef(header), body, ChannelNumber, Connection.MaxPayloadSize));
         }
-        
+
         public ValueTask TransmitAsync<TMethod, THeader>(in TMethod cmd, in THeader header, ReadOnlyMemory<byte> body)
             where TMethod : struct, IOutgoingAmqpMethod
             where THeader : IAmqpHeader
diff --git a/projects/RabbitMQ.Client/client/impl/SocketFrameHandler.cs b/projects/RabbitMQ.Client/client/impl/SocketFrameHandler.cs
@@ -62,6 +62,28 @@ public static async Task TimeoutAfter(this Task task, TimeSpan timeout)
             }
 #endif
         }
+
+        public static async ValueTask TimeoutAfter(this ValueTask task, TimeSpan timeout)
+        {
+            if (!task.IsCompletedSuccessfully)
+            {
+                var actualTask = task.AsTask();
+#if NET6_0_OR_GREATER
+                await actualTask.WaitAsync(timeout).ConfigureAwait(false);
+#else
+                if (actualTask == await Task.WhenAny(actualTask, Task.Delay(timeout)).ConfigureAwait(false))
+                {
+                    await actualTask.ConfigureAwait(false);
+                }
+                else
+                {
+                    Task supressErrorTask =
+     actualTask.ContinueWith((t, s) => t.Exception.Handle(e => true), null, CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
+                    throw new TimeoutException();
+                }
+#endif
+            }
+        }
     }
 
     internal sealed class SocketFrameHandler : IFrameHandler
diff --git a/projects/TestApplications/CreateChannel/CreateChannel.csproj b/projects/TestApplications/CreateChannel/CreateChannel.csproj
@@ -1,8 +1,15 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
+  <PropertyGroup Condition="$([MSBuild]::IsOSPlatform('Windows'))">
+    <TargetFrameworks>net6.0;net472</TargetFrameworks>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="!$([MSBuild]::IsOSPlatform('Windows'))">
+    <TargetFramework>net6.0</TargetFramework>
+  </PropertyGroup>
+  
   <PropertyGroup>
     <OutputType>Exe</OutputType>
-    <TargetFrameworks>net6.0;net472</TargetFrameworks>
   </PropertyGroup>
 
   <ItemGroup>
diff --git a/projects/TestApplications/MassPublish/MassPublish.csproj b/projects/TestApplications/MassPublish/MassPublish.csproj
@@ -1,8 +1,15 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
+  <PropertyGroup Condition="$([MSBuild]::IsOSPlatform('Windows'))">
+    <TargetFrameworks>net6.0;net472</TargetFrameworks>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="!$([MSBuild]::IsOSPlatform('Windows'))">
+    <TargetFramework>net6.0</TargetFramework>
+  </PropertyGroup>
+  
   <PropertyGroup>
     <OutputType>Exe</OutputType>
-    <TargetFrameworks>net6.0;net472</TargetFrameworks>
   </PropertyGroup>
 
   <ItemGroup>
diff --git a/projects/Unit/TestConcurrentAccessWithSharedConnection.cs b/projects/Unit/TestConcurrentAccessWithSharedConnection.cs
@@ -86,7 +86,7 @@ public void TestConcurrentChannelOpenAndPublishingSize1024()
         {
             TestConcurrentChannelOpenAndPublishingWithBodyOfSize(1024);
         }
-        
+
         [Fact]
         public Task TestConcurrentChannelOpenAndPublishingWithBlankMessagesAsync()
         {
@@ -125,7 +125,7 @@ internal void TestConcurrentChannelOpenAndPublishingWithBodyOfSize(int length, i
         {
             TestConcurrentChannelOpenAndPublishingWithBody(new byte[length], iterations);
         }
-        
+
         internal Task TestConcurrentChannelOpenAndPublishingWithBodyOfSizeAsync(int length, int iterations = 30)
         {
             return TestConcurrentChannelOpenAndPublishingWithBodyAsync(new byte[length], iterations);
@@ -147,7 +147,7 @@ internal void TestConcurrentChannelOpenAndPublishingWithBody(byte[] body, int it
                 ch.WaitForConfirmsAsync(cts.Token).GetAwaiter().GetResult();
             }, iterations);
         }
-        
+
         internal Task TestConcurrentChannelOpenAndPublishingWithBodyAsync(byte[] body, int iterations)
         {
             return TestConcurrentChannelOperationsAsync(async (conn) =>
@@ -170,7 +170,7 @@ internal void TestConcurrentChannelOperations(Action<IConnection> actions,
         {
             TestConcurrentChannelOperations(actions, iterations, _completionTimeout);
         }
-        
+
         internal Task TestConcurrentChannelOperationsAsync(Func<IConnection, Task> actions, int iterations)
         {
             return TestConcurrentChannelOperationsAsync(actions, iterations, _completionTimeout);
@@ -198,7 +198,7 @@ internal void TestConcurrentChannelOperations(Action<IConnection> actions,
             // closing the connection
             Assert.True(_conn.IsOpen);
         }
-        
+
         internal async Task TestConcurrentChannelOperationsAsync(Func<IConnection, Task> actions,
             int iterations, TimeSpan timeout)
         {
diff --git a/projects/Unit/Unit.csproj b/projects/Unit/Unit.csproj
@@ -1,7 +1,14 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
-  <PropertyGroup>
+  <PropertyGroup Condition="$([MSBuild]::IsOSPlatform('Windows'))">
     <TargetFrameworks>net6.0;net472</TargetFrameworks>
+  </PropertyGroup>
+  
+  <PropertyGroup Condition="!$([MSBuild]::IsOSPlatform('Windows'))">
+    <TargetFramework>net6.0</TargetFramework>
+  </PropertyGroup>
+  
+  <PropertyGroup>
     <AssemblyOriginatorKeyFile>../rabbit.snk</AssemblyOriginatorKeyFile>
     <SignAssembly>true</SignAssembly>
     <LangVersion>latest</LangVersion>

Original file line number	Diff line number	Diff line change
`@@ -109,7 +109,7 @@ public override void _Private_ConnectionOpen(string virtualHost)`
`109`	`109`	`{`
`110`	`110`	`ChannelSend(new ConnectionOpen(virtualHost));`
`111`	`111`	`}`
`112`		`-`
	`112`	`+`
`113`	`113`	`public override ValueTask _Private_ConnectionOpenAsync(string virtualHost)`
`114`	`114`	`{`
`115`	`115`	`return ModelSendAsync(new ConnectionOpen(virtualHost));`
Original file line number	Diff line number	Diff line change
`@@ -241,29 +241,9 @@ private async Task CloseAsync(ShutdownEventArgs reason, bool abort)`
`241`	`241`	`}`
`242`	`242`	`}`
`243`	`243`
`244`		`- internal void ConnectionOpen(string virtualHost)`
	`244`	`+ internal async ValueTask ConnectionOpenAsync(string virtualHost)`
`245`	`245`	`{`
`246`		`- var k = new SimpleBlockingRpcContinuation();`
`247`		`- lock (_rpcLock)`
`248`		`- {`
`249`		`- Enqueue(k);`
`250`		`- try`
`251`		`- {`
`252`		`- _Private_ConnectionOpen(virtualHost);`
`253`		`- }`
`254`		`- catch (AlreadyClosedException)`
`255`		`- {`
`256`		`- // let continuation throw OperationInterruptedException,`
`257`		`- // which is a much more suitable exception before connection`
`258`		`- // negotiation finishes`
`259`		`- }`
`260`		`- k.GetReply(HandshakeContinuationTimeout);`
`261`		`- }`
`262`		`- }`
`263`		`-`
`264`		`- internal ValueTask ConnectionOpenAsync(string virtualHost)`
`265`		`- {`
`266`		`- return _Private_ConnectionOpenAsync(virtualHost);`
	`246`	`+ await _Private_ConnectionOpenAsync(virtualHost).TimeoutAfter(HandshakeContinuationTimeout);`
`267`	`247`	`}`
`268`	`248`
`269`	`249`	`internal async ValueTask<ConnectionSecureOrTune> ConnectionSecureOkAsync(byte[] response)`
`@@ -401,7 +381,7 @@ protected void ChannelSend<T>(in T method) where T : struct, IOutgoingAmqpMethod`
`401`	`381`	`{`
`402`	`382`	`Session.Transmit(in method);`
`403`	`383`	`}`
`404`		`-`
	`384`	`+`
`405`	`385`	`[MethodImpl(MethodImplOptions.AggressiveInlining)]`
`406`	`386`	`protected ValueTask ModelSendAsync<T>(in T method) where T : struct, IOutgoingAmqpMethod`
`407`	`387`	`{`
`@@ -419,7 +399,7 @@ protected void ChannelSend<TMethod, THeader>(in TMethod method, in THeader heade`
`419`	`399`	`}`
`420`	`400`	`Session.Transmit(in method, in header, body);`
`421`	`401`	`}`
`422`		`-`
	`402`	`+`
`423`	`403`	`[MethodImpl(MethodImplOptions.AggressiveInlining)]`
`424`	`404`	`protected ValueTask ModelSendAsync<TMethod, THeader>(in TMethod method, in THeader header, ReadOnlyMemory<byte> body)`
`425`	`405`	`where TMethod : struct, IOutgoingAmqpMethod`
`@@ -429,7 +409,7 @@ protected ValueTask ModelSendAsync<TMethod, THeader>(in TMethod method, in THead`
`429`	`409`	`{`
`430`	`410`	`_flowControlBlock.Wait();`
`431`	`411`	`}`
`432`		`-`
	`412`	`+`
`433`	`413`	`return Session.TransmitAsync(in method, in header, body);`
`434`	`414`	`}`
`435`	`415`
`@@ -950,7 +930,7 @@ public void BasicPublish<TProperties>(CachedString exchange, CachedString routin`
`950`	`930`	`var cmd = new BasicPublishMemory(exchange.Bytes, routingKey.Bytes, mandatory, default);`
`951`	`931`	`ChannelSend(in cmd, in basicProperties, body);`
`952`	`932`	`}`
`953`		`-`
	`933`	`+`
`954`	`934`	`public ValueTask BasicPublishAsync<TProperties>(string exchange, string routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)`
`955`	`935`	`where TProperties : IReadOnlyBasicProperties, IAmqpHeader`
`956`	`936`	`{`
`@@ -965,7 +945,7 @@ public ValueTask BasicPublishAsync<TProperties>(string exchange, string routingK`
`965`	`945`	`var cmd = new BasicPublish(exchange, routingKey, mandatory, default);`
`966`	`946`	`return ModelSendAsync(in cmd, in basicProperties, body);`
`967`	`947`	`}`
`968`		`-`
	`948`	`+`
`969`	`949`	`public ValueTask BasicPublishAsync<TProperties>(CachedString exchange, CachedString routingKey, in TProperties basicProperties, ReadOnlyMemory<byte> body, bool mandatory)`
`970`	`950`	`where TProperties : IReadOnlyBasicProperties, IAmqpHeader`
`971`	`951`	`{`
Original file line number	Diff line number	Diff line change
`@@ -408,7 +408,7 @@ internal void Write(ReadOnlyMemory<byte> memory)`
`408`	`408`	`task.AsTask().GetAwaiter().GetResult();`
`409`	`409`	`}`
`410`	`410`	`}`
`411`		`-`
	`411`	`+`
`412`	`412`	`internal ValueTask WriteAsync(ReadOnlyMemory<byte> memory)`
`413`	`413`	`{`
`414`	`414`	`return _frameHandler.WriteAsync(memory);`
Original file line number	Diff line number	Diff line change
`@@ -162,7 +162,7 @@ public void Transmit<TMethod, THeader>(in TMethod cmd, in THeader header, ReadOn`
`162`	`162`
`163`	`163`	`Connection.Write(Framing.SerializeToFrames(ref Unsafe.AsRef(cmd), ref Unsafe.AsRef(header), body, ChannelNumber, Connection.MaxPayloadSize));`
`164`	`164`	`}`
`165`		`-`
	`165`	`+`
`166`	`166`	`public ValueTask TransmitAsync<TMethod, THeader>(in TMethod cmd, in THeader header, ReadOnlyMemory<byte> body)`
`167`	`167`	`where TMethod : struct, IOutgoingAmqpMethod`
`168`	`168`	`where THeader : IAmqpHeader`