Merge pull request #38 from ievsiukovi/master

Race condition when complete connection async send

Author: sdanzan

kafka-sharp/kafka-sharp/Network/Connection.cs

@@ -368,6 +368,20 @@ private void CleanSend()
             _sendContext.Promise = null;
         }
 
+        private void CompleteSend()
+        {
+            var promise = _sendContext.Promise;
+            CleanSend();
+            promise.SetResult(SuccessResult);
+        }
+
+        private void FaultSend(Exception ex)
+        {
+            var promise = _sendContext.Promise;
+            CleanSend();
+            promise.SetException(ex);
+        }
+
         // Async send loop body
         private static void OnSendCompleted(ISocket sender, ISocketAsyncEventArgs saea)
         {
@@ -381,9 +395,8 @@ private static void OnSendCompleted(ISocket sender, ISocketAsyncEventArgs saea)
 
             if (saea.SocketError != SocketError.Success)
             {
-                connection._sendContext.Promise.SetException(new TransportException(TransportError.WriteError,
-                    new SocketException((int) saea.SocketError)));
-                connection.CleanSend();
+                connection.FaultSend(new TransportException(TransportError.WriteError,
+                    new SocketException((int)saea.SocketError)));
                 return;
             }
 
@@ -402,8 +415,7 @@ private static void OnSendCompleted(ISocket sender, ISocketAsyncEventArgs saea)
                 }
                 else
                 {
-                    connection._sendContext.Promise.SetResult(SuccessResult);
-                    connection.CleanSend();
+                    connection.CompleteSend();
                 }
             }
         }
@@ -430,8 +442,7 @@ private void LoopSend(int from, int count)
             }
             catch (Exception ex)
             {
-                CleanSend();
-                _sendContext.Promise.SetException(new TransportException(TransportError.WriteError, ex));
+                FaultSend(new TransportException(TransportError.WriteError, ex));
             }
         }