Current section
Files
Jump to
Current section
Files
src/grpcbox_client.erl
%%%-------------------------------------------------------------------
%% @doc grpc client
%% @end
%%%-------------------------------------------------------------------
-module(grpcbox_client).
-export([unary/6,
unary/5,
stream/4,
stream/5,
send/2,
recv_headers/1,
recv_headers/2,
recv_data/1,
recv_data/2,
recv_trailers/1,
recv_trailers/2,
close_and_recv/1,
close_send/1]).
-include_lib("chatterbox/include/http2.hrl").
-include("grpcbox.hrl").
-type options() :: #{channel => grpcbox_channel:t(),
encoding => grpcbox:encoding()}.
-type unary_interceptor() :: term().
-type stream_interceptor() :: term().
-type interceptor() :: unary_interceptor() | stream_interceptor().
-type stream() :: #{channel => pid(),
stream_id => stream_id(),
stream_pid => pid(),
monitor_ref => reference(),
service_def => #grpcbox_def{},
encoding => grpcbox:encoding()}.
-export_type([stream/0,
options/0,
unary_interceptor/0,
stream_interceptor/0,
interceptor/0]).
get_channel(Options, Type) ->
Channel = maps:get(channel, Options, default_channel),
grpcbox_channel:pick(Channel, Type).
unary(Ctx, Service, Method, Input, Def, Options) ->
unary(Ctx, filename:join([<<>>, Service, Method]), Input, Def, Options).
unary(Ctx, Path, Input, Def, Options) ->
case get_channel(Options, unary) of
{ok, {Channel, Interceptor}} ->
Handler = fun(Ctx1, Input1) ->
unary_handler(Ctx1, Channel, Path, Input1, Def, Options)
end,
case Interceptor of
undefined ->
Handler(Ctx, Input);
_ ->
Interceptor(Ctx, Channel, Handler, Path, Input, Def, Options)
end;
{error, _Reason}=Error ->
Error
end.
unary_handler(Ctx, Channel, Path, Input, Def, Options) ->
try
case grpcbox_client_stream:send_request(Ctx, Channel, Path, Input, Def, Options) of
{ok, _Conn, Stream, Pid} ->
Ref = erlang:monitor(process, Pid),
S = #{channel => Channel,
stream_id => Stream,
stream_pid => Pid,
monitor_ref => Ref,
service_def => Def},
case recv_end(S, 5000) of
eos ->
case recv_headers(S, 0) of
{ok, Headers} ->
case recv_trailers(S) of
{ok, {<<"0">>, _, Metadata}} ->
case recv_data(S, 0) of
{ok, Data} ->
{ok, Data, #{headers => Headers,
trailers => Metadata}};
stream_finished ->
{ok, <<>>, #{headers => Headers,
trailers => Metadata}}
end;
{ok, {Status, Message, Trailers}} ->
{error, {Status, Message}, #{headers => Headers,
trailers => Trailers}};
{error, _}=Error ->
Error
end;
{http_error, Status, Headers} ->
%% different from an `error' in that it isn't from the grpc layer
{http_error, {Status, <<>>}, #{headers => Headers,
trailers => #{}}};
{error, _}=Error ->
Error
end;
{error, _}=Error ->
Error
end;
{error, {shutdown, econnrefused}} ->
{error, econnrefused};
{error, _}=Error ->
Error
end
catch
error:{badmatch, {error, {shutdown,econnrefused}}} ->
{error, econnrefused};
throw:{error, _}=E ->
E
end.
%% no input: bidrectional
stream(Ctx, Path, Def, Options) ->
case get_channel(Options, stream) of
{ok, {Channel, Interceptor}} ->
case Interceptor of
undefined ->
grpcbox_client_stream:new_stream(Ctx, Channel, Path, Def, Options);
#{new_stream := NewStream} ->
case NewStream(Ctx, Channel, Path, Def, fun grpcbox_client_stream:new_stream/5, Options) of
{ok, S} ->
{ok, S#{stream_interceptor => Interceptor}};
{error, _}=Error ->
Error
end;
_ ->
grpcbox_client_stream:new_stream(Ctx, Channel, Path, Def, Options)
end;
{error, _Reason}=Error ->
Error
end.
close_and_recv(Stream) ->
close_send(Stream),
case recv_end(Stream, 5000) of
eos ->
recv_data(Stream, 0);
{error, _}=Error ->
Error
end.
close_send(#{channel := Conn,
stream_id := StreamId}) ->
ok = h2_connection:send_body(Conn, StreamId, <<>>, [{send_end_stream, true}]).
%% h2_connection:send_trailers(Conn, StreamId, [], [{send_end_stream, true}]).
send(Stream=#{stream_interceptor := #{send_msg := SendMsg}}, Input) ->
SendMsg(Stream, fun grpcbox_client_stream:send_msg/2, Input);
send(Stream, Input) ->
grpcbox_client_stream:send_msg(Stream, Input).
%% input given, stream response
stream(Ctx, Path, Input, Def, Options) ->
case get_channel(Options, stream) of
{ok, {Channel, _Interceptor}} ->
case
grpcbox_client_stream:send_request(Ctx, Channel, Path,
Input, Def, Options)
of
{ok, Conn, Stream, Pid} ->
Ref = erlang:monitor(process, Pid),
{ok, #{channel => Conn,
stream_id => Stream,
stream_pid => Pid,
monitor_ref => Ref,
service_def => Def}};
{error, _Reason} = Error ->
Error
end;
{error, _Reason} = Error ->
Error
end.
recv_data(Stream) ->
recv_data(Stream, 500).
recv_data(Stream=#{stream_interceptor := #{recv_msg := RecvMsg}}, Timeout) ->
RecvMsg(Stream, fun grpcbox_client_stream:recv_msg/2, Timeout);
recv_data(Stream, Timeout) ->
grpcbox_client_stream:recv_msg(Stream, Timeout).
recv_headers(S) ->
recv_headers(S, 500).
recv_headers(S, Timeout) ->
case recv(headers, S, Timeout) of
{ok, Headers} ->
case maps:get(<<":status">>, Headers, undefined) of
<<"200">> ->
{ok, Headers};
ErrorStatus ->
{http_error, ErrorStatus, Headers}
end;
{error, _Reason}=Error ->
Error
end.
recv_trailers(S) ->
recv_trailers(S, 500).
recv_trailers(S, Timeout) ->
recv(trailers, S, Timeout).
recv(Type, #{stream_id := Id,
monitor_ref := Ref,
stream_pid := Pid}, Timeout) ->
receive
{Type, Id, V} ->
{ok, V};
{'DOWN', Ref, process, Pid, _Reason} ->
receive
{trailers, Id, {Status, Message, Metadata}} ->
{ok, {Status, Message, Metadata}}
after 0 ->
{error, unknown}
end
after Timeout ->
{error, timeout}
end.
recv_end(#{stream_id := StreamId,
stream_pid := Pid,
monitor_ref := Ref}, Timeout) ->
receive
{eos, StreamId} ->
erlang:demonitor(Ref, [flush]),
receive
{'END_STREAM', StreamId} ->
eos
after Timeout -> %% actually, this Timeout will never happen because of outer receive Timeout
{error, eos}
end;
{'DOWN', Ref, process, Pid, normal} ->
%% this is sent by h2_connection after the stream process has ended
receive
{'END_STREAM', StreamId} ->
eos
after Timeout ->
{error, {stream_down, normal}}
end;
{'DOWN', Ref, process, Pid, Reason} ->
{error, {stream_down, Reason}}
after Timeout ->
{error, timeout}
end.