Packages
macula
0.37.2
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_system/macula_gateway_pubsub_router.erl
%%%-------------------------------------------------------------------
%%% @doc
%%% Macula Gateway Pub/Sub Router - DHT-Routed Message Distribution
%%%
%%% Handles distribution of pub/sub messages to both local and remote
%%% subscribers using multi-hop Kademlia DHT routing (v0.7.8+).
%%%
%%% Responsibilities:
%%% - Deliver messages to local subscribers via QUIC streams
%%% - Query DHT for remote subscribers
%%% - Route messages via DHT multi-hop (pubsub_route protocol)
%%% - Wrap PUBLISH messages in pubsub_route envelopes
%%%
%%% Extracted from macula_gateway.erl (v0.7.9) for better separation of concerns.
%%%
%%% @end
%%%-------------------------------------------------------------------
-module(macula_gateway_pubsub_router).
-include_lib("kernel/include/logger.hrl").
%% API
-export([
distribute/5
]).
%% Exported for testing
-ifdef(TEST).
-export([
deliver_to_local_subscribers/2,
route_to_remote_subscribers/5,
parse_endpoint/1,
parse_endpoint_host/2
]).
-endif.
%%%===================================================================
%%% API Functions
%%%===================================================================
%% @doc Distribute pub/sub message to both local and remote subscribers.
%% Uses DHT routing for remote subscribers (multi-hop Kademlia).
%% For connected clients, uses existing bidirectional streams instead of mesh connections.
-spec distribute(
LocalSubscribers :: [quicer:stream_handle()],
PubMsg :: map(),
LocalNodeId :: binary(),
Mesh :: pid(),
Clients :: pid()
) -> ok.
distribute(LocalSubscribers, PubMsg, LocalNodeId, Mesh, Clients) ->
Topic = maps:get(<<"topic">>, PubMsg),
%% Deliver to local and remote subscribers
deliver_to_local_subscribers(LocalSubscribers, PubMsg),
route_to_remote_subscribers(Topic, PubMsg, LocalNodeId, Mesh, Clients),
ok.
%%%===================================================================
%%% Internal Functions - Local Delivery
%%%===================================================================
%% @private
%% @doc Deliver message to local subscribers via QUIC streams.
-spec deliver_to_local_subscribers([quicer:stream_handle()], map()) -> ok.
deliver_to_local_subscribers([], _PubMsg) ->
ok;
deliver_to_local_subscribers(Subscribers, PubMsg) ->
PubBinary = macula_protocol_encoder:encode(publish, PubMsg),
lists:foreach(fun(Stream) -> send_to_local_stream(Stream, PubBinary) end, Subscribers),
ok.
%% @private
%% @doc Send message to a single local subscriber stream.
%% Handles both QUIC stream references and handler PIDs.
-spec send_to_local_stream(quicer:stream_handle() | pid(), binary()) -> ok.
send_to_local_stream(Stream, Binary) when is_pid(Stream) ->
%% Stream is a handler PID - send Erlang message
%% Decode the binary back to map for Erlang message
case macula_protocol_decoder:decode(Binary) of
{ok, {publish, PubMsg}} ->
Topic = maps:get(<<"topic">>, PubMsg),
Payload = maps:get(<<"payload">>, PubMsg),
Stream ! {publish, Topic, Payload},
ok;
{error, _DecodeReason} ->
ok
end;
send_to_local_stream(Stream, Binary) ->
%% Stream is a QUIC reference - send via QUIC
_ = macula_quic:send(Stream, Binary),
ok.
%%%===================================================================
%%% Internal Functions - DHT Remote Routing
%%%===================================================================
%% @private
%% @doc Route message to remote subscribers via direct P2P (v0.14.0+).
%% Each node queries its own local DHT for subscribers and sends directly.
%% DHT propagation ensures subscriptions are replicated to k closest nodes.
%% Bootstrap is NOT a broker - it's just a seed node for DHT.
-spec route_to_remote_subscribers(binary(), map(), binary(), pid(), pid()) -> ok.
route_to_remote_subscribers(Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
%% Query LOCAL DHT for subscribers and route directly to them (P2P)
%% Subscriptions are replicated via DHT propagation to k closest nodes
route_via_local_dht(Topic, PubMsg, LocalNodeId, Mesh, Clients).
%% @private
%% @doc Route via local DHT lookup - primary P2P routing path (v0.14.0+).
%% All nodes use this - subscriptions replicate via DHT to k closest nodes.
%% Uses subscriber cache for 5-10x latency improvement on hot topics.
-spec route_via_local_dht(binary(), map(), binary(), pid(), pid()) -> ok.
route_via_local_dht(Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
%% Try cache first for O(1) lookup
CacheResult = macula_subscriber_cache:lookup(Topic),
?LOG_DEBUG("[PubSubRouter] Cache lookup for ~s: ~p", [Topic, CacheResult]),
handle_cache_result(CacheResult, Topic, PubMsg, LocalNodeId, Mesh, Clients).
%% @private Handle cache lookup result
handle_cache_result({ok, []}, Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
%% Cache hit but empty - retry DHT (subscriptions may have arrived)
?LOG_DEBUG("[PubSubRouter] Cache empty for ~s, invalidating and retrying DHT", [Topic]),
macula_subscriber_cache:invalidate(Topic),
lookup_and_cache_subscribers(Topic, PubMsg, LocalNodeId, Mesh, Clients);
handle_cache_result({ok, CachedSubscribers}, Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
%% Cache hit with subscribers - use cached subscribers
?LOG_DEBUG("[PubSubRouter] Cache hit for ~s with ~p subscribers", [Topic, length(CachedSubscribers)]),
route_to_each_subscriber(CachedSubscribers, Topic, PubMsg, LocalNodeId, Mesh, Clients);
handle_cache_result({miss, _TopicKey}, Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
%% Cache miss - do DHT lookup and cache result
?LOG_DEBUG("[PubSubRouter] Cache miss for ~s, querying DHT", [Topic]),
lookup_and_cache_subscribers(Topic, PubMsg, LocalNodeId, Mesh, Clients).
%% @private
%% @doc Lookup subscribers from DHT and cache the result.
%% Uses rate-limiting to prevent discovery storms during high-frequency publishing.
-spec lookup_and_cache_subscribers(binary(), map(), binary(), pid(), pid()) -> ok.
lookup_and_cache_subscribers(Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
%% Check rate-limiting before querying DHT
case macula_subscriber_cache:should_query_dht(Topic) of
false ->
%% Rate-limited - skip DHT query to prevent discovery storms
?LOG_WARNING("[PubSubRouter] DHT query rate-limited for ~s", [Topic]),
ok;
true ->
%% Allowed to query - perform DHT lookup
do_dht_lookup(Topic, PubMsg, LocalNodeId, Mesh, Clients)
end.
%% @private
%% @doc Actually perform DHT lookup (after rate-limit check passes).
-spec do_dht_lookup(binary(), map(), binary(), pid(), pid()) -> ok.
do_dht_lookup(Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
TopicKey = crypto:hash(sha256, Topic),
%% Record that we're doing a DHT query (for rate-limiting)
macula_subscriber_cache:record_dht_query(Topic),
?LOG_DEBUG("[PubSubRouter] DHT lookup for ~s", [Topic]),
case macula_gateway_dht:lookup_value(TopicKey) of
{ok, RemoteSubscribers} ->
?LOG_DEBUG("[PubSubRouter] DHT found ~p subscriber(s) for ~s: ~p",
[length(RemoteSubscribers), Topic, RemoteSubscribers]),
%% Cache for future lookups
macula_subscriber_cache:store(Topic, RemoteSubscribers),
%% Store subscriber endpoints in direct routing table for future direct P2P
store_subscriber_routes(RemoteSubscribers),
route_to_each_subscriber(RemoteSubscribers, Topic, PubMsg, LocalNodeId, Mesh, Clients);
{error, not_found} ->
%% DO NOT cache empty results - subscriptions may arrive later (timing issue)
?LOG_DEBUG("[PubSubRouter] DHT returned not_found for ~s", [Topic]),
ok
end.
%% @private
%% @doc Store subscriber endpoints in direct routing table for direct P2P routing.
-spec store_subscriber_routes([map()]) -> ok.
store_subscriber_routes([]) ->
ok;
store_subscriber_routes([Subscriber | Rest]) ->
macula_direct_routing:store_from_subscriber(Subscriber),
store_subscriber_routes(Rest).
%% @private
%% @doc Route message to each remote subscriber via DHT.
-spec route_to_each_subscriber([map()], binary(), map(), binary(), pid(), pid()) -> ok.
route_to_each_subscriber([], _Topic, _PubMsg, _LocalNodeId, _Mesh, _Clients) ->
ok;
route_to_each_subscriber([Subscriber | Rest], Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
%% Log subscriber info for debugging node_id mismatch
SubNodeId = case maps:find(node_id, Subscriber) of
{ok, N} -> N;
error -> case maps:find(<<"node_id">>, Subscriber) of
{ok, N2} -> N2;
error -> <<"unknown">>
end
end,
?LOG_DEBUG("[PubSubRouter] Routing to subscriber node_id=~s, local_node_id=~s, topic=~s",
[binary:encode_hex(SubNodeId), binary:encode_hex(LocalNodeId), Topic]),
route_to_single_subscriber(Subscriber, Topic, PubMsg, LocalNodeId, Mesh, Clients),
route_to_each_subscriber(Rest, Topic, PubMsg, LocalNodeId, Mesh, Clients).
%% @private
%% @doc Route message to a single remote subscriber (if node_id present).
%% Checks if subscriber is a connected client first - if so, uses existing bidirectional stream.
%% Otherwise creates mesh connection.
%% Handles both atom keys (from local DHT storage) and binary keys (from protocol).
-spec route_to_single_subscriber(map(), binary(), map(), binary(), pid(), pid()) -> ok.
route_to_single_subscriber(#{node_id := DestNodeId} = Subscriber, Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
%% Atom key version (from local DHT storage)
route_to_subscriber_impl(DestNodeId, Subscriber, node_id, Topic, PubMsg, LocalNodeId, Mesh, Clients);
route_to_single_subscriber(#{<<"node_id">> := DestNodeId} = Subscriber, Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
%% Binary key version (from protocol)
route_to_subscriber_impl(DestNodeId, Subscriber, <<"node_id">>, Topic, PubMsg, LocalNodeId, Mesh, Clients);
route_to_single_subscriber(_Subscriber, _Topic, _PubMsg, _LocalNodeId, _Mesh, _Clients) ->
%% Subscriber missing node_id - skip silently
ok.
%% @private
%% @doc Implementation of routing to a single subscriber.
-spec route_to_subscriber_impl(binary(), map(), atom() | binary(), binary(), map(), binary(), pid(), pid()) -> ok.
route_to_subscriber_impl(DestNodeId, _Subscriber, _EndpointKey, _Topic, _PubMsg, LocalNodeId, _Mesh, _Clients)
when DestNodeId =:= LocalNodeId ->
%% SKIP: This is our own node_id - already delivered locally
ok;
route_to_subscriber_impl(DestNodeId, Subscriber, EndpointKey, Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
?LOG_INFO("[PubSubRouter] route_to ~s endpoint_key=~p subscriber_keys=~p",
[binary:encode_hex(DestNodeId), EndpointKey, maps:keys(Subscriber)]),
do_route_to_subscriber(DestNodeId, Subscriber, EndpointKey, Topic, PubMsg, LocalNodeId, Mesh, Clients).
%% @private
%% @doc Actually route message to subscriber after authorization.
-spec do_route_to_subscriber(binary(), map(), atom() | binary(), binary(), map(), binary(), pid(), pid()) -> ok.
do_route_to_subscriber(DestNodeId, Subscriber, EndpointKey, Topic, PubMsg, LocalNodeId, Mesh, Clients) ->
Payload = maps:get(<<"payload">>, PubMsg),
Qos = maps:get(<<"qos">>, PubMsg, 0),
%% Create PUBLISH message for this subscriber
PublishMsg = #{
<<"topic">> => Topic,
<<"payload">> => Payload,
<<"qos">> => Qos,
<<"retain">> => false,
<<"message_id">> => crypto:strong_rand_bytes(16)
},
%% Check if this is a connected client (has existing bidirectional stream)
?LOG_DEBUG("[PubSubRouter] Checking if node ~s is connected client",
[binary:encode_hex(DestNodeId)]),
case macula_gateway_clients:get_client_stream(Clients, DestNodeId) of
{ok, Stream} ->
%% Subscriber is a connected client - send directly via their stream
?LOG_DEBUG("[PubSubRouter] SENDING to connected client ~s via stream ~p",
[binary:encode_hex(DestNodeId), Stream]),
send_to_client_stream(Stream, PublishMsg);
not_found ->
?LOG_DEBUG("[PubSubRouter] Node ~s NOT found as connected client",
[binary:encode_hex(DestNodeId)]),
%% Not a connected client - must be remote gateway, use mesh connection
%% Get endpoint using same key type as node_id (atom or binary)
EndpointUrl = case EndpointKey of
endpoint -> maps:get(endpoint, Subscriber, undefined);
<<"endpoint">> -> maps:get(<<"endpoint">>, Subscriber, undefined);
node_id -> maps:get(endpoint, Subscriber, undefined);
<<"node_id">> -> maps:get(<<"endpoint">>, Subscriber, undefined)
end,
PubSubRouteMsg = macula_pubsub_routing:wrap_publish(LocalNodeId, DestNodeId, PublishMsg, 10),
send_via_dht(DestNodeId, EndpointUrl, PubSubRouteMsg, Mesh)
end.
%% @private
%% @doc Send pubsub_route message via DHT mesh connection.
%% First checks direct routing table for cached endpoint, then falls back to provided endpoint.
%% Parses endpoint URL to extract address and port.
-spec send_via_dht(binary(), binary() | undefined, map(), pid()) -> ok.
send_via_dht(DestNodeId, EndpointUrl, PubSubRouteMsg, Mesh) ->
%% Try direct routing table first for cached endpoint
ResolvedEndpoint = resolve_endpoint(DestNodeId, EndpointUrl),
send_to_resolved_endpoint(DestNodeId, ResolvedEndpoint, PubSubRouteMsg, Mesh).
%% @private
%% @doc Resolve endpoint: try direct routing cache first, then fall back to provided endpoint.
-spec resolve_endpoint(binary(), binary() | undefined) -> {direct, binary()} | {provided, binary()} | not_found.
resolve_endpoint(DestNodeId, EndpointUrl) ->
case macula_direct_routing:lookup(DestNodeId) of
{ok, CachedEndpoint} ->
{direct, CachedEndpoint};
miss ->
resolve_provided_endpoint(EndpointUrl)
end.
%% @private
%% @doc Resolve provided endpoint or return not_found.
-spec resolve_provided_endpoint(binary() | undefined) -> {provided, binary()} | not_found.
resolve_provided_endpoint(undefined) ->
not_found;
resolve_provided_endpoint(EndpointUrl) when is_binary(EndpointUrl) ->
{provided, EndpointUrl};
resolve_provided_endpoint(_) ->
not_found.
%% @private
%% @doc Send to resolved endpoint.
-spec send_to_resolved_endpoint(binary(), {direct | provided, binary()} | not_found, map(), pid()) -> ok.
send_to_resolved_endpoint(DestNodeId, not_found, _PubSubRouteMsg, _Mesh) ->
?LOG_WARNING("[PubSubRouter] No endpoint for subscriber ~s — dropping",
[binary:encode_hex(DestNodeId)]),
ok;
send_to_resolved_endpoint(DestNodeId, {_Source, EndpointUrl}, PubSubRouteMsg, Mesh) ->
case parse_endpoint(EndpointUrl) of
{ok, {Address, Port}} ->
?LOG_WARNING("[PubSubRouter] Sending to ~s (~s) via ~s:~p",
[binary:encode_hex(DestNodeId), EndpointUrl, Address, Port]),
case macula_gateway_mesh:get_or_create_connection(Mesh, DestNodeId, {Address, Port}) of
{ok, Stream} ->
?LOG_WARNING("[PubSubRouter] Connected! Sending pubsub_route to ~s", [EndpointUrl]),
send_route_message(Stream, PubSubRouteMsg);
{error, Reason} ->
?LOG_WARNING("[PubSubRouter] Connection failed to ~s: ~p", [EndpointUrl, Reason]),
ok
end;
{error, _ParseReason} ->
ok
end.
%% @private
%% @doc Send PUBLISH message directly to connected client stream.
-spec send_to_client_stream(reference(), map()) -> ok.
send_to_client_stream(Stream, PublishMsg) ->
PubBinary = macula_protocol_encoder:encode(publish, PublishMsg),
_ = macula_quic:send(Stream, PubBinary),
ok.
%% @private
%% @doc Send encoded pubsub_route message to stream.
%% NOTE: We DON'T close streams here - the stream stays open for the
%% QUIC connection's lifetime. Closing a stream doesn't close the connection,
%% and quicer handles stream cleanup on connection close.
-spec send_route_message(reference(), map()) -> ok.
send_route_message(Stream, PubSubRouteMsg) ->
RouteMsg = macula_protocol_encoder:encode(pubsub_route, PubSubRouteMsg),
_ = macula_quic:send(Stream, RouteMsg),
ok.
%%%===================================================================
%%% Node ID Helpers (Removed - not needed)
%%%===================================================================
%% Per Kademlia spec: node IDs are always 256-bit (32-byte) raw binary.
%% Hex encoding is ONLY for display/logging, never for internal storage.
%% DHT stores and returns raw binary node IDs.
%%%===================================================================
%%% Endpoint Parsing Helper
%%%===================================================================
%% @private
%% @doc Parse endpoint URL (e.g., "https://192.168.1.100:4433") to {Address, Port}.
%% Returns {ok, {Address, Port}} or {error, Reason}.
-spec parse_endpoint(binary()) -> {ok, {inet:ip_address() | list(), inet:port_number()}} | {error, term()}.
parse_endpoint(EndpointUrl) when is_binary(EndpointUrl) ->
case uri_string:parse(EndpointUrl) of
#{host := Host, port := Port} when is_integer(Port) ->
parse_endpoint_host(Host, Port);
#{host := Host} ->
parse_endpoint_host(Host, 4433); % Default QUIC port
_Other ->
{error, {invalid_endpoint_format, EndpointUrl}}
end;
parse_endpoint(_Other) ->
{error, invalid_endpoint_type}.
%% @private
%% Parse hostname to IP address tuple.
parse_endpoint_host(Host, Port) when is_list(Host) ->
case inet:parse_address(Host) of
{ok, IpTuple} ->
{ok, {IpTuple, Port}};
{error, _} ->
%% Not an IP address, treat as hostname
{ok, {Host, Port}}
end;
parse_endpoint_host(Host, Port) when is_binary(Host) ->
parse_endpoint_host(binary_to_list(Host), Port).