public static Task<IEnumerable<KeyValuePair<string, string>>> GetHeaders(ISocket socket)
        {
            var headers = new MemoryStream ();
            return Util.ReadExactly (socket, 10).ContinueWith (t =>
            {
                t.PropagateExceptions ();
                byte[] lenBuffer = t.Result;

                int pos = Array.IndexOf (lenBuffer, (byte)':');

                var lenString = Encoding.UTF8.GetString (lenBuffer, 0, pos);

                var len = int.Parse (lenString) + 1;

                pos++;
                var alreadyReadLen = lenBuffer.Length - pos;
                headers.Write (lenBuffer, pos, alreadyReadLen);

                return Util.ReadDataAsync (socket, headers, len - alreadyReadLen);
            }).Unwrap ().ContinueWith (t2 =>
            {
                t2.PropagateExceptions ();
                var rv = new List<KeyValuePair<string, string>> ();

                var headerPairs = Encoding.UTF8.GetString (headers.ToArray ()).Split ((char)0);
                for (var i = 0; i < headerPairs.Length - 1; i += 2)
                    rv.Add (new KeyValuePair<string, string> (headerPairs[i], headerPairs[i + 1]));

                return Util.TaskFromResult ((IEnumerable<KeyValuePair<string, string>>)rv);
            }).Unwrap ();
        }

        public static void CloseSock(ISocket socket, int timeout = 0)
        {
            if (socket.GetSocket().ProtocolType == ProtocolType.Tcp) socket.GetSocket().Shutdown(SocketShutdown.Both);

            if (timeout == 0) socket.Close();
            else socket.Close(timeout);
        }

 public SocketSender(IFireflyService service, ISocket socket)
 {
     _service = service;
     _socket = socket;
     _state = State.Immediate;
     _socketEvent = _service.Memory.AllocSocketEvent();
 }

		protected NodeBase(ICluster owner, IPEndPoint endpoint, IFailurePolicy failurePolicy, ISocket socket)
		{
			this.owner = owner;
			this.endpoint = endpoint;
			this.socket = socket;
			this.failurePolicy = failurePolicy;
			this.name = endpoint.ToString();

			failLock = new Object();
			writeQueue = new ConcurrentQueue<Data>();
			readQueue = new Queue<Data>();

			mustReconnect = true;
			IsAlive = true;

			counterEnqueuePerSec = Metrics.Meter("node write enqueue/sec", endpoint.ToString(), Interval.Seconds);
			counterDequeuePerSec = Metrics.Meter("node write dequeue/sec", endpoint.ToString(), Interval.Seconds);
			counterOpReadPerSec = Metrics.Meter("node op read/sec", endpoint.ToString(), Interval.Seconds);
			counterWriteQueue = Metrics.Counter("write queue length", endpoint.ToString());
			counterReadQueue = Metrics.Counter("read queue length", endpoint.ToString());

			counterWritePerSec = Metrics.Meter("node write/sec", endpoint.ToString(), Interval.Seconds);
			counterErrorPerSec = Metrics.Meter("node in error/sec", endpoint.ToString(), Interval.Seconds);
			counterItemCount = Metrics.Counter("commands", endpoint.ToString());
			gaugeSendSpeed = Metrics.Gauge("send speed", endpoint.ToString());
		}

 public StreamProxy(ISocket inStream, ISocket outStream, int bufferSize = 16384)
 {
     Logger.Debug("Setting up Stream Proxy from " + inStream.GetType() + " to " + outStream.GetType() + "(bufsize=" + bufferSize + ")");
     _inStream = inStream;
     _outStream = outStream;
     _bufferSize = bufferSize;
 }

 public void EngageWith(ISocket socket)
 {
     Interlocked.Exchange(ref _flushingInProgress, 0);
     Interlocked.Exchange(ref _writingOffset, 0);
     Interlocked.Exchange(ref _socketException, null);
     _asyncSocket.EngageWith(socket);
 }

 public Receiver(ISocket socket, Action<string> messageAction, Action closeAction)
 {
     _socket = socket;
     _messageAction = messageAction;
     _closeAction = closeAction;
     _queue = new Queue<byte>();
 }

 public Connection(IServerTrace trace, AppDelegate app, ISocket socket, Action<ISocket> disconnected)
 {
     _trace = trace;
     _app = app;
     _socket = socket;
     _disconnected = disconnected;
 }

 public ClacksClient(ISocket socket) {
     if(socket == null) {
         throw new ArgumentNullException("socket");
     }
     _socket = socket;
     _receiver = new ResponseReceiver(_socket);
 }

        private void OnClientConnect(ISocket clientSocket)
        {
            FleckLog.Debug(String.Format("Client connected from {0}:{1}", clientSocket.RemoteIpAddress, clientSocket.RemotePort.ToString()));
            ListenForClients();

            WebSocketConnection connection = null;

            connection = new WebSocketConnection(
                clientSocket,
                _config,
                bytes => RequestParser.Parse(bytes, _scheme),
                r => HandlerFactory.BuildHandler(r,
                                                 s => connection.OnMessage(s),
                                                 connection.Close,
                                                 b => connection.OnBinary(b),
                                                 b => connection.OnPing(b),
                                                 b => connection.OnPong(b)),
                s => SubProtocolNegotiator.Negotiate(SupportedSubProtocols, s));

            if (IsSecure)
            {
                FleckLog.Debug("Authenticating Secure Connection");
                clientSocket
                    .Authenticate(Certificate,
                                  connection.StartReceiving,
                                  e => FleckLog.Warn("Failed to Authenticate", e));
            }
            else
            {
                connection.StartReceiving();
            }
        }

 /// <summary>
 /// Constructs a new KayakContext with the provided `ISocket`, `IKayakServerRequest`, and 
 /// `IKayakServerResponse` objects.
 /// </summary>
 public KayakContext(ISocket socket, IKayakServerRequest request, IKayakServerResponse response)
 {
     Socket = socket;
     Request = request;
     Response = response;
     Items = new Dictionary<object, object>();
 }

        public static KayakContext CreateContext(ISocket socket)
        {
            var request = new KayakServerRequest(socket);
            var response = new KayakServerResponse(socket);

            return new KayakContext(socket, request, response);
        }

        public IEnumerable<object> BeginRequestInternal(ISocket socket)
        {
            var bufferHeaders = new ContinuationState<LinkedList<ArraySegment<byte>>>(socket.BufferHeaders());
            yield return bufferHeaders;

            var headerBuffers = bufferHeaders.Result;

            Dictionary<string, object> env = new Dictionary<string, object>();

            var bodyDataReadWithHeaders = headerBuffers.Last.Value;
            headerBuffers.RemoveLast();

            var headersString = headerBuffers.GetString();
            var reader = new StringReader(headersString);
            var requestLine = reader.ReadRequestLine();
            var headers = reader.ReadHeaders();

            env["Owin.RequestMethod"] = requestLine.Verb;
            env["Owin.RequestUri"] = requestLine.RequestUri;
            env["Owin.RequestHeaders"] = headers;
            env["Owin.BaseUri"] = "";
            env["Owin.RemoteEndPoint"] = socket.RemoteEndPoint;
            env["Owin.RequestBody"] = CreateReadBody(socket, bodyDataReadWithHeaders);

            // TODO provide better values
            env["Owin.ServerName"] = "";
            env["Owin.ServerPort"] = 0;
            env["Owin.UriScheme"] = "http";

            yield return env;
        }

 public void EngageWith(ISocket socket)
 {
     _socket = socket;
     _notIoArgs.AcceptSocket = _socket.UnderlyingSocket;
     _receiveArgs.AcceptSocket = _socket.UnderlyingSocket;
     _sendArgs.AcceptSocket = _socket.UnderlyingSocket;
 }

        private void OnClientConnect(ISocket clientSocket)
        {
            FleckLog.Debug("Client Connected");
            ListenForClients();

            WebSocketConnection connection = null;

            connection = new WebSocketConnection(
                clientSocket,
                _config,
                bytes => RequestParser.Parse(bytes, _scheme),
                r => HandlerFactory.BuildHandler(r,
                                                 s => connection.OnMessage(s),
                                                 connection.Close,
                                                 b => connection.OnBinary(b)));

            if (IsSecure)
            {
                FleckLog.Debug("Authenticating Secure Connection");
                clientSocket
                    .Authenticate(_x509Certificate,
                                  connection.StartReceiving,
                                  e => FleckLog.Warn("Failed to Authenticate", e));
            }
            else
            {
                connection.StartReceiving();
            }
        }

        public NodeType SendReceiveAndVerify(ISocket socket, TimeSpan timeout)
        {
            var greetingMessageNegotiator = NodeGreetingMessageNegotiatorFactory.CreateFromSocket(socket);

            var tcpSocket = socket as TcpSocket;
            var sendReceiveTimeout = new Tuple<int, int>(0, 0);
            if (tcpSocket != null)
            {
                sendReceiveTimeout = new Tuple<int, int>(tcpSocket.TcpClient.SendTimeout,
                    tcpSocket.TcpClient.ReceiveTimeout);

                tcpSocket.TcpClient.SendTimeout = timeout.ToMillisOrZero();
                tcpSocket.TcpClient.ReceiveTimeout = timeout.ToMillisOrZero();
            }

            try
            {
                greetingMessageNegotiator.WriteGreeting(_greetingMessage);

                var readGreeting = greetingMessageNegotiator.VerifyRemoteGreeting(_expectedRemoteNodeTypes);
                return readGreeting.NodeType;
            }
            catch (IOException)
            {
                throw new TimeoutException("Timeout occurred negotiating after connection had been established");
            }
            finally
            {
                if (tcpSocket != null)
                {
                    tcpSocket.TcpClient.SendTimeout = sendReceiveTimeout.Item1;
                    tcpSocket.TcpClient.ReceiveTimeout = sendReceiveTimeout.Item2;
                }
            }
        }

        public static bool SC_LOGIN_RESULT_Handler(ISocket sock, byte[] buffer)
        {
            SC_LOGIN_RESULT_Packet packet = PacketHelper.ParsePacketStruct<SC_LOGIN_RESULT_Packet>(buffer);
            Int16 readLen = PacketHelper.ParseBodyLen(buffer);
            if (readLen < PacketHelper.Size(packet))
            {
                Console.WriteLine("read length is too small({0})", readLen);
                return true;    // socket close
            }

            if (packet.Result == 1)
            {
                Console.WriteLine("Login success");
                Program.loginResult = 1;
                Program.m_alldone.Set();
            }
            else if (packet.Result == 0)
            {
                Console.WriteLine("Login fail");
                Program.loginResult = 0;
                Program.m_alldone.Set();
            }
            else
            {
                Console.WriteLine("unknown result({0})", packet.Result);
                return true;
            }

            return false;
        }

        public Func<byte[], int, int, Action<Action<int>, Action<Exception>>> CreateReadBody(
            ISocket socket,
            ArraySegment<byte> bodyDataReadWithHeaders)
        {
            return (buffer, offset, count) =>
                {
                    if (bodyDataReadWithHeaders.Count > 0)
                    {
                        int bytesRead;

                        bytesRead = Math.Min(bodyDataReadWithHeaders.Count, count);
                        Buffer.BlockCopy(bodyDataReadWithHeaders.Array, bodyDataReadWithHeaders.Offset, buffer, offset, bytesRead);

                        if (bytesRead < bodyDataReadWithHeaders.Count)
                            bodyDataReadWithHeaders =
                                new ArraySegment<byte>(
                                    bodyDataReadWithHeaders.Array,
                                    bodyDataReadWithHeaders.Offset + bytesRead,
                                    bodyDataReadWithHeaders.Count - bytesRead);
                        else
                            bodyDataReadWithHeaders = default(ArraySegment<byte>);

                        return (r, e) => r(bytesRead);
                    }
                    else if (socket != null)
                    {
                        return socket.Read(buffer, offset, count);
                    }
                    else
                    {
                        return (r, e) => r(0);
                    }
                };
        }

 public RpcClient(ISocket socket)
 {
     if(socket == null) {
         throw new ArgumentNullException("socket");
     }
     _socket = socket;
 }

 public Connection(IFireflyService services, AppDelegate app, ISocket socket, Action<ISocket> disconnected)
 {
     _services = services;
     _app = app;
     _socket = socket;
     _socketSender = new SocketSender(_services, _socket);
     _disconnected = disconnected;
 }