Async SASL (#618)

* Async SASL

* Fix netmf build on vs2015

Author: xinchen10

csproj/Amqp.NetFX40.csproj

@@ -147,6 +147,9 @@
       <Link>Transactions\ResourceManager.cs</Link>
     </Compile>
     <Compile Include="..\src\Net\BufferManager.cs" />
+    <Compile Include="..\src\Sasl\AsyncSaslProfile.cs">
+      <Link>Sasl\AsyncSaslProfile.cs</Link>
+    </Compile>
     <Compile Include="..\src\Types\IStringDecoder.cs">
       <Link>Types\IStringDecoder.cs</Link>
     </Compile>

csproj/Amqp.NetFX45.csproj

@@ -141,6 +141,9 @@
       <Link>Transactions\ResourceManager.cs</Link>
     </Compile>
     <Compile Include="..\src\Net\BufferManager.cs" />
+    <Compile Include="..\src\Sasl\AsyncSaslProfile.cs">
+      <Link>Sasl\AsyncSaslProfile.cs</Link>
+    </Compile>
     <Compile Include="..\src\Types\IStringDecoder.cs">
       <Link>Types\IStringDecoder.cs</Link>
     </Compile>

csproj/Amqp.Uwp.csproj

@@ -226,6 +226,9 @@
       <Link>Properties\Version.cs</Link>
     </Compile>
     <Compile Include="..\src\ReceiverLink.cs" />
+    <Compile Include="..\src\Sasl\AsyncSaslProfile.cs">
+      <Link>Sasl\AsyncSaslProfile.cs</Link>
+    </Compile>
     <Compile Include="..\src\Sasl\SaslChallenge.cs">
       <Link>Sasl\SaslChallenge.cs</Link>
     </Compile>

src/Amqp.Net.csproj

@@ -9,6 +9,7 @@
     <DefineConstants>$(DefineConstants);NETFX</DefineConstants>
     <IntermediateOutputPath>..\obj\$(Configuration)\$(MSBuildProjectName)\</IntermediateOutputPath>
     <OutputPath>..\bin\$(Configuration)\$(MSBuildProjectName)\</OutputPath>
+    <DocumentationFile>..\bin\$(Configuration)\$(MSBuildProjectName)\$(AssemblyName).XML</DocumentationFile>
   </PropertyGroup>
 
   <ItemGroup>

src/Net/AsyncPump.cs

@@ -38,7 +38,10 @@ public void Start(Connection connection, Action<Exception> onException = null)
             Task task = this.StartAsync(connection, onException);
         }
 
