Packages
macula
0.7.4
7.1.0
7.0.0
6.0.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.8.0
4.7.1
4.7.0
4.6.0
4.5.0
4.4.10
4.4.9
4.4.8
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.1
4.3.0
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.1
4.1.0
4.0.0
3.16.0
3.15.3
3.15.2
3.15.1
3.14.0
3.13.0
3.12.1
3.12.0
3.11.1
3.11.0
3.10.3
3.10.2
3.10.1
3.9.0
3.8.0
3.7.0
3.5.0
3.4.0
3.3.0
3.2.0
3.1.0
3.0.0
2.1.1
2.1.0
2.0.0
1.5.2
1.5.1
1.4.30
1.4.29
1.4.28
1.4.27
1.4.26
1.4.25
1.4.24
1.4.23
1.4.22
1.4.21
1.4.20
1.4.19
1.4.18
1.4.17
1.4.16
1.4.15
1.4.14
1.4.13
1.4.11
1.4.10
1.4.9
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.1
1.3.0
1.2.0
1.1.0
1.0.10
1.0.9
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.48.6
0.48.5
0.48.4
0.48.3
0.48.2
0.48.1
0.48.0
0.47.1
0.47.0
0.46.3
0.46.1
0.46.0
0.45.3
0.45.2
0.45.1
0.45.0
0.44.2
0.44.1
0.44.0
0.43.3
0.43.2
0.43.1
0.43.0
0.42.9
0.42.8
0.42.7
0.42.6
0.42.5
0.42.4
0.42.3
0.42.2
0.42.1
0.42.0
0.41.1
0.41.0
0.40.1
0.40.0
0.39.9
0.39.8
0.39.7
0.39.6
0.39.5
0.39.4
0.39.3
0.39.2
0.39.1
0.39.0
0.38.8
0.38.7
0.38.6
0.38.5
0.38.4
0.38.3
0.38.2
0.38.1
0.38.0
0.37.7
0.37.6
0.37.5
0.37.4
0.37.3
0.37.2
0.37.1
0.37.0
0.36.6
0.36.5
0.36.4
0.36.3
0.36.2
0.36.1
0.36.0
0.35.4
0.35.3
0.35.2
0.35.1
0.35.0
0.34.1
0.34.0
0.33.1
0.33.0
0.32.5
0.32.4
0.32.3
0.32.2
0.32.1
0.32.0
0.31.9
0.31.8
0.31.7
0.31.6
0.31.5
0.31.4
0.31.3
0.31.2
0.31.1
0.31.0
0.30.10
0.30.9
0.30.8
0.30.7
0.30.6
0.30.5
0.30.4
0.30.3
0.30.2
0.30.1
0.30.0
0.29.0
0.28.3
0.28.2
0.28.1
0.28.0
0.27.1
0.27.0
0.26.1
0.26.0
0.25.6
0.25.5
0.25.4
0.25.3
0.25.2
0.25.1
0.25.0
0.24.6
0.24.5
0.24.4
0.24.3
0.24.2
0.24.1
0.24.0
0.23.3
0.23.2
0.23.1
0.23.0
0.22.12
0.22.11
0.22.10
0.22.9
0.22.8
0.22.7
0.22.6
0.22.5
0.22.4
0.22.3
0.22.2
0.22.1
0.22.0
0.21.7
0.21.6
0.21.5
0.21.4
0.21.2
0.21.1
0.21.0
0.20.25
0.20.24
0.20.23
0.20.22
0.20.21
0.20.20
0.20.19
0.20.18
0.20.17
0.20.16
0.20.15
0.20.14
0.20.13
0.20.12
0.20.11
0.20.10
0.20.9
0.20.8
0.20.7
0.20.6
0.20.5
0.20.3
0.20.2
0.20.1
0.20.0
0.19.2
0.19.1
0.19.0
0.18.1
0.18.0
0.17.4
0.17.3
0.17.2
0.17.1
0.17.0
0.16.6
0.16.5
0.16.4
0.16.3
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.3
0.14.2
0.14.1
0.14.0
0.12.6
0.12.5
0.12.3
0.11.3
0.10.2
0.10.1
0.10.0
0.9.2
0.9.1
0.9.0
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.10
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.30
0.7.29
0.7.28
0.7.27
0.7.26
0.7.25
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.4
0.3.3
0.3.2
0.3.1
Macula HTTP/3 Mesh SDK — connect, subscribe, publish, call, advertise
Current section
Files
Jump to
Current section
Files
src/macula_gateway_rpc_router.erl
%%%-------------------------------------------------------------------
%%% @doc
%%% RPC Router Module - handles routed RPC messages (CALL/REPLY).
%%%
%%% Responsibilities:
%%% - Process routed CALL messages delivered locally
%%% - Process routed REPLY messages delivered locally
%%% - Send REPLY back via routing path
%%% - Forward rpc_route messages to next hop
%%% - Coordinate between RPC handler, mesh, and routing modules
%%%
%%% Pattern: Stateless delegation module
%%% - No GenServer (no state to manage)
%%% - Pure functions coordinating between modules
%%% - Consistent error handling ({ok, Result} | {error, Reason})
%%%
%%% Extracted from macula_gateway.erl (Phase 11)
%%% @end
%%%-------------------------------------------------------------------
-module(macula_gateway_rpc_router).
%% API
-export([
handle_routed_call/5,
handle_routed_reply/4,
send_reply_via_routing/4,
forward_rpc_route/3
]).
%%%===================================================================
%%% API
%%%===================================================================
%% @doc Handle routed CALL message delivered locally.
%% Looks up RPC handler, invokes it, sends reply via routing path.
-spec handle_routed_call(map(), map(), binary(), pid(), pid()) ->
ok | {error, term()}.
handle_routed_call(CallMsg, RpcRouteMsg, NodeId, RpcPid, MeshPid) ->
io:format("[RPC Router] Processing routed CALL message~n"),
%% Extract source node ID from rpc_route envelope
#{<<"source_node_id">> := SourceNodeId} = RpcRouteMsg,
%% Extract call data
#{<<"procedure">> := Procedure,
<<"args">> := ArgsJson,
<<"call_id">> := CallId} = CallMsg,
%% Look up handler via macula_gateway_rpc
case macula_gateway_rpc:get_handler(RpcPid, Procedure) of
not_found ->
io:format("[RPC Router] No handler for procedure: ~p~n", [Procedure]),
ErrorReply = #{
call_id => CallId,
error => #{
code => <<"no_handler">>,
message => <<"No handler registered for ", Procedure/binary>>
}
},
send_reply_via_routing(ErrorReply, SourceNodeId, NodeId, MeshPid);
{ok, Handler} ->
invoke_handler_and_reply(Handler, ArgsJson, CallId, SourceNodeId, NodeId, MeshPid)
end.
%% @doc Handle routed REPLY message delivered locally.
%% Routes to connection via gproc (local node) or to client stream (remote client).
-spec handle_routed_reply(map(), map(), binary(), map()) ->
ok | {error, term()}.
handle_routed_reply(ReplyMsg, RpcRouteMsg, NodeId, ClientStreams) ->
io:format("[RPC Router] Processing routed REPLY message~n"),
%% Extract destination node ID
#{<<"destination_node_id">> := DestinationNodeId} = RpcRouteMsg,
%% Check if this is for local node or remote client
case DestinationNodeId of
NodeId ->
%% Local node - send to connection via gproc (send full rpc_route message)
deliver_reply_to_local_connection(RpcRouteMsg);
_RemoteNodeId ->
%% Remote client - send to client stream
deliver_reply_to_remote_client(ReplyMsg, RpcRouteMsg, ClientStreams)
end.
%% @doc Send REPLY back via routing path.
%% Wraps reply in rpc_route envelope and routes to destination.
%% Crashes on routing failures - exposes mesh/routing issues immediately.
-spec send_reply_via_routing(map(), binary(), binary(), pid()) -> ok.
send_reply_via_routing(ReplyMsg, DestNodeId, NodeId, MeshPid) ->
io:format("[RPC Router] Sending reply via routing to ~p~n",
[binary:encode_hex(DestNodeId)]),
%% Get routing server PID (let it crash if not available)
RoutingServerPid = whereis(macula_routing_server),
case RoutingServerPid of
undefined ->
io:format("[RPC Router] Routing server not available~n"),
error(routing_server_not_available);
_ ->
%% Wrap reply in rpc_route envelope (with MaxHops = 10)
RpcRouteMsg = macula_rpc_routing:wrap_reply(NodeId, DestNodeId, ReplyMsg, 10),
%% Route the reply (let it crash on errors)
case macula_rpc_routing:route_or_deliver(NodeId, RpcRouteMsg, RoutingServerPid) of
{deliver, _, _} ->
%% Should not happen for replies going back, but handle it
io:format("[RPC Router] Reply is for local node (unexpected)~n"),
ok;
{forward, NextHopNodeInfo, UpdatedRpcRouteMsg} ->
%% Forward to next hop (let it crash on errors)
forward_rpc_route(NextHopNodeInfo, UpdatedRpcRouteMsg, MeshPid);
{error, Reason} ->
io:format("[RPC Router] Routing error: ~p~n", [Reason]),
error({routing_failed, Reason})
end
end.
%% @doc Forward rpc_route message to next hop.
%% Gets mesh connection and sends encoded message.
%% Crashes on connection/encoding failures - exposes mesh/protocol issues.
-spec forward_rpc_route(map(), map(), pid()) -> ok.
forward_rpc_route(NextHopNodeInfo, RpcRouteMsg, MeshPid) ->
io:format("[RPC Router] Forwarding rpc_route to next hop~n"),
%% Extract next hop info (let it crash on missing fields)
#{<<"node_id">> := NextHopNodeId,
<<"address">> := AddressBin,
<<"port">> := Port} = NextHopNodeInfo,
%% Parse address
Address = parse_address(AddressBin),
%% Get or create mesh connection (let it crash on errors)
{ok, Stream} = macula_gateway_mesh:get_or_create_connection(MeshPid, NextHopNodeId, {Address, Port}),
%% Encode and send rpc_route message (let it crash on errors)
EncodedMsg = macula_protocol_encoder:encode(rpc_route, RpcRouteMsg),
macula_quic:send(Stream, EncodedMsg),
io:format("[RPC Router] Forwarded rpc_route successfully~n"),
ok.
%%%===================================================================
%%% Internal functions
%%%===================================================================
%% @private
%% @doc Invoke RPC handler and send reply.
invoke_handler_and_reply(Handler, ArgsJson, CallId, SourceNodeId, NodeId, MeshPid) ->
try
%% Decode JSON args
Args = json:decode(ArgsJson),
%% Invoke handler
Result = Handler(Args),
%% Create success reply
Reply = #{
call_id => CallId,
result => encode_json(Result)
},
io:format("[RPC Router] Handler invoked successfully~n"),
send_reply_via_routing(Reply, SourceNodeId, NodeId, MeshPid)
catch
Class:Reason:Stacktrace ->
io:format("[RPC Router] Handler error: ~p:~p~n~p~n",
[Class, Reason, Stacktrace]),
%% Create error reply
ErrorReply = #{
call_id => CallId,
error => #{
code => <<"handler_error">>,
message => format_error(Class, Reason)
}
},
send_reply_via_routing(ErrorReply, SourceNodeId, NodeId, MeshPid)
end.
%% @private
%% @doc Deliver reply to local connection via gproc.
%% Sends the full rpc_route message to all local connections.
deliver_reply_to_local_connection(RpcRouteMsg) ->
io:format("[RPC Router] Delivering reply to local connections~n"),
%% Look up all local macula_connection processes via gproc
case gproc:lookup_pids({p, l, macula_connection}) of
[] ->
io:format("[RPC Router] WARNING: No local connection process found~n"),
{error, connection_not_found};
Pids ->
io:format("[RPC Router] Found ~p local connection process(es)~n", [length(Pids)]),
%% Send to all local connections (usually just one)
lists:foreach(fun(Pid) ->
gen_server:cast(Pid, {rpc_route_reply, RpcRouteMsg})
end, Pids),
ok
end.
%% @private
%% @doc Deliver reply to remote client via stream.
%% Sends the full rpc_route message (not just the reply).
deliver_reply_to_remote_client(_ReplyMsg, RpcRouteMsg, ClientStreams) ->
%% Extract connection ID from rpc_route
case maps:get(<<"connection_id">>, RpcRouteMsg, undefined) of
undefined ->
io:format("[RPC Router] No connection_id in rpc_route~n"),
{error, no_connection_id};
ConnectionId ->
%% Look up client stream
case maps:get(ConnectionId, ClientStreams, undefined) of
undefined ->
io:format("[RPC Router] Client stream not found for connection ~p~n",
[ConnectionId]),
{error, client_not_found};
StreamPid ->
%% Encode and send the full rpc_route message
RpcRouteBinary = macula_protocol_encoder:encode(rpc_route, RpcRouteMsg),
macula_quic:send(StreamPid, RpcRouteBinary),
io:format("[RPC Router] Delivered reply to remote client~n"),
ok
end
end.
%% @private
%% @doc Parse address binary to tuple or string.
parse_address(AddressBin) when is_binary(AddressBin) ->
%% Try to parse as IP address
case inet:parse_address(binary_to_list(AddressBin)) of
{ok, IpTuple} -> IpTuple;
{error, _} -> binary_to_list(AddressBin) % Treat as hostname
end;
parse_address(Address) ->
Address.
%% @private
%% @doc Encode result to JSON binary.
encode_json(Term) when is_binary(Term) ->
Term;
encode_json(Term) ->
macula_utils:encode_json(Term).
%% @private
%% @doc Format error for reply message.
format_error(Class, Reason) ->
iolist_to_binary(io_lib:format("~p: ~p", [Class, Reason])).