ConfigureAwait

docs/examples/RabbitMQ.Next.Examples.SimpleConsumer/Program.cs

@@ -14,7 +14,8 @@ static async Task Main()
 
         var connection = await ConnectionBuilder.Default
             .Endpoint("amqp://test2:test2@localhost:5672/")
-            .ConnectAsync();
+            .ConnectAsync()
+            .ConfigureAwait(false);
 
         Console.WriteLine("Connection opened");
 
@@ -33,7 +34,7 @@ static async Task Main()
         await consumer.ConsumeAsync(async message =>
         {
             Console.WriteLine($"[{DateTimeOffset.Now.TimeOfDay}] Message received via '{message.Exchange}' exchange: {message.Content<string>()}");
-        } ,cancellation.Token);
+        } ,cancellation.Token).ConfigureAwait(false);
     }
 
     private static Task MonitorKeypressAsync(CancellationTokenSource cancellation)

docs/examples/RabbitMQ.Next.Examples.SimplePublisher/Program.cs

@@ -14,7 +14,8 @@ static async Task Main()
 
         var connection = await ConnectionBuilder.Default
             .Endpoint("amqp://test2:test2@localhost:5672/")
-            .ConnectAsync();
+            .ConnectAsync()
+            .ConfigureAwait(false);
 
         Console.WriteLine("Connection opened");
 
@@ -33,7 +34,7 @@ static async Task Main()
 
             try
             {
-                await publisher.PublishAsync(input);
+                await publisher.PublishAsync(input).ConfigureAwait(false);
             }
             catch (Exception e)
             {

docs/examples/RabbitMQ.Next.Examples.TopologyBuilder/Program.cs

@@ -15,7 +15,8 @@ static async Task Main()
 
             var connection = await ConnectionBuilder.Default
                 .Endpoint("amqp://test:pass@localhost:5672/")
-                .ConnectAsync();
+                .ConnectAsync()
+                .ConfigureAwait(false);
 
             Console.WriteLine("Connection opened");
 

src/RabbitMQ.Next.Abstractions/ConnectionExtensions.cs

@@ -9,7 +9,7 @@ public static class ConnectionExtensions
     public static Task UseChannelAsync<TState>(this IConnection connection, TState state, Func<TState, IChannel, Task> fn)
         => connection.UseChannelAsync((state, fn), async (st, ch) =>
         {
-            await st.fn(st.state, ch);
+            await st.fn(st.state, ch).ConfigureAwait(false);
             return true;
         });
 
@@ -23,14 +23,14 @@ public static async Task<TResult> UseChannelAsync<TState, TResult>(this IConnect
         IChannel channel = null;
         try
         {
-            channel = await connection.OpenChannelAsync();
-            return await fn(state, channel);
+            channel = await connection.OpenChannelAsync().ConfigureAwait(false);
+            return await fn(state, channel).ConfigureAwait(false);
         }
         finally
         {
             if (channel != null && !channel.Completion.IsCompleted)
             {
-                await channel.CloseAsync();
+                await channel.CloseAsync().ConfigureAwait(false);
             }
         }
     }

src/RabbitMQ.Next.Consumer/Consumer.cs

@@ -44,7 +44,7 @@ public Consumer(
     }
 
     public async ValueTask DisposeAsync()
-        => await this.CancelConsumeAsync();
+        => await this.CancelConsumeAsync().ConfigureAwait(false);
 
     public async Task ConsumeAsync(Func<IDeliveredMessage, ValueTask> handler, CancellationToken cancellation)
     {
@@ -58,15 +58,15 @@ public async Task ConsumeAsync(Func<IDeliveredMessage, ValueTask> handler, Cance
             throw new ArgumentNullException(nameof(handler));
         }
 
-        await this.InitConsumerAsync(handler);
+        await this.InitConsumerAsync(handler).ConfigureAwait(false);
 
         try
         {
-            await Task.WhenAny(cancellation.AsTask(), this.channel.Completion);
+            await Task.WhenAny(cancellation.AsTask(), this.channel.Completion).ConfigureAwait(false);
         }
         finally
         {
-            await this.CancelConsumeAsync();   
+            await this.CancelConsumeAsync().ConfigureAwait(false);   
         }
     }
 
@@ -80,37 +80,38 @@ private async Task CancelConsumeAsync()
         for (var i = 0; i < this.queues.Count; i++)
         {
             var queue = this.queues[i];
-            await this.channel.SendAsync<CancelMethod, CancelOkMethod>(new CancelMethod(queue.ConsumerTag));
+            await this.channel.SendAsync<CancelMethod, CancelOkMethod>(new CancelMethod(queue.ConsumerTag)).ConfigureAwait(false);
         }
 
         if (this.acknowledgement != null)
         {
-            await this.acknowledgement.DisposeAsync();
+            await this.acknowledgement.DisposeAsync().ConfigureAwait(false);
         }
 
-        await this.channel.CloseAsync();
+        await this.channel.CloseAsync().ConfigureAwait(false);
 
         this.acknowledgement = null;
         this.channel = null;
     }
 
     private async Task InitConsumerAsync(Func<IDeliveredMessage, ValueTask> handler)
     {
-        this.channel = await this.connection.OpenChannelAsync();
+        this.channel = await this.connection.OpenChannelAsync().ConfigureAwait(false);
         this.acknowledgement = this.acknowledgementFactory(this.channel);
 
         var deliverHandler = new DeliverMessageHandler(handler, this.acknowledgement, this.serializer, this.onPoisonMessage, this.concurrencyLevel);
         this.channel.WithMessageHandler(deliverHandler);
 
-        await this.channel.SendAsync<QosMethod, QosOkMethod>(new QosMethod(this.prefetchSize, this.prefetchCount, false));
+        await this.channel.SendAsync<QosMethod, QosOkMethod>(new QosMethod(this.prefetchSize, this.prefetchCount, false)).ConfigureAwait(false);
 
         for (var i = 0; i < this.queues.Count; i++)
         {
             var queue = this.queues[i];
             var response = await this.channel.SendAsync<ConsumeMethod, ConsumeOkMethod>(
                 new ConsumeMethod(
                     queue.Queue, queue.ConsumerTag, queue.NoLocal, this.acknowledgement == null,
-                    queue.Exclusive, queue.Arguments));
+                    queue.Exclusive, queue.Arguments))
+                .ConfigureAwait(false);
 
             queue.ConsumerTag = response.ConsumerTag;
         }

