Merge pull request #2677 from MediaBrowser/beta

Beta

3 files changed, 186 insertions, 168 deletions

diff --git a/Emby.Common.Implementations/Net/NetAcceptSocket.cs b/Emby.Common.Implementations/Net/NetAcceptSocket.cs
index 82e7e9b00..5e831ac7a 100644
--- a/Emby.Common.Implementations/Net/NetAcceptSocket.cs
+++ b/Emby.Common.Implementations/Net/NetAcceptSocket.cs
@@ -97,7 +97,6 @@ namespace Emby.Common.Implementations.Net
             _acceptor.StartAccept();
         }
 
-#if NET46
         public Task SendFile(string path, byte[] preBuffer, byte[] postBuffer, CancellationToken cancellationToken)
         {
             var options = TransmitFileOptions.UseDefaultWorkerThread;
@@ -109,6 +108,18 @@ namespace Emby.Common.Implementations.Net
             return completionSource.Task;
         }
 
+        public IAsyncResult BeginSendFile(string path, byte[] preBuffer, byte[] postBuffer, AsyncCallback callback, object state)
+        {
+            var options = TransmitFileOptions.UseDefaultWorkerThread;
+
+            return Socket.BeginSendFile(path, preBuffer, postBuffer, options, new AsyncCallback(FileSendCallback), state);
+        }
+
+        public void EndSendFile(IAsyncResult result)
+        {
+            Socket.EndSendFile(result);
+        }
+
         private void FileSendCallback(IAsyncResult ar)
         {
             // Retrieve the socket from the state object.
@@ -117,25 +128,23 @@ namespace Emby.Common.Implementations.Net
             var client = data.Item1;
             var path = data.Item2;
             var taskCompletion = data.Item3;
-        
+
             // Complete sending the data to the remote device.
-        try {
-            client.EndSendFile(ar);
-        taskCompletion.TrySetResult(true);
-}
-        catch(SocketException ex){
-        _logger.Info("Socket.SendFile failed for {0}. error code {1}", path, ex.SocketErrorCode);
-        taskCompletion.TrySetException(ex);
-}catch(Exception ex){
-        taskCompletion.TrySetException(ex);
-}
-        }
-#else
-        public Task SendFile(string path, byte[] preBuffer, byte[] postBuffer, CancellationToken cancellationToken)
-        {
-            throw new NotImplementedException();
+            try
+            {
+                client.EndSendFile(ar);
+                taskCompletion.TrySetResult(true);
+            }
+            catch (SocketException ex)
+            {
+                _logger.Info("Socket.SendFile failed for {0}. error code {1}", path, ex.SocketErrorCode);
+                taskCompletion.TrySetException(ex);
+            }
+            catch (Exception ex)
+            {
+                taskCompletion.TrySetException(ex);
+            }
         }
-#endif
 
         public void Dispose()
         {
diff --git a/Emby.Common.Implementations/Net/SocketFactory.cs b/Emby.Common.Implementations/Net/SocketFactory.cs
index 39f236afa..0a1232a40 100644
--- a/Emby.Common.Implementations/Net/SocketFactory.cs
+++ b/Emby.Common.Implementations/Net/SocketFactory.cs
@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Net;
 using System.Net.Sockets;
@@ -188,16 +189,7 @@ namespace Emby.Common.Implementations.Net
 
             try
             {
-#if NET46
-				retVal.ExclusiveAddressUse = false;
-#else
-                // The ExclusiveAddressUse acceptSocket option is a Windows-specific option that, when set to "true," tells Windows not to allow another acceptSocket to use the same local address as this acceptSocket
-                // See https://github.com/dotnet/corefx/pull/11509 for more details
-                if (System.Runtime.InteropServices.RuntimeInformation.IsOSPlatform(System.Runtime.InteropServices.OSPlatform.Windows))
-				{
-					retVal.ExclusiveAddressUse = false;
-				}
-#endif
+                retVal.ExclusiveAddressUse = false;
                 //retVal.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Broadcast, true);
                 retVal.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
                 retVal.SetSocketOption(SocketOptionLevel.IP, SocketOptionName.MulticastTimeToLive, multicastTimeToLive);
@@ -217,5 +209,89 @@ namespace Emby.Common.Implementations.Net
                 throw;
             }
         }