-        public async Task PumpAsync(uint maxFrameSize, Func<ProtocolHeader, bool> onHeader, Func<ByteBuffer, bool> onBuffer)
+        public async Task PumpAsync(uint maxFrameSize,
+            Func<ProtocolHeader, bool> onHeader,
+            Func<ByteBuffer, bool> onBuffer,
+            Func<ByteBuffer, Task<bool>> onBufferAsync = null)
         {
             byte[] header = new byte[FixedWidth.ULong];
 
@@ -73,7 +76,10 @@ public async Task PumpAsync(uint maxFrameSize, Func<ProtocolHeader, bool> onHead
                     buffer.Append(frameSize);
                     Trace.WriteBuffer("RECV {0}", buffer.Buffer, buffer.Offset, buffer.Length);
 
-                    if (!onBuffer(buffer))
+                    var pending = onBufferAsync != null ?
+                        await onBufferAsync(buffer).ConfigureAwait(false) :
+                        onBuffer(buffer);
+                    if (!pending)
                     {
                         break;
                     }

src/Net/TaskExtensions.cs

@@ -511,9 +511,30 @@ await pump.PumpAsync(
                     saslProfile.OnHeader(myHeader, header);
                     return true;
                 },
-                buffer =>
+                null,
+                async buffer =>
                 {
-                    return saslProfile.OnFrame(hostname, writer, buffer, out code);
+                    bool shouldContinue = saslProfile.OnFrame(hostname, buffer, out var response, out code);
+                    if (response != null)
+                    {
+                        var responseTask = response as AsyncSaslProfile.ResponseTask;
+                        if (responseTask != null)
+                        {
+                            response = await responseTask.Task;
+                            if (response != null && response.Descriptor.Code == Codec.SaslOutcome.Code)
+                            {
+                                code = ((SaslOutcome)response).Code;
+                                shouldContinue = false;
+                            }
+                        }
+
+                        if (response != null)
+                        {
+                            SaslProfile.SendCommand(writer, response);
+                        }
+                    }
+
+                    return shouldContinue;
                 }).ConfigureAwait(false);
 
             await writer.FlushAsync().ConfigureAwait(false);

src/Sasl/AsyncSaslProfile.cs

@@ -0,0 +1,80 @@
+//  ------------------------------------------------------------------------------------
+//  Copyright (c) Microsoft Corporation
+//  All rights reserved. 
+//  
+//  Licensed under the Apache License, Version 2.0 (the ""License""); you may not use this 
+//  file except in compliance with the License. You may obtain a copy of the License at 
+//  http://www.apache.org/licenses/LICENSE-2.0  
+//  
+//  THIS CODE IS PROVIDED *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, 
+//  EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY IMPLIED WARRANTIES OR 
+//  CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE, MERCHANTABLITY OR 
+//  NON-INFRINGEMENT. 
+// 
+//  See the Apache Version 2.0 License for specific language governing permissions and 
+//  limitations under the License.
+//  ------------------------------------------------------------------------------------
+
+namespace Amqp.Sasl
+{
+    using System;
+    using System.Runtime.CompilerServices;
+    using System.Threading.Tasks;
+    using Amqp.Types;
+
+    /// <summary>
+    /// A <seealso cref="SaslProfile"/> that processes commands asynchronously.
+    /// </summary>
+    public abstract class AsyncSaslProfile : SaslProfile
+    {
+        /// <summary>
+        /// <inheritdoc cref="SaslProfile.SaslProfile(Symbol)"/>
+        /// </summary>
+        public AsyncSaslProfile(Symbol mechanism)
+            : base(mechanism)
+        {
+        }
+
+        /// <summary>
+        /// Processes the received command asynchronously and returns a response. If returns
+        /// null, the SASL handshake completes.
+        /// </summary>
+        /// <param name="command">The SASL command received from the peer.</param>
+        /// <returns>A Task that returns a SASL command as a response to the incoming command.</returns>
+        protected abstract Task<DescribedList> OnCommandAsync(DescribedList command);
+
+        /// <summary>
+        /// <inheritdoc cref="SaslProfile.OnCommand(DescribedList)"/>
+        /// </summary>
+        protected sealed override DescribedList OnCommand(DescribedList command)
+        {
+            var task = this.OnCommandAsync(command);
+            return new ResponseTask(task);
+        }
+
+        // Just a wrapper of task that also satisfies the OnCommand contract.
+        internal sealed class ResponseTask : DescribedList
+        {
+            static readonly Descriptor descriptor = new Descriptor(ulong.MaxValue, string.Empty);
+            readonly Task<DescribedList> task;
+
+            public ResponseTask(Task<DescribedList> task)
+                : base(descriptor, 0)
+            {
+                this.task = task;
+            }
+
+            public Task<DescribedList> Task => this.task;
+
+            internal override void ReadField(ByteBuffer buffer, int index, byte formatCode)
+            {
+                throw new NotImplementedException();
+            }
+
+            internal override void WriteField(ByteBuffer buffer, int index)
+            {
+                throw new NotImplementedException();
+            }
+        }
+    }
+}

src/Sasl/SaslProfile.cs

@@ -75,17 +75,20 @@ internal ITransport Open(string hostname, ITransport transport)
             this.OnHeader(myHeader, theirHeader);
 
             SaslCode code = SaslCode.SysTemp;
-            while (true)
+            bool shouldContinue = true;
+            while (shouldContinue)
             {
                 ByteBuffer buffer = Reader.ReadFrameBuffer(transport, new byte[4], MaxFrameSize);
                 if (buffer == null)
                 {
                     throw new OperationCanceledException(Fx.Format(SRAmqp.TransportClosed, transport.GetType().Name));
                 }
 
-                if (!this.OnFrame(hostname, transport, buffer, out code))
+                DescribedList response = null;
+                shouldContinue = this.OnFrame(hostname, buffer, out response, out code);
+                if (response != null)
                 {
-                    break;
+                    SendCommand(transport, response);
                 }
             }
 
@@ -113,7 +116,7 @@ internal ProtocolHeader Start(ITransport transport, DescribedList command)
 
             if (command != null)
             {
-                this.SendCommand(transport, command);
+                SendCommand(transport, command);
             }
 
             return myHeader;
@@ -129,14 +132,15 @@ internal void OnHeader(ProtocolHeader myHeader, ProtocolHeader theirHeader)
             }
         }
 
-        internal bool OnFrame(string hostname, ITransport transport, ByteBuffer buffer, out SaslCode code)
+        internal bool OnFrame(string hostname, ByteBuffer buffer, out DescribedList response, out SaslCode code)
         {
             ushort channel;
             DescribedList command;
             Frame.Decode(buffer, out channel, out command);
             Trace.WriteLine(TraceLevel.Frame, "RECV {0}", command);
 
             bool shouldContinue = true;
+            response = null;
             if (command.Descriptor.Code == Codec.SaslOutcome.Code)
             {
                 code = ((SaslOutcome)command).Code;
@@ -162,24 +166,16 @@ internal bool OnFrame(string hostname, ITransport transport, ByteBuffer buffer,
                     throw new AmqpException(ErrorCode.NotImplemented, mechanisms.ToString());
                 }
 
-                DescribedList init = this.GetStartCommand(hostname);
-                if (init != null)
-                {
-                    this.SendCommand(transport, init);
-                }
+                response = this.GetStartCommand(hostname);
             }
             else
             {
                 code = SaslCode.Ok;
-                DescribedList response = this.OnCommand(command);
-                if (response != null)
+                response = this.OnCommand(command);
+                if (response != null && response.Descriptor.Code == Codec.SaslOutcome.Code)
                 {
-                    this.SendCommand(transport, response);
-                    if (response.Descriptor.Code == Codec.SaslOutcome.Code)
-                    {
-                        code = ((SaslOutcome)response).Code;
-                        shouldContinue = false;
-                    }
+                    code = ((SaslOutcome)response).Code;
+                    shouldContinue = false;
                 }
             }
 
@@ -231,7 +227,7 @@ protected virtual bool Match(Symbol mechanism)
         /// <returns>A SASL command as a response to the incoming command.</returns>
         protected abstract DescribedList OnCommand(DescribedList command);
 
-        void SendCommand(ITransport transport, DescribedList command)
+        internal static void SendCommand(ITransport transport, DescribedList command)
         {
             ByteBuffer buffer = new ByteBuffer(Frame.CmdBufferSize, true);
             Frame.Encode(buffer, FrameType.Sasl, 0, command);