Move moving around...

zomp/lib/otpr-zx/0.1.0/src/zx_conn.erl

@@ -69,3 +69,203 @@ connect(Parent, Debug, {Host, Port}) ->
 
 confirm(Parent, Debug, Socket) ->
 
+
+-spec connect_user(realm()) -> gen_tcp:socket() | no_return().
+%% @private
+%% Connect to a given realm, whatever method is required.
+
+connect_user(Realm) ->
+    ok = log(info, "Connecting to realm ~ts...", [Realm]),
+    Hosts = 
+        case file:consult(zx_lib:hosts_cache_file(Realm)) of
+            {ok, Cached}    -> Cached;
+            {error, enoent} -> []
+        end,
+    connect_user(Realm, Hosts).
+
+
+-spec connect_user(realm(), [host()]) -> gen_tcp:socket() | no_return().
+%% @private
+%% Try to connect to a subordinate host, if there are none then connect to prime.
+
+connect_user(Realm, []) ->
+    {Host, Port} = zx_lib:get_prime(Realm),
+    HostString =
+        case io_lib:printable_unicode_list(Host) of
+            true  -> Host;
+            false -> inet:ntoa(Host)
+        end,
+    ok = log(info, "Trying prime at ~ts:~160tp", [HostString, Port]),
+    case gen_tcp:connect(Host, Port, connect_options(), 5000) of
+        {ok, Socket} ->
+            confirm_user(Realm, Socket, []);
+        {error, Error} ->
+            ok = log(warning, "Connection problem with prime: ~tp", [Error]),
+            halt(0)
+    end;
+connect_user(Realm, Hosts = [Node = {Host, Port} | Rest]) ->
+    ok = log(info, "Trying node at ~ts:~tp", [inet:ntoa(Host), Port]),
+    case gen_tcp:connect(Host, Port, connect_options(), 5000) of
+        {ok, Socket} ->
+            confirm_user(Realm, Socket, Hosts);
+        {error, Error} ->
+            ok = log(warning, "Connection problem with ~tp: ~tp", [Node, Error]),
+            connect_user(Realm, Rest)
+    end.
+
+
+-spec confirm_user(Realm, Socket, Hosts) -> Socket | no_return()
+    when Realm  :: realm(),
+         Socket :: gen_tcp:socket(),
+         Hosts  :: [host()].
+%% @private
+%% Confirm the zomp node can handle "OTPR USER 1" and is accepting connections or try
+%% another node.
+
+confirm_user(Realm, Socket, Hosts) ->
+    {ok, {Addr, Port}} = inet:peername(Socket),
+    Host = inet:ntoa(Addr),
+    ok = gen_tcp:send(Socket, <<"OTPR USER 1">>),
+    receive
+        {tcp, Socket, Bin} ->
+            case binary_to_term(Bin) of
+                ok ->
+                    ok = log(info, "Connected to ~ts:~p", [Host, Port]),
+                    confirm_serial(Realm, Socket, Hosts);
+                {redirect, Next} ->
+                    ok = log(info, "Redirected..."),
+                    ok = disconnect(Socket),
+                    connect_user(Realm, Next ++ Hosts)
+            end;
+        {tcp_closed, Socket} ->
+            halt_on_unexpected_close()
+        after 5000 ->
+            ok = log(warning, "Host ~ts:~p timed out.", [Host, Port]),
+            ok = disconnect(Socket),
+            connect_user(Realm, Hosts)
+    end.
+
+
+-spec confirm_serial(Realm, Socket, Hosts) -> Socket | no_return()
+    when Realm  :: realm(),
+         Socket :: gen_tcp:socket(),
+         Hosts  :: [host()].
+%% @private
+%% Confirm that the connected host has a valid serial for the realm zx is trying to
+%% reach, and if not retry on another node.
+
+confirm_serial(Realm, Socket, Hosts) ->
+    SerialFile = filename:join(zx_lib:zomp_home(), "realm.serials"),
+    Serials =
+        case file:consult(SerialFile) of
+            {ok, Ss}        -> Ss;
+            {error, enoent} -> []
+        end,
+    Serial =
+        case lists:keyfind(Realm, 1, Serials) of
+            false      -> 0;
+            {Realm, S} -> S
+        end,
+    ok = send(Socket, {latest, Realm}),
+    receive
+        {tcp, Socket, Bin} ->
+            case binary_to_term(Bin) of
+                {ok, Serial} ->
+                    ok = log(info, "Node's serial same as ours."),
+                    Socket;
+                {ok, Current} when Current > Serial ->
+                    ok = log(info, "Node's serial newer than ours. Storing."),
+                    NewSerials = lists:keystore(Realm, 1, Serials, {Realm, Current}),
+                    {ok, Host} = inet:peername(Socket),
+                    CacheFile = zx_lib:hosts_cache_file(Realm),
+                    ok = zx_lib:write_terms(CacheFile, [Host | Hosts]),
+                    ok = zx_lib:write_terms(SerialFile, NewSerials),
+                    Socket;
+                {ok, Current} when Current < Serial ->
+                    log(info, "Our serial: ~tp, node serial: ~tp.", [Serial, Current]),
+                    ok = log(info, "Node's serial older than ours. Trying another."),
+                    ok = disconnect(Socket),
+                    connect_user(Realm, Hosts);
+                {error, bad_realm} ->
+                    ok = log(info, "Node is no longer serving realm. Trying another."),
+                    ok = disconnect(Socket),
+                    connect_user(Realm, Hosts)
+            end;
+        {tcp_closed, Socket} ->
+            halt_on_unexpected_close()
+        after 5000 ->
+            ok = log(info, "Host timed out on confirm_serial. Trying another."),
+            ok = disconnect(Socket),
+            connect_user(Realm, Hosts)
+    end.
+    
+
+
+-spec halt_on_unexpected_close() -> no_return().
+
+halt_on_unexpected_close() ->
+    ok = log(warning, "Socket closed unexpectedly."),
+    halt(1).
+
+
+
+
+
+
+-spec fetch(Socket, PackageID) -> Result
+    when Socket    :: gen_tcp:socket(),
+         PackageID :: package_id(),
+         Result    :: ok.
+%% @private
+%% Download a package to the local cache.
+
+fetch(Socket, PackageID) ->
+    {ok, LatestID} = request_zrp(Socket, PackageID),
+    ok = receive_zrp(Socket, LatestID),
+    log(info, "Fetched ~ts", [package_string(LatestID)]).
+
+
+-spec request_zrp(Socket, PackageID) -> Result
+    when Socket    :: gen_tcp:socket(),
+         PackageID :: package_id(),
+         Result    :: {ok, Latest :: package_id()}
+                    | {error, Reason :: timeout | term()}.
+
+request_zrp(Socket, PackageID) ->
+    ok = send(Socket, {fetch, PackageID}),
+    receive
+        {tcp, Socket, Bin} ->
+            case binary_to_term(Bin) of
+                {sending, LatestID} ->
+                    {ok, LatestID};
+                Error = {error, Reason} ->
+                    PackageString = package_string(PackageID),
+                    Message = "Error receiving package ~ts: ~tp",
+                    ok = log(info, Message, [PackageString, Reason]),
+                    Error
+            end;
+        {tcp_closed, Socket} ->
+            halt_on_unexpected_close()
+        after 60000 ->
+            {error, timeout}
+   end.
+
+
+-spec receive_zrp(Socket, PackageID) -> Result
+    when Socket    :: gen_tcp:socket(),
+         PackageID :: package_id(),
+         Result    :: ok | {error, timeout}.
+
+receive_zrp(Socket, PackageID) ->
+    receive
+        {tcp, Socket, Bin} ->
+            ZrpPath = filename:join("zrp", zx_lib:namify_zrp(PackageID)),
+            ok = file:write_file(ZrpPath, Bin),
+            ok = send(Socket, ok),
+            log(info, "Wrote ~ts", [ZrpPath]);
+        {tcp_closed, Socket} ->
+            halt_on_unexpected_close()
+        after 60000 ->
+            ok = log(error, "Timeout in socket receive for ~tp", [PackageID]),
+            {error, timeout}
+    end.