+
+        public Stream CreateNetworkStream(ISocket socket, bool ownsSocket)
+        {
+            var netSocket = (UdpSocket)socket;
+
+            return new SocketStream(netSocket.Socket, ownsSocket);
+        }
     }
+
+    public class SocketStream : Stream
+    {
+        private readonly Socket _socket;
+
+        public SocketStream(Socket socket, bool ownsSocket)
+        {
+            _socket = socket;
+        }
+
+        public override void Flush()
+        {
+        }
+
+        public override bool CanRead
+        {
+            get { return true; }
+        }
+        public override bool CanSeek
+        {
+            get { return false; }
+        }
+        public override bool CanWrite
+        {
+            get { return true; }
+        }
+        public override long Length
+        {
+            get { throw new NotImplementedException(); }
+        }
+        public override long Position
+        {
+            get { throw new NotImplementedException(); }
+            set { throw new NotImplementedException(); }
+        }
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            _socket.Send(buffer, offset, count, SocketFlags.None);
+        }
+
+        public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
+        {
+            return _socket.BeginSend(buffer, offset, count, SocketFlags.None, callback, state);
+        }
+
+        public override void EndWrite(IAsyncResult asyncResult)
+        {
+            _socket.EndSend(asyncResult);
+        }
+
+        public override void SetLength(long value)
+        {
+            throw new NotImplementedException();
+        }
+
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            throw new NotImplementedException();
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            return _socket.Receive(buffer, offset, count, SocketFlags.None);
+        }
+
+        public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
+        {
+            return _socket.BeginReceive(buffer, offset, count, SocketFlags.None, callback, state);
+        }
+
+        public override int EndRead(IAsyncResult asyncResult)
+        {
+            return _socket.EndReceive(asyncResult);
+        }
+    }
+
 }
diff --git a/Emby.Common.Implementations/Net/UdpSocket.cs b/Emby.Common.Implementations/Net/UdpSocket.cs
index f9181eb6a..578610b4c 100644
--- a/Emby.Common.Implementations/Net/UdpSocket.cs
+++ b/Emby.Common.Implementations/Net/UdpSocket.cs
@@ -14,11 +14,16 @@ namespace Emby.Common.Implementations.Net
     // THIS IS A LINKED FILE - SHARED AMONGST MULTIPLE PLATFORMS	
     // Be careful to check any changes compile and work for all platform projects it is shared in.
 
