%-*-Mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*- % ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et nomod: %%% %%%------------------------------------------------------------------------ %%% @doc %%% ==Cowboy CloudI HTTP Handler== %%% @end %%% %%% MIT License %%% %%% Copyright (c) 2012-2018 Michael Truog %%% %%% Permission is hereby granted, free of charge, to any person obtaining a %%% copy of this software and associated documentation files (the "Software"), %%% to deal in the Software without restriction, including without limitation %%% the rights to use, copy, modify, merge, publish, distribute, sublicense, %%% and/or sell copies of the Software, and to permit persons to whom the %%% Software is furnished to do so, subject to the following conditions: %%% %%% The above copyright notice and this permission notice shall be included in %%% all copies or substantial portions of the Software. %%% %%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR %%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, %%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE %%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER %%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING %%% FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER %%% DEALINGS IN THE SOFTWARE. %%% %%% @author Michael Truog %%% @copyright 2012-2018 Michael Truog %%% @version 1.7.4 {@date} {@time} %%%------------------------------------------------------------------------ -module(cloudi_http_cowboy_handler). -author('mjtruog at protonmail dot com'). %-behaviour(cowboy_http_handler). %-behaviour(cowboy_websocket_handler). %% external interface %% cowboy_http_handler callbacks -export([init/3, handle/2, info/3, terminate/3]). %% cowboy_websocket_handler callbacks -export([websocket_init/3, websocket_handle/3, websocket_info/3, websocket_terminate/3]). -include_lib("cloudi_core/include/cloudi_logger.hrl"). -include_lib("cloudi_core/include/cloudi_service_children.hrl"). -include("cloudi_http_cowboy_handler.hrl"). -record(websocket_state, { % for service requests entering CloudI websocket_connect_trans_id = undefined :: undefined | cloudi_service:trans_id(), name_incoming :: string(), name_outgoing :: cloudi_service:service_name(), request_info :: list({binary(), binary()}) | binary(), % for a service request exiting CloudI response_pending = false :: boolean(), response_timer = undefined :: undefined | reference(), request_pending = undefined :: undefined | cloudi:message_service_request(), response_lookup :: undefined | #{any() := {cloudi:message_service_request(), reference()}}, recv_timeouts :: undefined | #{cloudi:trans_id() := reference()}, queued :: undefined | pqueue4:pqueue4( cloudi:message_service_request()) }). %%%------------------------------------------------------------------------ %%% External interface functions %%%------------------------------------------------------------------------ %%%------------------------------------------------------------------------ %%% Callback functions from cowboy_http_handler %%%------------------------------------------------------------------------ init(_Transport, Req0, #cowboy_state{use_websockets = UseWebSockets} = State) when UseWebSockets =:= true; UseWebSockets =:= exclusively -> case upgrade_request(Req0) of {websocket, Req1} -> {upgrade, protocol, cowboy_websocket, Req1, State#cowboy_state{use_websockets = true}}; {undefined, Req1} -> if UseWebSockets =:= exclusively -> {shutdown, Req1, State}; true -> {ok, Req1, State} end; {Upgrade, Req1} -> ?LOG_ERROR("Unknown protocol: ~w", [Upgrade]), {shutdown, Req1, State} end; init(_Transport, Req, #cowboy_state{use_websockets = false} = State) -> {ok, Req, State}. handle(Req0, #cowboy_state{output_type = OutputType, content_type_forced = ContentTypeForced, content_types_accepted = ContentTypesAccepted, set_x_forwarded_for = SetXForwardedFor, status_code_timeout = StatusCodeTimeout, query_get_format = QueryGetFormat, use_host_prefix = UseHostPrefix, use_client_ip_prefix = UseClientIpPrefix, use_x_method_override = UseXMethodOverride, use_method_suffix = UseMethodSuffix } = State) -> RequestStartMicroSec = ?LOG_WARN_APPLY(fun request_time_start/0, []), {MethodHTTP, Req1} = cowboy_req:method(Req0), {HeadersIncoming0, Req2} = cowboy_req:headers(Req1), Method = if UseXMethodOverride =:= true -> case lists:keyfind(<<"x-http-method-override">>, 1, HeadersIncoming0) of {_, MethodOverride} -> MethodOverride; false -> MethodHTTP end; UseXMethodOverride =:= false -> MethodHTTP end, {QS, Req4} = if MethodHTTP =:= <<"GET">> -> if QueryGetFormat =:= text_pairs -> {QSVals, Req3} = cowboy_req:qs_vals(Req2), if (OutputType =:= external) orelse (OutputType =:= binary) orelse (OutputType =:= list) -> % cloudi_request_info text_pairs format {get_query_string_external(QSVals), Req3}; OutputType =:= internal -> % cloudi_key_value format {QSVals, Req3} end; QueryGetFormat =:= raw -> cowboy_req:qs(Req2) end; true -> % query strings only handled for GET methods {undefined, Req2} end, {PathRaw, Req5} = cowboy_req:path(Req4), {{ClientIpAddr, ClientPort} = Client, Req6} = cowboy_req:peer(Req5), {NameIncoming, ReqN} = service_name_incoming(UseClientIpPrefix, UseHostPrefix, PathRaw, Client, Req6), RequestAccepted = if ContentTypesAccepted =:= undefined -> true; true -> header_accept_check(HeadersIncoming0, ContentTypesAccepted) end, if RequestAccepted =:= false -> HttpCode = 406, {ok, Req} = cowboy_req:reply(HttpCode, ReqN), ?LOG_WARN_APPLY(fun request_time_end_error/6, [HttpCode, MethodHTTP, NameIncoming, undefined, RequestStartMicroSec, not_acceptable]), {ok, Req, State}; RequestAccepted =:= true -> NameOutgoing = if UseMethodSuffix =:= false -> NameIncoming; Method =:= <<"GET">> -> NameIncoming ++ "/get"; Method =:= <<"POST">> -> NameIncoming ++ "/post"; Method =:= <<"PUT">> -> NameIncoming ++ "/put"; Method =:= <<"DELETE">> -> NameIncoming ++ "/delete"; Method =:= <<"HEAD">> -> NameIncoming ++ "/head"; Method =:= <<"OPTIONS">> -> NameIncoming ++ "/options"; Method =:= <<"PATCH">> -> NameIncoming ++ "/connect"; Method =:= <<"TRACE">> -> NameIncoming ++ "/trace"; Method =:= <<"CONNECT">> -> NameIncoming ++ "/connect"; true -> % handle custom methods, if they occur NameIncoming ++ [$/ | cloudi_string:lowercase(erlang:binary_to_list(Method))] end, PeerShort = erlang:list_to_binary(inet_parse:ntoa(ClientIpAddr)), PeerLong = cloudi_ip_address:to_binary(ClientIpAddr), PeerPort = erlang:integer_to_binary(ClientPort), HeadersIncoming1 = [{<<"peer">>, PeerShort}, {<<"peer-port">>, PeerPort}, {<<"source-address">>, PeerLong}, {<<"source-port">>, PeerPort}, {<<"url-path">>, PathRaw} | HeadersIncoming0], HeadersIncomingN = if SetXForwardedFor =:= true -> case lists:keyfind(<<"x-forwarded-for">>, 1, HeadersIncoming0) of false -> [{<<"x-forwarded-for">>, PeerShort} | HeadersIncoming1]; _ -> HeadersIncoming1 end; SetXForwardedFor =:= false -> HeadersIncoming1 end, Body = if MethodHTTP =:= <<"GET">> -> % only the query string is provided as the % body of a GET request passed within the Request parameter % of a CloudI service request, which prevents misuse of GET QS; (MethodHTTP =:= <<"HEAD">>) orelse (MethodHTTP =:= <<"OPTIONS">>) orelse (MethodHTTP =:= <<"TRACE">>) orelse (MethodHTTP =:= <<"CONNECT">>) -> <<>>; true -> % POST, PUT, DELETE or anything else case header_content_type(HeadersIncoming0) of <<"application/zip">> -> 'application_zip'; <<"multipart/", _/binary>> -> 'multipart'; _ -> 'normal' end end, case handle_request(NameOutgoing, HeadersIncomingN, Body, ReqN, State) of {{cowboy_response, HeadersOutgoing, Response}, ReqN0, NewState} -> {HttpCode, Req} = handle_response(NameIncoming, HeadersOutgoing, Response, ReqN0, OutputType, ContentTypeForced), ?LOG_TRACE_APPLY(fun request_time_end_success/5, [HttpCode, MethodHTTP, NameIncoming, NameOutgoing, RequestStartMicroSec]), {ok, Req, NewState}; {{cowboy_error, timeout}, ReqN0, NewState} -> HttpCode = StatusCodeTimeout, {ok, Req} = if HttpCode =:= 405 -> % currently not providing a list of valid methods % (a different HTTP status code is a better % choice, since this service name may not exist) HeadersOutgoing = [{<<"allow">>, <<"">>}], cowboy_req:reply(HttpCode, HeadersOutgoing, ReqN0); true -> cowboy_req:reply(HttpCode, ReqN0) end, ?LOG_WARN_APPLY(fun request_time_end_error/6, [HttpCode, MethodHTTP, NameIncoming, NameOutgoing, RequestStartMicroSec, timeout]), {ok, Req, NewState}; {{cowboy_error, Reason}, ReqN0, NewState} -> HttpCode = 500, {ok, Req} = cowboy_req:reply(HttpCode, ReqN0), ?LOG_WARN_APPLY(fun request_time_end_error/6, [HttpCode, MethodHTTP, NameIncoming, NameOutgoing, RequestStartMicroSec, Reason]), {ok, Req, NewState} end end. info(Message, Req, State) -> ?LOG_WARN("ignored ~p", [Message]), {ok, Req, State}. terminate(_Reason, _Req, _State) -> ok. websocket_init(_Transport, Req0, #cowboy_state{scope = Scope, prefix = Prefix, timeout_websocket = TimeoutWebSocket, output_type = OutputType, set_x_forwarded_for = SetXForwardedFor, websocket_connect = WebSocketConnect, websocket_ping = WebSocketPing, websocket_protocol = WebSocketProtocol, websocket_name_unique = WebSocketNameUnique, websocket_subscriptions = WebSocketSubscriptions, use_websockets = true, use_host_prefix = UseHostPrefix, use_client_ip_prefix = UseClientIpPrefix, use_method_suffix = UseMethodSuffix } = State) -> {Method, Req1} = cowboy_req:method(Req0), {HeadersIncoming0, Req2} = cowboy_req:headers(Req1), {PathRaw, Req3} = cowboy_req:path(Req2), {{ClientIpAddr, ClientPort} = Client, Req4} = cowboy_req:peer(Req3), {NameIncoming, ReqN} = service_name_incoming(UseClientIpPrefix, UseHostPrefix, PathRaw, Client, Req4), NameOutgoing = if UseMethodSuffix =:= false -> NameIncoming; Method =:= <<"GET">> -> NameIncoming ++ "/get" end, % can not turn-off the /websocket suffix, since it would otherwise % cause a conflict with service requests coming from HTTP into CloudI % when UseMethodSuffix == false PathRawStr = erlang:binary_to_list(PathRaw), NameWebSocket = PathRawStr ++ "/websocket", % service requests are only received if they relate to % the service's prefix SubscribeWebSocket = lists:prefix(Prefix, NameWebSocket), HeadersIncoming1 = if SubscribeWebSocket =:= true -> [{<<"service-name">>, erlang:list_to_binary(NameWebSocket)} | HeadersIncoming0]; SubscribeWebSocket =:= false -> HeadersIncoming0 end, PeerShort = erlang:list_to_binary(inet_parse:ntoa(ClientIpAddr)), PeerLong = cloudi_ip_address:to_binary(ClientIpAddr), PeerPort = erlang:integer_to_binary(ClientPort), HeadersIncoming2 = [{<<"peer">>, PeerShort}, {<<"peer-port">>, PeerPort}, {<<"source-address">>, PeerLong}, {<<"source-port">>, PeerPort}, {<<"url-path">>, PathRaw} | HeadersIncoming1], HeadersIncomingN = if SetXForwardedFor =:= true -> case lists:keyfind(<<"x-forwarded-for">>, 1, HeadersIncoming0) of false -> [{<<"x-forwarded-for">>, PeerShort} | HeadersIncoming2]; _ -> HeadersIncoming2 end; SetXForwardedFor =:= false -> HeadersIncoming2 end, RequestInfo = if (OutputType =:= external) orelse (OutputType =:= binary) -> headers_external_incoming(HeadersIncomingN); (OutputType =:= internal) orelse (OutputType =:= list) -> HeadersIncomingN end, WebSocketPingStatus = if WebSocketPing =:= undefined -> undefined; is_integer(WebSocketPing) -> erlang:send_after(WebSocketPing, self(), {websocket_ping, WebSocketPing}), received end, ResponseLookup = if WebSocketProtocol /= undefined -> #{}; true -> undefined end, RecvTimeouts = if WebSocketProtocol =:= undefined -> #{}; true -> undefined end, Queued = if WebSocketProtocol =:= undefined -> pqueue4:new(); true -> undefined end, if SubscribeWebSocket =:= true -> % initiate an asynchronous close if the websocket must be unique OldConnectionMonitors = if WebSocketNameUnique =:= true -> case cpg:get_members(Scope, NameWebSocket, infinity) of {ok, _, OldConnections} -> lists:map(fun(OldConnection) -> cloudi_service_http_cowboy:close(OldConnection), erlang:monitor(process, OldConnection) end, OldConnections); {error, _} -> [] end; WebSocketNameUnique =:= false -> [] end, % service requests are only received if they relate to % the service's prefix ok = cpg:join(Scope, NameWebSocket, self(), infinity), % block on the websocket close if the connection must be unique if WebSocketNameUnique =:= true -> lists:foreach(fun(OldConnectionMonitor) -> receive {'DOWN', OldConnectionMonitor, process, _, _} -> ok end end, OldConnectionMonitors); WebSocketNameUnique =:= false -> ok end, if WebSocketSubscriptions =:= undefined -> ok; true -> % match websocket_subscriptions to determine if a % more subscriptions should occur, possibly % using parameters in a pattern template % for the subscription case trie:find_match(PathRawStr, WebSocketSubscriptions) of error -> ok; {ok, Pattern, Functions} -> Parameters = cloudi_service_name: parse(PathRawStr, Pattern), websocket_subscriptions(Functions, Parameters, Scope) end end; SubscribeWebSocket =:= false -> ok end, {ok, ReqN, websocket_connect_check(WebSocketConnect, State#cowboy_state{ websocket_ping = WebSocketPingStatus, websocket_subscriptions = undefined, websocket_state = #websocket_state{ name_incoming = NameIncoming, name_outgoing = NameOutgoing, request_info = RequestInfo, response_lookup = ResponseLookup, recv_timeouts = RecvTimeouts, queued = Queued}}), TimeoutWebSocket}. websocket_handle({ping, _Payload}, Req, State) -> % cowboy automatically responds with pong {ok, Req, State}; websocket_handle({pong, _Payload}, Req, State) -> {ok, Req, State#cowboy_state{websocket_ping = received}}; websocket_handle({WebSocketResponseType, ResponseBinary}, Req, #cowboy_state{output_type = OutputType, websocket_protocol = undefined, use_websockets = true, websocket_state = #websocket_state{ request_info = ResponseInfo, response_pending = true, response_timer = ResponseTimer, request_pending = T } = WebSocketState } = State) when WebSocketResponseType =:= text; WebSocketResponseType =:= binary -> Response = if (OutputType =:= external) orelse (OutputType =:= internal) orelse (OutputType =:= binary) -> ResponseBinary; (OutputType =:= list) -> erlang:binary_to_list(ResponseBinary) end, websocket_handle_outgoing_response(T, ResponseTimer, ResponseInfo, Response), websocket_process_queue(Req, State#cowboy_state{websocket_state = WebSocketState#websocket_state{ response_pending = false, response_timer = undefined, request_pending = undefined} }); websocket_handle({WebSocketRequestType, RequestBinary}, Req, #cowboy_state{dispatcher = Dispatcher, timeout_sync = TimeoutSync, output_type = OutputType, websocket_protocol = undefined, use_websockets = true, websocket_state = #websocket_state{ name_incoming = NameIncoming, name_outgoing = NameOutgoing, request_info = RequestInfo, response_pending = false} } = State) when WebSocketRequestType =:= text; WebSocketRequestType =:= binary -> Request = if (OutputType =:= external) orelse (OutputType =:= internal) orelse (OutputType =:= binary) -> RequestBinary; (OutputType =:= list) -> erlang:binary_to_list(RequestBinary) end, ResponseBinaryF = fun(Data) -> true = (((OutputType =:= external) orelse (OutputType =:= internal)) andalso (is_binary(Data) orelse is_list(Data))) orelse ((OutputType =:= binary) andalso is_binary(Data)) orelse ((OutputType =:= list) andalso is_list(Data)), Data end, websocket_handle_incoming_request(Dispatcher, NameOutgoing, RequestInfo, Request, TimeoutSync, ResponseBinaryF, WebSocketRequestType, Req, NameIncoming, State); websocket_handle({WebSocketRequestType, RequestBinary}, Req, #cowboy_state{dispatcher = Dispatcher, timeout_sync = TimeoutSync, websocket_protocol = WebSocketProtocol, use_websockets = true, websocket_state = #websocket_state{ name_incoming = NameIncoming, name_outgoing = NameOutgoing, request_info = Info, response_pending = false, response_lookup = ResponseLookup } = WebSocketState } = State) when WebSocketRequestType =:= text; WebSocketRequestType =:= binary -> {LookupID, LookupData, Value} = case WebSocketProtocol(incoming, RequestBinary) of {incoming, Request} -> {undefined, undefined, Request}; {ID, Response} -> case maps:find(ID, ResponseLookup) of {ok, ResponseData} -> {ID, ResponseData, Response}; error -> {undefined, timeout, undefined} end end, case LookupData of undefined -> % an incoming service request ResponseF = fun(ProtocolData) -> {_, Data} = WebSocketProtocol(outgoing, ProtocolData), Data end, websocket_handle_incoming_request(Dispatcher, NameOutgoing, Info, Value, TimeoutSync, ResponseF, WebSocketRequestType, Req, NameIncoming, State); timeout -> % a response arrived but has already timed-out {ok, Req, State}; {T, ResponseTimer} -> % a response to an outgoing service request that has finished websocket_handle_outgoing_response(T, ResponseTimer, Info, Value), {ok, Req, State#cowboy_state{websocket_state = WebSocketState#websocket_state{ response_lookup = maps:remove(LookupID, ResponseLookup)} }} end. websocket_info({response_timeout, ID}, Req, #cowboy_state{use_websockets = true, websocket_state = #websocket_state{ response_pending = false, response_lookup = ResponseLookup } = WebSocketState } = State) -> {ok, Req, State#cowboy_state{websocket_state = WebSocketState#websocket_state{ response_lookup = maps:remove(ID, ResponseLookup)} }}; websocket_info(response_timeout, Req, #cowboy_state{websocket_protocol = undefined, use_websockets = true, websocket_state = #websocket_state{ response_pending = true} = WebSocketState } = State) -> websocket_process_queue(Req, State#cowboy_state{websocket_state = WebSocketState#websocket_state{ response_pending = false, response_timer = undefined, request_pending = undefined} }); websocket_info({Type, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Source}, Req, #cowboy_state{output_type = OutputType, websocket_output_type = WebSocketOutputType, websocket_protocol = undefined, use_websockets = true, websocket_state = #websocket_state{ response_pending = false} = WebSocketState } = State) when ((((OutputType =:= external) orelse (OutputType =:= internal)) andalso (is_binary(Request) orelse is_list(Request))) orelse ((OutputType =:= binary) andalso is_binary(Request)) orelse ((OutputType =:= list) andalso is_list(Request))), (Type =:= 'cloudi_service_send_async' orelse Type =:= 'cloudi_service_send_sync') -> ResponseTimer = erlang:send_after(Timeout, self(), response_timeout), T = {Type, Name, Pattern, undefined, undefined, Timeout, Priority, TransId, Source}, NewState = State#cowboy_state{ websocket_state = WebSocketState#websocket_state{ response_pending = true, response_timer = ResponseTimer, request_pending = T}}, case websocket_terminate_check(RequestInfo) of true when Request == <<>> -> {reply, close, Req, NewState}; true -> {reply, [{WebSocketOutputType, Request}, close], Req, NewState}; false -> {reply, {WebSocketOutputType, Request}, Req, NewState} end; websocket_info({Type, Name, Pattern, RequestInfo, RequestProtocol, Timeout, Priority, TransId, Source}, Req, #cowboy_state{websocket_output_type = WebSocketOutputType, websocket_protocol = WebSocketProtocol, use_websockets = true, websocket_state = #websocket_state{ response_pending = false, response_lookup = ResponseLookup } = WebSocketState } = State) when (Type =:= 'cloudi_service_send_async' orelse Type =:= 'cloudi_service_send_sync') -> {ID, Request} = WebSocketProtocol(outgoing, RequestProtocol), T = {Type, Name, Pattern, undefined, undefined, Timeout, Priority, TransId, Source}, ResponseTimer = erlang:send_after(Timeout, self(), {response_timeout, ID}), NewResponseLookup = maps:put(ID, {T, ResponseTimer}, ResponseLookup), NewState = State#cowboy_state{ websocket_state = WebSocketState#websocket_state{ response_lookup = NewResponseLookup}}, case websocket_terminate_check(RequestInfo) of true when Request == <<>> -> {reply, close, Req, NewState}; true -> {reply, [{WebSocketOutputType, Request}, close], Req, NewState}; false -> {reply, {WebSocketOutputType, Request}, Req, NewState} end; websocket_info({Type, _, _, _, Request, Timeout, Priority, TransId, _} = T, Req, #cowboy_state{output_type = OutputType, websocket_protocol = undefined, use_websockets = true, websocket_state = #websocket_state{ response_pending = true, recv_timeouts = RecvTimeouts, queued = Queue } = WebSocketState } = State) when ((((OutputType =:= external) orelse (OutputType =:= internal)) andalso (is_binary(Request) orelse is_list(Request))) orelse ((OutputType =:= binary) andalso is_binary(Request)) orelse ((OutputType =:= list) andalso is_list(Request))), (Type =:= 'cloudi_service_send_async' orelse Type =:= 'cloudi_service_send_sync'), (Timeout > 0) -> {ok, Req, State#cowboy_state{ websocket_state = WebSocketState#websocket_state{ recv_timeouts = maps:put(TransId, erlang:send_after(Timeout, self(), {'cloudi_service_recv_timeout', Priority, TransId}), RecvTimeouts), queued = pqueue4:in(T, Priority, Queue)} }}; websocket_info({Type, Name, _, _, _, Timeout, _, TransId, _}, Req, #cowboy_state{output_type = OutputType, websocket_protocol = undefined, use_websockets = true} = State) when Type =:= 'cloudi_service_send_async'; Type =:= 'cloudi_service_send_sync' -> if Timeout > 0 -> ?LOG_ERROR("output ~p config ignoring service request to ~s (~s)", [OutputType, Name, uuid:uuid_to_string(TransId)]); true -> ok end, {ok, Req, State}; websocket_info({'cloudi_service_recv_timeout', Priority, TransId}, Req, #cowboy_state{websocket_protocol = undefined, use_websockets = true, websocket_state = #websocket_state{ recv_timeouts = RecvTimeouts, queued = Queue } = WebSocketState } = State) -> F = fun({_, {_, _, _, _, _, _, _, Id, _}}) -> Id == TransId end, {_, NewQueue} = pqueue4:remove_unique(F, Priority, Queue), {ok, Req, State#cowboy_state{ websocket_state = WebSocketState#websocket_state{ recv_timeouts = maps:remove(TransId, RecvTimeouts), queued = NewQueue} }}; websocket_info({'cloudi_service_return_async', _, _, <<>>, <<>>, _, TransId, _}, Req, #cowboy_state{use_websockets = true, websocket_state = #websocket_state{ websocket_connect_trans_id = TransId } = WebSocketState } = State) -> {ok, Req, State#cowboy_state{ websocket_state = WebSocketState#websocket_state{ websocket_connect_trans_id = undefined} }}; websocket_info({'cloudi_service_return_async', _, _, ResponseInfo, Response, _, TransId, _}, Req, #cowboy_state{output_type = OutputType, websocket_output_type = WebSocketOutputType, websocket_protocol = WebSocketProtocol, use_websockets = true, websocket_state = #websocket_state{ websocket_connect_trans_id = TransId } = WebSocketState } = State) -> WebSocketResponse = if Response == <<>> -> % websocket_connect is special because a % <<>> response will not be sent back to the websocket % since this is the response to an event rather than a % request/response pair undefined; WebSocketProtocol =:= undefined -> true = ((((OutputType =:= external) orelse (OutputType =:= internal)) andalso (is_binary(Response) orelse is_list(Response))) orelse ((OutputType =:= binary) andalso is_binary(Response)) orelse ((OutputType =:= list) andalso is_list(Response))), {WebSocketOutputType, Response}; is_function(WebSocketProtocol) -> {_, ResponseBinary} = WebSocketProtocol(outgoing, Response), {WebSocketOutputType, ResponseBinary} end, NewState = State#cowboy_state{ websocket_state = WebSocketState#websocket_state{ websocket_connect_trans_id = undefined}}, case websocket_terminate_check(ResponseInfo) of true -> if WebSocketResponse =:= undefined -> {reply, close, Req, NewState}; true -> {reply, [WebSocketResponse, close], Req, NewState} end; false -> if WebSocketResponse =:= undefined -> {ok, Req, NewState}; true -> {reply, WebSocketResponse, Req, NewState} end end; websocket_info({websocket_ping, WebSocketPing}, Req, #cowboy_state{websocket_ping = WebSocketPingStatus} = State) -> if WebSocketPingStatus =:= undefined -> {shutdown, Req, State}; WebSocketPingStatus =:= received -> erlang:send_after(WebSocketPing, self(), {websocket_ping, WebSocketPing}), {reply, {ping, <<>>}, Req, State#cowboy_state{websocket_ping = undefined}} end; websocket_info({cowboy_error, shutdown}, Req, State) -> % from cloudi_service_http_cowboy:close/1 {shutdown, Req, State}; websocket_info(Info, Req, #cowboy_state{use_websockets = true} = State) -> ?LOG_ERROR("Invalid websocket request state: \"~p\"", [Info]), {ok, Req, State}. websocket_terminate(Reason, _Req, #cowboy_state{ websocket_disconnect = WebSocketDisconnect} = State) -> websocket_disconnect_check(WebSocketDisconnect, Reason, State), ok. %%%------------------------------------------------------------------------ %%% Private functions %%%------------------------------------------------------------------------ upgrade_request(Req0) -> case cowboy_req:parse_header(<<"connection">>, Req0) of {undefined, _, Req1} -> {undefined, Req1}; {ok, C, Req1} -> case lists:member(<<"upgrade">>, C) of true -> {ok, [U0 | _], Req2} = cowboy_req:parse_header(<<"upgrade">>, Req1), try erlang:binary_to_existing_atom(U0, utf8) of U1 -> {U1, Req2} catch error:badarg -> % non-atom is ignored and logged {U0, Req2} end; false -> {undefined, Req1} end end. header_accept_check(Headers, ContentTypesAccepted) -> case lists:keyfind(<<"accept">>, 1, Headers) of false -> true; {<<"accept">>, Value} -> case binary:match(Value, ContentTypesAccepted) of nomatch -> false; _ -> true end end. header_content_type(Headers) -> case lists:keyfind(<<"content-type">>, 1, Headers) of false -> <<>>; {<<"content-type">>, Value} -> hd(binary:split(Value, <<";">>)) end. % format for external services, http headers passed as key-value pairs headers_external_incoming(L) -> erlang:iolist_to_binary(headers_external_incoming_text(L)). headers_external_incoming_text([] = L) -> L; headers_external_incoming_text([{K, V} | L]) when is_binary(K) -> [[K, 0, V, 0] | headers_external_incoming_text(L)]. headers_external_outgoing(<<>>) -> []; headers_external_outgoing([] = ResponseInfo) -> ResponseInfo; headers_external_outgoing([{_, _} | _] = ResponseInfo) -> % assumes key/value within tuple are iodata() % (cowboy can error if this is not true) ResponseInfo; headers_external_outgoing(ResponseInfo) when is_binary(ResponseInfo) -> headers_external_outgoing_text(binary:split(ResponseInfo, <<0>>, [global])). headers_external_outgoing_text([<<>>]) -> []; headers_external_outgoing_text([K, <<>>]) -> [{K, <<>>}]; headers_external_outgoing_text([K, V | L]) -> [{K, V} | headers_external_outgoing_text(L)]. get_query_string_external([]) -> <<>>; get_query_string_external(QsVals) -> erlang:iolist_to_binary(get_query_string_external_text(QsVals)). get_query_string_external_text([] = L) -> L; get_query_string_external_text([{K, V} | L]) -> if V =:= true -> [[K, 0, <<"true">>, 0] | get_query_string_external_text(L)]; is_binary(V) -> [[K, 0, V, 0] | get_query_string_external_text(L)] end. request_time_start() -> cloudi_timestamp:microseconds_monotonic(). request_time_end_success(HttpCode, Method, NameIncoming, NameOutgoing, RequestStartMicroSec) -> ?LOG_TRACE("~w ~s ~s (to ~s) ~p ms", [HttpCode, Method, NameIncoming, NameOutgoing, (cloudi_timestamp:microseconds_monotonic() - RequestStartMicroSec) / 1000.0]). request_time_end_error(HttpCode, Method, NameIncoming, NameOutgoing, RequestStartMicroSec, Reason) -> RequestTime = (cloudi_timestamp:microseconds_monotonic() - RequestStartMicroSec) / 1000.0, if NameOutgoing =:= undefined -> ?LOG_WARN("~w ~s ~s ~p ms: ~p", [HttpCode, Method, NameIncoming, RequestTime, Reason]); true -> ?LOG_WARN("~w ~s ~s (to ~s) ~p ms: ~p", [HttpCode, Method, NameIncoming, NameOutgoing, RequestTime, Reason]) end. websocket_time_start() -> cloudi_timestamp:microseconds_monotonic(). websocket_time_end_success(NameIncoming, NameOutgoing, RequestStartMicroSec) -> ?LOG_TRACE("~s (to ~s) ~p ms", [NameIncoming, NameOutgoing, (cloudi_timestamp:microseconds_monotonic() - RequestStartMicroSec) / 1000.0]). websocket_time_end_error(NameIncoming, NameOutgoing, RequestStartMicroSec, Reason) -> ?LOG_WARN("~s (to ~s) ~p ms: ~p", [NameIncoming, NameOutgoing, (cloudi_timestamp:microseconds_monotonic() - RequestStartMicroSec) / 1000.0, Reason]). websocket_request_end(Name, NewTimeout, OldTimeout) -> ?LOG_TRACE("~s ~p ms", [Name, OldTimeout - NewTimeout]). handle_request(Name, Headers, 'normal', Req, #cowboy_state{ timeout_body = TimeoutBody, length_body_read = LengthBodyRead, length_body_chunk = LengthBodyChunk} = State) -> BodyOpts = [{length, LengthBodyChunk}, {read_length, LengthBodyRead}, {read_timeout, TimeoutBody}], {ok, Body, NextReq} = handle_request_body(Req, BodyOpts), handle_request(Name, Headers, Body, NextReq, State); handle_request(Name, Headers, 'application_zip', Req, #cowboy_state{ timeout_body = TimeoutBody, length_body_read = LengthBodyRead, length_body_chunk = LengthBodyChunk} = State) -> BodyOpts = [{length, LengthBodyChunk}, {read_length, LengthBodyRead}, {read_timeout, TimeoutBody}], {ok, Body, NextReq} = handle_request_body(Req, BodyOpts), handle_request(Name, Headers, zlib:unzip(Body), NextReq, State); handle_request(Name, Headers, 'multipart', Req, #cowboy_state{ dispatcher = Dispatcher, timeout_async = TimeoutAsync, timeout_part_header = TimeoutPartHeader, length_part_header_read = LengthPartHeaderRead, length_part_header_chunk = LengthPartHeaderChunk, timeout_part_body = TimeoutPartBody, length_part_body_read = LengthPartBodyRead, length_part_body_chunk = LengthPartBodyChunk, parts_destination_lock = PartsDestinationLock} = State) -> DestinationLock = if PartsDestinationLock =:= true -> cloudi_service:get_pid(Dispatcher, Name, TimeoutAsync); PartsDestinationLock =:= false -> {ok, undefined} end, case DestinationLock of {ok, Destination} -> Self = self(), PartHeaderOpts = [{length, LengthPartHeaderChunk}, {read_length, LengthPartHeaderRead}, {read_timeout, TimeoutPartHeader}], PartBodyOpts = [{length, LengthPartBodyChunk}, {read_length, LengthPartBodyRead}, {read_timeout, TimeoutPartBody}], MultipartId = erlang:list_to_binary(erlang:pid_to_list(Self)), handle_request_multipart(Name, lists:keysort(1, Headers), Destination, Self, PartHeaderOpts, PartBodyOpts, MultipartId, Req, State); {error, timeout} -> {{cowboy_error, timeout}, Req, State} end; handle_request(Name, Headers, Body, Req, #cowboy_state{ dispatcher = Dispatcher, timeout_sync = TimeoutSync, output_type = OutputType} = State) -> RequestInfo = if (OutputType =:= external) orelse (OutputType =:= binary) -> headers_external_incoming(Headers); (OutputType =:= internal) orelse (OutputType =:= list) -> Headers end, Request = if (OutputType =:= external) orelse (OutputType =:= internal) orelse (OutputType =:= binary) -> Body; (OutputType =:= list) -> erlang:binary_to_list(Body) end, case send_sync_minimal(Dispatcher, Name, RequestInfo, Request, TimeoutSync, self()) of {ok, ResponseInfo, Response} -> HeadersOutgoing = headers_external_outgoing(ResponseInfo), {{cowboy_response, HeadersOutgoing, Response}, Req, State}; {error, timeout} -> {{cowboy_error, timeout}, Req, State} end. handle_request_body(Req, BodyOpts) -> case cowboy_req:body(Req, BodyOpts) of {ok, _, _} = Success -> Success; {error, Reason} -> erlang:exit({cowboy_error, Reason}) end. handle_request_multipart(Name, Headers, Destination, Self, PartHeaderOpts, PartBodyOpts, MultipartId, Req0, State) -> case cowboy_req:part(Req0, PartHeaderOpts) of {ok, HeadersPart, ReqN} -> handle_request_multipart([], 0, Name, Headers, HeadersPart, Destination, Self, PartHeaderOpts, PartBodyOpts, MultipartId, ReqN, State); {done, ReqN} -> {{cowboy_error, multipart_empty}, ReqN, State} end. handle_request_multipart(TransIdList, I, Name, Headers, HeadersPart, Destination, Self, PartHeaderOpts, PartBodyOpts, MultipartId, Req0, State) -> case handle_request_multipart_send([], I, Name, Headers, HeadersPart, Destination, Self, PartHeaderOpts, PartBodyOpts, MultipartId, Req0, State) of {{ok, TransId}, undefined, ReqN, NewState} -> handle_request_multipart_receive(lists:reverse([TransId | TransIdList]), ReqN, NewState); {{ok, TransId}, HeadersPartNext, ReqN, NewState} -> handle_request_multipart([TransId | TransIdList], I + 1, Name, Headers, HeadersPartNext, Destination, Self, PartHeaderOpts, PartBodyOpts, MultipartId, ReqN, NewState) end. headers_merge(HeadersPart, Headers) -> lists:keymerge(1, lists:keysort(1, HeadersPart), Headers). handle_request_multipart_send(PartBodyList, I, Name, Headers, HeadersPart0, Destination, Self, PartHeaderOpts, PartBodyOpts, MultipartId, Req0, #cowboy_state{ dispatcher = Dispatcher, timeout_async = TimeoutAsync, output_type = OutputType} = State) -> case cowboy_req:part_body(Req0, PartBodyOpts) of {ok, PartBodyChunkLast, Req1} -> PartBody = if PartBodyList == [] -> PartBodyChunkLast; true -> erlang:iolist_to_binary(lists:reverse([PartBodyChunkLast | PartBodyList])) end, {HeadersPartNextN, ReqN} = case cowboy_req:part(Req1, PartHeaderOpts) of {ok, HeadersPartNext0, Req2} -> {HeadersPartNext0, Req2}; {done, Req2} -> {undefined, Req2} end, % each multipart part becomes a separate service request % however, the non-standard HTTP header request data provides % information to handle the sequence concurrently % (use multipart_destination_lock (defaults to true) if you need % the same destination used for each part) HeadersPart1 = headers_merge(HeadersPart0, Headers), HeadersPartN = if HeadersPartNextN =:= undefined -> [% socket pid as a string {<<"x-multipart-id">>, MultipartId}, % 0-based index {<<"x-multipart-index">>, erlang:integer_to_binary(I)}, % yes, this is the last part {<<"x-multipart-last">>, <<"true">>} | HeadersPart1]; true -> [% socket pid as a string {<<"x-multipart-id">>, MultipartId}, % 0-based index {<<"x-multipart-index">>, erlang:integer_to_binary(I)} | HeadersPart1] end, RequestInfo = if (OutputType =:= external) orelse (OutputType =:= binary) -> headers_external_incoming(HeadersPartN); (OutputType =:= internal) orelse (OutputType =:= list) -> HeadersPartN end, Request = if (OutputType =:= external) orelse (OutputType =:= internal) orelse (OutputType =:= binary) -> PartBody; (OutputType =:= list) -> erlang:binary_to_list(PartBody) end, SendResult = send_async_minimal(Dispatcher, Name, RequestInfo, Request, TimeoutAsync, Destination, Self), {SendResult, HeadersPartNextN, ReqN, State}; {more, PartBodyChunk, Req1} -> handle_request_multipart_send([PartBodyChunk | PartBodyList], I, Name, Headers, HeadersPart0, Destination, Self, PartHeaderOpts, PartBodyOpts, MultipartId, Req1, State) end. handle_request_multipart_receive_results([], _, [Error | _], Req, State) -> {Error, Req, State}; handle_request_multipart_receive_results([], [Success | _], _, Req, State) -> {Success, Req, State}; handle_request_multipart_receive_results([{ResponseInfo, Response, _} | ResponseList], SuccessList, ErrorList, Req, State) -> HeadersOutgoing = headers_external_outgoing(ResponseInfo), Status = case lists:keyfind(<<"status">>, 1, HeadersOutgoing) of {_, V} -> erlang:binary_to_integer(hd(binary:split(V, <<" ">>))); false -> 200 end, if (Status >= 200) andalso (Status =< 299) -> handle_request_multipart_receive_results(ResponseList, [{cowboy_response, HeadersOutgoing, Response} | SuccessList], ErrorList, Req, State); true -> handle_request_multipart_receive_results(ResponseList, SuccessList, [{cowboy_response, HeadersOutgoing, Response} | ErrorList], Req, State) end. handle_request_multipart_receive([_ | _] = TransIdList, Req, #cowboy_state{ timeout_sync = TimeoutSync} = State) -> case recv_asyncs_minimal(TimeoutSync, TransIdList) of {ok, ResponseList} -> handle_request_multipart_receive_results(ResponseList, [], [], Req, State); {error, timeout} -> {{cowboy_error, timeout}, Req, State} end. handle_response(NameIncoming, HeadersOutgoing0, Response, ReqN, OutputType, ContentTypeForced) -> ResponseBinary = if (((OutputType =:= external) orelse (OutputType =:= internal)) andalso (is_binary(Response) orelse is_list(Response))) orelse ((OutputType =:= binary) andalso is_binary(Response)) -> Response; ((OutputType =:= list) andalso is_list(Response)) -> erlang:iolist_to_binary(Response) end, {HttpCode, HeadersOutgoingN} = case lists:keytake(<<"status">>, 1, HeadersOutgoing0) of false -> {200, HeadersOutgoing0}; {value, {_, Status}, HeadersOutgoing1} when is_binary(Status) -> {erlang:binary_to_integer(hd(binary:split(Status, <<" ">>))), HeadersOutgoing1} end, ResponseHeadersOutgoing = if HeadersOutgoingN =/= [] -> HeadersOutgoingN; ContentTypeForced =/= undefined -> [{<<"content-type">>, ContentTypeForced}]; true -> Extension = filename:extension(NameIncoming), if Extension == [] -> [{<<"content-type">>, <<"text/html">>}]; true -> {AttachmentGuess, ContentType} = case cloudi_response_info: lookup_content_type(binary, Extension) of error -> {attachment, <<"application/octet-stream">>}; {ok, AttachmentGuessContentTypeTuple} -> AttachmentGuessContentTypeTuple end, if AttachmentGuess =:= attachment, HttpCode >= 200, HttpCode < 300, HttpCode /= 204 -> [{<<"content-disposition">>, ["attachment; filename=\"", filename:basename(NameIncoming), "\""]}, {<<"content-type">>, ContentType}]; true -> [{<<"content-type">>, ContentType}] end end end, {ok, Req} = cowboy_req:reply(HttpCode, ResponseHeadersOutgoing, ResponseBinary, ReqN), {HttpCode, Req}. service_name_incoming(UseClientIpPrefix, UseHostPrefix, PathRaw, Client, Req0) when UseClientIpPrefix =:= true, UseHostPrefix =:= true -> {HostRaw, Req1} = cowboy_req:host(Req0), {service_name_incoming_merge(Client, HostRaw, PathRaw), Req1}; service_name_incoming(UseClientIpPrefix, UseHostPrefix, PathRaw, Client, Req0) when UseClientIpPrefix =:= true, UseHostPrefix =:= false -> {service_name_incoming_merge(Client, undefined, PathRaw), Req0}; service_name_incoming(UseClientIpPrefix, UseHostPrefix, PathRaw, _Client, Req0) when UseClientIpPrefix =:= false, UseHostPrefix =:= true -> {HostRaw, Req1} = cowboy_req:host(Req0), {service_name_incoming_merge(undefined, HostRaw, PathRaw), Req1}; service_name_incoming(UseClientIpPrefix, UseHostPrefix, PathRaw, _Client, Req0) when UseClientIpPrefix =:= false, UseHostPrefix =:= false -> {service_name_incoming_merge(undefined, undefined, PathRaw), Req0}. service_name_incoming_merge(undefined, undefined, PathRaw) -> erlang:binary_to_list(PathRaw); service_name_incoming_merge(undefined, HostRaw, PathRaw) -> erlang:binary_to_list(<>); service_name_incoming_merge({ClientIpAddr, _ClientPort}, undefined, PathRaw) -> cloudi_ip_address:to_string(ClientIpAddr) ++ erlang:binary_to_list(PathRaw); service_name_incoming_merge({ClientIpAddr, _ClientPort}, HostRaw, PathRaw) -> cloudi_ip_address:to_string(ClientIpAddr) ++ erlang:binary_to_list(<<$/, HostRaw/binary, PathRaw/binary>>). websocket_terminate_check(<<>>) -> false; websocket_terminate_check([]) -> false; websocket_terminate_check(ResponseInfo) -> HeadersOutgoing = headers_external_outgoing(ResponseInfo), case lists:keyfind(<<"connection">>, 1, HeadersOutgoing) of {<<"connection">>, <<"close">>} -> true; {<<"connection">>, _} -> false; false -> false end. websocket_connect_request(OutputType) when OutputType =:= external; OutputType =:= internal; OutputType =:= binary -> <<"CONNECT">>; websocket_connect_request(OutputType) when OutputType =:= list -> "CONNECT". websocket_connect_check(undefined, State) -> State; websocket_connect_check({async, WebSocketConnectName}, #cowboy_state{ dispatcher = Dispatcher, timeout_async = TimeoutAsync, output_type = OutputType, websocket_state = #websocket_state{ request_info = RequestInfo} = WebSocketState } = State) -> Request = websocket_connect_request(OutputType), case send_async_minimal(Dispatcher, WebSocketConnectName, RequestInfo, Request, TimeoutAsync, self()) of {ok, TransId} -> State#cowboy_state{ websocket_state = WebSocketState#websocket_state{ websocket_connect_trans_id = TransId}}; {error, timeout} -> State end; websocket_connect_check({sync, WebSocketConnectName}, #cowboy_state{ dispatcher = Dispatcher, timeout_sync = TimeoutSync, output_type = OutputType, websocket_state = #websocket_state{ request_info = RequestInfo} = WebSocketState } = State) -> Self = self(), case send_async_minimal(Dispatcher, WebSocketConnectName, RequestInfo, websocket_connect_request(OutputType), TimeoutSync, Self) of {ok, TransId} -> case recv_async_minimal(TimeoutSync, TransId) of {ok, ResponseInfo, Response} -> % must provide the response after the websocket_init is done Self ! {'cloudi_service_return_async', WebSocketConnectName, WebSocketConnectName, ResponseInfo, Response, TimeoutSync, TransId, Self}, State#cowboy_state{ websocket_state = WebSocketState#websocket_state{ websocket_connect_trans_id = TransId}}; {error, timeout} -> State end; {error, timeout} -> State end. websocket_disconnect_request(OutputType) when OutputType =:= external; OutputType =:= internal; OutputType =:= binary -> <<"DISCONNECT">>; websocket_disconnect_request(OutputType) when OutputType =:= list -> "DISCONNECT". websocket_disconnect_request_info_reason({remote, CloseCode, CloseBinary}) when is_integer(CloseCode) -> [<<"remote,">>, erlang:integer_to_binary(CloseCode), <<",">>, CloseBinary]; websocket_disconnect_request_info_reason({ReasonType, ReasonDescription}) -> [erlang:atom_to_binary(ReasonType, utf8), <<",">>, erlang:atom_to_binary(ReasonDescription, utf8)]. websocket_disconnect_request_info(Reason, RequestInfo, OutputType) when OutputType =:= external; OutputType =:= binary -> erlang:iolist_to_binary([<<"disconnection">>, 0, websocket_disconnect_request_info_reason(Reason), 0, RequestInfo]); websocket_disconnect_request_info(Reason, RequestInfo, OutputType) when OutputType =:= internal; OutputType =:= list -> [{<<"disconnection">>, websocket_disconnect_request_info_reason(Reason)} | RequestInfo]. websocket_disconnect_check(undefined, _, _) -> ok; websocket_disconnect_check({async, WebSocketDisconnectName}, Reason, #cowboy_state{ dispatcher = Dispatcher, timeout_async = TimeoutAsync, output_type = OutputType, websocket_state = #websocket_state{ request_info = RequestInfo}}) -> send_async_minimal(Dispatcher, WebSocketDisconnectName, websocket_disconnect_request_info(Reason, RequestInfo, OutputType), websocket_disconnect_request(OutputType), TimeoutAsync, self()), ok; websocket_disconnect_check({sync, WebSocketDisconnectName}, Reason, #cowboy_state{ dispatcher = Dispatcher, timeout_sync = TimeoutSync, output_type = OutputType, websocket_state = #websocket_state{ request_info = RequestInfo}}) -> send_sync_minimal(Dispatcher, WebSocketDisconnectName, websocket_disconnect_request_info(Reason, RequestInfo, OutputType), websocket_disconnect_request(OutputType), TimeoutSync, self()), ok. websocket_subscriptions([], _, _) -> ok; websocket_subscriptions([F | Functions], Parameters, Scope) -> case F(Parameters) of {ok, NameWebSocket} -> ok = cpg:join(Scope, NameWebSocket, self(), infinity); {error, _} -> ok end, websocket_subscriptions(Functions, Parameters, Scope). websocket_handle_incoming_request(Dispatcher, NameOutgoing, RequestInfo, Request, TimeoutSync, ResponseF, WebSocketRequestType, Req, NameIncoming, State) -> RequestStartMicroSec = ?LOG_WARN_APPLY(fun websocket_time_start/0, []), case send_sync_minimal(Dispatcher, NameOutgoing, RequestInfo, Request, TimeoutSync, self()) of {ok, ResponseInfo, Response} -> ?LOG_TRACE_APPLY(fun websocket_time_end_success/3, [NameIncoming, NameOutgoing, RequestStartMicroSec]), case websocket_terminate_check(ResponseInfo) of true when Response == <<>> -> {reply, close, Req, State}; true -> {reply, [{WebSocketRequestType, ResponseF(Response)}, close], Req, State}; false -> {reply, {WebSocketRequestType, ResponseF(Response)}, Req, State} end; {error, timeout} -> ?LOG_WARN_APPLY(fun websocket_time_end_error/4, [NameIncoming, NameOutgoing, RequestStartMicroSec, timeout]), {reply, {WebSocketRequestType, <<>>}, Req, State} end. websocket_handle_outgoing_response({SendType, Name, Pattern, _, _, OldTimeout, _, TransId, Source}, ResponseTimer, ResponseInfo, Response) -> Timeout = case erlang:cancel_timer(ResponseTimer) of false -> 0; V -> V end, ReturnType = if SendType =:= 'cloudi_service_send_async' -> 'cloudi_service_return_async'; SendType =:= 'cloudi_service_send_sync' -> 'cloudi_service_return_sync' end, Source ! {ReturnType, Name, Pattern, ResponseInfo, Response, Timeout, TransId, Source}, ?LOG_TRACE_APPLY(fun websocket_request_end/3, [Name, Timeout, OldTimeout]). websocket_process_queue(Req, #cowboy_state{websocket_state = #websocket_state{ response_pending = false, recv_timeouts = RecvTimeouts, queued = Queue} = WebSocketState} = State) -> case pqueue4:out(Queue) of {empty, NewQueue} -> {ok, Req, State#cowboy_state{websocket_state = WebSocketState#websocket_state{queued = NewQueue}}}; {{value, {Type, Name, Pattern, RequestInfo, Request, _, Priority, TransId, Pid}}, NewQueue} -> Timeout = case erlang:cancel_timer(maps:get(TransId, RecvTimeouts)) of false -> 0; V -> V end, websocket_info({Type, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Pid}, Req, State#cowboy_state{websocket_state = WebSocketState#websocket_state{ recv_timeouts = maps:remove(TransId, RecvTimeouts), queued = NewQueue}}) end.