Packages
macula
0.6.5
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_connection.erl.backup
%%%-------------------------------------------------------------------
%%% @doc
%%% Macula SDK connection manager.
%%%
%%% Manages individual HTTP/3 (QUIC) connections to Macula mesh nodes.
%%% Handles connection lifecycle, authentication, message sending/receiving,
%%% and subscription management.
%%% @end
%%%-------------------------------------------------------------------
-module(macula_connection).
-behaviour(gen_server).
%% API
-export([
start_link/2,
stop/1,
publish/3,
publish/4,
subscribe/3,
unsubscribe/2,
call/3,
call/4,
advertise/4,
unadvertise/2
]).
%% gen_server callbacks
-export([
init/1,
handle_call/3,
handle_cast/2,
handle_info/2,
terminate/2,
code_change/3
]).
%% Exported for testing
-ifdef(TEST).
-export([
next_message_id/1,
topic_matches/3,
decode_messages/2,
process_message/2
]).
-endif.
-include_lib("kernel/include/logger.hrl").
-include("macula_config.hrl").
-record(state, {
url :: binary(),
opts :: map(),
realm :: binary(),
node_id :: binary(),
%% Supervision tree (Phase 7)
supervisor_pid :: undefined | pid(),
connection_manager_pid :: undefined | pid(),
pubsub_handler_pid :: undefined | pid(),
rpc_handler_pid :: undefined | pid(),
advertisement_manager_pid :: undefined | pid(),
%% Legacy fields - will be removed after Phase 7 complete
%% QUIC connection and stream
connection :: undefined | pid(),
stream :: undefined | pid(),
%% Connection state
status :: connecting | connected | disconnected | error,
%% Subscriptions: #{SubscriptionRef => {Topic, Callback}}
subscriptions :: #{reference() => {binary(), fun((map()) -> ok)}},
%% Pending RPC calls: #{CallId => {From, Timer}}
pending_calls :: #{binary() => {term(), reference()}},
%% Pending DHT queries: #{QueryKey => {From, Timer}}
pending_queries :: #{binary() => {term(), reference()}},
%% Pending publish acknowledgments for QoS 1: #{MsgId => {Topic, Payload, QoS, RetryCount, TimerRef}}
pending_pubacks :: #{binary() => {binary(), binary(), integer(), integer(), reference()}},
%% Message ID counter
msg_id_counter :: non_neg_integer(),
%% Receive buffer for partial messages
recv_buffer :: binary(),
%% Service registry for DHT advertisement
service_registry :: macula_service_registry:registry(),
%% Advertised services with re-advertisement timers
%% #{Procedure => #{handler, metadata, ttl, timer_ref}}
advertised_services :: #{binary() => #{
handler := fun((map()) -> {ok, term()} | {error, term()}),
metadata := map(),
ttl := pos_integer(),
timer_ref := reference()
}},
%% Advertised subscriptions with re-advertisement timers
%% #{Topic => #{sub_ref, ttl, timer_ref}}
advertised_subscriptions :: #{binary() => #{
sub_ref := reference(),
ttl := pos_integer(),
timer_ref := reference()
}},
%% Topic pattern matching configuration
topic_separator :: binary(), % Default: <<".">>
topic_wildcard_single :: binary(), % Default: <<"*">> (matches one segment)
topic_wildcard_multi :: binary(), % Default: <<"**">> (matches zero or more segments)
%% Provider selection strategy state
provider_selector_state :: map(),
%% Connection cache for multi-endpoint RPC
%% #{Endpoint => #{connection, stream, last_used}}
endpoint_connections :: #{binary() => #{
connection := pid(),
stream := pid(),
last_used := integer()
}},
%% Pending subscriber discovery queries (for mesh-wide pub/sub)
%% #{QueryId => {Topic, Payload, Qos, Opts}}
pending_subscriber_queries :: #{binary() => {binary(), binary(), integer(), map()}}
}).
-define(CONNECT_RETRY_DELAY, 1000).
%%%===================================================================
%%% API
%%%===================================================================
%% @doc Start a client connection to a Macula mesh.
-spec start_link(binary(), map()) -> {ok, pid()} | {error, term()}.
start_link(Url, Opts) ->
gen_server:start_link(?MODULE, {Url, Opts}, []).
%% @doc Stop the client connection.
-spec stop(pid()) -> ok.
stop(Client) ->
gen_server:stop(Client).
%% @doc Publish an event through this client (no options).
-spec publish(pid(), binary(), map() | binary()) -> ok | {error, term()}.
publish(Client, Topic, Data) ->
publish(Client, Topic, Data, #{}).
%% @doc Publish an event through this client with options.
-spec publish(pid(), binary(), map() | binary(), map()) -> ok | {error, term()}.
publish(Client, Topic, Data, Opts) ->
gen_server:call(Client, {publish, Topic, Data, Opts}, ?DEFAULT_TIMEOUT).
%% @doc Subscribe to a topic through this client.
-spec subscribe(pid(), binary(), fun((map()) -> ok)) ->
{ok, reference()} | {error, term()}.
subscribe(Client, Topic, Callback) ->
gen_server:call(Client, {subscribe, Topic, Callback}, ?DEFAULT_TIMEOUT).
%% @doc Unsubscribe from a topic.
-spec unsubscribe(pid(), reference()) -> ok | {error, term()}.
unsubscribe(Client, SubRef) ->
gen_server:call(Client, {unsubscribe, SubRef}, ?DEFAULT_TIMEOUT).
%% @doc Make an RPC call through this client (default timeout).
-spec call(pid(), binary(), map() | list()) -> {ok, term()} | {error, term()}.
call(Client, Procedure, Args) ->
call(Client, Procedure, Args, #{}).
%% @doc Make an RPC call through this client with options.
-spec call(pid(), binary(), map() | list(), map()) ->
{ok, term()} | {error, term()}.
call(Client, Procedure, Args, Opts) ->
Timeout = maps:get(timeout, Opts, ?CALL_TIMEOUT),
gen_server:call(Client, {call, Procedure, Args, Opts}, Timeout + 1000).
%% @doc Advertise a service that this client provides.
%%
%% Registers a handler function locally and publishes the service to the DHT
%% so other clients can discover and call it.
-spec advertise(pid(), binary(), macula_service_registry:handler_fn(), map()) ->
{ok, reference()} | {error, term()}.
advertise(Client, Procedure, Handler, Opts) ->
gen_server:call(Client, {advertise, Procedure, Handler, Opts}, ?DEFAULT_TIMEOUT).
%% @doc Stop advertising a service.
%%
%% Removes the local handler and stops advertising to the DHT.
-spec unadvertise(pid(), binary()) -> ok | {error, term()}.
unadvertise(Client, Procedure) ->
gen_server:call(Client, {unadvertise, Procedure}, ?DEFAULT_TIMEOUT).
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
%% @private
init({Url, Opts}) ->
%% Parse URL to extract host and port
{Host, Port} = macula_utils:parse_url(Url),
%% Get realm (required)
Realm = get_realm_from_opts(Opts),
%% Generate or get node ID
NodeId = maps:get(node_id, Opts, macula_utils:generate_node_id()),
%% Get provider selection strategy from options (default: random)
SelectionStrategy = maps:get(provider_selection_strategy, Opts, random),
%% Get topic pattern matching configuration (defaults: dot-separated with * wildcards)
TopicSeparator = maps:get(topic_separator, Opts, <<".">>),
TopicWildcardSingle = maps:get(topic_wildcard_single, Opts, <<"*">>),
TopicWildcardMulti = maps:get(topic_wildcard_multi, Opts, <<"**">>),
%% Phase 7: Start supervision tree
?LOG_INFO("[Phase 7] Starting supervision tree for ~s", [Url]),
%% Prepare opts for handlers (includes node_id, realm, url)
HandlerOpts = Opts#{
node_id => NodeId,
realm => Realm,
url => Url,
host => Host,
port => Port,
provider_strategy => SelectionStrategy
},
%% Start supervision tree
{ok, SupPid} = macula_connection_sup:start_link(Url, HandlerOpts),
%% Look up child PIDs from supervisor
Children = supervisor:which_children(SupPid),
ConnMgrPid = find_child_pid(Children, connection_manager),
PubSubPid = find_child_pid(Children, pubsub_handler),
RpcPid = find_child_pid(Children, rpc_handler),
AdvMgrPid = find_child_pid(Children, advertisement_manager),
?LOG_INFO("[Phase 7] Supervision tree started - ConnMgr: ~p, PubSub: ~p, RPC: ~p, AdvMgr: ~p",
[ConnMgrPid, PubSubPid, RpcPid, AdvMgrPid]),
State = #state{
url = Url,
opts = Opts#{host => Host, port => Port},
realm = Realm,
node_id = NodeId,
%% Phase 7: New supervision tree fields
supervisor_pid = SupPid,
connection_manager_pid = ConnMgrPid,
pubsub_handler_pid = PubSubPid,
rpc_handler_pid = RpcPid,
advertisement_manager_pid = AdvMgrPid,
%% Legacy fields (still needed for backward compatibility during Phase 7)
connection = undefined,
stream = undefined,
status = connecting,
subscriptions = #{},
pending_calls = #{},
pending_queries = #{},
pending_pubacks = #{},
msg_id_counter = 0,
recv_buffer = <<>>,
service_registry = macula_service_registry:new(),
advertised_services = #{},
advertised_subscriptions = #{},
topic_separator = TopicSeparator,
topic_wildcard_single = TopicWildcardSingle,
topic_wildcard_multi = TopicWildcardMulti,
provider_selector_state = #{
strategy => SelectionStrategy,
counters => #{}
},
endpoint_connections = #{},
pending_subscriber_queries = #{}
},
%% Register this connection process with gproc so the gateway can find it
gproc:reg({p, l, macula_connection}),
%% Schedule periodic cache pruning (every 60 seconds)
erlang:send_after(60000, self(), prune_caches),
%% NOTE: Connection is now managed by connection_manager, not here
%% The old "self() ! connect" is removed
{ok, State}.
%% @private
%% Phase 7.3: Delegate to pubsub_handler
handle_call({publish, Topic, Data, Opts}, _From, State) ->
Result = macula_pubsub_handler:publish(State#state.pubsub_handler_pid, Topic, Data, Opts),
{reply, Result, State};
%% Phase 7.3: Delegate to pubsub_handler
handle_call({subscribe, Topic, Callback}, _From, State) ->
Result = macula_pubsub_handler:subscribe(State#state.pubsub_handler_pid, Topic, Callback),
{reply, Result, State};
%% Phase 7.3: Delegate to pubsub_handler
handle_call({unsubscribe, SubRef}, _From, State) ->
Result = macula_pubsub_handler:unsubscribe(State#state.pubsub_handler_pid, SubRef),
{reply, Result, State};
%% Phase 7.3: Delegate to rpc_handler
handle_call({call, Procedure, Args, Opts}, _From, State) ->
Result = macula_rpc_handler:call(State#state.rpc_handler_pid, Procedure, Args, Opts),
{reply, Result, State};
%% Phase 7.3: Delegate to advertisement_manager
handle_call({advertise, Procedure, Handler, Opts}, _From, State) ->
Result = macula_advertisement_manager:advertise_service(State#state.advertisement_manager_pid, Procedure, Handler, Opts),
{reply, Result, State};
%% Phase 7.3: Delegate to advertisement_manager
handle_call({unadvertise, Procedure}, _From, State) ->
Result = macula_advertisement_manager:unadvertise_service(State#state.advertisement_manager_pid, Procedure),
{reply, Result, State};
handle_call(_Request, _From, State) ->
{reply, {error, unknown_request}, State}.
%% @private
handle_cast({do_publish, PublishMsg, Qos, BinaryTopic, Payload, _Opts, MsgId}, State) ->
%% Spawn the publish send to avoid blocking the gen_server
%% Even async_send can block at NIF level or due to flow control
Stream = State#state.stream,
NodeId = State#state.node_id,
spawn(fun() ->
case send_message_raw(publish, PublishMsg, Stream) of
ok ->
ok;
{error, Reason} ->
?LOG_WARNING("[~s] Failed to send publish: ~p", [NodeId, Reason])
end
end),
%% If QoS 1, track for acknowledgment
State2 = track_qos_if_needed(Qos, MsgId, BinaryTopic, Payload, State),
%% Trigger async discovery of remote subscribers (mesh-wide pub/sub)
%% Using cast to make the entire discovery process asynchronous
gen_server:cast(self(), {discover_subscribers, BinaryTopic, Payload, Qos, _Opts}),
{noreply, State2};
%% @private
%% Handle async discovery of remote subscribers (mesh-wide pub/sub)
%% Spawn the entire discovery process to avoid any blocking
handle_cast({discover_subscribers, Topic, Payload, Qos, Opts}, State) ->
%% Spawn the entire discovery process to completely avoid blocking
spawn(fun() ->
_State2 = discover_remote_subscribers(Topic, Payload, Qos, Opts, State),
%% Send updated state back to gen_server if needed (for now, discovery is fire-and-forget)
ok
end),
{noreply, State};
%% @doc Handle routed RPC REPLY from local gateway
handle_cast({rpc_route_reply, RpcRouteMsg}, State) ->
io:format("[Connection ~s] Received routed REPLY from gateway~n", [State#state.node_id]),
%% Process the rpc_route message using existing handler
NewState = process_message({rpc_route, RpcRouteMsg}, State),
{noreply, NewState};
handle_cast(_Msg, State) ->
{noreply, State}.
%% @private
handle_info(connect, State) ->
case do_connect(State) of
{ok, State2} ->
{noreply, State2};
{error, Reason} ->
?LOG_ERROR("Connection failed: ~p, retrying in ~p ms",
[Reason, ?CONNECT_RETRY_DELAY]),
erlang:send_after(?CONNECT_RETRY_DELAY, self(), connect),
{noreply, State#state{status = error}}
end;
handle_info({quic, Data, Stream, _Props}, State) when is_binary(Data) ->
%% Received data from QUIC stream (main or endpoint connection)
MainStream = State#state.stream,
EndpointConnections = State#state.endpoint_connections,
%% Check if this is the main stream or an endpoint stream
IsMainStream = (Stream =:= MainStream),
IsEndpointStream = lists:any(fun({_Endpoint, #{stream := S}}) -> S =:= Stream end,
maps:to_list(EndpointConnections)),
StreamType = if
IsMainStream -> "main";
IsEndpointStream -> "endpoint";
true -> "unknown"
end,
?LOG_INFO("Received ~p bytes on ~s stream ~p", [byte_size(Data), StreamType, Stream]),
%% Valid stream - process the data, otherwise ignore
handle_stream_data(IsMainStream orelse IsEndpointStream, Data, Stream, State);
%% Handle QUIC control messages (non-binary Data)
handle_info({quic, ControlMsg, _Stream, _Props}, State) when is_atom(ControlMsg) ->
%% QUIC control message (send_shutdown_complete, etc.) - ignore
?LOG_DEBUG("Ignoring QUIC control message: ~p", [ControlMsg]),
{noreply, State};
handle_info({call_timeout, CallId}, State) ->
%% RPC call timed out
case maps:get(CallId, State#state.pending_calls, undefined) of
undefined ->
{noreply, State};
{From, _Timer} ->
%% Simple timeout without failover context
?LOG_WARNING("RPC call ~s timed out (no failover available)", [CallId]),
gen_server:reply(From, {error, timeout}),
PendingCalls = maps:remove(CallId, State#state.pending_calls),
{noreply, State#state{pending_calls = PendingCalls}};
{From, _Timer, FailoverContext} ->
%% Timeout with failover context - try next provider
#{
procedure := Procedure,
args := Args,
opts := Opts,
all_providers := AllProviders,
excluded_providers := ExcludedProviders,
attempt := Attempt
} = FailoverContext,
?LOG_WARNING("RPC call to ~s timed out on attempt ~p, trying next provider",
[Procedure, Attempt]),
%% Remove this call from pending
PendingCalls = maps:remove(CallId, State#state.pending_calls),
State2 = State#state{pending_calls = PendingCalls},
%% Retry with next provider
case do_remote_call_with_failover(Procedure, Args, Opts, From,
AllProviders, ExcludedProviders,
Attempt + 1, State2) of
{noreply, State3} ->
{noreply, State3};
{reply, Error, State3} ->
%% All providers exhausted or max attempts reached
gen_server:reply(From, Error),
{noreply, State3}
end
end;
handle_info({find_value_timeout, ServiceKey}, State) ->
%% DHT FIND_VALUE query timed out
case maps:get(ServiceKey, State#state.pending_queries, undefined) of
undefined ->
%% Already handled or cancelled
{noreply, State};
{From, Procedure, Args, Opts, _Registry, _Timer} ->
%% Query timed out - fall back to direct call
?LOG_WARNING("FIND_VALUE query for service ~s timed out. Attempting direct call.", [Procedure]),
PendingQueries = maps:remove(ServiceKey, State#state.pending_queries),
State2 = State#state{pending_queries = PendingQueries},
do_direct_call(Procedure, Args, Opts, From, State2)
end;
handle_info({readvertise, Procedure}, State) ->
%% Re-advertisement timer fired - re-publish to DHT
AdvertisedServices = State#state.advertised_services,
case maps:get(Procedure, AdvertisedServices, undefined) of
undefined ->
%% Service was unadvertised, timer should have been cancelled but wasn't
?LOG_WARNING("Re-advertisement timer fired for unadvertised service: ~s", [Procedure]),
{noreply, State};
#{metadata := Metadata, ttl := TTL} = ServiceInfo ->
%% Re-publish to DHT
ProviderInfo = #{
node_id => State#state.node_id,
endpoint => State#state.url,
metadata => Metadata
},
case macula_service_registry:publish_to_dht(macula_routing_server, Procedure, ProviderInfo, TTL, 20) of
ok ->
?LOG_DEBUG("Re-advertised service ~s to DHT", [Procedure]);
{error, Reason} ->
?LOG_WARNING("Failed to re-advertise service ~s to DHT: ~p", [Procedure, Reason])
end,
%% Schedule next re-advertisement
ReadvInterval = max(10, TTL - 60) * 1000,
TimerRef = erlang:send_after(ReadvInterval, self(), {readvertise, Procedure}),
?LOG_DEBUG("Scheduled next re-advertisement for ~s in ~p seconds", [Procedure, ReadvInterval div 1000]),
%% Update timer reference
ServiceInfo2 = ServiceInfo#{timer_ref => TimerRef},
AdvertisedServices2 = AdvertisedServices#{Procedure => ServiceInfo2},
{noreply, State#state{advertised_services = AdvertisedServices2}}
end;
%% Re-advertise subscription in DHT (periodic refresh)
handle_info({resubscribe, Topic}, State) ->
case maps:get(Topic, State#state.advertised_subscriptions, undefined) of
undefined ->
%% Subscription was removed, don't re-advertise
?LOG_DEBUG("[~s] Skipping re-subscription for ~s (no longer subscribed)",
[State#state.node_id, Topic]),
{noreply, State};
SubInfo ->
%% Cancel old timer
OldTimerRef = maps:get(timer_ref, SubInfo),
erlang:cancel_timer(OldTimerRef),
%% Re-advertise (same as initial advertisement)
SubRef = maps:get(sub_ref, SubInfo),
State2 = advertise_subscription_in_dht(Topic, SubRef, State),
?LOG_DEBUG("[~s] Re-advertised subscription for topic ~s",
[State#state.node_id, Topic]),
{noreply, State2}
end;
handle_info({puback_timeout, MsgId}, State) ->
%% Handle PUBACK timeout - retry or give up
PendingPubacks = State#state.pending_pubacks,
case maps:get(MsgId, PendingPubacks, undefined) of
undefined ->
%% Already acknowledged or removed
{noreply, State};
{Topic, Payload, Qos, RetryCount, _OldTimerRef} ->
NewRetryCount = RetryCount + 1,
if
NewRetryCount >= ?PUBACK_MAX_RETRIES ->
%% Max retries reached, give up
?LOG_ERROR("[~s] PUBACK timeout for message ~p (topic: ~s) after ~p retries - giving up",
[State#state.node_id, MsgId, Topic, NewRetryCount]),
{noreply, State#state{pending_pubacks = maps:remove(MsgId, PendingPubacks)}};
true ->
%% Retry sending the message
?LOG_WARNING("[~s] PUBACK timeout for message ~p (topic: ~s) - retry ~p/~p",
[State#state.node_id, MsgId, Topic, NewRetryCount, ?PUBACK_MAX_RETRIES]),
%% Rebuild and resend publish message
PublishMsg = #{
topic => Topic,
payload => Payload,
qos => Qos,
retain => false,
message_id => MsgId
},
case send_message(publish, PublishMsg, State) of
ok ->
%% Start new timer
NewTimerRef = erlang:send_after(?PUBACK_TIMEOUT, self(), {puback_timeout, MsgId}),
%% Update pending with new retry count and timer
PendingPubacks2 = PendingPubacks#{
MsgId => {Topic, Payload, Qos, NewRetryCount, NewTimerRef}
},
{noreply, State#state{pending_pubacks = PendingPubacks2}};
{error, Reason} ->
?LOG_ERROR("[~s] Failed to retry PUBACK message ~p: ~p",
[State#state.node_id, MsgId, Reason]),
{noreply, State#state{pending_pubacks = maps:remove(MsgId, PendingPubacks)}}
end
end
end;
%% Periodic cache pruning
handle_info(prune_caches, State) ->
Registry = State#state.service_registry,
%% Prune expired entries from both service cache and subscriber cache
{Registry2, ServicesPruned} = macula_service_registry:prune_expired_services(Registry),
{Registry3, SubscribersPruned} = macula_service_registry:prune_expired_subscribers(Registry2),
TotalPruned = ServicesPruned + SubscribersPruned,
case TotalPruned of
0 ->
ok; % No entries expired
_ ->
?LOG_DEBUG("[~s] Cache pruning: removed ~p service entries, ~p subscriber entries",
[State#state.node_id, ServicesPruned, SubscribersPruned])
end,
%% Schedule next pruning in 60 seconds
erlang:send_after(60000, self(), prune_caches),
{noreply, State#state{service_registry = Registry3}};
handle_info(Info, State) ->
%% Ignore unexpected messages
?LOG_DEBUG("Unhandled handle_info message: ~p", [Info]),
{noreply, State}.
%% @private
terminate(_Reason, #state{stream = Stream, connection = Conn, endpoint_connections = EndpointConns}) ->
%% Clean up QUIC resources
catch macula_quic:close(Stream),
catch macula_quic:close(Conn),
%% Close all endpoint connections
lists:foreach(fun({_Endpoint, #{connection := C, stream := S}}) ->
catch macula_quic:close(S),
catch macula_quic:close(C)
end, maps:to_list(EndpointConns)),
ok.
%% @private
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%%%===================================================================
%%% Internal functions
%%%===================================================================
%% @doc Extract and normalize realm from options (pattern matching on type).
-spec get_realm_from_opts(map()) -> binary().
get_realm_from_opts(Opts) ->
normalize_realm(maps:get(realm, Opts, undefined)).
normalize_realm(undefined) ->
error({missing_required_option, realm});
normalize_realm(Realm) when is_binary(Realm) ->
Realm;
normalize_realm(Realm) when is_list(Realm) ->
list_to_binary(Realm);
normalize_realm(Realm) when is_atom(Realm) ->
atom_to_binary(Realm).
%% @doc Find child PID from supervisor children list (Phase 7 helper).
-spec find_child_pid(list(), atom()) -> pid().
find_child_pid(Children, ChildId) ->
extract_child_pid(lists:keyfind(ChildId, 1, Children), ChildId).
extract_child_pid({_, Pid, _Type, _Modules}, _ChildId) when is_pid(Pid) ->
Pid;
extract_child_pid({_, undefined, _Type, _Modules}, ChildId) ->
error({child_not_started, ChildId});
extract_child_pid(false, ChildId) ->
error({child_not_found, ChildId}).
%% @doc Establish QUIC connection and perform handshake.
-spec do_connect(#state{}) -> {ok, #state{}} | {error, term()}.
do_connect(State) ->
#{host := Host, port := Port} = State#state.opts,
%% Connect via QUIC
QuicOpts = [
{alpn, ["macula"]},
{verify, none} %% TODO: Add proper TLS verification
],
ConnectResult = try
macula_quic:connect(Host, Port, QuicOpts, ?DEFAULT_TIMEOUT)
catch
_:Error ->
{error, Error}
end,
case ConnectResult of
{ok, Conn} ->
%% Open bidirectional stream
case macula_quic:open_stream(Conn) of
{ok, Stream} ->
%% Set stream to active mode so we receive messages
case quicer:setopt(Stream, active, true) of
ok ->
?LOG_DEBUG("Stream set to active mode");
{error, SetOptErr} ->
?LOG_WARNING("Failed to set stream active: ~p", [SetOptErr])
end,
%% Send CONNECT message
%% Include advertise endpoint for peer-to-peer connections
LocalEndpoint = case application:get_env(macula, advertise_endpoint) of
{ok, Endpoint} when is_binary(Endpoint) -> Endpoint;
_ ->
%% Construct from NODE_HOST env var
NodeHost = list_to_binary(os:getenv("NODE_HOST", "localhost")),
<<"https://", NodeHost/binary, ":9443">>
end,
ConnectMsg = #{
version => <<"1.0">>,
node_id => State#state.node_id,
realm_id => State#state.realm,
capabilities => [pubsub, rpc],
endpoint => LocalEndpoint
},
case send_message_raw(connect, ConnectMsg, Stream) of
ok ->
?LOG_INFO("Connected to Macula mesh: ~s:~p", [Host, Port]),
%% Add connected server node to DHT routing table
try
%% Parse the server's endpoint URL to get address
ServerAddress = parse_server_endpoint(State#state.url),
%% Use correct node_info structure with address field
ServerNodeInfo = #{
node_id => crypto:strong_rand_bytes(32), %% We'll get actual node_id from CONNECTED response
address => ServerAddress
},
macula_routing_server:add_node(macula_routing_server, ServerNodeInfo),
?LOG_DEBUG("Added server node to DHT routing table with address: ~p", [ServerAddress])
catch
_:E ->
?LOG_WARNING("Failed to add server node to DHT: ~p", [E])
end,
{ok, State#state{
connection = Conn,
stream = Stream,
status = connected
}};
{error, Reason} ->
macula_quic:close(Stream),
macula_quic:close(Conn),
{error, {handshake_failed, Reason}}
end;
{error, Reason} ->
macula_quic:close(Conn),
{error, {stream_open_failed, Reason}}
end;
{error, Reason} ->
{error, {connection_failed, Reason}};
{error, Type, Details} ->
{error, {connection_failed, {Type, Details}}};
Other ->
{error, {connection_failed, Other}}
end.
%% @doc Send a protocol message through the established stream.
-spec send_message(atom(), map(), #state{}) -> ok | {error, term()}.
send_message(Type, Msg, #state{stream = Stream}) ->
send_message_raw(Type, Msg, Stream).
%% @doc Send a protocol message through a stream (raw).
-spec send_message_raw(atom(), map(), pid()) -> ok | {error, term()}.
send_message_raw(Type, Msg, Stream) ->
try
io:format("[send_message_raw] Type=~p, Msg=~p~n", [Type, Msg]),
Binary = macula_protocol_encoder:encode(Type, Msg),
macula_quic:async_send(Stream, Binary)
catch
error:Reason:Stacktrace ->
io:format("[send_message_raw] ERROR: ~p~nStacktrace: ~p~n", [Reason, Stacktrace]),
{error, {encode_error, Reason}}
end.
%% @doc Dispatch stream data based on validity (pattern matching on boolean).
handle_stream_data(true, Data, _Stream, State) ->
handle_received_data(Data, State);
handle_stream_data(false, _Data, Stream, State) ->
?LOG_WARNING("Received data from unknown stream: ~p", [Stream]),
{noreply, State}.
%% @doc Track QoS 1 message for acknowledgment (pattern matching on QoS level).
track_qos_if_needed(1, MsgId, BinaryTopic, Payload, State) ->
%% Start timeout timer
TimerRef = erlang:send_after(?PUBACK_TIMEOUT_MS, self(), {puback_timeout, MsgId}),
%% Store pending acknowledgment
PendingPubacks = State#state.pending_pubacks,
State#state{
pending_pubacks = PendingPubacks#{
MsgId => {BinaryTopic, Payload, 1, 0, TimerRef}
}
};
track_qos_if_needed(_Qos, _MsgId, _BinaryTopic, _Payload, State) ->
State.
%% @doc Handle received data from the stream.
-spec handle_received_data(binary(), #state{}) -> {noreply, #state{}}.
handle_received_data(Data, State) ->
%% Append to receive buffer
Buffer = <<(State#state.recv_buffer)/binary, Data/binary>>,
%% Try to decode messages
{Messages, RemainingBuffer} = decode_messages(Buffer, []),
%% Process each message
State2 = lists:foldl(fun process_message/2, State, Messages),
{noreply, State2#state{recv_buffer = RemainingBuffer}}.
%% @doc Decode all complete messages from buffer.
-spec decode_messages(binary(), list()) -> {list(), binary()}.
decode_messages(Buffer, Acc) when byte_size(Buffer) < 8 ->
%% Not enough for header
{lists:reverse(Acc), Buffer};
decode_messages(<<_Version:8, _TypeId:8, _Flags:8, _Reserved:8,
PayloadLen:32/big-unsigned, Rest/binary>> = Buffer, Acc) ->
case byte_size(Rest) of
ActualLen when ActualLen >= PayloadLen ->
%% We have a complete message
case macula_protocol_decoder:decode(Buffer) of
{ok, {Type, Msg}} ->
%% Skip this message and continue
<<_:8/binary, _Payload:PayloadLen/binary, Remaining/binary>> = Buffer,
decode_messages(Remaining, [{Type, Msg} | Acc]);
{error, _Reason} ->
%% Decode error, skip this message
<<_:8/binary, _Payload:PayloadLen/binary, Remaining/binary>> = Buffer,
decode_messages(Remaining, Acc)
end;
_ ->
%% Incomplete message, wait for more data
{lists:reverse(Acc), Buffer}
end.
%% @doc Process a received message.
-spec process_message({atom(), map()}, #state{}) -> #state{}.
process_message({publish, Msg}, State) ->
%% Handle incoming publish (for subscriptions)
%% Support both atom and binary keys from MessagePack decoding
Topic = case maps:get(topic, Msg, undefined) of
undefined -> maps:get(<<"topic">>, Msg);
T -> T
end,
Payload = case maps:get(payload, Msg, undefined) of
undefined -> maps:get(<<"payload">>, Msg);
P -> P
end,
?LOG_DEBUG("[~s] Received PUBLISH message: topic=~s, payload_size=~p bytes",
[State#state.node_id, Topic, byte_size(Payload)]),
%% Find matching subscriptions and invoke callbacks
SubscriptionMatches = lists:filter(
fun({_SubRef, {SubTopic, _Callback}}) ->
topic_matches(SubTopic, Topic, State)
end,
maps:to_list(State#state.subscriptions)
),
case SubscriptionMatches of
[] ->
?LOG_DEBUG("[~s] No matching subscriptions for topic: ~s",
[State#state.node_id, Topic]);
Matches ->
?LOG_INFO("[~s] Found ~p subscription(s) for topic: ~s",
[State#state.node_id, length(Matches), Topic]),
%% Invoke each matching callback
lists:foreach(fun({SubRef, {SubTopic, Callback}}) ->
spawn(fun() ->
try
%% Decode payload
Event = macula_utils:decode_json(Payload),
?LOG_DEBUG("[~s] Invoking callback for subscription ~p (topic: ~s)",
[State#state.node_id, SubRef, SubTopic]),
%% Invoke callback
Callback(Event),
?LOG_DEBUG("[~s] Callback completed successfully for subscription ~p",
[State#state.node_id, SubRef])
catch
Class:Reason:Stacktrace ->
?LOG_ERROR("[~s] Callback error for subscription ~p: ~p:~p~nStacktrace: ~p",
[State#state.node_id, SubRef, Class, Reason, Stacktrace])
end
end)
end, Matches)
end,
State;
process_message({call, Msg}, State) ->
%% Handle incoming RPC call (for locally advertised services)
#{procedure := Procedure, args := Args, call_id := CallId} = Msg,
%% Look up local handler
case macula_service_registry:get_local_handler(State#state.service_registry, Procedure) of
{ok, Handler} ->
%% Decode args
DecodedArgs = macula_utils:decode_json(Args),
%% Execute handler (spawn to avoid blocking)
spawn(fun() ->
try
Result = Handler(DecodedArgs),
send_rpc_reply(CallId, Result, State)
catch
_:Error ->
send_rpc_error(CallId, Error, State)
end
end),
State;
not_found ->
%% Service not found locally
spawn(fun() -> send_rpc_error(CallId, service_not_found, State) end),
State
end;
process_message({reply, Msg}, State) ->
%% Handle RPC reply
%% MessagePack decoder returns binary keys
#{<<"call_id">> := CallId} = Msg,
case maps:get(CallId, State#state.pending_calls, undefined) of
undefined ->
?LOG_WARNING("[~s] Received REPLY for unknown call_id: ~p",
[State#state.node_id, binary:encode_hex(CallId)]),
State;
{From, Timer} ->
%% Simple reply without failover context
erlang:cancel_timer(Timer),
case maps:get(<<"result">>, Msg, undefined) of
undefined ->
%% Error response
Error = maps:get(<<"error">>, Msg, <<"Unknown error">>),
gen_server:reply(From, {error, Error}),
State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)};
Result ->
%% Success response
DecodedResult = macula_utils:decode_json(Result),
gen_server:reply(From, {ok, DecodedResult}),
State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)}
end;
{From, Timer, FailoverContext} ->
%% Reply with failover context - check for errors and retry if needed
erlang:cancel_timer(Timer),
case maps:get(<<"result">>, Msg, undefined) of
undefined ->
%% Error response - try failover
#{
procedure := Procedure,
args := Args,
opts := Opts,
all_providers := AllProviders,
excluded_providers := ExcludedProviders,
attempt := Attempt
} = FailoverContext,
Error = maps:get(<<"error">>, Msg, <<"Unknown error">>),
?LOG_WARNING("RPC call to ~s failed with error: ~p (attempt ~p), trying next provider",
[Procedure, Error, Attempt]),
%% Remove this call from pending
PendingCalls = maps:remove(CallId, State#state.pending_calls),
State2 = State#state{pending_calls = PendingCalls},
%% Retry with next provider
case do_remote_call_with_failover(Procedure, Args, Opts, From,
AllProviders, ExcludedProviders,
Attempt + 1, State2) of
{noreply, State3} ->
State3;
{reply, ErrorReply, State3} ->
%% All providers exhausted or max attempts reached
gen_server:reply(From, ErrorReply),
State3
end;
Result ->
%% Success response
DecodedResult = macula_utils:decode_json(Result),
gen_server:reply(From, {ok, DecodedResult}),
State#state{pending_calls = maps:remove(CallId, State#state.pending_calls)}
end
end;
process_message({pong, _Msg}, State) ->
%% TODO: Handle ping/pong for keepalive
State;
process_message({rpc_route, RpcRouteMsg}, State) ->
%% Handle routed RPC messages (CALL or REPLY)
LocalNodeId = State#state.node_id,
case macula_rpc_routing:should_deliver_locally(LocalNodeId, RpcRouteMsg) of
true ->
%% Message is for us - unwrap and deliver
%% MessagePack decoder returns binary keys
#{<<"payload_type">> := PayloadType, <<"payload">> := Payload} = RpcRouteMsg,
case PayloadType of
<<"reply">> ->
%% Routed REPLY coming back from provider
?LOG_INFO("[~s] Received routed REPLY", [State#state.node_id]),
process_message({reply, Payload}, State);
<<"call">> ->
%% Routed CALL - forward to gateway for handler processing
?LOG_INFO("[~s] Connection received routed CALL, forwarding to gateway",
[State#state.node_id]),
case whereis(macula_gateway) of
undefined ->
?LOG_WARNING("[~s] No gateway running, cannot process routed CALL",
[State#state.node_id]),
State;
GatewayPid ->
%% Send the entire rpc_route message to gateway for processing
gen_server:cast(GatewayPid, {process_rpc_route, RpcRouteMsg}),
State
end
end;
false ->
%% Not for us - this shouldn't happen at connection level
%% (gateway does the forwarding)
?LOG_WARNING("[~s] Connection received rpc_route not for us (destination: ~p)",
[State#state.node_id,
binary:encode_hex(maps:get(<<"destination_node_id">>, RpcRouteMsg))]),
State
end;
process_message({find_value_reply, Msg}, State) ->
%% Handle FIND_VALUE_REPLY messages from DHT
io:format("[~s] ==> RECEIVED FIND_VALUE_REPLY MESSAGE <=~n", [State#state.node_id]),
handle_find_value_reply(Msg, State);
process_message({store_reply, _Msg}, State) ->
%% STORE replies don't need handling on client side
State;
process_message({puback, Msg}, State) ->
%% Handle PUBACK acknowledgment for QoS 1
MsgId = maps:get(message_id, Msg),
PendingPubacks = State#state.pending_pubacks,
case maps:get(MsgId, PendingPubacks, undefined) of
undefined ->
?LOG_WARNING("[~s] Received PUBACK for unknown message_id: ~p",
[State#state.node_id, MsgId]),
State;
{Topic, _Payload, _Qos, _RetryCount, TimerRef} ->
%% Cancel timeout timer
erlang:cancel_timer(TimerRef),
%% Remove from pending
?LOG_DEBUG("[~s] PUBACK received for message ~p (topic: ~s)",
[State#state.node_id, MsgId, Topic]),
State#state{pending_pubacks = maps:remove(MsgId, PendingPubacks)}
end;
process_message(OtherMsg, State) ->
%% Ignore unknown message types
io:format("[~s] Unknown message: ~p~n", [State#state.node_id, OtherMsg]),
State.
%% @doc Handle incoming FIND_VALUE_REPLY from DHT.
-spec handle_find_value_reply(map(), #state{}) -> #state{}.
handle_find_value_reply(Msg, State) ->
%% Check if this is for subscriber discovery or RPC service discovery
HasSubscriberQuery = maps:size(State#state.pending_subscriber_queries) > 0,
%% Decode the DHT protocol message
case macula_routing_protocol:decode_find_value_reply(Msg) of
{ok, {value, Provider}} when is_map(Provider) ->
%% Single provider returned - normalize to list and convert binary keys to atoms
Providers = [macula_utils:normalize_provider(Provider)],
?LOG_INFO("[~s] Received FIND_VALUE_REPLY with ~p provider(s)",
[State#state.node_id, length(Providers)]),
%% Log provider details for debugging
lists:foreach(fun(P) ->
NodeId = maps:get(node_id, P, <<"unknown">>),
Endpoint = maps:get(endpoint, P, <<"unknown">>),
?LOG_INFO("[~s] Provider: node_id=~s, endpoint=~s",
[State#state.node_id, NodeId, Endpoint])
end, Providers),
%% Route to appropriate handler
case HasSubscriberQuery of
true -> handle_subscriber_discovery_reply(Providers, State);
false -> continue_pending_query_with_providers(Providers, State)
end;
{ok, {value, Providers}} when is_list(Providers) ->
%% List of providers returned - normalize all providers
NormalizedProviders = lists:map(fun macula_utils:normalize_provider/1, Providers),
?LOG_INFO("[~s] Received FIND_VALUE_REPLY with ~p provider(s)",
[State#state.node_id, length(NormalizedProviders)]),
%% Log provider details for debugging
lists:foreach(fun(P) ->
NodeId = maps:get(node_id, P, <<"unknown">>),
Endpoint = maps:get(endpoint, P, <<"unknown">>),
?LOG_INFO("[~s] Provider: node_id=~s, endpoint=~s",
[State#state.node_id, NodeId, Endpoint])
end, NormalizedProviders),
%% Route to appropriate handler
case HasSubscriberQuery of
true -> handle_subscriber_discovery_reply(NormalizedProviders, State);
false -> continue_pending_query_with_providers(NormalizedProviders, State)
end;
{ok, {nodes, _Nodes}} ->
%% Service/subscribers not found
?LOG_INFO("[~s] Received FIND_VALUE_REPLY with nodes (not found in DHT)",
[State#state.node_id]),
case HasSubscriberQuery of
true -> handle_subscriber_discovery_reply([], State);
false -> continue_pending_query_with_no_providers(State)
end;
{error, Reason} ->
?LOG_WARNING("[~s] Failed to decode FIND_VALUE_REPLY: ~p",
[State#state.node_id, Reason]),
case HasSubscriberQuery of
true -> handle_subscriber_discovery_reply([], State);
false -> continue_pending_query_with_no_providers(State)
end
end.
%% @doc Continue pending query with discovered providers.
-spec continue_pending_query_with_providers(list(), #state{}) -> #state{}.
continue_pending_query_with_providers(Providers, State) ->
io:format("[~s] *** continue_pending_query_with_providers called with ~p providers ***~n",
[State#state.node_id, length(Providers)]),
case maps:to_list(State#state.pending_queries) of
[{ServiceKey, {From, Procedure, Args, Opts, Registry, Timer}} | _] ->
io:format("[~s] Found pending query for procedure ~s~n", [State#state.node_id, Procedure]),
%% Cancel timeout timer
erlang:cancel_timer(Timer),
%% Cache providers in service registry
CacheTTL = 60, % Cache for 1 minute
UpdatedRegistry = macula_service_registry:cache_service(Registry, Procedure, Providers, CacheTTL),
State2 = State#state{
service_registry = UpdatedRegistry,
pending_queries = maps:remove(ServiceKey, State#state.pending_queries)
},
%% Continue with the RPC call
io:format("[~s] About to call do_remote_call for ~s with ~p providers~n",
[State#state.node_id, Procedure, length(Providers)]),
case do_remote_call(Procedure, Args, Opts, From, Providers, State2) of
{noreply, State3} ->
io:format("[~s] do_remote_call returned {noreply, State}~n", [State#state.node_id]),
State3;
{reply, Reply, State3} ->
io:format("[~s] do_remote_call returned {reply, ~p, State}~n", [State#state.node_id, Reply]),
gen_server:reply(From, Reply),
State3
end;
[] ->
%% No pending query - this is unexpected but not fatal
?LOG_WARNING("Received FIND_VALUE_REPLY with providers but no pending query"),
State
end.
%% @doc Continue pending query with no providers (fall back to direct call).
-spec continue_pending_query_with_no_providers(#state{}) -> #state{}.
continue_pending_query_with_no_providers(State) ->
case maps:to_list(State#state.pending_queries) of
[{ServiceKey, {From, Procedure, Args, Opts, _Registry, Timer}} | _] ->
%% Cancel timeout timer
erlang:cancel_timer(Timer),
%% Remove from pending queries
State2 = State#state{pending_queries = maps:remove(ServiceKey, State#state.pending_queries)},
%% Fall back to direct call
?LOG_WARNING("Service ~s not found in DHT. Attempting direct call to connected endpoint.", [Procedure]),
case do_direct_call(Procedure, Args, Opts, From, State2) of
{noreply, State3} -> State3;
{reply, Reply, State3} -> gen_server:reply(From, Reply), State3
end;
[] ->
%% No pending query - this is unexpected but not fatal
?LOG_WARNING("Received FIND_VALUE_REPLY with no providers and no pending query"),
State
end.
%% @doc Handle subscriber discovery reply from DHT.
-spec handle_subscriber_discovery_reply(list(), #state{}) -> #state{}.
handle_subscriber_discovery_reply(Subscribers, State) ->
%% Get the first pending subscriber query (FIFO - should only be one at a time)
case maps:to_list(State#state.pending_subscriber_queries) of
[{QueryId, {Topic, Payload, Qos, _Opts}} | _] ->
?LOG_DEBUG("[~s] Handling subscriber discovery reply for topic ~s (query_id: ~s, ~p subscribers)",
[State#state.node_id, Topic, QueryId, length(Subscribers)]),
%% Remove from pending queries
PendingQueries = maps:remove(QueryId, State#state.pending_subscriber_queries),
State2 = State#state{pending_subscriber_queries = PendingQueries},
%% Cache discovered subscribers (TTL: 60 seconds)
CacheTTL = 60,
Registry = State2#state.service_registry,
Registry2 = macula_service_registry:cache_subscribers(Registry, Topic, Subscribers, CacheTTL),
State3 = State2#state{service_registry = Registry2},
%% Route to discovered remote subscribers
route_to_remote_subscribers(Topic, Payload, Qos, Subscribers, State3),
State3;
[] ->
%% No pending subscriber query - this is unexpected but not fatal
?LOG_WARNING("[~s] Received subscriber discovery reply but no pending query",
[State#state.node_id]),
State
end.
%% @doc Route a published message to remote subscribers.
-spec route_to_remote_subscribers(binary(), binary(), integer(), list(), #state{}) -> ok.
route_to_remote_subscribers(_Topic, _Payload, _Qos, [], State) ->
?LOG_DEBUG("[~s] No remote subscribers found for topic", [State#state.node_id]),
ok;
route_to_remote_subscribers(Topic, Payload, Qos, Subscribers, State) ->
?LOG_INFO("[~s] Routing message to ~p remote subscriber(s) for topic ~s",
[State#state.node_id, length(Subscribers), Topic]),
%% Route message to each remote subscriber
lists:foreach(fun(Subscriber) ->
Endpoint = maps:get(endpoint, Subscriber, <<"unknown">>),
NodeId = maps:get(node_id, Subscriber, <<"unknown">>),
%% Send the message to this subscriber
case send_to_remote_subscriber(Topic, Payload, Qos, Endpoint, NodeId, State) of
ok ->
?LOG_INFO("[~s] Successfully sent message to remote subscriber ~s at ~s",
[State#state.node_id, NodeId, Endpoint]);
{error, Reason} ->
?LOG_WARNING("[~s] Failed to send message to remote subscriber ~s: ~p",
[State#state.node_id, NodeId, Reason])
end
end, Subscribers),
ok.
%% @doc Send a publish message to a single remote subscriber endpoint.
-spec send_to_remote_subscriber(binary(), binary(), integer(), binary(), binary(), #state{}) ->
ok | {error, term()}.
send_to_remote_subscriber(Topic, Payload, Qos, Endpoint, NodeId, State) ->
%% Get or create connection to subscriber endpoint using connection pool
Pool = State#state.endpoint_connections,
case macula_connection_pool:get_or_create_connection(
Endpoint,
State#state.node_id,
State#state.realm,
Pool
) of
{ok, _Conn, Stream, _UpdatedPool} ->
%% Note: Not updating state with pool - maintains existing behavior
%% where connections weren't cached across calls (pre-existing issue)
%% Generate message ID for this publish
{MsgId, _State2} = next_message_id(State),
%% Build publish message
PublishMsg = #{
topic => Topic,
payload => Payload,
qos => Qos,
message_id => MsgId,
retain => false
},
%% Send via the subscriber's stream
case send_message_raw(publish, PublishMsg, Stream) of
ok ->
?LOG_DEBUG("[~s] Sent PUBLISH to remote subscriber ~s (msg_id: ~p)",
[State#state.node_id, NodeId, MsgId]),
ok;
{error, Reason} ->
?LOG_WARNING("[~s] Failed to send PUBLISH to ~s: ~p",
[State#state.node_id, Endpoint, Reason]),
{error, {send_failed, Reason}}
end;
{error, Reason, _Pool} ->
?LOG_WARNING("[~s] Failed to connect to remote subscriber ~s: ~p",
[State#state.node_id, Endpoint, Reason]),
{error, {connection_failed, Reason}}
end.
%% @doc Get next message ID.
-spec next_message_id(#state{}) -> {binary(), #state{}}.
next_message_id(State) ->
Counter = State#state.msg_id_counter,
{MsgId, NewCounter} = macula_utils:next_message_id(Counter),
{MsgId, State#state{msg_id_counter = NewCounter}}.
%% @doc Check if a published topic matches a subscription topic pattern.
%% Supports configurable wildcards (defaults: dot-separated with * and **):
%% - '*' matches a single segment (configurable via topic_wildcard_single)
%% - '**' matches multiple segments (configurable via topic_wildcard_multi)
%% - '.' as separator (configurable via topic_separator)
%% - exact match otherwise
-spec topic_matches(binary(), binary(), #state{}) -> boolean().
topic_matches(Pattern, Topic, State) ->
Separator = State#state.topic_separator,
WildcardSingle = State#state.topic_wildcard_single,
WildcardMulti = State#state.topic_wildcard_multi,
macula_utils:topic_matches(Pattern, Topic, Separator, WildcardSingle, WildcardMulti).
%% @doc Send RPC reply for successful handler execution.
-spec send_rpc_reply(binary(), term(), #state{}) -> ok | {error, term()}.
send_rpc_reply(CallId, {ok, Result}, State) ->
%% Success result
EncodedResult = macula_utils:encode_json(Result),
ReplyMsg = #{
call_id => CallId,
result => EncodedResult
},
send_message(reply, ReplyMsg, State);
send_rpc_reply(CallId, {error, Reason}, State) ->
%% Error result
send_rpc_error(CallId, Reason, State).
%% @doc Send RPC error reply.
-spec send_rpc_error(binary(), term(), #state{}) -> ok | {error, term()}.
send_rpc_error(CallId, Reason, State) ->
ErrorMsg = #{
call_id => CallId,
error => macula_utils:ensure_binary(io_lib:format("~p", [Reason]))
},
send_message(error, ErrorMsg, State).
%% @doc Make RPC call to a remote provider (with service discovery).
%% Supports automatic failover to alternate providers on failure.
-spec do_remote_call(binary(), term(), map(), term(), list(), #state{}) ->
{noreply, #state{}} | {reply, {error, term()}, #state{}}.
do_remote_call(Procedure, Args, Opts, From, Providers, State) ->
io:format("[~s] do_remote_call ENTER~n", [State#state.node_id]),
Result = do_remote_call_with_failover(Procedure, Args, Opts, From, Providers, [], 1, State),
io:format("[~s] do_remote_call EXIT with ~p~n", [State#state.node_id, element(1, Result)]),
Result.
%% @doc Make RPC call with failover support.
%% ExcludedProviders is a list of node_ids that have already been tried and failed.
%% Attempt is the current attempt number (starts at 1).
-spec do_remote_call_with_failover(binary(), term(), map(), term(), list(), list(), pos_integer(), #state{}) ->
{noreply, #state{}} | {reply, {error, term()}, #state{}}.
do_remote_call_with_failover(Procedure, Args, Opts, From, Providers, ExcludedProviders, Attempt, State) ->
io:format("[~s] do_remote_call_with_failover ENTER: ~p providers, attempt ~p~n",
[State#state.node_id, length(Providers), Attempt]),
%% Get max attempts from options (default: 3 or number of providers, whichever is smaller)
MaxAttempts = maps:get(max_attempts, Opts, min(3, length(Providers))),
io:format("[~s] MaxAttempts: ~p~n", [State#state.node_id, MaxAttempts]),
%% Filter out excluded providers
AvailableProviders = case ExcludedProviders of
[] ->
Providers;
_ ->
lists:filter(fun(#{node_id := NodeId}) ->
not lists:member(NodeId, ExcludedProviders)
end, Providers)
end,
io:format("[~s] AvailableProviders: ~p~n", [State#state.node_id, length(AvailableProviders)]),
%% Check if we have any providers left
case AvailableProviders of
[] ->
?LOG_ERROR("All providers exhausted for service ~s after ~p attempts",
[Procedure, Attempt - 1]),
{reply, {error, all_providers_failed}, State};
_ when Attempt > MaxAttempts ->
?LOG_ERROR("Max attempts (~p) reached for service ~s", [MaxAttempts, Procedure]),
{reply, {error, max_attempts_exceeded}, State};
_ ->
%% Use provider selection strategy to pick a provider
SelectorState = State#state.provider_selector_state,
Strategy = maps:get(strategy, SelectorState, random),
%% Set the current service ID for round-robin tracking
SelectorState2 = SelectorState#{current_service_id => Procedure},
case macula_provider_selector:select_provider(AvailableProviders, Strategy, SelectorState2) of
{ok, Provider, SelectorState3} ->
#{node_id := NodeId} = Provider,
io:format("[~s] Provider selected: ~s~n", [State#state.node_id, binary:encode_hex(NodeId)]),
AttemptLog = case Attempt of
1 -> "";
N -> io_lib:format(" (attempt ~p/~p)", [N, MaxAttempts])
end,
?LOG_INFO("Calling service ~s via DHT routing to provider ~s~s (strategy: ~p, ~p providers available)",
[Procedure, binary:encode_hex(NodeId), AttemptLog, Strategy, length(AvailableProviders)]),
%% Update state with new selector state
State2 = State#state{provider_selector_state = SelectorState3},
io:format("[~s] About to call do_remote_call_to_provider~n", [State#state.node_id]),
%% Attempt the call and store failover context
case do_remote_call_to_provider(Procedure, Args, Opts, From,
Provider, Providers, ExcludedProviders,
Attempt, State2) of
{noreply, State3} ->
io:format("[~s] do_remote_call_to_provider returned {noreply, State}~n", [State#state.node_id]),
{noreply, State3};
{reply, Error, State3} ->
io:format("[~s] do_remote_call_to_provider returned {reply, ~p, State}~n", [State#state.node_id, Error]),
{reply, Error, State3}
end;
{error, no_providers} ->
?LOG_ERROR("No providers available for service ~s", [Procedure]),
{reply, {error, no_providers}, State}
end
end.
%% @doc Make direct RPC call to currently connected endpoint.
-spec do_direct_call(binary(), term(), map(), term(), #state{}) ->
{noreply, #state{}} | {reply, {error, term()}, #state{}}.
do_direct_call(Procedure, Args, Opts, From, State) ->
%% Generate call ID
{CallId, State2} = next_message_id(State),
%% Encode args
EncodedArgs = case Args of
A when is_binary(A) -> A;
A when is_map(A) -> macula_utils:encode_json(A);
A when is_list(A) -> macula_utils:encode_json(A)
end,
%% Build call message (RPC call) - use CallId directly as binary
CallMsg = #{
procedure => macula_utils:ensure_binary(Procedure),
args => EncodedArgs,
call_id => CallId
},
case send_message(call, CallMsg, State2) of
ok ->
%% Set up timeout timer
Timeout = maps:get(timeout, Opts, ?CALL_TIMEOUT),
Timer = erlang:send_after(Timeout, self(), {call_timeout, CallId}),
%% Store pending call (without failover context)
PendingCalls = maps:put(CallId, {From, Timer}, State2#state.pending_calls),
State3 = State2#state{pending_calls = PendingCalls},
{noreply, State3};
{error, Reason} ->
{reply, {error, Reason}, State2}
end.
%% @doc Make RPC call to a specific provider endpoint with failover context.
-spec do_remote_call_to_provider(binary(), term(), map(), term(), map(), list(), list(), pos_integer(), #state{}) ->
{noreply, #state{}} | {reply, {error, term()}, #state{}}.
do_remote_call_to_provider(Procedure, Args, Opts, From, Provider, AllProviders, ExcludedProviders, Attempt, State) ->
io:format("[~s] do_remote_call_to_provider ENTER~n", [State#state.node_id]),
%% Extract provider info
#{node_id := ProviderNodeId} = Provider,
io:format("[~s] Provider node_id: ~s~n", [State#state.node_id, binary:encode_hex(ProviderNodeId)]),
%% -------------------------------------------------------------------
%% NEW: DHT-routed RPC using rpc_route message
%% Instead of creating direct connection, route through mesh
%% -------------------------------------------------------------------
%% Generate call ID
{CallId, State2} = next_message_id(State),
io:format("[~s] Generated CallId: ~p~n", [State#state.node_id, CallId]),
%% Encode args
EncodedArgs = case Args of
A when is_binary(A) -> A;
A when is_map(A) -> macula_utils:encode_json(A);
A when is_list(A) -> macula_utils:encode_json(A)
end,
%% Build call message
CallMsg = #{
procedure => macula_utils:ensure_binary(Procedure),
args => EncodedArgs,
call_id => CallId
},
%% Wrap in rpc_route envelope for DHT routing
LocalNodeId = State2#state.node_id,
MaxHops = 10,
io:format("[~s] About to wrap in rpc_route envelope~n", [State#state.node_id]),
RpcRouteMsg = macula_rpc_routing:wrap_call(LocalNodeId, ProviderNodeId, CallMsg, MaxHops),
io:format("[~s] RPC route envelope created~n", [State#state.node_id]),
?LOG_INFO("Sending routed RPC call to provider ~p (call_id: ~p, max_hops: ~p)",
[binary:encode_hex(ProviderNodeId), CallId, MaxHops]),
io:format("[~s] About to send_message_raw rpc_route~n", [State#state.node_id]),
%% Send via main mesh stream (not direct endpoint connection)
case send_message_raw(rpc_route, RpcRouteMsg, State2#state.stream) of
ok ->
io:format("[~s] send_message_raw returned ok~n", [State#state.node_id]),
?LOG_INFO("Sent routed RPC call (call_id: ~p)", [CallId]),
%% Set up timeout timer
Timeout = maps:get(timeout, Opts, ?CALL_TIMEOUT),
Timer = erlang:send_after(Timeout, self(), {call_timeout, CallId}),
%% Store pending call WITH failover context
FailoverContext = #{
procedure => Procedure,
args => Args,
opts => Opts,
all_providers => AllProviders,
excluded_providers => [ProviderNodeId | ExcludedProviders],
attempt => Attempt,
provider_node_id => ProviderNodeId % Store for reply routing
},
PendingCalls = maps:put(CallId, {From, Timer, FailoverContext}, State2#state.pending_calls),
State3 = State2#state{pending_calls = PendingCalls},
io:format("[~s] do_remote_call_to_provider EXIT with {noreply, State}~n", [State#state.node_id]),
{noreply, State3};
{error, Reason} ->
io:format("[~s] send_message_raw returned error: ~p~n", [State#state.node_id, Reason]),
?LOG_WARNING("Failed to send routed RPC call: ~p", [Reason]),
{reply, {error, {send_failed, Reason}}, State2}
end.
%%%===================================================================
%%% Multi-Endpoint Connection Management
%%%===================================================================
%% Note: Endpoint connection management now handled by macula_connection_pool module
%%%===================================================================
%%% DHT-Based Subscriber Discovery
%%%===================================================================
%% @doc Query DHT for remote subscribers to a topic (exact match only)
%% Note: This implementation supports exact-match topics only. Wildcard subscription
%% discovery across the mesh is deferred to a future enhancement.
-spec discover_remote_subscribers(binary(), binary(), integer(), map(), #state{}) -> #state{}.
discover_remote_subscribers(Topic, Payload, Qos, _Opts, State) ->
NodeId = State#state.node_id,
Registry = State#state.service_registry,
%% Check subscriber cache first (pattern matching for cache hit/miss)
case macula_service_registry:discover_subscribers(Registry, Topic) of
{ok, Subscribers, Registry2} ->
%% Cache hit - route to cached subscribers immediately
?LOG_DEBUG("[~s] Cache hit for subscribers to topic: ~s (~p subscribers)",
[NodeId, Topic, length(Subscribers)]),
route_to_remote_subscribers(Topic, Payload, Qos, Subscribers, State),
State#state{service_registry = Registry2};
{cache_miss, Registry2} ->
%% Cache miss - query DHT asynchronously
State2 = State#state{service_registry = Registry2},
query_dht_for_subscribers(Topic, Payload, Qos, State2)
end.
%% @doc Query DHT for subscribers (extracted for clarity).
-spec query_dht_for_subscribers(binary(), binary(), integer(), #state{}) -> #state{}.
query_dht_for_subscribers(Topic, Payload, Qos, State) ->
NodeId = State#state.node_id,
%% Create DHT key from topic (same as subscription advertisement)
TopicKey = crypto:hash(sha256, Topic),
%% Generate message ID for tracking the query
MsgId = next_message_id(State),
State2 = State#state{msg_id_counter = MsgId},
?LOG_DEBUG("[~s] Querying DHT for remote subscribers to topic: ~s (MsgId: ~s)",
[NodeId, Topic, MsgId]),
%% Send FIND_VALUE query to DHT asynchronously
%% The response will come back via handle_info({quic, Data, ...})
spawn(fun() ->
try
FindValueMsg = macula_routing_protocol:encode_find_value(TopicKey),
case send_message(find_value, FindValueMsg, State2) of
ok ->
?LOG_DEBUG("[~s] Sent FIND_VALUE for topic ~s", [NodeId, Topic]);
{error, SendError} ->
?LOG_WARNING("[~s] Failed to send FIND_VALUE for topic ~s: ~p",
[NodeId, Topic, SendError])
end
catch
_:QueryError:Stack ->
?LOG_WARNING("[~s] Failed to query DHT for topic ~s: ~p~nStack: ~p",
[NodeId, Topic, QueryError, Stack])
end
end),
%% Track the pending query with topic/payload/qos so we can route when response arrives
PendingQueries = State2#state.pending_subscriber_queries,
PendingQueries2 = PendingQueries#{MsgId => {Topic, Payload, Qos, #{}}},
State2#state{pending_subscriber_queries = PendingQueries2}.
%%%===================================================================
%%% DHT Subscription Advertisement
%%%===================================================================
%% @doc Advertise a topic subscription in the DHT for mesh-wide pub/sub routing
-spec advertise_subscription_in_dht(binary(), reference(), #state{}) -> #state{}.
advertise_subscription_in_dht(Topic, SubRef, State) ->
%% Default TTL for subscriptions is 300 seconds (5 minutes)
TTL = 300,
%% Create DHT key by hashing the topic (same as service registry pattern)
TopicKey = crypto:hash(sha256, Topic),
%% Create subscriber value with endpoint info for routing messages back
SubscriberValue = #{
node_id => State#state.node_id,
endpoint => State#state.url,
ttl => TTL
},
%% Send STORE message to DHT
?LOG_INFO("[~s] Advertising subscription to topic ~s in DHT", [State#state.node_id, Topic]),
try
StoreMsg = macula_routing_protocol:encode_store(TopicKey, SubscriberValue),
case send_message(store, StoreMsg, State) of
ok ->
?LOG_DEBUG("[~s] Successfully stored subscription for ~s in DHT",
[State#state.node_id, Topic]);
{error, SendError} ->
?LOG_WARNING("[~s] Failed to send STORE for subscription ~s: ~p",
[State#state.node_id, Topic, SendError])
end
catch
_:DhtError ->
?LOG_WARNING("[~s] Failed to advertise subscription ~s in DHT: ~p (continuing)",
[State#state.node_id, Topic, DhtError])
end,
%% Schedule re-advertisement (TTL - 60 seconds, minimum 10 seconds)
ResubInterval = max(10, TTL - 60) * 1000, % milliseconds
TimerRef = erlang:send_after(ResubInterval, self(), {resubscribe, Topic}),
?LOG_DEBUG("[~s] Scheduled re-subscription for ~s in ~p seconds",
[State#state.node_id, Topic, ResubInterval div 1000]),
%% Store subscription info for re-advertisement
AdvertisedSubscriptions = State#state.advertised_subscriptions,
SubInfo = #{
sub_ref => SubRef,
ttl => TTL,
timer_ref => TimerRef
},
AdvertisedSubscriptions2 = AdvertisedSubscriptions#{Topic => SubInfo},
State#state{advertised_subscriptions = AdvertisedSubscriptions2}.
%% @doc Parse server endpoint URL to {IP, Port} tuple.
%% Uses idiomatic Erlang pattern matching on function heads.
-spec parse_server_endpoint(binary()) -> {{byte(), byte(), byte(), byte()}, inet:port_number()}.
parse_server_endpoint(Endpoint) when is_binary(Endpoint) ->
try
case uri_string:parse(Endpoint) of
#{host := Host, port := Port} when is_integer(Port) ->
HostStr = binary_to_list(Host),
case inet:getaddr(HostStr, inet) of
{ok, IPTuple} -> {IPTuple, Port};
{error, _Reason} ->
%% Fallback to placeholder if DNS fails
{{0,0,0,0}, Port}
end;
#{host := Host} ->
%% No port specified, use default 9443
HostStr = binary_to_list(Host),
case inet:getaddr(HostStr, inet) of
{ok, IPTuple} -> {IPTuple, 9443};
{error, _Reason} ->
{{0,0,0,0}, 9443}
end;
_ ->
%% Invalid URL format
{{0,0,0,0}, 0}
end
catch
_:_ ->
%% Parse error, return placeholder
{{0,0,0,0}, 0}
end;
parse_server_endpoint(_Other) ->
{{0,0,0,0}, 0}.