diff options
Diffstat (limited to 'src')
-rw-r--r-- | src/gun.app.src | 2 | ||||
-rw-r--r-- | src/gun.erl | 91 | ||||
-rw-r--r-- | src/gun_app.erl | 2 | ||||
-rw-r--r-- | src/gun_http.erl | 207 | ||||
-rw-r--r-- | src/gun_spdy.erl | 38 | ||||
-rw-r--r-- | src/gun_sup.erl | 2 | ||||
-rw-r--r-- | src/gun_ws.erl | 125 |
7 files changed, 372 insertions, 95 deletions
diff --git a/src/gun.app.src b/src/gun.app.src index cb9f2c0..30fd964 100644 --- a/src/gun.app.src +++ b/src/gun.app.src @@ -1,4 +1,4 @@ -%% Copyright (c) 2013-2014, Loïc Hoguin <[email protected]> +%% Copyright (c) 2013-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 diff --git a/src/gun.erl b/src/gun.erl index 71af26e..b7260bc 100644 --- a/src/gun.erl +++ b/src/gun.erl @@ -1,4 +1,4 @@ -%% Copyright (c) 2013-2014, Loïc Hoguin <[email protected]> +%% Copyright (c) 2013-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 @@ -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() }). @@ -244,9 +247,8 @@ await(ServerPid, StreamRef, Timeout, MRef) -> {response, IsFin, Status, Headers}; {gun_data, ServerPid, StreamRef, IsFin, Data} -> {data, IsFin, Data}; - {gun_push, ServerPid, StreamRef, AssocToStreamRef, - Method, Host, Path, Headers} -> - {push, AssocToStreamRef, Method, Host, Path, Headers}; + {gun_push, ServerPid, StreamRef, NewStreamRef, Method, Host, Path, Headers} -> + {push, NewStreamRef, Method, Host, Path, Headers}; {gun_error, ServerPid, StreamRef, Reason} -> {error, Reason}; {gun_error, ServerPid, Reason} -> @@ -338,14 +340,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 +435,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 +467,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 +479,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 +505,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 +518,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 +535,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 +555,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_app.erl b/src/gun_app.erl index 95def87..61a3ecf 100644 --- a/src/gun_app.erl +++ b/src/gun_app.erl @@ -1,4 +1,4 @@ -%% Copyright (c) 2013-2014, Loïc Hoguin <[email protected]> +%% Copyright (c) 2013-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 diff --git a/src/gun_http.erl b/src/gun_http.erl index 86fc436..745c2a9 100644 --- a/src/gun_http.erl +++ b/src/gun_http.erl @@ -1,4 +1,4 @@ -%% Copyright (c) 2014, Loïc Hoguin <[email protected]> +%% Copyright (c) 2014-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 @@ -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() @@ -47,6 +50,9 @@ init(Owner, Socket, Transport, [{version, Version}]) -> #http_state{owner=Owner, socket=Socket, transport=Transport, version=Version}. +%% Stop looping when we got no more data. +handle(<<>>, State) -> + State; %% Close when server responds and we don't have any open streams. handle(_, #http_state{streams=[]}) -> close; @@ -70,11 +76,11 @@ handle(Data, State=#http_state{in=body_chunked, in_state=InState, {more, Data2, InState2} -> send_data_if_alive(Data2, State, nofin), State#http_state{buffer= <<>>, in_state=InState2}; - {more, Data2, _Length, InState2} -> + {more, Data2, Length, InState2} when is_integer(Length) -> %% @todo See if we can recv faster than one message at a time. send_data_if_alive(Data2, State, nofin), State#http_state{buffer= <<>>, in_state=InState2}; - {more, Data2, _Length, Rest, InState2} -> + {more, Data2, Rest, InState2} -> %% @todo See if we can recv faster than one message at a time. send_data_if_alive(Data2, State, nofin), State#http_state{buffer=Rest, in_state=InState2}; @@ -125,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 = response_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(Version, 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) -> @@ -177,39 +192,38 @@ close_streams(Owner, [{StreamRef, _}|Tail]) -> close_streams(Owner, Tail). %% We can only keep-alive by sending an empty line in-between streams. -keepalive(State=#http_state{socket=Socket, transport=Transport}) -> +keepalive(State=#http_state{socket=Socket, transport=Transport, out=head}) -> Transport:send(Socket, <<"\r\n">>), + State; +keepalive(State) -> State. request(State=#http_state{socket=Socket, transport=Transport, version=Version, - out=head}, StreamRef, Method, Host, Path, Headers) -> - Headers2 = case Version of - 'HTTP/1.0' -> lists:keydelete(<<"transfer-encoding">>, 1, Headers); - 'HTTP/1.1' -> Headers - end, + out=head}, StreamRef, Method, Host, Port, Path, Headers) -> + Headers2 = 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. - Conn = conn_from_headers(Headers2), - Out = io_from_headers(Version, Headers2), + Conn = conn_from_headers(Version, Headers2), + Out = request_io_from_headers(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. - Conn = conn_from_headers(Headers2), + Conn = conn_from_headers(Version, Headers2), Transport:send(Socket, [ cow_http:request(Method, Path, Version, [ - {<<"content-length">>, integer_to_list(iolist_size(Body))} + {<<"content-length">>, integer_to_binary(iolist_size(Body))} |Headers3]), Body]), new_stream(State#http_state{connection=Conn}, StreamRef). @@ -278,8 +292,10 @@ error_stream_not_found(State=#http_state{owner=Owner}) -> %% Headers information retrieval. -conn_from_headers(Headers) -> +conn_from_headers(Version, Headers) -> case lists:keyfind(<<"connection">>, 1, Headers) of + false when Version =:= 'HTTP/1.0' -> + close; false -> keepalive; {_, ConnHd} -> @@ -290,7 +306,20 @@ conn_from_headers(Headers) -> end end. -io_from_headers(Version, Headers) -> +request_io_from_headers(Headers) -> + case lists:keyfind(<<"content-length">>, 1, Headers) of + {_, <<"0">>} -> + head; + {_, Length} -> + {body, cow_http_hd:parse_content_length(Length)}; + _ -> + case lists:keymember(<<"content-type">>, 1, Headers) of + true -> body_chunked; + false -> head + end + end. + +response_io_from_headers(Version, Headers) -> case lists:keyfind(<<"content-length">>, 1, Headers) of {_, <<"0">>} -> head; @@ -329,3 +358,91 @@ 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 + ], + IsSecure = Transport:secure(), + Headers3 = case lists:keymember(<<"host">>, 1, Headers) of + true -> Headers2; + false when Port =:= 80, not IsSecure -> [{<<"host">>, Host}|Headers2]; + false when Port =:= 443, IsSecure -> [{<<"host">>, Host}|Headers2]; + false -> [{<<"host">>, [Host, $:, integer_to_binary(Port)]}|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_spdy.erl b/src/gun_spdy.erl index 5ada67a..7651584 100644 --- a/src/gun_spdy.erl +++ b/src/gun_spdy.erl @@ -1,4 +1,4 @@ -%% Copyright (c) 2013-2014, Loïc Hoguin <[email protected]> +%% Copyright (c) 2013-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 @@ -18,8 +18,8 @@ -export([handle/2]). -export([close/1]). -export([keepalive/1]). --export([request/6]). -export([request/7]). +-export([request/8]). -export([data/4]). -export([cancel/2]). @@ -89,8 +89,8 @@ handle_frame(Rest, State=#spdy_state{owner=Owner, case get_stream_by_id(AssocToStreamID, State) of #stream{ref=AssocToStreamRef} -> StreamRef = make_ref(), - Owner ! {gun_push, self(), StreamRef, - AssocToStreamRef, Method, Host, Path, Headers}, + Owner ! {gun_push, self(), AssocToStreamRef, + StreamRef, Method, Host, Path, Headers}, handle_loop(Rest, new_stream(StreamID, StreamRef, not IsFin, false, Version, State)); false -> @@ -189,30 +189,44 @@ keepalive(State=#spdy_state{socket=Socket, transport=Transport, Transport:send(Socket, cow_spdy:ping(PingID)), State#spdy_state{ping_id=PingID + 2}. -%% @todo Allow overriding the host when doing requests. +%% @todo Always https scheme? request(State=#spdy_state{socket=Socket, transport=Transport, zdef=Zdef, - stream_id=StreamID}, StreamRef, Method, Host, Path, Headers) -> - Out = false =/= lists:keyfind(<<"content-type">>, 1, Headers), + stream_id=StreamID}, StreamRef, Method, Host, Port, Path, Headers) -> + {Host2, Headers2} = prepare_request(Headers, Host, Port), + Out = (false =/= lists:keyfind(<<"content-type">>, 1, Headers2)) + orelse (false =/= lists:keyfind(<<"content-length">>, 1, Headers2)), Transport:send(Socket, cow_spdy:syn_stream(Zdef, StreamID, 0, not Out, false, 0, - Method, <<"https">>, Host, Path, <<"HTTP/1.1">>, Headers)), + Method, <<"https">>, Host2, Path, <<"HTTP/1.1">>, Headers2)), new_stream(StreamID, StreamRef, true, Out, <<"HTTP/1.1">>, State#spdy_state{stream_id=StreamID + 2}). %% @todo Handle Body > 16MB. (split it out into many frames) +%% @todo Always https scheme? request(State=#spdy_state{socket=Socket, transport=Transport, zdef=Zdef, - stream_id=StreamID}, StreamRef, Method, Host, Path, Headers, Body) -> - Headers2 = lists:keystore(<<"content-length">>, 1, Headers, - {<<"content-length">>, integer_to_list(iolist_size(Body))}), + stream_id=StreamID}, StreamRef, Method, Host, Port, Path, Headers, Body) -> + {Host2, Headers2} = prepare_request(Headers, Host, Port), + Headers3 = lists:keystore(<<"content-length">>, 1, Headers2, + {<<"content-length">>, integer_to_binary(iolist_size(Body))}), Transport:send(Socket, [ cow_spdy:syn_stream(Zdef, StreamID, 0, false, false, 0, - Method, <<"https">>, Host, Path, <<"HTTP/1.1">>, Headers2), + Method, <<"https">>, Host2, Path, <<"HTTP/1.1">>, Headers3), cow_spdy:data(StreamID, true, Body) ]), new_stream(StreamID, StreamRef, true, false, <<"HTTP/1.1">>, State#spdy_state{stream_id=StreamID + 2}). +prepare_request(Headers, Host, Port) -> + Headers2 = lists:keydelete(<<"keep-alive">>, 1, + lists:keydelete(<<"proxy-connection">>, 1, + lists:keydelete(<<"transfer-encoding">>, 1, + lists:keydelete(<<"connection">>, 1, Headers)))), + case lists:keytake(<<"host">>, 1, Headers2) of + false -> {[Host, $:, integer_to_binary(Port)], Headers2}; + {value, {_, Host1}, Headers3} -> {Host1, Headers3} + end. + data(State=#spdy_state{socket=Socket, transport=Transport}, StreamRef, IsFin, Data) -> case get_stream_by_ref(StreamRef, State) of diff --git a/src/gun_sup.erl b/src/gun_sup.erl index 8c0455f..c1b34fa 100644 --- a/src/gun_sup.erl +++ b/src/gun_sup.erl @@ -1,4 +1,4 @@ -%% Copyright (c) 2013-2014, Loïc Hoguin <[email protected]> +%% Copyright (c) 2013-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 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. |