reorganizing because zx needs to feel special
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,208 @@
|
||||
-module(vanillae_fetcher).
|
||||
-vsn("0.1.0").
|
||||
-author("Craig Everett <ceverett@tsuriai.jp>").
|
||||
-copyright("Craig Everett <ceverett@tsuriai.jp>").
|
||||
-license("MIT").
|
||||
|
||||
-export([connect/4]).
|
||||
|
||||
-include("$zx_include/zx_logger.hrl").
|
||||
|
||||
|
||||
connect(Node = {Host, Port}, Request, From, Timeout) ->
|
||||
Timer = erlang:send_after(Timeout, self(), timeout),
|
||||
Options = [{mode, binary}, {nodelay, true}, {active, once}],
|
||||
case gen_tcp:connect(Host, Port, Options, 3000) of
|
||||
{ok, Sock} -> do(Request, Sock, Node, From, Timer);
|
||||
Error -> gen_server:reply(From, Error)
|
||||
end.
|
||||
|
||||
do(Request, Sock, Node, From, Timer) ->
|
||||
Formed = unicode:characters_to_list(form(Request, Node)),
|
||||
case gen_tcp:send(Sock, Formed) of
|
||||
ok -> await(Sock, From, Timer);
|
||||
Error -> gen_server:reply(From, Error)
|
||||
end.
|
||||
|
||||
await(Sock, From, Timer) ->
|
||||
receive
|
||||
{tcp, Sock, Bin} ->
|
||||
parse(Bin, Sock, From, Timer);
|
||||
{tcp_closed, Sock} ->
|
||||
ok = erlang:cancel_timer(Timer, [{async, true}]),
|
||||
gen_server:reply(From, {error, enotconn});
|
||||
timeout ->
|
||||
gen_server:reply(From, {error, timeout})
|
||||
after 120000 ->
|
||||
gen_server:reply(From, {error, timeout})
|
||||
end.
|
||||
|
||||
|
||||
form({get, Path}, Node) ->
|
||||
["GET ", Path, " HTTP/1.1\r\n",
|
||||
"Host: ", host_string(Node), "\r\n",
|
||||
"User-Agent: Vanillae/0.1.0\r\n",
|
||||
"Accept: */*\r\n\r\n"];
|
||||
form({post, Path, Payload}, Node) ->
|
||||
ByteSize = integer_to_list(byte_size(Payload)),
|
||||
["POST ", Path, " HTTP/1.1\r\n",
|
||||
"Host: ", host_string(Node), "\r\n",
|
||||
"Content-Type: application/json\r\n",
|
||||
"Content-Length: ", ByteSize, "\r\n",
|
||||
"User-Agent: Vanillae/0.1.0\r\n",
|
||||
"Accept: */*\r\n\r\n",
|
||||
Payload].
|
||||
|
||||
|
||||
host_string({Address, Port}) when is_list(Address) ->
|
||||
PortS = integer_to_list(Port),
|
||||
[Address, ":", PortS];
|
||||
host_string({Address, Port}) when is_atom(Address) ->
|
||||
AddressS = atom_to_list(Address),
|
||||
PortS = integer_to_list(Port),
|
||||
[AddressS, ":", PortS];
|
||||
host_string({Address, Port}) ->
|
||||
AddressS = inet:ntoa(Address),
|
||||
PortS = integer_to_list(Port),
|
||||
[AddressS, ":", PortS].
|
||||
|
||||
|
||||
parse(Received, Sock, From, Timer) ->
|
||||
case Received of
|
||||
<<"HTTP/1.1 200 OK\r\n", Tail/binary>> ->
|
||||
parse2(200, Tail, Sock, From, Timer);
|
||||
<<"HTTP/1.1 400 Bad Request\r\n", Tail/binary>> ->
|
||||
parse2(400, Tail, Sock, From, Timer);
|
||||
<<"HTTP/1.1 404 Not Found\r\n", Tail/binary>> ->
|
||||
parse2(404, Tail, Sock, From, Timer);
|
||||
<<"HTTP/1.1 500 Internal Server Error\r\n", Tail/binary>> ->
|
||||
parse2(500, Tail, Sock, From, Timer);
|
||||
_ ->
|
||||
ok = zx_net:disconnect(Sock),
|
||||
ok = erlang:cancel_timer(Timer, [{async, true}]),
|
||||
gen_server:reply(From, {error, {received, Received}})
|
||||
end.
|
||||
|
||||
parse2(Code, Received, Sock, From, Timer) ->
|
||||
case read_headers(Sock, Received) of
|
||||
{ok, Headers, Rest} -> consume(Code, Rest, Headers, Sock, From, Timer);
|
||||
Error -> gen_server:reply(From, Error)
|
||||
end.
|
||||
|
||||
|
||||
consume(Code, Rest, Headers, Sock, From, Timer) ->
|
||||
case maps:find(<<"content-length">>, Headers) of
|
||||
error ->
|
||||
ok = erlang:cancel_timer(Timer, [{async, true}]),
|
||||
gen_server:reply(From, {error, {headers, Headers}});
|
||||
{ok, <<"0">>} ->
|
||||
ok = erlang:cancel_timer(Timer, [{async, true}]),
|
||||
Result = case Code =:= 200 of true -> ok; false -> {error, Code} end,
|
||||
gen_server:reply(From, Result);
|
||||
{ok, Size} ->
|
||||
try
|
||||
Length = binary_to_integer(Size),
|
||||
consume2(Length, Rest, Sock, From, Timer)
|
||||
catch
|
||||
error:badarg ->
|
||||
ok = erlang:cancel_timer(Timer, [{async, true}]),
|
||||
gen_server:reply(From, {error, {headers, Headers}})
|
||||
end
|
||||
end.
|
||||
|
||||
consume2(Length, Received, Sock, From, Timer) ->
|
||||
Size = byte_size(Received),
|
||||
if
|
||||
Size == Length ->
|
||||
ok = erlang:cancel_timer(Timer, [{async, true}]),
|
||||
ok = zx_net:disconnect(Sock),
|
||||
Result = zj:decode(Received),
|
||||
gen_server:reply(From, Result);
|
||||
Size < Length ->
|
||||
consume3(Length, Received, Sock, From, Timer);
|
||||
Size > Length ->
|
||||
ok = erlang:cancel_timer(Timer, [{async, true}]),
|
||||
gen_server:reply(From, {error, bad_length})
|
||||
end.
|
||||
|
||||
consume3(Length, Received, Sock, From, Timer) ->
|
||||
ok = inet:setopts(Sock, [{active, once}]),
|
||||
receive
|
||||
{tcp, Sock, Bin} ->
|
||||
consume2(Length, <<Received/binary, Bin/binary>>, Sock, From, Timer);
|
||||
timeout ->
|
||||
gen_server:reply(From, {error, {timeout, Received}})
|
||||
end.
|
||||
|
||||
|
||||
read_headers(Socket, <<"\r">>) ->
|
||||
ok = inet:setopts(Socket, [{active, once}]),
|
||||
receive
|
||||
{tcp, Socket, Bin} -> read_headers(Socket, <<"\r", Bin/binary>>);
|
||||
timeout -> {error, timeout}
|
||||
after 120000 -> {error, timeout}
|
||||
end;
|
||||
read_headers(_, <<"\r\n", Received/binary>>) ->
|
||||
log(info, "~p Headers died at: ~p", [?LINE, Received]),
|
||||
{error, headers};
|
||||
read_headers(Socket, Received) ->
|
||||
read_hkey(Socket, Received, <<>>, #{}).
|
||||
|
||||
read_hkey(Socket, <<Char, Rest/binary>>, Acc, Headers)
|
||||
when $A =< Char, Char =< $Z ->
|
||||
read_hkey(Socket, Rest, <<Acc/binary, (Char + 32)>>, Headers);
|
||||
read_hkey(Socket, <<Char, Rest/binary>>, Acc, Headers)
|
||||
when 32 =< Char, Char =< 57;
|
||||
59 =< Char, Char =< 126 ->
|
||||
read_hkey(Socket, Rest, <<Acc/binary, Char>>, Headers);
|
||||
read_hkey(Socket, <<":", Rest/binary>>, Key, Headers) ->
|
||||
skip_hblanks(Socket, Rest, Key, Headers);
|
||||
read_hkey(_, <<"\r\n", Rest/binary>>, <<>>, Headers) ->
|
||||
{ok, Headers, Rest};
|
||||
read_hkey(Socket, <<>>, Acc, Headers) ->
|
||||
ok = inet:setopts(Socket, [{active, once}]),
|
||||
receive
|
||||
{tcp, Socket, Bin} -> read_hkey(Socket, Bin, Acc, Headers);
|
||||
timeout -> {error, timeout}
|
||||
after 120000 -> {error, timeout}
|
||||
end;
|
||||
read_hkey(_, Received, _, _) ->
|
||||
log(info, "~p Headers died at: ~p", [?LINE, Received]),
|
||||
{error, headers}.
|
||||
|
||||
skip_hblanks(Socket, <<" ", Rest/binary>>, Key, Headers) ->
|
||||
skip_hblanks(Socket, Rest, Key, Headers);
|
||||
skip_hblanks(Socket, <<>>, Key, Headers) ->
|
||||
ok = inet:setopts(Socket, [{active, once}]),
|
||||
receive
|
||||
{tcp, Socket, Bin} -> skip_hblanks(Socket, Bin, Key, Headers);
|
||||
timeout -> {error, timeout}
|
||||
after 120000 -> {error, timeout}
|
||||
end;
|
||||
skip_hblanks(_, Received = <<"\r", _/binary>>, _, _) ->
|
||||
log(info, "~p Headers died at: ~p", [?LINE, Received]),
|
||||
{error, headers};
|
||||
skip_hblanks(_, Received = <<"\n", _/binary>>, _, _) ->
|
||||
log(info, "~p Headers died at: ~p", [?LINE, Received]),
|
||||
{error, headers};
|
||||
skip_hblanks(Socket, Rest, Key, Headers) ->
|
||||
read_hval(Socket, Rest, <<>>, Key, Headers).
|
||||
|
||||
read_hval(_, Received = <<"\r\n", _/binary>>, <<>>, _, _) ->
|
||||
log(info, "~p Headers died at: ~p", [?LINE, Received]),
|
||||
{error, headers};
|
||||
read_hval(Socket, <<"\r\n", Rest/binary>>, Val, Key, Headers) ->
|
||||
read_hkey(Socket, Rest, <<>>, maps:put(Key, Val, Headers));
|
||||
read_hval(Socket, <<Char, Rest/binary>>, Acc, Key, Headers)
|
||||
when 32 =< Char, Char =< 126 ->
|
||||
read_hval(Socket, Rest, <<Acc/binary, Char>>, Key, Headers);
|
||||
read_hval(Socket, <<>>, Val, Key, Headers) ->
|
||||
ok = inet:setopts(Socket, [{active, once}]),
|
||||
receive
|
||||
{tcp, Socket, Bin} -> read_hval(Socket, Bin, Val, Key, Headers);
|
||||
timeout -> {error, timeout}
|
||||
after 120000 -> {error, timeout}
|
||||
end;
|
||||
read_hval(_, Received, _, _, _) ->
|
||||
log(info, "~p Headers died at: ~p", [?LINE, Received]),
|
||||
{error, headers}.
|
||||
@@ -0,0 +1,241 @@
|
||||
%%% @doc
|
||||
%%% Vanillae Request Manager for Erlang
|
||||
%%%
|
||||
%%% This process is responsible for remembering the configured nodes and dispatching
|
||||
%%% requests to them. Request dispatch is made in a round-robin fashion with forwarded
|
||||
%%% gen_server return `From' values passed to the request worker instead of being
|
||||
%%% responded to directly by the manager itself (despite requests being generated as
|
||||
%%% gen_server:call/3s.
|
||||
%%% @end
|
||||
|
||||
-module(vanillae_man).
|
||||
-vsn("0.1.0").
|
||||
-behavior(gen_server).
|
||||
-author("Craig Everett <ceverett@tsuriai.jp>").
|
||||
-copyright("Craig Everett <ceverett@tsuriai.jp>").
|
||||
-license("MIT").
|
||||
|
||||
%% Admin functions
|
||||
-export([network_id/0, network_id/1,
|
||||
ae_nodes/0, ae_nodes/1,
|
||||
timeout/0, timeout/1]).
|
||||
|
||||
%% The whole point of this module:
|
||||
-export([request/1, request/2]).
|
||||
|
||||
%% gen_server goo
|
||||
-export([start_link/0]).
|
||||
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
|
||||
code_change/3, terminate/2]).
|
||||
|
||||
%% TODO: Make logging more flexible
|
||||
-include("$zx_include/zx_logger.hrl").
|
||||
|
||||
|
||||
%%% Type and Record Definitions
|
||||
|
||||
-record(fetcher,
|
||||
{pid = none :: none | pid(),
|
||||
mon = none :: none | reference(),
|
||||
time = none :: none | erlang:timestamp(),
|
||||
node = none :: none | vanilae:ae_node(),
|
||||
from = none :: none | gen_server:from(),
|
||||
req = none :: none | binary()}).
|
||||
|
||||
-record(s,
|
||||
{network_id = "ae_mainnet" :: string(),
|
||||
ae_nodes = {[], []} :: {[vanillae:ae_node()], [vanillae:ae_node()]},
|
||||
fetchers = [] :: [#fetcher{}],
|
||||
timeout = 5000 :: pos_integer()}).
|
||||
|
||||
|
||||
-type state() :: #s{}.
|
||||
|
||||
|
||||
|
||||
%%% Service Interface
|
||||
|
||||
-spec network_id() -> Name
|
||||
when Name :: vanillae:network_id().
|
||||
|
||||
network_id() ->
|
||||
gen_server:call(?MODULE, network_id).
|
||||
|
||||
|
||||
-spec network_id(Name) -> ok
|
||||
when Name :: vanillae:network_id().
|
||||
|
||||
network_id(Name) ->
|
||||
gen_server:cast(?MODULE, {network_id, Name}).
|
||||
|
||||
|
||||
-spec ae_nodes() -> Used
|
||||
when Used :: [vanillae:ae_nodes()].
|
||||
|
||||
ae_nodes() ->
|
||||
gen_server:call(?MODULE, ae_nodes).
|
||||
|
||||
|
||||
-spec ae_nodes(ToUse) -> ok
|
||||
when ToUse :: [vanillae:ae_nodes()].
|
||||
|
||||
ae_nodes(ToUse) ->
|
||||
gen_server:cast(?MODULE, {ae_nodes, ToUse}).
|
||||
|
||||
|
||||
-spec timeout() -> Value
|
||||
when Value :: pos_integer().
|
||||
|
||||
timeout() ->
|
||||
gen_server:call(?MODULE, timeout).
|
||||
|
||||
|
||||
-spec timeout(Value) -> ok
|
||||
when Value :: pos_integer().
|
||||
|
||||
timeout(Value) when 0 < Value, Value =< 120000 ->
|
||||
gen_server:cast(?MODULE, {timeout, Value}).
|
||||
|
||||
|
||||
-spec request(Path) -> {ok, Value} | {error, Reason}
|
||||
when Path :: unicode:charlist(),
|
||||
Value :: map(),
|
||||
Reason :: vanillae:ae_error().
|
||||
|
||||
request(Path) ->
|
||||
gen_server:call(?MODULE, {request, {get, Path}}, infinity).
|
||||
|
||||
|
||||
-spec request(Path, Data) -> {ok, Value} | {error, Reason}
|
||||
when Path :: unicode:charlist(),
|
||||
Data :: unicode:charlist(),
|
||||
Value :: map(),
|
||||
Reason :: vanillae:ae_error().
|
||||
|
||||
request(Path, Data) ->
|
||||
gen_server:call(?MODULE, {request, {post, Path, Data}}, infinity).
|
||||
|
||||
|
||||
|
||||
%%% Startup Functions
|
||||
|
||||
|
||||
-spec start_link() -> Result
|
||||
when Result :: {ok, pid()}
|
||||
| {error, Reason :: term()}.
|
||||
%% @private
|
||||
%% This should only ever be called by v_clients (the service-level supervisor).
|
||||
|
||||
start_link() ->
|
||||
gen_server:start_link({local, ?MODULE}, ?MODULE, none, []).
|
||||
|
||||
|
||||
-spec init(none) -> {ok, state()}.
|
||||
%% @private
|
||||
%% Called by the supervisor process to give the process a chance to perform any
|
||||
%% preparatory work necessary for proper function.
|
||||
|
||||
init(none) ->
|
||||
ok = io:format("Starting.~n"),
|
||||
State = #s{},
|
||||
{ok, State}.
|
||||
|
||||
|
||||
|
||||
%%% gen_server Message Handling Callbacks
|
||||
|
||||
|
||||
handle_call({request, Request}, From, State) ->
|
||||
NewState = do_request(Request, From, State),
|
||||
{noreply, NewState};
|
||||
handle_call(network_id, _, State = #s{network_id = Name}) ->
|
||||
{reply, Name, State};
|
||||
handle_call(ae_nodes, _, State = #s{ae_nodes = {Wait, Used}}) ->
|
||||
Nodes = lists:append(Wait, Used),
|
||||
{reply, Nodes, State};
|
||||
handle_call(timeout, _, State = #s{timeout = Value}) ->
|
||||
{reply, Value, State};
|
||||
handle_call(Unexpected, From, State) ->
|
||||
ok = log(warning, "Unexpected call from ~tp: ~tp~n", [From, Unexpected]),
|
||||
{noreply, State}.
|
||||
|
||||
|
||||
handle_cast({network_id, Name}, State) ->
|
||||
{noreply, State#s{network_id = Name}};
|
||||
handle_cast({ae_nodes, []}, State) ->
|
||||
{noreply, State#s{ae_nodes = none}};
|
||||
handle_cast({ae_nodes, ToUse}, State) ->
|
||||
{noreply, State#s{ae_nodes = {ToUse, []}}};
|
||||
handle_cast({timeout, Value}, State) ->
|
||||
{noreply, State#s{timeout = Value}};
|
||||
handle_cast(Unexpected, State) ->
|
||||
ok = log(warning, "Unexpected cast: ~tp~n", [Unexpected]),
|
||||
{noreply, State}.
|
||||
|
||||
|
||||
handle_info({'DOWN', Mon, process, PID, Info}, State) ->
|
||||
NewState = handle_down(PID, Mon, Info, State),
|
||||
{noreply, NewState};
|
||||
handle_info(Unexpected, State) ->
|
||||
ok = log("Unexpected info: ~tp~n", [Unexpected]),
|
||||
{noreply, State}.
|
||||
|
||||
|
||||
handle_down(_, Mon, normal, State = #s{fetchers = Fetchers}) ->
|
||||
NewFetchers = lists:keydelete(Mon, #fetcher.mon, Fetchers),
|
||||
State#s{fetchers = NewFetchers};
|
||||
handle_down(PID, Mon, Info, State = #s{fetchers = Fetchers}) ->
|
||||
case lists:keytake(Mon, #fetcher.mon, Fetchers) of
|
||||
{value, #fetcher{time = Time, node = Node, from = From, req = R}, Remaining} ->
|
||||
TS = calendar:system_time_to_rfc3339(Time, [{unit, nanosecond}]),
|
||||
Format =
|
||||
"ERROR ~s: Fetcher process ~p making request to ~p exited with ~p~n"
|
||||
"Request contents:~n~n"
|
||||
"~s",
|
||||
Formatted = io_lib:format(Format, [TS, PID, Node, Info, R]),
|
||||
Message = unicode:characters_to_list(Formatted),
|
||||
ok = gen_server:reply(From, {error, Message}),
|
||||
State#s{fetchers = Remaining};
|
||||
false ->
|
||||
Unexpected = {'DOWN', Mon, process, PID, Info},
|
||||
ok = log(warning, "Unexpected info: ~w", [Unexpected]),
|
||||
State
|
||||
end.
|
||||
|
||||
|
||||
|
||||
|
||||
%%% OTP Service Functions
|
||||
|
||||
code_change(_, State, _) ->
|
||||
{ok, State}.
|
||||
|
||||
|
||||
terminate(_, _) ->
|
||||
ok.
|
||||
|
||||
|
||||
|
||||
%%% Doer Functions
|
||||
|
||||
do_request(_, From, State = #s{ae_nodes = {[], []}}) ->
|
||||
ok = gen_server:reply(From, {error, no_nodes}),
|
||||
State;
|
||||
do_request(Request,
|
||||
From,
|
||||
State = #s{fetchers = Fetchers,
|
||||
ae_nodes = {[Node | Rest], Used},
|
||||
timeout = Timeout}) ->
|
||||
Now = erlang:system_time(nanosecond),
|
||||
Fetcher = fun() -> vanillae_fetcher:connect(Node, Request, From, Timeout) end,
|
||||
{PID, Mon} = spawn_monitor(Fetcher),
|
||||
New = #fetcher{pid = PID,
|
||||
mon = Mon,
|
||||
time = Now,
|
||||
node = Node,
|
||||
from = From,
|
||||
req = Request},
|
||||
State#s{fetchers = [New | Fetchers], ae_nodes = {Rest, [Node | Used]}};
|
||||
do_request(Request, From, State = #s{ae_nodes = {[], Used}}) ->
|
||||
Fresh = lists:reverse(Used),
|
||||
do_request(Request, From, State#s{ae_nodes = {Fresh, []}}).
|
||||
Reference in New Issue
Block a user