-    internal sealed class UdpSocket : DisposableManagedObjectBase, ISocket
+    public sealed class UdpSocket : DisposableManagedObjectBase, ISocket
     {
         private Socket _Socket;
         private int _LocalPort;
 
+        public Socket Socket
+        {
+            get { return _Socket; }
+        }
+
         private readonly SocketAsyncEventArgs _receiveSocketAsyncEventArgs = new SocketAsyncEventArgs()
         {
             SocketFlags = SocketFlags.None
@@ -116,129 +121,104 @@ namespace Emby.Common.Implementations.Net
             private set;
         }
 
-        public Task<SocketReceiveResult> ReceiveAsync(CancellationToken cancellationToken)
+        public IAsyncResult BeginReceive(byte[] buffer, int offset, int count, AsyncCallback callback)
         {
-            ThrowIfDisposed();
-            var tcs = new TaskCompletionSource<SocketReceiveResult>();
             EndPoint receivedFromEndPoint = new IPEndPoint(IPAddress.Any, 0);
 
-            var state = new AsyncReceiveState(_Socket, receivedFromEndPoint);
-            state.TaskCompletionSource = tcs;
-
-            cancellationToken.Register(() => tcs.TrySetCanceled());
-
-            _receiveSocketAsyncEventArgs.RemoteEndPoint = receivedFromEndPoint;
-            _currentReceiveTaskCompletionSource = tcs;
-
-            try
-            {
-                var willRaiseEvent = _Socket.ReceiveFromAsync(_receiveSocketAsyncEventArgs);
-
-                if (!willRaiseEvent)
-                {
-                    _receiveSocketAsyncEventArgs_Completed(this, _receiveSocketAsyncEventArgs);
-                }
-            }
-            catch (Exception ex)
-            {
-                tcs.TrySetException(ex);
-            }
-
-            return tcs.Task;
+            return _Socket.BeginReceiveFrom(buffer, offset, count, SocketFlags.None, ref receivedFromEndPoint, callback, buffer);
         }
 
-        public Task SendAsync(byte[] buffer, int size, IpEndPointInfo endPoint, CancellationToken cancellationToken)
+        public SocketReceiveResult EndReceive(IAsyncResult result)
         {
-            ThrowIfDisposed();
-
-            if (buffer == null) throw new ArgumentNullException("messageData");
-            if (endPoint == null) throw new ArgumentNullException("endPoint");
+            IPEndPoint sender = new IPEndPoint(IPAddress.Any, 0);
+            EndPoint remoteEndPoint = (EndPoint)sender;
 
-            var ipEndPoint = NetworkManager.ToIPEndPoint(endPoint);
+            var receivedBytes = _Socket.EndReceiveFrom(result, ref remoteEndPoint);
 
-#if NETSTANDARD1_6
+            var buffer = (byte[]) result.AsyncState;
 
-            if (size != buffer.Length)
+            return new SocketReceiveResult
             {
-                byte[] copy = new byte[size];
-                Buffer.BlockCopy(buffer, 0, copy, 0, size);
-                buffer = copy;
-            }
-
-            cancellationToken.ThrowIfCancellationRequested();
+                ReceivedBytes = receivedBytes,
+                RemoteEndPoint = ToIpEndPointInfo((IPEndPoint)remoteEndPoint),
+                Buffer = buffer,
+                LocalIPAddress = LocalIPAddress
+            };
+        }
 
-            _Socket.SendTo(buffer, ipEndPoint);
-            return Task.FromResult(true);
-#else
-            var taskSource = new TaskCompletionSource<bool>();
+        public Task<SocketReceiveResult> ReceiveAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            var taskCompletion = new TaskCompletionSource<SocketReceiveResult>();
 
-            try
+            Action<IAsyncResult> callback = callbackResult =>
             {
-                _Socket.BeginSendTo(buffer, 0, size, SocketFlags.None, ipEndPoint, result =>
+                try
                 {
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        taskSource.TrySetCanceled();
-                        return;
-                    }
-                    try
-                    {
-                        _Socket.EndSend(result);
-                        taskSource.TrySetResult(true);
-                    }
-                    catch (Exception ex)
-                    {
-                        taskSource.TrySetException(ex);
-                    }
+                    taskCompletion.TrySetResult(EndReceive(callbackResult));
+                }
+                catch (Exception ex)
+                {
+                    taskCompletion.TrySetException(ex);
+                }
+            };
 
-                }, null);
-            }
-            catch (Exception ex)
+            var result = BeginReceive(buffer, offset, count, new AsyncCallback(callback));
+
+            if (result.CompletedSynchronously)
             {
-                taskSource.TrySetException(ex);
+                callback(result);
             }
 
-            return taskSource.Task;
-#endif
-            //ThrowIfDisposed();
+            cancellationToken.Register(() => taskCompletion.TrySetCanceled());
+
+            return taskCompletion.Task;
+        }
 
-            //if (buffer == null) throw new ArgumentNullException("messageData");
-            //if (endPoint == null) throw new ArgumentNullException("endPoint");
+        public Task<SocketReceiveResult> ReceiveAsync(CancellationToken cancellationToken)
+        {
+            var buffer = new byte[8192];
 
-            //cancellationToken.ThrowIfCancellationRequested();
+            return ReceiveAsync(buffer, 0, buffer.Length, cancellationToken);
+        }
 