src/RabbitMQ.Next.Consumer/DeliverMessageHandler.cs

@@ -54,19 +54,19 @@ public void Release(Exception ex = null)
     private async Task ProcessDeliveredMessagesAsync()
     {
         var reader = this.deliverChannel.Reader;
-        while (await reader.WaitToReadAsync())
+        while (await reader.WaitToReadAsync().ConfigureAwait(false))
         {
             while (reader.TryRead(out var delivered))
             {
                 var message = new DeliveredMessage(this.serializer, delivered.method, delivered.payload);
                 try
                 {
-                    await this.messageHandler(message);
-                    await this.acknowledgement.AckAsync(delivered.method.DeliveryTag);
+                    await this.messageHandler(message).ConfigureAwait(false);
+                    await this.acknowledgement.AckAsync(delivered.method.DeliveryTag).ConfigureAwait(false);
                 }
                 catch (Exception)
                 {
-                    await this.acknowledgement.NackAsync(delivered.method.DeliveryTag, this.onPoisonMessage == PoisonMessageMode.Requeue);
+                    await this.acknowledgement.NackAsync(delivered.method.DeliveryTag, this.onPoisonMessage == PoisonMessageMode.Requeue).ConfigureAwait(false);
                 }
                 finally
                 {

src/RabbitMQ.Next.Publisher/ConfirmMessageMiddleware.cs

@@ -27,9 +27,9 @@ public ValueTask InitAsync(IChannel channel, CancellationToken cancellation)
 
     public async ValueTask<ulong> InvokeAsync<TContent>(TContent content, IMessageBuilder message, CancellationToken cancellation)
     {
-        var deliveryTag = await this.next.InvokeAsync(content, message, cancellation);
+        var deliveryTag = await this.next.InvokeAsync(content, message, cancellation).ConfigureAwait(false);
 
-        var confirmed = await this.confirms.WaitForConfirmAsync(deliveryTag, cancellation);
+        var confirmed = await this.confirms.WaitForConfirmAsync(deliveryTag, cancellation).ConfigureAwait(false);
         if (!confirmed)
         {
             // todo: provide some useful info here

src/RabbitMQ.Next.Publisher/Publisher.cs

@@ -52,7 +52,7 @@ public async ValueTask DisposeAsync()
 
         if (ch != null)
         {
-            await ch.CloseAsync();
+            await ch.CloseAsync().ConfigureAwait(false);
         }
     }
 
@@ -79,8 +79,8 @@ private async Task InternalPublishAsync<TContent>(TContent content, MessageBuild
         {
             this.CheckDisposed();
 
-            var pipeline = this.publishPipeline ?? await this.InitializeAsync(cancellation);
-            await pipeline.InvokeAsync(content, message, cancellation);
+            var pipeline = this.publishPipeline ?? await this.InitializeAsync(cancellation).ConfigureAwait(false);
+            await pipeline.InvokeAsync(content, message, cancellation).ConfigureAwait(false);
         }
         finally
         {
@@ -104,7 +104,7 @@ private async ValueTask<IPublishMiddleware> InitializeAsync(CancellationToken ca
             throw new InvalidOperationException("Connection should be in Open state to use the API");
         }
 
-        await this.channelOpenSync.WaitAsync(cancellationToken);
+        await this.channelOpenSync.WaitAsync(cancellationToken).ConfigureAwait(false);
 
         try
         {
@@ -113,9 +113,9 @@ private async ValueTask<IPublishMiddleware> InitializeAsync(CancellationToken ca
                 return this.publishPipeline;
             }
 
-            this.channel = await this.connection.OpenChannelAsync(cancellationToken);
+            this.channel = await this.connection.OpenChannelAsync(cancellationToken).ConfigureAwait(false);
             // ensure target exchange exists
-            await this.channel.SendAsync<DeclareMethod, DeclareOkMethod>(new DeclareMethod(this.exchange), cancellationToken);
+            await this.channel.SendAsync<DeclareMethod, DeclareOkMethod>(new DeclareMethod(this.exchange), cancellationToken).ConfigureAwait(false);
 
             IReturnMiddleware returnPipeline = new VoidReturnMiddleware();
             for (var i = 0; i < this.returnMiddlewares.Count; i++)
@@ -126,12 +126,12 @@ private async ValueTask<IPublishMiddleware> InitializeAsync(CancellationToken ca
             this.channel.WithMessageHandler(new ReturnMessageHandler(returnPipeline, this.serializer));
 
             this.publishPipeline = new InternalMessagePublisher(this.exchange, this.serializer);
-            await this.publishPipeline.InitAsync(this.channel, default);
+            await this.publishPipeline.InitAsync(this.channel, default).ConfigureAwait(false);
 
             for (var i = 0; i < this.publishMiddlewares.Count; i++)
             {
                 this.publishPipeline = this.publishMiddlewares[i].Invoke(this.publishPipeline);
-                await this.publishPipeline.InitAsync(this.channel, default);
+                await this.publishPipeline.InitAsync(this.channel, default).ConfigureAwait(false);
             }
 
             return this.publishPipeline;
@@ -140,7 +140,7 @@ private async ValueTask<IPublishMiddleware> InitializeAsync(CancellationToken ca
         {
             if (this.channel != null)
             {
-                await this.channel.CloseAsync();
+                await this.channel.CloseAsync().ConfigureAwait(false);
             }
 
             throw;

src/RabbitMQ.Next.Publisher/ReturnMessageHandler.cs

@@ -50,13 +50,13 @@ public void Release(Exception ex = null)
     private async Task ProcessReturnedMessagesAsync()
     {
         var reader = this.returnChannel.Reader;
-        while (await reader.WaitToReadAsync())
+        while (await reader.WaitToReadAsync().ConfigureAwait(false))
         {
             while (reader.TryRead(out var returned))
             {
                 try
                 {
-                    await this.returnPipeline.InvokeAsync(returned, default); 
+                    await this.returnPipeline.InvokeAsync(returned, default).ConfigureAwait(false); 
                 }
                 finally
                 {

src/RabbitMQ.Next.TopologyBuilder/ConnectionExtensions.cs

@@ -13,7 +13,7 @@ public static async Task ConfigureAsync(this IConnection connection, Func<ITopol
         }
 
         using var topologyBuilder = new TopologyBuilder(connection);
-        await builder.Invoke(topologyBuilder);
+        await builder.Invoke(topologyBuilder).ConfigureAwait(false);
     }
 
     public static async Task ConfigureAsync<TState>(this IConnection connection, TState state, Func<ITopologyBuilder, TState, Task> builder)
@@ -24,6 +24,6 @@ public static async Task ConfigureAsync<TState>(this IConnection connection, TSt
         }
 
         using var topologyBuilder = new TopologyBuilder(connection);
-        await builder.Invoke(topologyBuilder, state);
+        await builder.Invoke(topologyBuilder, state).ConfigureAwait(false);
     }
 }

src/RabbitMQ.Next/Buffers/IMemoryAccessor.cs

@@ -14,6 +14,4 @@ internal interface IMemoryAccessor : IDisposable
     IMemoryAccessor Append(IMemoryAccessor next);
 
     void WriteTo(Stream stream);
-
-    void CopyTo(Span<byte> destination);
 }

src/RabbitMQ.Next/Buffers/MemoryAccessor.cs

@@ -34,9 +34,4 @@ public void WriteTo(Stream stream)
         ArgumentNullException.ThrowIfNull(stream);
         stream.Write(this.memory, 0, this.memory.Length);
     }
-
-    public void CopyTo(Span<byte> destination)
-    {
-        this.memory.CopyTo(destination);
-    }
 }

src/RabbitMQ.Next/Buffers/PooledMemoryAccessor.cs

@@ -81,14 +81,6 @@ public void WriteTo(Stream stream)
         stream.Write(this.memory, this.offset, this.Size);
     }
 
-    public void CopyTo(Span<byte> destination)
-    {
-        if (this.Size > 0)
-        {
-            this.Memory.Span.CopyTo(destination);
-        }
-    }
-
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
     private void CheckDisposed()
     {

src/RabbitMQ.Next/Buffers/SharedMemory.cs

@@ -159,12 +159,6 @@ public void WriteTo(Stream stream)
             stream.Write(this.owner.memory, this.offset, this.Size);
         }
 
-        public void CopyTo(Span<byte> destination)
-        {
-            this.CheckDisposed();
-            this.Memory.Span.CopyTo(destination);
-        }
-
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void CheckDisposed()
         {

src/RabbitMQ.Next/Channels/Channel.cs

@@ -89,8 +89,8 @@ public async Task<TResponse> SendAsync<TRequest, TResponse>(TRequest request, Ca
         where TRequest : struct, IOutgoingMethod where TResponse : struct, IIncomingMethod
     {
         var waitTask = this.WaitAsync<TResponse>(cancellation);
-        await this.SendAsync(request, cancellation);
-        return await waitTask;
+        await this.SendAsync(request, cancellation).ConfigureAwait(false);
+        return await waitTask.ConfigureAwait(false);
     }
 
     public async Task<ulong> PublishAsync<TState>(
@@ -116,7 +116,7 @@ public async Task<ulong> PublishAsync<TState>(
             this.messageBuilderPool.Return(messageBuilder);   
         }
 
-        await this.WriteToSocketAsync(memory, cancellation);
+        await this.WriteToSocketAsync(memory, cancellation).ConfigureAwait(false);
         var deliveryTag = Interlocked.Increment(ref this.lastDeliveryTag);
 
         return deliveryTag;
@@ -157,7 +157,7 @@ public async Task<TMethod> WaitAsync<TMethod>(CancellationToken cancellation = d
 
         try
         {
-            return await waitHandler.WaitTask;
+            return await waitHandler.WaitTask.ConfigureAwait(false);
         }
         finally
         {
@@ -167,13 +167,13 @@ public async Task<TMethod> WaitAsync<TMethod>(CancellationToken cancellation = d
 
     public async Task CloseAsync(Exception ex = null)
     {
-        await this.SendAsync<CloseMethod, CloseOkMethod>(new CloseMethod((ushort) ReplyCode.Success, string.Empty, MethodId.Unknown));
+        await this.SendAsync<CloseMethod, CloseOkMethod>(new CloseMethod((ushort) ReplyCode.Success, string.Empty, MethodId.Unknown)).ConfigureAwait(false);
         this.TryComplete(ex);
     }
 
     public async Task CloseAsync(ushort statusCode, string description, MethodId failedMethodId)
     {
-        await this.SendAsync<CloseMethod, CloseOkMethod>(new CloseMethod(statusCode, description, failedMethodId));
+        await this.SendAsync<CloseMethod, CloseOkMethod>(new CloseMethod(statusCode, description, failedMethodId)).ConfigureAwait(false);
         this.TryComplete(new ChannelException(statusCode, description, failedMethodId));
     }
 

src/RabbitMQ.Next/ConnectionFactory.cs

@@ -18,7 +18,7 @@ public async Task<IConnection> ConnectAsync(ConnectionSettings settings, Cancell
         var memoryPool = new DefaultObjectPool<byte[]>(new MemoryPoolPolicy(bufferSize), 100);
 
         var connection = new Connection(settings, memoryPool);
-        await connection.OpenConnectionAsync(cancellation);
+        await connection.OpenConnectionAsync(cancellation).ConfigureAwait(false);
         return connection;
     }
 }

src/RabbitMQ.Next/RabbitMQ.Next.csproj

@@ -16,11 +16,5 @@
   <ItemGroup>
     <ProjectReference Include="..\RabbitMQ.Next.Abstractions\RabbitMQ.Next.Abstractions.csproj" />
   </ItemGroup>
-
-  <ItemGroup>
-    <Compile Remove="Buffers\MemoryBlock.cs" />
-    <Compile Remove="Buffers\MemoryBlockPool.cs" />
-    <Compile Remove="Buffers\StaticMemoryAccessor.cs" />
-  </ItemGroup>
 
 </Project>