WIP but works -- if you add it to your sup tree directly.

2022-09-13 20:22:02 +09:00
parent 962a817341
commit 5a6a6fb6c3
9 changed files with 1472 additions and 712 deletions
@@ -4,6 +4,5 @@
              {included_applications,[]},
              {applications,[stdlib,kernel]},
              {vsn,"0.1.0"},
-              {modules,[v_client,v_client_man,v_client_sup,v_clients,v_sup,
+              {modules,[vanillae,vanillae_fetcher,vanillae_man]},
                        vanillae]},
              {mod,{vanillae,[]}}]}.
@@ -1,204 +0,0 @@
 %%% @doc
 %%% Vanillae for Erlang Client
 %%%
 %%% An extremely naive (currently Telnet) client handler.
 %%% Unlike other modules that represent discrete processes, this one does not adhere
 %%% to any OTP behavior. It does, however, adhere to OTP.
 %%%
 %%% In some cases it is more comfortable to write socket handlers or a certain
 %%% category of state machines as "pure" Erlang processes. This approach is made
 %%% OTP-able by use of the proc_lib module, which is the underlying library used
 %%% to write the stdlib's behaviors like gen_server, gen_statem, gen_fsm, etc.
 %%%
 %%% http://erlang.org/doc/design_principles/spec_proc.html
 %%% @end
 -module(v_client).
 -vsn("0.1.0").
 -author("Craig Everett <ceverett@tsuriai.jp>").
 -copyright("Craig Everett <ceverett@tsuriai.jp>").
 -license("MIT").
 -export([start/1]).
 -export([start_link/1, init/2]).
 -export([system_continue/3, system_terminate/4,
         system_get_state/1, system_replace_state/2]).
 %%% Type and Record Definitions
 -record(s, {socket = none :: none | gen_tcp:socket()}).
 %% An alias for the state record above. Aliasing state can smooth out annoyances
 %% that can arise from using the record directly as its own type all over the code.
 -type state() :: #s{}.
 %%% Service Interface
 -spec start(ListenSocket) -> Result
    when ListenSocket :: gen_tcp:socket(),
         Result       :: {ok, pid()}
                       | {error, Reason},
         Reason       :: {already_started, pid()}
                       | {shutdown, term()}
                       | term().
 %% @private
 %% How the v_client_man or a prior v_client kicks things off.
 %% This is called in the context of v_client_man or the prior v_client.
 start(ListenSocket) ->
    v_client_sup:start_acceptor(ListenSocket).
 -spec start_link(ListenSocket) -> Result
    when ListenSocket :: gen_tcp:socket(),
         Result       :: {ok, pid()}
                       | {error, Reason},
         Reason       :: {already_started, pid()}
                       | {shutdown, term()}
                       | term().
 %% @private
 %% This is called by the v_client_sup. While start/1 is called to iniate a startup
 %% (essentially requesting a new worker be started by the supervisor), this is
 %% actually called in the context of the supervisor.
 start_link(ListenSocket) ->
    proc_lib:start_link(?MODULE, init, [self(), ListenSocket]).
 -spec init(Parent, ListenSocket) -> no_return()
    when Parent       :: pid(),
         ListenSocket :: gen_tcp:socket().
 %% @private
 %% This is the first code executed in the context of the new worker itself.
 %% This function does not have any return value, as the startup return is
 %% passed back to the supervisor by calling proc_lib:init_ack/2.
 %% We see the initial form of the typical arity-3 service loop form here in the
 %% call to listen/3.
 init(Parent, ListenSocket) ->
    ok = io:format("~p Listening.~n", [self()]),
    Debug = sys:debug_options([]),
    ok = proc_lib:init_ack(Parent, {ok, self()}),
    listen(Parent, Debug, ListenSocket).
 -spec listen(Parent, Debug, ListenSocket) -> no_return()
    when Parent       :: pid(),
         Debug        :: [sys:dbg_opt()],
         ListenSocket :: gen_tcp:socket().
 %% @private
 %% This function waits for a TCP connection. The owner of the socket is still
 %% the v_client_man (so it can still close it on a call to v_client_man:ignore/0),
 %% but the only one calling gen_tcp:accept/1 on it is this process. Closing the socket
 %% is one way a manager process can gracefully unblock child workers that are blocking
 %% on a network accept.
 %%
 %% Once it makes a TCP connection it will call start/1 to spawn its successor.
 listen(Parent, Debug, ListenSocket) ->
    case gen_tcp:accept(ListenSocket) of
        {ok, Socket} ->
            {ok, _} = start(ListenSocket),
            {ok, Peer} = inet:peername(Socket),
            ok = io:format("~p Connection accepted from: ~p~n", [self(), Peer]),
            ok = v_client_man:enroll(),
            State = #s{socket = Socket},
            loop(Parent, Debug, State);
        {error, closed} ->
            ok = io:format("~p Retiring: Listen socket closed.~n", [self()]),
            exit(normal)
     end.
 -spec loop(Parent, Debug, State) -> no_return()
    when Parent :: pid(),
         Debug  :: [sys:dbg_opt()],
         State  :: state().
 %% @private
 %% The service loop itself. This is the service state. The process blocks on receive
 %% of Erlang messages, TCP segments being received themselves as Erlang messages.
 loop(Parent, Debug, State = #s{socket = Socket}) ->
    ok = inet:setopts(Socket, [{active, once}]),
    receive
        {tcp, Socket, <<"bye\r\n">>} ->
            ok = io:format("~p Client saying goodbye. Bye!~n", [self()]),
            ok = gen_tcp:send(Socket, "Bye!\r\n"),
            ok = gen_tcp:shutdown(Socket, read_write),
            exit(normal);
        {tcp, Socket, Message} ->
            ok = io:format("~p received: ~tp~n", [self(), Message]),
            ok = v_client_man:echo(Message),
            loop(Parent, Debug, State);
        {relay, Sender, Message} when Sender == self() ->
            ok = gen_tcp:send(Socket, ["Message from YOU: ", Message]),
            loop(Parent, Debug, State);
        {relay, Sender, Message} ->
            From = io_lib:format("Message from ~tp: ", [Sender]),
            ok = gen_tcp:send(Socket, [From, Message]),
            loop(Parent, Debug, State);
        {tcp_closed, Socket} ->
            ok = io:format("~p Socket closed, retiring.~n", [self()]),
            exit(normal);
        {system, From, Request} ->
            sys:handle_system_msg(Request, From, Parent, ?MODULE, Debug, State);
        Unexpected ->
            ok = io:format("~p Unexpected message: ~tp", [self(), Unexpected]),
            loop(Parent, Debug, State)
    end.
 -spec system_continue(Parent, Debug, State) -> no_return()
    when Parent :: pid(),
         Debug  :: [sys:dbg_opt()],
         State  :: state().
 %% @private
 %% The function called by the OTP internal functions after a system message has been
 %% handled. If the worker process has several possible states this is one place
 %% resumption of a specific state can be specified and dispatched.
 system_continue(Parent, Debug, State) ->
    loop(Parent, Debug, State).
 -spec system_terminate(Reason, Parent, Debug, State) -> no_return()
    when Reason :: term(),
         Parent :: pid(),
         Debug  :: [sys:dbg_opt()],
         State  :: state().
 %% @private
 %% Called by the OTP inner bits to allow the process to terminate gracefully.
 %% Exactly when and if this is callback gets called is specified in the docs:
 %% See: http://erlang.org/doc/design_principles/spec_proc.html#msg
 system_terminate(Reason, _Parent, _Debug, _State) ->
    exit(Reason).
 -spec system_get_state(State) -> {ok, State}
    when State :: state().
 %% @private
 %% This function allows the runtime (or anything else) to inspect the running state
 %% of the worker process at any arbitrary time.
 system_get_state(State) -> {ok, State}.
 -spec system_replace_state(StateFun, State) -> {ok, NewState, State}
    when StateFun :: fun(),
         State    :: state(),
         NewState :: term().
 %% @private
 %% This function allows the system to update the process state in-place. This is most
 %% useful for state transitions between code types, like when performing a hot update
 %% (very cool, but sort of hard) or hot patching a running system (living on the edge!).
 system_replace_state(StateFun, State) ->
    {ok, StateFun(State), State}.
@@ -1,298 +0,0 @@
 %%% @doc
 %%% Vanillae for Erlang Client Manager
 %%%
 %%% This is the "manager" part of the service->worker pattern.
 %%% It keeps track of who is connected and can act as a router among the workers.
 %%% Having this process allows us to abstract and customize service-level concepts
 %%% (the high-level ideas we care about in terms of solving an external problem in the
 %%% real world) and keep them separate from the lower-level details of supervision that
 %%% OTP should take care of for us.
 %%% @end
 -module(v_client_man).
 -vsn("0.1.0").
 -behavior(gen_server).
 -author("Craig Everett <ceverett@tsuriai.jp>").
 -copyright("Craig Everett <ceverett@tsuriai.jp>").
 -license("MIT").
 -export([listen/1, ignore/0]).
 -export([enroll/0, echo/1]).
 -export([start_link/0]).
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         code_change/3, terminate/2]).
 %%% Type and Record Definitions
 -record(s, {port_num = none :: none | inet:port_number(),
            listener = none :: none | gen_tcp:socket(),
            clients  = []   :: [pid()]}).
 -type state()   :: #s{}.
 %%% Service Interface
 -spec listen(PortNum) -> Result
    when PortNum :: inet:port_number(),
         Result  :: ok
                  | {error, Reason},
         Reason  :: {listening, inet:port_number()}.
 %% @doc
 %% Tell the service to start listening on a given port.
 %% Only one port can be listened on at a time in the current implementation, so
 %% an error is returned if the service is already listening.
 listen(PortNum) ->
    gen_server:call(?MODULE, {listen, PortNum}).
 -spec ignore() -> ok.
 %% @doc
 %% Tell the service to stop listening.
 %% It is not an error to call this function when the service is not listening.
 ignore() ->
    gen_server:cast(?MODULE, ignore).
 %%% Client Process Interface
 -spec enroll() -> ok.
 %% @doc
 %% Clients register here when they establish a connection.
 %% Other processes can enroll as well.
 enroll() ->
    gen_server:cast(?MODULE, {enroll, self()}).
 -spec echo(Message) -> ok
    when Message :: string().
 %% @doc
 %% The function that tells the manager to broadcast a message to all clients.
 %% This can broadcast arbitrary strings to clients from non-clients as well.
 echo(Message) ->
    gen_server:cast(?MODULE, {echo, Message, self()}).
 %%% 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
 -spec handle_call(Message, From, State) -> Result
    when Message  :: term(),
         From     :: {pid(), reference()},
         State    :: state(),
         Result   :: {reply, Response, NewState}
                   | {noreply, State},
         Response :: ok
                   | {error, {listening, inet:port_number()}},
         NewState :: state().
 %% @private
 %% The gen_server:handle_call/3 callback.
 %% See: http://erlang.org/doc/man/gen_server.html#Module:handle_call-3
 handle_call({listen, PortNum}, _, State) ->
    {Response, NewState} = do_listen(PortNum, State),
    {reply, Response, NewState};
 handle_call(Unexpected, From, State) ->
    ok = io:format("~p Unexpected call from ~tp: ~tp~n", [self(), From, Unexpected]),
    {noreply, State}.
 -spec handle_cast(Message, State) -> {noreply, NewState}
    when Message  :: term(),
         State    :: state(),
         NewState :: state().
 %% @private
 %% The gen_server:handle_cast/2 callback.
 %% See: http://erlang.org/doc/man/gen_server.html#Module:handle_cast-2
 handle_cast({enroll, Pid}, State) ->
    NewState = do_enroll(Pid, State),
    {noreply, NewState};
 handle_cast({echo, Message, Sender}, State) ->
    ok = do_echo(Message, Sender, State),
    {noreply, State};
 handle_cast(ignore, State) ->
    NewState = do_ignore(State),
    {noreply, NewState};
 handle_cast(Unexpected, State) ->
    ok = io:format("~p Unexpected cast: ~tp~n", [self(), Unexpected]),
    {noreply, State}.
 -spec handle_info(Message, State) -> {noreply, NewState}
    when Message  :: term(),
         State    :: state(),
         NewState :: state().
 %% @private
 %% The gen_server:handle_info/2 callback.
 %% See: http://erlang.org/doc/man/gen_server.html#Module:handle_info-2
 handle_info({'DOWN', Mon, process, Pid, Reason}, State) ->
    NewState = handle_down(Mon, Pid, Reason, State),
    {noreply, NewState};
 handle_info(Unexpected, State) ->
    ok = io:format("~p Unexpected info: ~tp~n", [self(), Unexpected]),
    {noreply, State}.
 %%% OTP Service Functions
 -spec code_change(OldVersion, State, Extra) -> Result
    when OldVersion :: {down, Version} | Version,
         Version    :: term(),
         State      :: state(),
         Extra      :: term(),
         Result     :: {ok, NewState}
                     | {error, Reason :: term()},
         NewState   :: state().
 %% @private
 %% The gen_server:code_change/3 callback.
 %% See: http://erlang.org/doc/man/gen_server.html#Module:code_change-3
 code_change(_, State, _) ->
    {ok, State}.
 -spec terminate(Reason, State) -> no_return()
    when Reason :: normal
                 | shutdown
                 | {shutdown, term()}
                 | term(),
         State  :: state().
 %% @private
 %% The gen_server:terminate/2 callback.
 %% See: http://erlang.org/doc/man/gen_server.html#Module:terminate-2
 terminate(_, _) ->
    ok.
 %%% Doer Functions
 -spec do_listen(PortNum, State) -> {Result, NewState}
    when PortNum  :: inet:port_number(),
         State    :: state(),
         Result   :: ok
                   | {error, Reason :: {listening, inet:port_number()}},
         NewState :: state().
 %% @private
 %% The "doer" procedure called when a "listen" message is received.
 do_listen(PortNum, State = #s{port_num = none}) ->
    SocketOptions =
        [inet6,
         {packet,    line},
         {active,    once},
         {mode,      binary},
         {keepalive, true},
         {reuseaddr, true}],
    {ok, Listener} = gen_tcp:listen(PortNum, SocketOptions),
    {ok, _} = v_client:start(Listener),
    {ok, State#s{port_num = PortNum, listener = Listener}};
 do_listen(_, State = #s{port_num = PortNum}) ->
    ok = io:format("~p Already listening on ~p~n", [self(), PortNum]),
    {{error, {listening, PortNum}}, State}.
 -spec do_ignore(State) -> NewState
    when State    :: state(),
         NewState :: state().
 %% @private
 %% The "doer" procedure called when an "ignore" message is received.
 do_ignore(State = #s{listener = none}) ->
    State;
 do_ignore(State = #s{listener = Listener}) ->
    ok = gen_tcp:close(Listener),
    State#s{port_num = none, listener = none}.
 -spec do_enroll(Pid, State) -> NewState
    when Pid      :: pid(),
         State    :: state(),
         NewState :: state().
 do_enroll(Pid, State = #s{clients = Clients}) ->
    case lists:member(Pid, Clients) of
        false ->
            Mon = monitor(process, Pid),
            ok = io:format("Monitoring ~tp @ ~tp~n", [Pid, Mon]),
            State#s{clients = [Pid | Clients]};
        true ->
            State
    end.
 -spec do_echo(Message, Sender, State) -> ok
    when Message :: string(),
         Sender  :: pid(),
         State   :: state().
 %% @private
 %% The "doer" procedure called when an "echo" message is received.
 do_echo(Message, Sender, #s{clients = Clients}) ->
    Send = fun(Client) -> Client ! {relay, Sender, Message} end,
    lists:foreach(Send, Clients).
 -spec handle_down(Mon, Pid, Reason, State) -> NewState
    when Mon      :: reference(),
         Pid      :: pid(),
         Reason   :: term(),
         State    :: state(),
         NewState :: state().
 %% @private
 %% Deal with monitors. When a new process enrolls as a client a monitor is set and
 %% the process is added to the client list. When the process terminates we receive
 %% a 'DOWN' message from the monitor. More sophisticated work managers typically have
 %% an "unenroll" function, but this echo service doesn't need one.
 handle_down(Mon, Pid, Reason, State = #s{clients = Clients}) ->
    case lists:member(Pid, Clients) of
        true ->
            NewClients = lists:delete(Pid, Clients),
            State#s{clients = NewClients};
        false ->
            Unexpected = {'DOWN', Mon, process, Pid, Reason},
            ok = io:format("~p Unexpected info: ~tp~n", [self(), Unexpected]),
            State
    end.
@@ -1,70 +0,0 @@
 %%% @doc
 %%% Vanillae for Erlang Client Supervisor
 %%%
 %%% This process supervises the client socket handlers themselves. It is a peer of the
 %%% v_client_man (the manager interface to this network service component),
 %%% and a child of the supervisor named v_clients.
 %%%
 %%% Because we don't know (or care) how many client connections the server may end up
 %%% handling this is a simple_one_for_one supervisor which can spawn and manage as
 %%% many identically defined workers as required, but cannot supervise any other types
 %%% of processes (one of the tradeoffs of the "simple" in `simple_one_for_one').
 %%%
 %%% http://erlang.org/doc/design_principles/sup_princ.html#id79244
 %%% @end
 -module(v_client_sup).
 -vsn("0.1.0").
 -behaviour(supervisor).
 -author("Craig Everett <ceverett@tsuriai.jp>").
 -copyright("Craig Everett <ceverett@tsuriai.jp>").
 -license("MIT").
 -export([start_acceptor/1]).
 -export([start_link/0]).
 -export([init/1]).
 -spec start_acceptor(ListenSocket) -> Result
    when ListenSocket :: gen_tcp:socket(),
         Result       :: {ok, pid()}
                       | {error, Reason},
         Reason       :: {already_started, pid()}
                       | {shutdown, term()}
                       | term().
 %% @private
 %% Spawns the first listener at the request of the v_client_man when
 %% vanillae:listen/1 is called, or the next listener at the request of the
 %% currently listening v_client when a connection is made.
 %%
 %% Error conditions, supervision strategies and other important issues are
 %% explained in the supervisor module docs:
 %% http://erlang.org/doc/man/supervisor.html
 start_acceptor(ListenSocket) ->
    supervisor:start_child(?MODULE, [ListenSocket]).
 -spec start_link() -> {ok, pid()}.
 %% @private
 %% This supervisor's own start function.
 start_link() ->
    supervisor:start_link({local, ?MODULE}, ?MODULE, none).
 -spec init(none) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
 %% @private
 %% The OTP init/1 function.
 init(none) ->
    RestartStrategy = {simple_one_for_one, 1, 60},
    Client    = {v_client,
                 {v_client, start_link, []},
                 temporary,
                 brutal_kill,
                 worker,
                 [v_client]},
    {ok, {RestartStrategy, [Client]}}.
@@ -1,48 +0,0 @@
 %%% @doc
 %%% Vanillae for Erlang Client Service Supervisor
 %%%
 %%% This is the service-level supervisor of the system. It is the parent of both the
 %%% client connection handlers and the client manager (which manages the client
 %%% connection handlers). This is the child of v_sup.
 %%%
 %%% See: http://erlang.org/doc/apps/kernel/application.html
 %%% @end
 -module(v_clients).
 -vsn("0.1.0").
 -behavior(supervisor).
 -author("Craig Everett <ceverett@tsuriai.jp>").
 -copyright("Craig Everett <ceverett@tsuriai.jp>").
 -license("MIT").
 -export([start_link/0]).
 -export([init/1]).
 -spec start_link() -> {ok, pid()}.
 %% @private
 %% This supervisor's own start function.
 start_link() ->
  supervisor:start_link({local, ?MODULE}, ?MODULE, none).
 -spec init(none) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
 %% @private
 %% The OTP init/1 function.
 init(none) ->
    RestartStrategy = {rest_for_one, 1, 60},
    ClientMan = {v_client_man,
                 {v_client_man, start_link, []},
                 permanent,
                 5000,
                 worker,
                 [v_client_man]},
    ClientSup = {v_client_sup,
                 {v_client_sup, start_link, []},
                 permanent,
                 5000,
                 supervisor,
                 [v_client_sup]},
    Children  = [ClientSup, ClientMan],
    {ok, {RestartStrategy, Children}}.
@@ -1,46 +0,0 @@
 %%% @doc
 %%% Vanillae for Erlang Top-level Supervisor
 %%%
 %%% The very top level supervisor in the system. It only has one service branch: the
 %%% client handling service. In a more complex system the client handling service would
 %%% only be one part of a larger system. Were this a game system, for example, the
 %%% item data management service would be a peer, as would a login credential provision
 %%% service, game world event handling, and so on.
 %%%
 %%% See: http://erlang.org/doc/design_principles/applications.html
 %%% See: http://zxq9.com/archives/1311
 %%% @end
 -module(v_sup).
 -vsn("0.1.0").
 -behaviour(supervisor).
 -author("Craig Everett <ceverett@tsuriai.jp>").
 -copyright("Craig Everett <ceverett@tsuriai.jp>").
 -license("MIT").
 -export([start_link/0]).
 -export([init/1]).
 -spec start_link() -> {ok, pid()}.
 %% @private
 %% This supervisor's own start function.
 start_link() ->
  supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 -spec init([]) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
 %% @private
 %% The OTP init/1 function.
 init([]) ->
    RestartStrategy = {one_for_one, 1, 60},
    Clients   = {v_clients,
                 {v_clients, start_link, []},
                 permanent,
                 5000,
                 supervisor,
                 [v_clients]},
    Children  = [Clients],
    {ok, {RestartStrategy, Children}}.
@@ -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, []}}).