pharpend/zx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed up zx_conn a bit

Browse Source
This commit is contained in:
Craig Everett
2018-06-05 18:56:14 +09:00
parent 3be0803721
commit 44d3bb37fc
8 changed files with 226 additions and 424 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zomp/lib/otpr/zx/0.1.0/src/zx.erl
+8 -3
View File
@@ -119,6 +119,9 @@ do(["list", "packages", Realm]) ->
do(["list", "versions", PackageName]) ->
ok = start(),
done(zx_local:list_versions(PackageName));
do(["latest", PackageString]) ->
ok = start(),
done(zx_local:latest(PackageString));
do(["import", "realm", RealmFile]) ->
done(zx_local:import_realm(RealmFile));
do(["drop", "dep", PackageString]) ->
@@ -389,12 +392,12 @@ install(PackageString) ->
install(PackageString, ID) ->
receive
{z_result, ID, done} ->
{result, ID, done} ->
ok;
{z_result, ID, {hops, Count}} ->
{result, ID, {hops, Count}} ->
ok = log(info, "~ts ~w hops away.", [PackageString, Count]),
install(PackageString, ID);
{z_result, ID, {error, Reason}} ->
{result, ID, {error, Reason}} ->
{error, Reason, 1}
after 60000 ->
{error, timeout, 62}
@@ -625,6 +628,8 @@ usage() ->
" zx accept PackageID~n"
" zx create realm~n"
" zx create realmfile Realm~n"
" zx takeover Realm~n"
" zx abdicate Realm~n"
"~n"
"Where~n"
" PackageID :: A string of the form Realm-Name[-Version]~n"
zomp/lib/otpr/zx/0.1.0/src/zx_conn.erl
+183 -305
View File
@@ -11,9 +11,6 @@
-copyright("Craig Everett <zxq9@zxq9.com>").
-license("GPL-3.0").
% FIXME
-compile(export_all).
-export([subscribe/2, unsubscribe/2, fetch/3, query/3]).
-export([start/1, stop/1]).
-export([start_link/1, init/2]).
@@ -21,13 +18,6 @@
-include("zx_logger.hrl").
%%% Types
%-type incoming() :: ping
% | {sub, Channel :: term(), Message :: term()}
% | term().
%%% Interface
-spec subscribe(Conn, Package) -> ok
@@ -52,6 +42,10 @@ unsubscribe(Conn, Package) ->
when Conn :: pid(),
ID :: zx_daemon:id(),
Object :: zx_daemon:object().
%% @doc
%% This requests a package be fetched from upstream.
%% Only to be called by zx_daemon.
%% Results must be returned with the ID via zx_daemon:result/2.
fetch(Conn, ID, Object) ->
Conn ! {fetch, ID, Object},
@@ -62,6 +56,10 @@ fetch(Conn, ID, Object) ->
when Conn :: pid(),
ID :: zx_daemon:id(),
Action :: zx_daemon:action().
%% @doc
%% Wraps any legal query.
%% Only to be called by zx_daemon.
%% Results must be returned with the ID via zx_daemon:result/2.
query(Conn, ID, Action) ->
Conn ! {query, ID, Action},
@@ -142,43 +140,30 @@ connect(Parent, Debug, {Host, Port}) ->
Debug :: [sys:dbg_opt()],
Socket :: gen_tcp:socket().
%% @private
%% Confirm the zomp node can handle "OTPR USER 1" and is accepting connections or try
%% Confirm the zomp node can handle "LEAF 1:" and is accepting connections or try
%% another node.
confirm_service(Parent, Debug, Socket) ->
ok = gen_tcp:send(Socket, <<"OTPR USER 1">>),
ok = gen_tcp:send(Socket, <<"LEAF 1:">>),
receive
{tcp, Socket, Bin} ->
case binary_to_term(Bin, [safe]) of
ok ->
query_realms(Parent, Debug, Socket);
{redirect, Hosts} ->
ok = zx_daemon:report({redirect, Hosts}),
ok = zx_net:disconnect(Socket),
terminate()
end;
{tcp_closed, Socket} ->
handle_unexpected_close()
after 5000 ->
handle_timeout(Socket)
end.
-spec query_realms(Parent, Debug, Socket) -> no_return()
when Parent :: pid(),
Debug :: [sys:dbg_opt()],
Socket :: gen_tcp:socket().
%% @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.
query_realms(Parent, Debug, Socket) ->
ok = zx_net:send(Socket, list),
receive
{tcp, Socket, Bin} ->
{ok, Realms} = binary_to_term(Bin, [safe]),
{tcp, Socket, <<0:8, RealmsBin/binary>>} ->
{ok, Realms} = zx_lib:b_to_ts(RealmsBin),
ok = zx_daemon:report({connected, Realms}),
loop(Parent, Debug, Socket);
{tcp, Socket, <<1:8, HostsBin/binary>>} ->
{ok, Hosts} = zx_lib:b_to_ts(HostsBin),
ok = zx_daemon:report({redirect, Hosts}),
ok = zx_net:disconnect(Socket),
terminate();
{tcp, Socket, <<2:8>>} ->
ok = zx_daemon:report({use_version, no_version}),
terminate();
{tcp, Socket, <<2:8, Version:16>>} ->
ok = zx_daemon:report({use_version, Version}),
terminate();
{tcp, Socket, <<3:8, Reason:16>>} ->
ok = zx_daemon:report({no_service, Reason}),
terminate();
{tcp_closed, Socket} ->
handle_unexpected_close()
after 5000 ->
@@ -200,28 +185,40 @@ query_realms(Parent, Debug, Socket) ->
%% of order or else whatever sequenced communication was happening will be corrupted.
loop(Parent, Debug, Socket) ->
ok = inet:setopts(Socket, [{active, once}]),
receive
{tcp, Socket, Bin} ->
ok = handle_message(Socket, Bin),
ok = inet:setopts(Socket, [{active, once}]),
{tcp, Socket, <<1:1, 0:7>>} ->
ok = pong(Socket),
loop(Parent, Debug, Socket);
{tcp, Socket, <<1:1, 1:7, SigSize:24, Sig:SigSize/binary, 9:8, Bin/binary>>} ->
ok = handle_package_update(Sig, Bin),
loop(Parent, Debug, Socket);
{tcp, Socket, <<1:1, 2:7, Bin/binary>>} ->
{ok, {Realm, Serial}} = zx_lib:b_to_ts(Bin),
ok = zx_daemon:report({serial_update, Realm, Serial}),
loop(Parent, Debug, Socket);
{tcp, Socket, Unexpected} ->
ok = log(warning, "Funky data from node: ~tp", [Unexpected]),
ok = zx_net:disconnect(Socket),
terminate();
{subscribe, Package} ->
ok = zx_net:send(Socket, {subscribe, Package}),
ok = do_subscribe(Socket, Package),
loop(Parent, Debug, Socket);
{unsubscribe, Package} ->
ok = zx_net:send(Socket, {unsubscribe, Package}),
loop(Parent, Debug, Socket);
{fetch, ID, Object} ->
{ok, Outcome} = handle_fetch(Socket, Object),
ok = zx_daemon:result(ID, Outcome),
ok = do_unsubscribe(Socket, Package),
loop(Parent, Debug, Socket);
{query, ID, Action} ->
{ok, Outcome} = handle_query(Socket, Action),
ok = zx_daemon:result(ID, Outcome),
Result = do_query(Socket, Action),
ok = zx_daemon:result(ID, Result),
loop(Parent, Debug, Socket);
{fetch, ID, PackageID} ->
ok = do_fetch(Socket, ID, PackageID),
loop(Parent, Debug, Socket);
stop ->
ok = zx_net:disconnect(Socket),
terminate();
{tcp_closed, Socket} ->
handle_unexpected_close();
Unexpected ->
ok = log(warning, "Unexpected message: ~tp", [Unexpected]),
loop(Parent, Debug, Socket)
@@ -229,185 +226,163 @@ loop(Parent, Debug, Socket) ->
%%% Idle Incoming Upstream Messages
-spec handle_message(Socket, Bin) -> ok | no_return()
when Socket :: gen_tcp:socket(),
Bin :: binary().
%% @private
%% Single point to convert a binary message to a safe internal message. Actual handling
%% of the converted message occurs in dispatch/2.
handle_message(Socket, Bin) ->
Message = binary_to_term(Bin, [safe]),
ok = log(info, "Received network message: ~tp", [Message]),
case binary_to_term(Bin, [safe]) of
ping ->
zx_net:send(Socket, pong);
{sub, Channel, Message} ->
log("Sub: ~tp - ~tp", [Channel, Message]);
{update, Message} ->
log("Update: ~tp", [Message]);
{redirect, Nodes} ->
log("Redirected to ~tp", [Nodes]);
Invalid ->
{ok, {Addr, Port}} = zomp:peername(Socket),
Host = inet:ntoa(Addr),
Warning = "Invalid message from ~s:~p: ~tp",
ok = log(warning, Warning, [Host, Port, Invalid]),
ok = zx_net:disconnect(Socket),
%% TODO: Pull in missing key chains.
handle_package_update(Sig, Bin) ->
Message = {_, _, _, KeyID, {Realm, Name} = PackageID} = zx_lib:b_to_ts(Bin),
Package = {Realm, Name},
{ok, Key} = zx_key:load(KeyID),
case zx_key:verify(Bin, Sig, Key) of
true ->
zx_daemon:notify(Package, {update, PackageID});
false ->
ok = log(error, "Received an unverified update message: ~tp", [Message]),
terminate()
end.
%%% Incoming Request Actions
-spec handle_request(Socket, Action) -> Result
when Socket :: gen_tcp:socket(),
Action :: term(),
Result :: {ok, Outcome :: term()}.
-spec do_subscribe(gen_tcp:socket(), zx:package()) -> ok | no_return().
handle_request(Socket, Action) ->
ok = zx_net:send(Socket, Action),
Response =
case element(1, Action) of
list ->
do_list(Action, Socket);
latest ->
do_latest(Action, Socket);
fetch ->
do_fetch(Action, Socket);
key ->
do_key(Action, Socket);
pending ->
do_pending(Action, Socket);
packagers ->
do_packagers(Action, Socket);
maintainers ->
do_maintainers(Action, Socket);
sysops ->
do_sysops(Action, Socket)
end,
handle_response(Socket, Response).
do_subscribe(Socket, Package) ->
Reference = term_to_binary(Package),
Message = <<0:1, 1:7, Reference/binary>>,
ok = gen_tcp:send(Socket, Message),
wait_ok(Socket).
handle_fetch(_, _) -> {error, nyi}.
-spec do_unsubscribe(gen_tcp:socket(), zx:package()) -> ok | no_return().
handle_query(_, _) -> {error, nyi}.
do_list(_, _) -> {error, nyi}.
do_latest(_, _) -> {error, nyi}.
do_fetch(_, _) -> {error, nyi}.
do_key(_, _) -> {error, nyi}.
do_pending(_, _) -> {error, nyi}.
do_packagers(_, _) -> {error, nyi}.
do_maintainers(_, _) -> {error, nyi}.
do_sysops(_, _) -> {error, nyi}.
do_unsubscribe(Socket, Package) ->
Reference = term_to_binary(Package),
Message = <<0:1, 2:7, Reference/binary>>,
ok = gen_tcp:send(Socket, Message),
wait_ok(Socket).
do_query(Socket, {list, Realm}) ->
Reference = term_to_binary(Realm),
Message = <<0:1, 3:7, Reference/binary>>,
send_query(Socket, Message);
do_query(Socket, {list, Realm, Name}) ->
Reference = term_to_binary({Realm, Name}),
Message = <<0:1, 4:7, Reference/binary>>,
send_query(Socket, Message);
do_query(Socket, {list, Realm, Name, Version}) ->
Reference = term_to_binary({Realm, Name, Version}),
Message = <<0:1, 4:7, Reference/binary>>,
send_query(Socket, Message);
do_query(Socket, {latest, Realm, Name}) ->
Reference = term_to_binary({Realm, Name}),
Message = <<0:1, 6:7, Reference/binary>>,
send_query(Socket, Message);
do_query(Socket, {latest, Realm, Name, Version}) ->
Reference = term_to_binary({Realm, Name, Version}),
Message = <<0:1, 6:7, Reference/binary>>,
send_query(Socket, Message).
handle_response(Socket, Command) ->
send_query(Socket, Message) ->
ok = gen_tcp:send(Socket, Message),
wait_query(Socket).
wait_query(Socket) ->
ok = inet:setopts(Socket, [{active, once}]),
receive
{tcp, Socket, Bin} ->
Outcome = binary_to_term(Bin, [safe]),
interpret_response(Socket, Outcome, Command);
{tcp_closed, Socket} ->
handle_unexpected_close()
after 5000 ->
handle_timeout(Socket)
{tcp, Socket, <<1:1, 0:7>>} ->
ok = pong(Socket),
wait_query(Socket);
{tcp, Socket, <<0:1, 0:7, Bin/binary>>} ->
{ok, Result} = zx_lib:b_to_ts(Bin),
Result;
{tcp, Socket, <<0:1, 1:7>>} ->
ok = zx_daemon:report(failed),
terminate();
{tcp, Socket, <<0:1, 2:7>>} ->
{error, bad_realm};
{tcp, Socket, <<0:1, 3:7>>} ->
{error, bad_package};
{tcp, Socket, <<0:1, 4:7>>} ->
{error, bad_version}
after 5000 ->
handle_timeout(Socket)
end.
interpret_response(Socket, ping, Command) ->
ok = zx_net:send(Socket, pong),
handle_response(Socket, Command);
interpret_response(Socket, {sub, Channel, Message}, Command) ->
ok = zx_daemon:notify(Channel, Message),
handle_response(Socket, Command).
%interpret_response(Socket, {update, Message}, Command) ->
%interpret_response(Socket, Response, list) ->
%interpret_response(Socket, Response, latest) ->
%interpret_response(Socket, Response, fetch) ->
%interpret_response(Socket, Response, key) ->
%interpret_response(Socket, Response, pending) ->
%interpret_response(Socket, Response, packagers) ->
%interpret_response(Socket, Response, maintainers) ->
%interpret_response(Socket, Response, sysops) ->
%
%
%
% case element(1, Action) of
% end,
pong(Socket) ->
gen_tcp:send(Socket, <<1:1, 0:7>>).
-spec fetch(Socket, PackageID) -> Result
-spec do_fetch(Socket, ID, PackageID) -> Result
when Socket :: gen_tcp:socket(),
ID :: zx_daemon:id(),
PackageID :: zx:package_id(),
Result :: ok.
%% @private
%% Download a package to the local cache.
fetch(Socket, PackageID) ->
{ok, LatestID} = request_zsp(Socket, PackageID),
ok = receive_zsp(Socket, LatestID),
Latest = zx_lib:package_string(LatestID),
log(info, "Fetched ~ts", [Latest]).
do_fetch(Socket, ID, PackageID) ->
Reference = binary_to_term(PackageID),
Message = <<0:1, 7:7, Reference/binary>>,
ok = gen_tcp:send(Socket, Message),
ok = wait_hops(Socket, ID),
{ok, Bin} = receive_zsp(Socket),
zx_daemon:result(ID, {done, Bin}).
-spec request_zsp(Socket, PackageID) -> Result
when Socket :: gen_tcp:socket(),
PackageID :: zx:package_id(),
Result :: {ok, Latest :: zx:package_id()}
| {error, Reason :: timeout | term()}.
request_zsp(Socket, PackageID) ->
ok = zx_net:send(Socket, {fetch, PackageID}),
wait_hops(Socket, ID) ->
ok = inet:setopts(Socket, [{active, once}]),
receive
{tcp, Socket, Bin} ->
case binary_to_term(Bin) of
{sending, LatestID} ->
{ok, LatestID};
Error = {error, Reason} ->
PackageString = zx_lib:package_string(PackageID),
Message = "Error receiving package ~ts: ~tp",
ok = log(info, Message, [PackageString, Reason]),
Error
end;
{tcp_closed, Socket} ->
handle_unexpected_close()
{tcp, Socket, <<0:1, 0:7, 0:8>>} ->
ok = inet:setopts(Socket, [{packet, 0}]),
gen_tcp:send(Socket, <<0:1, 0:7>>);
{tcp, Socket, <<0:1, 0:7, Distance:8>>} ->
ok = zx_daemon:result(ID, {hops, Distance}),
wait_hops(Socket, ID);
{tcp, Socket, <<0:1, 1:7>>} ->
{error, bad_message};
{tcp, Socket, <<0:1, 2:7>>} ->
{error, bad_realm};
{tcp, Socket, <<0:1, 3:7>>} ->
{error, bad_package};
{tcp, Socket, <<0:1, 4:7>>} ->
{error, bad_version};
{tcp, Socket, <<0:1, 5:7>>} ->
handle_timeout(Socket)
after 60000 ->
{error, timeout}
end.
-spec receive_zsp(Socket, PackageID) -> Result
when Socket :: gen_tcp:socket(),
PackageID :: zx:package_id(),
Result :: ok | {error, timeout}.
receive_zsp(Socket, PackageID) ->
receive
{tcp, Socket, Bin} ->
ZrpPath = zx_lib:zsp_path(PackageID),
ok = file:write_file(ZrpPath, Bin),
ok = zx_net:send(Socket, ok),
log(info, "Wrote ~ts", [ZrpPath]);
{tcp_closed, Socket} ->
handle_unexpected_close()
after 60000 ->
ok = log(error, "Timeout in socket receive for ~tp", [PackageID]),
{error, timeout}
handle_timeout(Socket)
end.
receive_zsp(Socket) ->
ok = inet:setopts(Socket, [{active, once}]),
receive
{tcp, Socket, <<Size:32, Bin:Size/binary>>} ->
{ok, Bin};
{tcp, Socket, <<Total:32, Bin/binary>>} ->
Size = byte_size(Bin),
receive_zsp(Socket, Total, Size, Bin)
after 5000 ->
handle_timeout(Socket)
end.
receive_zsp(Socket, Total, SoFar, Bin) when Total > SoFar ->
ok = inet:setopts(Socket, [{active, once}]),
receive
{tcp, Socket, New} ->
Size = byte_size(New),
NewBin = <<Bin/binary, New/binary>>,
receive_zsp(Socket, Total, Size + SoFar, NewBin)
after 5000 ->
handle_timeout(Socket)
end;
receive_zsp(Socket, Total, Total, Bin) ->
ok = inet:setopts(Socket, [{packet, 4}]),
ok = gen_tcp:send(Socket, <<0:1, 0:7>>),
{ok, Bin}.
%%% Terminal handlers
@@ -426,6 +401,15 @@ handle_timeout(Socket) ->
terminate().
-spec wait_ok(gen_tcp:socket()) -> ok | no_return().
wait_ok(Socket) ->
receive
{tcp, Socket, <<0:1, 0:7>>} -> ok
after 5000 -> handle_timeout(Socket)
end.
-spec terminate() -> no_return().
%% @private
%% Convenience wrapper around the suicide call.
@@ -435,109 +419,3 @@ handle_timeout(Socket) ->
terminate() ->
exit(normal).
%-spec do_query_latest(Object, State) -> {Result, NewState}
% when Object :: zx:package() | zx:package_id(),
% State :: state(),
% Result :: {ok, zx:version()}
% | {error, Reason},
% Reason :: bad_realm
% | bad_package
% | bad_version,
% NewState :: state().
%% @private
%% Queries a zomp realm for the latest version of a package or package
%% version (complete or incomplete version number).
%
%do_query_latest(Socket, {Realm, Name}) ->
% ok = zx_net:send(Socket, {latest, Realm, Name}),
% receive
% {tcp, Socket, Bin} -> binary_to_term(Bin)
% after 5000 -> {error, timeout}
% end;
%do_query_latest(Socket, {Realm, Name, Version}) ->
% ok = zx_net:send(Socket, {latest, Realm, Name, Version}),
% receive
% {tcp, Socket, Bin} -> binary_to_term(Bin)
% after 5000 -> {error, timeout}
% end.
%-spec do_fetch(PackageIDs, State) -> NewState
% when PackageIDs :: [zx:package_id()],
% State :: state(),
% NewState :: state(),
% Result :: ok
% | {error, Reason},
% Reason :: bad_realm
% | bad_package
% | bad_version
% | network.
%% @private
%%
%
%do_fetch(PackageIDs, State) ->
% FIXME: Need to create a job queue divided by realm and dispatched to connectors,
% and cleared from the master pending queue kept here by the daemon as the
% workers succeed. Basic task queue management stuff... which never existed
% in ZX before... grrr...
% case scrub(PackageIDs) of
% [] ->
% ok;
% Needed ->
% Partitioned = partition_by_realm(Needed),
% EnsureDeps =
% fun({Realm, Packages}) ->
% ok = zx_conn:queue_package(Pid, Realm, Packages),
% log(info, "Disconnecting from realm: ~ts", [Realm])
% end,
% lists:foreach(EnsureDeps, Partitioned)
% end.
%
%
%partition_by_realm(PackageIDs) ->
% PartitionMap = lists:foldl(fun partition_by_realm/2, #{}, PackageIDs),
% maps:to_list(PartitionMap).
%
%
%partition_by_realm({R, P, V}, M) ->
% maps:update_with(R, fun(Ps) -> [{P, V} | Ps] end, [{P, V}], M).
%
%
%ensure_deps(_, _, []) ->
% ok;
%ensure_deps(Socket, Realm, [{Name, Version} | Rest]) ->
% ok = ensure_dep(Socket, {Realm, Name, Version}),
% ensure_deps(Socket, Realm, Rest).
%
%
%-spec ensure_dep(gen_tcp:socket(), package_id()) -> ok | no_return().
%% @private
%% Given an PackageID as an argument, check whether its package file exists in the
%% system cache, and if not download it. Should return `ok' whenever the file is
%% sourced, but exit with an error if it cannot locate or acquire the package.
%
%ensure_dep(Socket, PackageID) ->
% ZrpFile = zx_lib:zsp_path(PackageID),
% ok =
% case filelib:is_regular(ZrpFile) of
% true -> ok;
% false -> fetch(Socket, PackageID)
% end,
% ok = install(PackageID),
% build(PackageID).
%
%
%-spec scrub(Deps) -> Scrubbed
% when Deps :: [package_id()],
% Scrubbed :: [package_id()].
%% @private
%% Take a list of dependencies and return a list of dependencies that are not yet
%% installed on the system.
%
%scrub([]) ->
% [];
%scrub(Deps) ->
% lists:filter(fun(PackageID) -> not zx_lib:installed(PackageID) end, Deps).
zomp/lib/otpr/zx/0.1.0/src/zx_daemon.erl
+14 -18
View File
@@ -235,14 +235,10 @@
| {list, zx:realm()}
| {list, zx:realm(), zx:name()}
| {list, zx:realm(), zx:name(), zx:version()}
| {latest, zx:realm(), zx:name()}
| {latest, zx:realm(), zx:name(), zx:version()}
| {fetch, zx:realm(), zx:name(), zx:version()}
| {fetchkey, zx:realm(), zx:key_name()}
| {pending, zx:realm(), zx:name()}
| {resigns, zx:realm()}
| {packagers, zx:realm(), zx:name()}
| {maintainers, zx:realm(), zx:name()}
| {sysops, zx:realm()}.
| {fetchkey, zx:realm(), zx:key_name()}.
% Outgoing Result Messages
%
@@ -251,7 +247,7 @@
%
% Subscription messages are a separate type below.
-type result() :: {z_result,
-type result() :: {result,
RequestID :: id(),
Message :: realm_list()
| package_list()
@@ -581,8 +577,8 @@ report(Message) ->
%% @private
%% Return a tagged result back to the daemon to be forwarded to the original requestor.
result(Reference, Result) ->
gen_server:cast(?MODULE, {result, Reference, Result}).
result(ID, Result) ->
gen_server:cast(?MODULE, {result, ID, Result}).
-spec notify(Package, Message) -> ok
@@ -681,8 +677,8 @@ handle_cast({report, Conn, Message}, State) ->
NextState = do_report(Conn, Message, State),
NewState = eval_queue(NextState),
{noreply, NewState};
handle_cast({result, Ref, Result}, State) ->
NextState = do_result(Ref, Result, State),
handle_cast({result, ID, Result}, State) ->
NextState = do_result(ID, Result, State),
NewState = eval_queue(NextState),
{noreply, NewState};
handle_cast({notify, Conn, Package, Update}, State) ->
@@ -982,7 +978,7 @@ do_result(ID, Result, State = #s{requests = Requests, dropped = Dropped, mx = MX
{Dropped, NextR, NextMX};
{Request, Rest} ->
Requestor = element(1, Request),
Requestor ! {z_result, ID, Result},
Requestor ! {result, ID, Result},
NextMX = mx_del_monitor(Requestor, {requestor, ID}, MX),
{Dropped, Rest, NextMX};
error ->
@@ -998,14 +994,14 @@ handle_fetch_result(ID, {done, Bin}, {Requestor, _}, Requests, MX) ->
ok -> done;
Error -> Error
end,
Requestor ! {z_result, ID, Result},
Requestor ! {result, ID, Result},
NextMX = mx_del_monitor(Requestor, {requestor, ID}, MX),
{NextMX, Requests};
handle_fetch_result(ID, Hops = {hops, _}, Request = {Requestor, _}, Requests, MX) ->
Requestor ! {z_result, ID, Hops},
Requestor ! {result, ID, Hops},
{MX, maps:put(ID, Request, Requests)};
handle_fetch_result(ID, Outcome, {Requestor, _}, Requests, MX) ->
Requestor ! {z_result, ID, Outcome},
Requestor ! {result, ID, Outcome},
NextMX = mx_del_monitor(Requestor, {requestor, ID}, MX),
{NextMX, Requests}.
@@ -1252,10 +1248,10 @@ do_fetch(PackageID, Requestor, State = #s{id = ID}) ->
{ok, Bin} ->
case do_import_package(Bin) of
ok ->
Requestor ! {z_result, ID, ok},
Requestor ! {result, ID, ok},
{ok, State};
Error ->
Requestor ! {z_result, ID, Error},
Requestor ! {result, ID, Error},
{ok, State}
end;
{error, enoent} ->
@@ -1263,7 +1259,7 @@ do_fetch(PackageID, Requestor, State = #s{id = ID}) ->
Action = {fetch, Realm, Name, Version},
do_request(Requestor, Action, State);
Error ->
Requestor ! {z_result, ID, Error}
Requestor ! {result, ID, Error}
end.
zomp/lib/otpr/zx/0.1.0/src/zx_lib.erl
+7 -1
View File
@@ -17,7 +17,7 @@
-export([zomp_dir/0, find_zomp_dir/0,
path/1, path/2, path/3, path/4, ppath/2,
force_dir/1, mktemp_dir/1,
list_realms/0,
list_realms/0, realm_exists/1,
get_prime/1, realm_meta/1,
read_project_meta/0, read_project_meta/1, read_package_meta/1,
write_project_meta/1, write_project_meta/2,
@@ -192,6 +192,12 @@ list_realms() ->
[filename:basename(filename:dirname(C)) || C <- filelib:wildcard(Pattern)].
-spec realm_exists(zx:realm()) -> boolean().
realm_exists(Realm) ->
lists:member(Realm, list_realms()).
-spec get_prime(Realm) -> Result
when Realm :: zx:realm(),
Result :: {ok, zx:host()}
zomp/lib/otpr/zx/0.1.0/src/zx_local.erl
+12 -6
View File
@@ -335,7 +335,8 @@ list_realms() ->
%% them to stdout.
list_packages(Realm) ->
case zx_daemon:list_packages(Realm) of
{ok, ID} = zx_daemon:list(Realm),
case wait_result(ID) of
{ok, []} ->
io:format("Realm ~tp has no packages available.~n", [Realm]);
{ok, Packages} ->
@@ -364,15 +365,16 @@ list_versions(PackageString) ->
end.
list_versions2(PackageID) ->
case zx_daemon:list_versions(PackageID) of
list_versions2({Realm, Name, Version}) ->
{ok, ID} = zx_daemon:list(Realm, Name, Version),
case wait_result(ID) of
{ok, []} ->
io:format("No versions available.~n");
{ok, Versions} ->
Print =
fun(Version) ->
{ok, VersionString} = zx_lib:version_to_string(Version),
io:format("~ts~n", [VersionString])
fun(V) ->
{ok, VS} = zx_lib:version_to_string(V),
io:format("~ts~n", [VS])
end,
lists:foreach(Print, Versions);
{error, bad_realm} ->
@@ -385,6 +387,10 @@ list_versions2(PackageID) ->
end.
wait_result(ID) ->
receive {result, ID, Result} -> Result end.
-spec import_realm(Path) -> zx:outcome()
when Path :: file:filename().
%% @private
zomp/lib/otpr/zx/0.1.0/src/zx_net.erl
+1 -15
View File
@@ -9,25 +9,11 @@
-copyright("Craig Everett <zxq9@zxq9.com>").
-license("GPL-3.0").
-export([send/2, disconnect/1]).
-export([disconnect/1]).
-include("zx_logger.hrl").
-spec send(Socket, Message) -> Result
when Socket :: gen_tcp:socket(),
Message :: term(),
Result :: ok
| {error, Reason},
Reason :: closed | inet:posix().
%% @doc
%% Packages an Erlang term and sends it to the indicated socket.
send(Socket, Message) ->
BinMessage = term_to_binary(Message),
gen_tcp:send(Socket, BinMessage).
-spec disconnect(Socket) -> ok
when Socket :: gen_tcp:socket().
%% @doc
zomp/lib/otpr/zx/0.1.0/src/zx_sys_conf.erl
+1 -1
View File
@@ -76,7 +76,7 @@ populate_data(List) ->
Retries =
case proplists:get_value(retries, List, 3) of
RT when is_integer(RT) and RT > 0 -> {RT, RT};
_ -> 3
_ -> {3, 3}
end,
MaxConn =
case proplists:get_value(maxconn, List, 5) of