zomp/lib/otpr-zx/0.1.0/src/zx_lib.erl

@@ -328,26 +328,25 @@ version_to_string(_) ->
 
 package_id(String) ->
     case dash_split(String) of
-        [Realm, Name, VersionString] ->
+        {ok, [Realm, Name, VersionString]} ->
             package_id(Realm, Name, VersionString);
-        [A, B] ->
+        {ok, [A, B]} ->
             case valid_lower0_9(B) of
                 true  -> package_id(A, B, "");
                 false -> package_id("otpr", A, B)
             end;
-        [Name] ->
+        {ok, [Name]} ->
             package_id("otpr", Name, "");
-        _ ->
+        error ->
             {error, invalid_package_string}
     end.
 
 
--spec dash_split(string()) -> [string()] | error.
+-spec dash_split(string()) -> {ok, [string()]} | error.
 %% @private
 %% An explicit, strict token split that ensures invalid names with leading, trailing or
 %% double dashes don't slip through (a problem discovered with using string:tokens/2
 %% and string:lexemes/2.
-%% Intended only as a helper function for package_id/1
 
 dash_split(String) ->
     dash_split(String, "", []).
@@ -360,7 +359,7 @@ dash_split([Char | Rest], Acc, Elements) ->
     dash_split(Rest, [Char | Acc], Elements);
 dash_split("", Acc, Elements) ->
     Element = lists:reverse(Acc),
-    lists:reverse([Element | Elements]);
+    {ok, lists:reverse([Element | Elements])};
 dash_split(_, _, _) ->
     error.
 
@@ -521,3 +520,33 @@ latest_compatible(Version, Versions) ->
 
 installed(PackageID) ->
     filelib:is_dir(package_dir(PackageID)).
+
+
+
+
+-spec realm_conf(Realm) -> RealmFileName
+    when Realm :: string(),
+         RealmFileName :: file:filename().
+%% @private
+%% Take a realm name, and return the name of the realm filename that would result.
+
+realm_conf(Realm) ->
+    Realm ++ ".realm".
+
+
+-spec load_realm_conf(Realm) -> Result
+    when Realm     :: realm(),
+         Result    :: {ok, RealmConf}
+                    | {error, Reason},
+         RealmConf :: list(),
+         Reason    :: badarg
+                    | terminated
+                    | system_limit
+                    | file:posix()
+                    | {Line :: integer(), Mod :: module(), Cause :: term()}.
+%% @private
+%% Load the config for the given realm or halt with an error.
+
+load_realm_conf(Realm) ->
+    Path = filename:join(zx_lib:zomp_home(), realm_conf(Realm)),
+    file:consult(Path).