Add initial Websocket support

All autobahntestsuite tests pass including the permessage-deflate
compression tests.

Some of the tests pass in a non-strict fashion. They are testing
for protocol errors and expect events to happen in a particular
order, which is not respected by Gun. Gun fails earlier than is
expected due to concurrent processing of frames.

The implementation when error occurs during handshake is probably
a bit rough at this point. The documentation is also incomplete
and/or wrong at this time, though this is the general state of
the Gun documentation and will be resolved in a separate commit.

3 files changed, 307 insertions, 62 deletions

diff --git a/src/gun.erl b/src/gun.erl
index 71af26e..3b98732 100644
--- a/src/gun.erl
+++ b/src/gun.erl
@@ -58,6 +58,7 @@
 %% Websocket.
 -export([ws_upgrade/2]).
 -export([ws_upgrade/3]).
+-export([ws_upgrade/4]).
 -export([ws_send/2]).
 
 %% Debug.
@@ -85,6 +86,8 @@
 	| {type, conn_type()}].
 -export_type([opts/0]).
 
+-type ws_opts() :: [{compress, boolean()}].
+
 -record(state, {
 	parent :: pid(),
 	owner :: pid(),
@@ -98,7 +101,7 @@
 	socket :: inet:socket() | ssl:sslsocket(),
 	transport :: module(),
 	protocol :: module(),
-	proto_opts :: gun_http:opts(), %% @todo Make a tuple with SPDY and WS too.
+	proto_opts :: gun_http:opts(), %% @todo Make a tuple with SPDY too.
 	protocol_state :: any()
 }).
 
@@ -338,14 +341,19 @@ cancel(ServerPid, StreamRef) ->
 
 %% Websocket.
 
--spec ws_upgrade(pid(), iodata()) -> ok.
+-spec ws_upgrade(pid(), iodata()) -> reference().
 ws_upgrade(ServerPid, Path) ->
-	ws_upgrade(ServerPid, Path, []).
+	ws_upgrade(ServerPid, Path, [], []).
 
--spec ws_upgrade(pid(), iodata(), headers()) -> ok.
+-spec ws_upgrade(pid(), iodata(), headers()) -> reference().
 ws_upgrade(ServerPid, Path, Headers) ->
-	_ = ServerPid ! {ws_upgrade, self(), Path, Headers},
-	ok.
+	ws_upgrade(ServerPid, Path, Headers, []).
+
+-spec ws_upgrade(pid(), iodata(), headers(), ws_opts()) -> reference().
+ws_upgrade(ServerPid, Path, Headers, Opts) ->
+	StreamRef = make_ref(),
+	_ = ServerPid ! {ws_upgrade, self(), StreamRef, Path, Headers, Opts},
+	StreamRef.
 
 -spec ws_send(pid(), ws_frame() | [ws_frame()]) -> ok.
 ws_send(ServerPid, Frames) ->