-            //var tcs = new TaskCompletionSource<int>();
+        public Task SendToAsync(byte[] buffer, int offset, int size, IpEndPointInfo endPoint, CancellationToken cancellationToken)
+        {
+            var taskCompletion = new TaskCompletionSource<int>();
 
-            //cancellationToken.Register(() => tcs.TrySetCanceled());
+            Action<IAsyncResult> callback = callbackResult =>
+            {
+                try
+                {
+                    taskCompletion.TrySetResult(EndSendTo(callbackResult));
+                }
+                catch (Exception ex)
+                {
+                    taskCompletion.TrySetException(ex);
+                }
+            };
 
-            //_sendSocketAsyncEventArgs.SetBuffer(buffer, 0, size);
-            //_sendSocketAsyncEventArgs.RemoteEndPoint = NetworkManager.ToIPEndPoint(endPoint);
-            //_currentSendTaskCompletionSource = tcs;
+            var result = BeginSendTo(buffer, offset, size, endPoint, new AsyncCallback(callback), null);
 
-            //var willRaiseEvent = _Socket.SendAsync(_sendSocketAsyncEventArgs);
+            if (result.CompletedSynchronously)
+            {
+                callback(result);
+            }
 
-            //if (!willRaiseEvent)
-            //{
-            //    _sendSocketAsyncEventArgs_Completed(this, _sendSocketAsyncEventArgs);
-            //}
+            cancellationToken.Register(() => taskCompletion.TrySetCanceled());
 
-            //return tcs.Task;
+            return taskCompletion.Task;
         }
 
-        public async Task SendWithLockAsync(byte[] buffer, int size, IpEndPointInfo endPoint, CancellationToken cancellationToken)
+        public IAsyncResult BeginSendTo(byte[] buffer, int offset, int size, IpEndPointInfo endPoint, AsyncCallback callback, object state)
         {
-            ThrowIfDisposed();
+            var ipEndPoint = NetworkManager.ToIPEndPoint(endPoint);
 
-            //await _sendLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+            return _Socket.BeginSendTo(buffer, offset, size, SocketFlags.None, ipEndPoint, callback, state);
+        }
 
-            try
-            {
-                await SendAsync(buffer, size, endPoint, cancellationToken).ConfigureAwait(false);
-            }
-            finally
-            {
-                //_sendLock.Release();
-            }
+        public int EndSendTo(IAsyncResult result)
+        {
+            return _Socket.EndSendTo(result);
         }
 
         protected override void Dispose(bool disposing)
@@ -273,52 +253,5 @@ namespace Emby.Common.Implementations.Net
 
             return NetworkManager.ToIpEndPointInfo(endpoint);
         }
-
-        private void ProcessResponse(IAsyncResult asyncResult)
-        {
-#if NET46
-            var state = asyncResult.AsyncState as AsyncReceiveState;
-            try
-            {
-                var bytesRead = state.Socket.EndReceiveFrom(asyncResult, ref state.RemoteEndPoint);
-
-                var ipEndPoint = state.RemoteEndPoint as IPEndPoint;
-                state.TaskCompletionSource.SetResult(
-                    new SocketReceiveResult
-                    {
-                        Buffer = state.Buffer,
-                        ReceivedBytes = bytesRead,
-                        RemoteEndPoint = ToIpEndPointInfo(ipEndPoint),
-                        LocalIPAddress = LocalIPAddress
-                    }
-                );
-            }
-            catch (ObjectDisposedException)
-            {
-                state.TaskCompletionSource.SetCanceled();
-            }
-            catch (Exception ex)
-            {
-                state.TaskCompletionSource.SetException(ex);
-            }
-#endif
-        }
-
-        private class AsyncReceiveState
-        {
-            public AsyncReceiveState(Socket socket, EndPoint remoteEndPoint)
-            {
-                this.Socket = socket;
-                this.RemoteEndPoint = remoteEndPoint;
-            }
-
-            public EndPoint RemoteEndPoint;
-            public byte[] Buffer = new byte[8192];
-
-            public Socket Socket { get; private set; }
-
-            public TaskCompletionSource<SocketReceiveResult> TaskCompletionSource { get; set; }
-
-        }
     }
 }