#if !NET20 using System; using System.Collections.Generic; using System.IO; using System.Linq; using System.Threading.Tasks; namespace Apewer.WebSocket { /// public class Connection { #region internal internal Connection(SocketWrapper socket, Action initialize, Func parseRequest, Func handlerFactory, Func, string> negotiateSubProtocol) { Socket = socket; OnOpen = () => { }; OnClose = () => { }; OnMessage = x => { }; OnBytes = x => { }; OnPing = x => Pong(x); OnPong = x => { }; OnError = x => { }; _initialize = initialize; _handlerFactory = handlerFactory; _parseRequest = parseRequest; _negotiateSubProtocol = negotiateSubProtocol; } internal SocketWrapper Socket { get; set; } private readonly Action _initialize; private readonly Func _handlerFactory; private readonly Func, string> _negotiateSubProtocol; readonly Func _parseRequest; internal ComposableHandler Handler { get; set; } private bool _closing; private bool _closed; private const int ReadSize = 1024 * 4; internal Action OnOpen { get; set; } internal Action OnClose { get; set; } internal Action OnMessage { get; set; } internal Action OnBytes { get; set; } internal Action OnPing { get; set; } internal Action OnPong { get; set; } internal Action OnError { get; set; } internal ConnectionInfo ConnectionInfo { get; private set; } /// private Task Send(T message, Func createFrame) { if (Handler == null) throw new InvalidOperationException("Cannot send before handshake."); if (!Available) { const string errorMessage = "Data sent while closing or after close. Ignoring."; WebSocketLog.Warn(errorMessage); var taskForException = new TaskCompletionSource(); taskForException.SetException(new ConnectionNotAvailableException(errorMessage)); return taskForException.Task; } var bytes = createFrame(message); return SendBytes(bytes); } /// internal void StartReceiving() { var data = new List(ReadSize); var buffer = new byte[ReadSize]; Read(data, buffer); } internal void Close(int code) { if (!Available) return; _closing = true; if (Handler == null) { CloseSocket(); return; } var bytes = Handler.FrameClose(code); if (bytes.Length == 0) CloseSocket(); else SendBytes(bytes, CloseSocket); } internal void CreateHandler(IEnumerable data) { var request = _parseRequest(data.ToArray()); if (request == null) return; Handler = _handlerFactory(request); if (Handler == null) return; var subProtocol = _negotiateSubProtocol(request.SubProtocols); ConnectionInfo = ConnectionInfo.Create(request, Socket.RemoteIpAddress, Socket.RemotePort, subProtocol); _initialize(this); var handshake = Handler.CreateHandshake(subProtocol); SendBytes(handshake, OnOpen); } private void Read(List data, byte[] buffer) { if (!Available) return; Socket.Receive(buffer, r => { if (r <= 0) { WebSocketLog.Debug("0 bytes read. Closing."); CloseSocket(); return; } WebSocketLog.Debug(r + " bytes read"); var readBytes = buffer.Take(r); if (Handler != null) { Handler.Receive(readBytes); } else { data.AddRange(readBytes); CreateHandler(data); } Read(data, buffer); }, HandleReadError); } private void HandleReadError(Exception e) { if (e is AggregateException) { var agg = e as AggregateException; HandleReadError(agg.InnerException); return; } if (e is ObjectDisposedException) { WebSocketLog.Debug("Swallowing ObjectDisposedException", e); return; } OnError(e); if (e is WebSocketException) { WebSocketLog.Debug("Error while reading", e); Close(((WebSocketException)e).StatusCode); } else if (e is SubProtocolNegotiationFailureException) { WebSocketLog.Debug(e.Message); Close(StatusCodes.ProtocolError); } else if (e is IOException) { WebSocketLog.Debug("Error while reading", e); Close(StatusCodes.AbnormalClosure); } else { WebSocketLog.Error("Application Error", e); Close(StatusCodes.InternalServerError); } } private Task SendBytes(byte[] bytes, Action callback = null) { return Socket.Send(bytes, () => { WebSocketLog.Debug("Sent " + bytes.Length + " bytes"); if (callback != null) callback(); }, e => { if (e is IOException) WebSocketLog.Debug("Failed to send. Disconnecting.", e); else WebSocketLog.Info("Failed to send. Disconnecting.", e); CloseSocket(); }); } private void CloseSocket() { _closing = true; OnClose(); _closed = true; Socket.Close(); Socket.Dispose(); _closing = false; } #endregion #region properties /// public bool Available { get { return !_closing && !_closed && Socket.Connected; } } /// 客户端 IP 地址。 public string Address { get { return (ConnectionInfo == null) ? "" : ConnectionInfo.ClientIpAddress; } } /// 客户端端口。 public int Port { get { return (ConnectionInfo == null) ? 0 : ConnectionInfo.ClientPort; } } /// 获取主机信息。例: /// 从请求 ws://127.0.0.1:8000/page?field=value /// 中获取 127.0.0.1:8000 public string Host { get { return (ConnectionInfo == null) ? "" : ConnectionInfo.Host; } } /// 获取源站信息。例: /// 从请求 ws://127.0.0.1:8000/page?field=value /// 中获取 http://localhost public string Origin { get { return (ConnectionInfo == null) ? "" : ConnectionInfo.Origin; } } /// 获取 URL 查询路径。例: /// 从请求 ws://127.0.0.1:8000/page?field=value /// 中获取 /page?field=value public string QueryPath { get { return (ConnectionInfo == null) ? "" : ConnectionInfo.Path; } } #endregion #region methods /// 发送文本。 /// public Task Send(params char[] message) { if (message == null || message.Length < 1) return null; if (message.Length == 1) { return Send(message[0].ToString(), Handler.FrameText); } else { var sb = new System.Text.StringBuilder(); foreach (var i in message) { if ((object)i != null) sb.Append(i); } var text = sb.ToString(); if (text.Length < 1) return null; return Send(text, Handler.FrameText); } } /// 发送文本。 /// public Task Send(params string[] message) { if (message == null || message.Length < 1) return null; if (message.Length == 1) { if (string.IsNullOrEmpty(message[0])) return null; return Send(message[0], Handler.FrameText); } else { var sb = new System.Text.StringBuilder(); foreach (var i in message) { if (i != null) sb.Append(i); } var text = sb.ToString(); if (text.Length < 1) return null; return Send(text, Handler.FrameText); } } /// 发送字节数组。 /// public Task Send(params byte[] message) { if (message == null || message.Length < 1) return null; return Send(message, Handler.FrameBytes); } /// 发送 PING。 /// public Task Ping(params byte[] message) { return Send(message, Handler.FramePing); } /// 发送 PONG。 /// public Task Pong(params byte[] message) { return Send(message, Handler.FramePong); } /// 关闭 Socket 连接。 public void Close() { Close(StatusCodes.NormalClosure); } #endregion } } #endif