@@ -428,8 +436,10 @@ connect(State=#state{owner=Owner, host=Host, port=Port, type=Type,
 			retry(State, Retries - 1)
 	end.
 
-retry(State=#state{keepalive_ref=KeepaliveRef}, Retries) when
-		is_reference(KeepaliveRef) ->
+%% Exit normally if the retry functionality has been disabled.
+retry(_, 0) ->
+	ok;
+retry(State=#state{keepalive_ref=KeepaliveRef}, Retries) when is_reference(KeepaliveRef) ->
 	_ = erlang:cancel_timer(KeepaliveRef),
 	%% Flush if we have a keepalive message
 	receive
@@ -458,7 +468,7 @@ before_loop(State=#state{keepalive=Keepalive}) ->
 	KeepaliveRef = erlang:send_after(Keepalive, self(), keepalive),
 	loop(State#state{keepalive_ref=KeepaliveRef}).
 
-loop(State=#state{parent=Parent, owner=Owner, host=Host,
+loop(State=#state{parent=Parent, owner=Owner, host=Host, port=Port,
 		retry=Retry, socket=Socket, transport=Transport,
 		protocol=Protocol, protocol_state=ProtoState}) ->
 	{OK, Closed, Error} = Transport:messages(),
@@ -470,7 +480,9 @@ loop(State=#state{parent=Parent, owner=Owner, host=Host,
 					Transport:close(Socket),
 					retry(State#state{socket=undefined, transport=undefined,
 						protocol=undefined}, Retry);
-				ProtoState2 ->
+				{upgrade, Protocol2, ProtoState2} ->
+					ws_loop(State#state{protocol=Protocol2, protocol_state=ProtoState2});
+			ProtoState2 ->
 					loop(State#state{protocol_state=ProtoState2})
 			end;
 		{Closed, Socket} ->
@@ -494,11 +506,11 @@ loop(State=#state{parent=Parent, owner=Owner, host=Host,
 			before_loop(State#state{protocol_state=ProtoState2});
 		{request, Owner, StreamRef, Method, Path, Headers} ->
 			ProtoState2 = Protocol:request(ProtoState,
-				StreamRef, Method, Host, Path, Headers),
+				StreamRef, Method, Host, Port, Path, Headers),
 			loop(State#state{protocol_state=ProtoState2});
 		{request, Owner, StreamRef, Method, Path, Headers, Body} ->
 			ProtoState2 = Protocol:request(ProtoState,
-				StreamRef, Method, Host, Path, Headers, Body),
+				StreamRef, Method, Host, Port, Path, Headers, Body),
 			loop(State#state{protocol_state=ProtoState2});
 		{data, Owner, StreamRef, IsFin, Data} ->
 			ProtoState2 = Protocol:data(ProtoState,
@@ -507,11 +519,10 @@ loop(State=#state{parent=Parent, owner=Owner, host=Host,
 		{cancel, Owner, StreamRef} ->
 			ProtoState2 = Protocol:cancel(ProtoState, StreamRef),
 			loop(State#state{protocol_state=ProtoState2});
-		{ws_upgrade, Owner, Path, Headers} when Protocol =/= gun_spdy ->
-			%% @todo
-			ProtoState2 = Protocol:ws_upgrade(ProtoState,
-				Path, Headers),
-			ws_loop(State#state{protocol=gun_ws, protocol_state=ProtoState2});
+		{ws_upgrade, Owner, StreamRef, Path, Headers, Opts} when Protocol =/= gun_spdy ->
+			ProtoState2 = Protocol:ws_upgrade(ProtoState, StreamRef, Host, Port, Path, Headers, Opts),
+			loop(State#state{protocol_state=ProtoState2});
+			%% @todo can fail if http/1.0
 		{shutdown, Owner} ->
 			%% @todo Protocol:shutdown?
 			ok;
@@ -525,9 +536,9 @@ loop(State=#state{parent=Parent, owner=Owner, host=Host,
 			element(2, Any) ! {gun_error, self(), {notowner,
 				"Operations are restricted to the owner of the connection."}},
 			loop(State);
-		{ws_upgrade, _, _, _} ->
-			Owner ! {gun_error, self(), {badstate,
-				"Websocket over SPDY isn't supported."}},
+		{ws_upgrade, _, StreamRef, _, _} ->
+			Owner ! {gun_error, self(), StreamRef, {badstate,
+				"Websocket is only supported over HTTP/1.1."}},
 			loop(State);
 		{ws_send, _, _} ->
 			Owner ! {gun_error, self(), {badstate,
@@ -545,23 +556,34 @@ ws_loop(State=#state{parent=Parent, owner=Owner, retry=Retry, socket=Socket,
 	ok = Transport:setopts(Socket, [{active, once}]),
 	receive
 		{OK, Socket, Data} ->
-			ProtoState2 = Protocol:handle(ProtoState, Data),
-			ws_loop(State#state{protocol_state=ProtoState2});
+			case Protocol:handle(Data, ProtoState) of
+				close ->
+					Transport:close(Socket),
+					retry(State#state{socket=undefined, transport=undefined, protocol=undefined}, Retry);
+				ProtoState2 ->
+					ws_loop(State#state{protocol_state=ProtoState2})
+			end;
 		{Closed, Socket} ->
 			Transport:close(Socket),
-			retry(State#state{socket=undefined, transport=undefined,
-				protocol=undefined}, Retry);
+			retry(State#state{socket=undefined, transport=undefined, protocol=undefined}, Retry);
 		{Error, Socket, _} ->
 			Transport:close(Socket),
-			retry(State#state{socket=undefined, transport=undefined,
-				protocol=undefined}, Retry);
-		%% @todo keepalive
-		{ws_send, Owner, Frames} when is_list(Frames) ->
-			todo; %% @todo
+			retry(State#state{socket=undefined, transport=undefined, protocol=undefined}, Retry);
+		%% Ignore any previous HTTP keep-alive.
+		keepalive ->
+			ws_loop(State);
+%		{ws_send, Owner, Frames} when is_list(Frames) ->
+%			todo; %% @todo
 		{ws_send, Owner, Frame} ->
-			{todo, Frame}; %% @todo
+			case Protocol:send(Frame, ProtoState) of
+				close ->
+					Transport:close(Socket),
+					retry(State#state{socket=undefined, transport=undefined, protocol=undefined}, Retry);
+				ProtoState2 ->
+					ws_loop(State#state{protocol_state=ProtoState2})
+			end;
 		{shutdown, Owner} ->
-			%% @todo Protocol:shutdown?
+			%% @todo Protocol:shutdown? %% @todo close frame
 			ok;
 		{system, From, Request} ->
 			sys:handle_system_msg(Request, From, Parent, ?MODULE, [],
diff --git a/src/gun_http.erl b/src/gun_http.erl
index bd6565c..f2a2d23 100644
--- a/src/gun_http.erl
+++ b/src/gun_http.erl
@@ -18,16 +18,19 @@
 -export([handle/2]).
 -export([close/1]).
 -export([keepalive/1]).
--export([request/6]).
 -export([request/7]).
+-export([request/8]).
 -export([data/4]).
 -export([cancel/2]).
+-export([ws_upgrade/7]).
 
 -type opts() :: [{version, cow_http:version()}].
 -export_type([opts/0]).
 
 -type io() :: head | {body, non_neg_integer()} | body_close | body_chunked.
 
+-type websocket_info() :: {websocket, reference(), binary(), [], []}. %% key, extensions, protocols
+
 -record(http_state, {
 	owner :: pid(),
 	socket :: inet:socket() | ssl:sslsocket(),
@@ -35,7 +38,7 @@
 	version = 'HTTP/1.1' :: cow_http:version(),
 	connection = keepalive :: keepalive | close,
 	buffer = <<>> :: binary(),
-	streams = [] :: [{reference(), boolean()}], %% ref + whether stream is alive
+	streams = [] :: [{reference() | websocket_info(), boolean()}], %% ref + whether stream is alive
 	in = head :: io(),
 	in_state :: {non_neg_integer(), non_neg_integer()},
 	out = head :: io()
@@ -128,31 +131,40 @@ handle_head(Data, State=#http_state{owner=Owner, version=ClientVersion,
 		connection=Conn, streams=[{StreamRef, IsAlive}|_]}) ->
 	{Version, Status, _, Rest} = cow_http:parse_status_line(Data),
 	{Headers, Rest2} = cow_http:parse_headers(Rest),
-	In = io_from_headers(Version, Headers),
-	IsFin = case In of head -> fin; _ -> nofin end,
-	case IsAlive of
-		false ->
-			ok;
-		true ->
-			Owner ! {gun_response, self(), StreamRef,
-				IsFin, Status, Headers},
-			ok
-	end,
-	Conn2 = if
-		Conn =:= close -> close;
-		Version =:= 'HTTP/1.0' -> close;
-		ClientVersion =:= 'HTTP/1.0' -> close;
-		true -> conn_from_headers(Headers)
-	end,
-	%% We always reset in_state even if not chunked.
-	if
-		IsFin =:= fin, Conn2 =:= close ->
-			close;
-		IsFin =:= fin ->
-			handle(Rest2, end_stream(State#http_state{in=In,
-				in_state={0, 0}, connection=Conn2}));
-		true ->
-			handle(Rest2, State#http_state{in=In, in_state={0, 0}, connection=Conn2})
+	case {Status, StreamRef} of
+		{101, {websocket, _, WsKey, WsExtensions, WsProtocols, WsOpts}} ->
+			ws_handshake(Rest2, State, Headers, WsKey, WsExtensions, WsProtocols, WsOpts);
+		_ ->
+			In = io_from_headers(Version, Headers),
+			IsFin = case In of head -> fin; _ -> nofin end,
+			case IsAlive of
+				false ->
+					ok;
+				true ->
+					StreamRef2 = case StreamRef of
+						{websocket, SR, _, _, _} -> SR;
+						_ -> StreamRef
+					end,
+					Owner ! {gun_response, self(), StreamRef2,
+						IsFin, Status, Headers},
+					ok
+			end,
+			Conn2 = if
+				Conn =:= close -> close;
+				Version =:= 'HTTP/1.0' -> close;
+				ClientVersion =:= 'HTTP/1.0' -> close;
+				true -> conn_from_headers(Headers)
+			end,
+			%% We always reset in_state even if not chunked.
+			if
+				IsFin =:= fin, Conn2 =:= close ->
+					close;
+				IsFin =:= fin ->
+					handle(Rest2, end_stream(State#http_state{in=In,
+						in_state={0, 0}, connection=Conn2}));
+				true ->
+					handle(Rest2, State#http_state{in=In, in_state={0, 0}, connection=Conn2})
+			end
 	end.
 
 send_data_if_alive(<<>>, _, nofin) ->
@@ -187,27 +199,28 @@ keepalive(State) ->
 	State.
 
 request(State=#http_state{socket=Socket, transport=Transport, version=Version,
-		out=head}, StreamRef, Method, Host, Path, Headers) ->
+		out=head}, StreamRef, Method, Host, Port, Path, Headers) ->
 	Headers2 = case Version of
 		'HTTP/1.0' -> lists:keydelete(<<"transfer-encoding">>, 1, Headers);
 		'HTTP/1.1' -> Headers
 	end,
 	Headers3 = case lists:keymember(<<"host">>, 1, Headers) of
-		false -> [{<<"host">>, Host}|Headers2];
+		false -> [{<<"host">>, [Host, $:, integer_to_binary(Port)]}|Headers2];
 		true -> Headers2
 	end,
 	%% We use Headers2 because this is the smallest list.
 	Conn = conn_from_headers(Headers2),
+	%% @todo This should probably also check for content-type like SPDY.
 	Out = io_from_headers(Version, Headers2),
 	Transport:send(Socket, cow_http:request(Method, Path, Version, Headers3)),
 	new_stream(State#http_state{connection=Conn, out=Out}, StreamRef).
 
 request(State=#http_state{socket=Socket, transport=Transport, version=Version,
-		out=head}, StreamRef, Method, Host, Path, Headers, Body) ->
+		out=head}, StreamRef, Method, Host, Port, Path, Headers, Body) ->
 	Headers2 = lists:keydelete(<<"content-length">>, 1,
 		lists:keydelete(<<"transfer-encoding">>, 1, Headers)),
 	Headers3 = case lists:keymember(<<"host">>, 1, Headers) of
-		false -> [{<<"host">>, Host}|Headers2];
+		false -> [{<<"host">>, [Host, $:, integer_to_binary(Port)]}|Headers2];
 		true -> Headers2
 	end,
 	%% We use Headers2 because this is the smallest list.
@@ -334,3 +347,88 @@ cancel_stream(State=#http_state{streams=Streams}, StreamRef) ->
 
 end_stream(State=#http_state{streams=[_|Tail]}) ->
 	State#http_state{in=head, streams=Tail}.
+
+%% Websocket upgrade.
+
+%% Ensure version is 1.1.
+ws_upgrade(#http_state{version='HTTP/1.0'}, _, _, _, _, _, _) ->
+	error; %% @todo
+ws_upgrade(State=#http_state{socket=Socket, transport=Transport, out=head},
+		StreamRef, Host, Port, Path, Headers, WsOpts) ->
+	%% @todo Add option for setting protocol.
+	{ExtHeaders, GunExtensions} = case maps:get(compress, WsOpts, false) of
+		true -> {[{<<"sec-websocket-extensions">>, <<"permessage-deflate; client_max_window_bits; server_max_window_bits=15">>}],
+			[<<"permessage-deflate">>]};
+		false -> {[], []}
+	end,
+	Key = cow_ws:key(),
+	Headers2 = [
+		{<<"connection">>, <<"upgrade">>},
+		{<<"upgrade">>, <<"websocket">>},
+		{<<"sec-websocket-version">>, <<"13">>},
+		{<<"sec-websocket-key">>, Key}
+		|ExtHeaders
+	],
+	Headers3 = case lists:keymember(<<"host">>, 1, Headers) of
+		false -> [{<<"host">>, [Host, $:, integer_to_binary(Port)]}|Headers2];
+		true -> Headers2
+	end,
+	Transport:send(Socket, cow_http:request(<<"GET">>, Path, 'HTTP/1.1', Headers3)),
+	new_stream(State#http_state{connection=keepalive, out=head},
+		{websocket, StreamRef, Key, GunExtensions, [], WsOpts}).
+
+ws_handshake(Buffer, State, Headers, Key, GunExtensions, GunProtocols, Opts) ->
+	%% @todo check upgrade, connection
+	case lists:keyfind(<<"sec-websocket-accept">>, 1, Headers) of
+		false ->
+			close;
+		{_, Accept} ->
+			case cow_ws:encode_key(Key) of
+				Accept -> ws_handshake_extensions(Buffer, State, Headers, GunExtensions, GunProtocols, Opts);
+				_ -> close
+			end
+	end.
+
+ws_handshake_extensions(Buffer, State, Headers, GunExtensions, GunProtocols, Opts) ->
+	case lists:keyfind(<<"sec-websocket-extensions">>, 1, Headers) of
+		false ->
+			ws_handshake_protocols(Buffer, State, Headers, #{}, GunProtocols);
+		{_, ExtHd} ->
+			case ws_validate_extensions(cow_http_hd:parse_sec_websocket_extensions(ExtHd), GunExtensions, #{}, Opts) of
+				close -> close;
+				Extensions -> ws_handshake_protocols(Buffer, State, Headers, Extensions, GunProtocols)
+			end
+	end.
+
+ws_validate_extensions([], _, Acc, _) ->
+	Acc;
+ws_validate_extensions([{Name = <<"permessage-deflate">>, Params}|Tail], GunExts, Acc, Opts) ->
+	case lists:member(Name, GunExts) of
+		true ->
+			case cow_ws:validate_permessage_deflate(Params, Acc, Opts) of
+				{ok, Acc2} -> ws_validate_extensions(Tail, GunExts, Acc2, Opts);
+				error -> close
+			end;
+		%% Fail the connection if extension was not requested.
+		false ->
+			close
+	end;
+%% Fail the connection on unknown extension.
+ws_validate_extensions(_, _, _, _) ->
+	close.
+
+%% @todo Validate protocols.
+ws_handshake_protocols(Buffer, State, _Headers, Extensions, _GunProtocols = []) ->
+	Protocols = [],
+	ws_handshake_end(Buffer, State, Extensions, Protocols).
+
+ws_handshake_end(Buffer, #http_state{owner=Owner, socket=Socket, transport=Transport}, Extensions, Protocols) ->
+	%% Send ourselves the remaining buffer, if any.
+	_ = case Buffer of
+		<<>> ->
+			ok;
+		_ ->
+			{OK, _, _} = Transport:messages(),
+			self() ! {OK, Socket, Buffer}
+	end,
+	gun_ws:init(Owner, Socket, Transport, Extensions, Protocols).
diff --git a/src/gun_ws.erl b/src/gun_ws.erl
new file mode 100644
index 0000000..5379362
--- /dev/null
+++ b/src/gun_ws.erl
@@ -0,0 +1,125 @@
+%% Copyright (c) 2015, Loïc Hoguin <[email protected]>
+%%
+%% Permission to use, copy, modify, and/or distribute this software for any
+%% purpose with or without fee is hereby granted, provided that the above
+%% copyright notice and this permission notice appear in all copies.
+%%
+%% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+%% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+%% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+%% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+%% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+%% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+%% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+-module(gun_ws).
+
+-export([init/5]).
+-export([handle/2]).
+-export([send/2]).
+
+-record(payload, {
+	type = undefined :: cow_ws:frame_type(),
+	rsv = undefined :: cow_ws:rsv(),
+	len = undefined :: non_neg_integer(),
+	mask_key = undefined :: cow_ws:mask_key(),
+	close_code = undefined :: undefined | cow_ws:close_code(),
+	unmasked = <<>> :: binary(),
+	unmasked_len = 0 :: non_neg_integer()
+}).
+
+-record(ws_state, {
+	owner :: pid(),
+	socket :: inet:socket() | ssl:sslsocket(),
+	transport :: module(),
+	buffer = <<>> :: binary(),
+	in = head :: head | #payload{} | close,
+	frag_state = undefined :: cow_ws:frag_state(),
+	frag_buffer = <<>> :: binary(),
+	utf8_state = 0 :: cow_ws:utf8_state(),
+	extensions = #{} :: cow_ws:extensions()
+}).
+
+%% @todo Protocols
+init(Owner, Socket, Transport, Extensions, _Protocols) ->
+	Owner ! {gun_ws_upgrade, self(), ok},
+	{upgrade, ?MODULE, #ws_state{owner=Owner, socket=Socket, transport=Transport, extensions=Extensions}}.
+
+%% Do not handle anything if we received a close frame.
+handle(_, State=#ws_state{in=close}) ->
+	State;
+%% Shortcut for common case when Data is empty after processing a frame.
+handle(<<>>, State=#ws_state{in=head}) ->
+	State;
+handle(Data, State=#ws_state{buffer=Buffer, in=head, frag_state=FragState, extensions=Extensions}) ->
+	Data2 = << Buffer/binary, Data/binary >>,
+	case cow_ws:parse_header(Data2, Extensions, FragState) of
+		{Type, FragState2, Rsv, Len, MaskKey, Rest} ->
+			handle(Rest, State#ws_state{buffer= <<>>,
+				in=#payload{type=Type, rsv=Rsv, len=Len, mask_key=MaskKey}, frag_state=FragState2});
+		more ->
+			State#ws_state{buffer=Data2};
+		error ->
+			close({error, badframe}, State)
+	end;
+handle(Data, State=#ws_state{in=In=#payload{type=Type, rsv=Rsv, len=Len, mask_key=MaskKey, close_code=CloseCode,
+		unmasked=Unmasked, unmasked_len=UnmaskedLen}, frag_state=FragState, utf8_state=Utf8State, extensions=Extensions}) ->
+	case cow_ws:parse_payload(Data, MaskKey, Utf8State, UnmaskedLen, Type, Len, FragState, Extensions, Rsv) of
+		{ok, CloseCode2, Payload, Utf8State2, Rest} ->
+			dispatch(Rest, State#ws_state{in=head, utf8_state=Utf8State2}, Type, << Unmasked/binary, Payload/binary >>, CloseCode2);
+		{ok, Payload, Utf8State2, Rest} ->
+			dispatch(Rest, State#ws_state{in=head, utf8_state=Utf8State2}, Type, << Unmasked/binary, Payload/binary >>, CloseCode);
+		{more, CloseCode2, Payload, Utf8State2} ->
+			State#ws_state{in=In#payload{close_code=CloseCode2, unmasked= << Unmasked/binary, Payload/binary >>,
+				len=Len - byte_size(Data), unmasked_len=2 + byte_size(Data)}, utf8_state=Utf8State2};
+		{more, Payload, Utf8State2} ->
+			State#ws_state{in=In#payload{unmasked= << Unmasked/binary, Payload/binary >>,
+				len=Len - byte_size(Data), unmasked_len=UnmaskedLen + byte_size(Data)}, utf8_state=Utf8State2};
+		Error = {error, _Reason} ->
+			close(Error, State)
+	end.
+
+dispatch(Rest, State=#ws_state{owner=Owner, frag_state=FragState, frag_buffer=SoFar},
+		Type0, Payload0, CloseCode0) ->
+	case cow_ws:make_frame(Type0, Payload0, CloseCode0, FragState) of
+		{fragment, nofin, _, Payload} ->
+			handle(Rest, State#ws_state{frag_buffer= << SoFar/binary, Payload/binary >>});
+		{fragment, fin, Type, Payload} ->
+			Owner ! {gun_ws, self(), {Type, << SoFar/binary, Payload/binary >>}},
+			handle(Rest, State#ws_state{frag_state=undefined, frag_buffer= <<>>});
+		ping ->
+			State2 = send(pong, State),
+			handle(Rest, State2);
+		{ping, Payload} ->
+			State2 = send({pong, Payload}, State),
+			handle(Rest, State2);
+		pong ->
+			handle(Rest, State);
+		{pong, _} ->
+			handle(Rest, State);
+		Frame ->
+			Owner ! {gun_ws, self(), Frame},
+			case Frame of
+				close -> handle(Rest, State#ws_state{in=close});
+				{close, _, _} -> handle(Rest, State#ws_state{in=close});
+				_ -> handle(Rest, State)
+			end
+	end.
+
+close(Reason, State) ->
+	case Reason of
+		Normal when Normal =:= stop; Normal =:= timeout ->
+			send({close, 1000, <<>>}, State);
+		{error, badframe} ->
+			send({close, 1002, <<>>}, State);
+		{error, badencoding} ->
+			send({close, 1007, <<>>}, State)
+	end.
+
+send(Frame, State=#ws_state{socket=Socket, transport=Transport, extensions=Extensions}) ->
+	Transport:send(Socket, cow_ws:masked_frame(Frame, Extensions)),
+	case Frame of
+		close -> close;
+		{close, _, _} -> close;
+		_ -> State
+	end.