macula_local_client (macula v0.20.5)

Local client for in-VM workloads to connect to macula_gateway

This module provides process-to-process communication between workloads running in the same BEAM VM as the Macula platform and the local gateway. Unlike macula_peer which creates QUIC connections, this connects directly to the local macula_gateway process.

Architecture: Phoenix/Elixir App → macula_local_client → macula_gateway ↓ (QUIC) Other Peers

Summary

Functions

advertise(Pid, Procedure, Handler)

Advertise an RPC service with default options

advertise(Pid, Procedure, Handler, Opts)

Advertise an RPC service with options

call(Pid, Procedure, Args)

Call an RPC procedure with default options

call(Pid, Procedure, Args, Opts)

Call an RPC procedure

connect(Url, Opts)

Connect to remote gateway (not supported for local client) For compatibility with macula_client_behaviour

connect_local(Opts)

Create a local client connection to the gateway

disconnect(Pid)

Disconnect the client

discover_subscribers(Pid, Topic)

Discover subscribers of a topic via DHT query

get_leader(Pid)

Get the current Platform Layer leader node ID

get_node_id(Pid)

Get the node ID of the local gateway

handle_call(_, From, State)

handle_cast(Msg, State)

handle_info(Info, State)

Handle pubsub events from gateway Gateway sends messages in format: {publish, Topic, Payload}

init(Opts)

propose_crdt_update(Pid, Key, Value)

Propose a CRDT update with default options (LWW-Register)

propose_crdt_update(Pid, Key, Value, Opts)

Propose a CRDT update with specific type

publish(Pid, Topic, Payload)

Publish an event to a topic

publish(Pid, Topic, Payload, Opts)

Publish an event to a topic with options

read_crdt(Pid, Key)

Read the current value of a CRDT-managed state entry

register_procedure(Pid, Procedure, Handler)

Register an RPC procedure (legacy API, use advertise/3 instead)

register_workload(Pid, Opts)

Register this workload with the Platform Layer

start_link(Opts)

Start a local client connection to the gateway (legacy API)

stop(Pid)

Stop the local client (legacy API)

subscribe(Pid, Topic, HandlerPid)

Subscribe to a topic

subscribe_leader_changes(Pid, Callback)

Subscribe to Platform Layer leader change notifications

terminate(Reason, State)

unadvertise(Pid, Procedure)

Unadvertise an RPC service

unregister_procedure(Pid, Procedure)

Unregister an RPC procedure (legacy API, use unadvertise/2 instead)

unsubscribe(Pid, SubRef)

Unsubscribe from a topic

Functions

advertise(Pid, Procedure, Handler)

-spec advertise(pid(), binary(), fun()) -> {ok, reference()} | {error, term()}.

Advertise an RPC service with default options

advertise(Pid, Procedure, Handler, Opts)

-spec advertise(pid(), binary(), fun(), map()) -> {ok, reference()} | {error, term()}.

Advertise an RPC service with options

call(Pid, Procedure, Args)

-spec call(pid(), binary(), list()) -> {ok, term()} | {error, term()}.

Call an RPC procedure with default options

call(Pid, Procedure, Args, Opts)

-spec call(pid(), binary(), list(), map()) -> {ok, term()} | {error, term()}.

Call an RPC procedure

connect(Url, Opts)

-spec connect(binary() | string(), map()) -> {error, not_supported}.

Connect to remote gateway (not supported for local client) For compatibility with macula_client_behaviour

connect_local(Opts)

-spec connect_local(map()) -> {ok, pid()} | {error, term()}.

Create a local client connection to the gateway

disconnect(Pid)

-spec disconnect(pid()) -> ok.

Disconnect the client

discover_subscribers(Pid, Topic)

-spec discover_subscribers(pid(), binary()) -> {ok, [binary()]} | {error, term()}.

Discover subscribers of a topic via DHT query

get_leader(Pid)

-spec get_leader(pid()) -> {ok, binary()} | {error, no_leader | term()}.

Get the current Platform Layer leader node ID

get_node_id(Pid)

-spec get_node_id(pid()) -> {ok, binary()} | {error, term()}.

Get the node ID of the local gateway

handle_call(_, From, State)

handle_cast(Msg, State)

handle_info(Info, State)

Handle pubsub events from gateway Gateway sends messages in format: {publish, Topic, Payload}

init(Opts)

propose_crdt_update(Pid, Key, Value)

-spec propose_crdt_update(pid(), binary(), term()) -> ok | {error, term()}.

Propose a CRDT update with default options (LWW-Register)

propose_crdt_update(Pid, Key, Value, Opts)

-spec propose_crdt_update(pid(), binary(), term(), map()) -> ok | {error, term()}.

Propose a CRDT update with specific type

publish(Pid, Topic, Payload)

-spec publish(pid(), binary(), map()) -> ok | {error, term()}.

Publish an event to a topic

publish(Pid, Topic, Payload, Opts)

-spec publish(pid(), binary(), map(), map()) -> ok | {error, term()}.

Publish an event to a topic with options

read_crdt(Pid, Key)

-spec read_crdt(pid(), binary()) -> {ok, term()} | {error, not_found | term()}.

Read the current value of a CRDT-managed state entry

register_procedure(Pid, Procedure, Handler)

-spec register_procedure(pid(), binary(), fun()) -> ok | {error, term()}.

Register an RPC procedure (legacy API, use advertise/3 instead)

register_workload(Pid, Opts)

-spec register_workload(pid(), map()) -> {ok, map()} | {error, term()}.

Register this workload with the Platform Layer

start_link(Opts)

-spec start_link(map()) -> {ok, pid()} | {error, term()}.

Start a local client connection to the gateway (legacy API)

stop(Pid)

-spec stop(pid()) -> ok.

Stop the local client (legacy API)

subscribe_leader_changes(Pid, Callback)

-spec subscribe_leader_changes(pid(), fun((map()) -> ok)) -> {ok, reference()} | {error, term()}.

Subscribe to Platform Layer leader change notifications

terminate(Reason, State)

unadvertise(Pid, Procedure)

-spec unadvertise(pid(), binary()) -> ok | {error, term()}.

Unadvertise an RPC service

unregister_procedure(Pid, Procedure)

-spec unregister_procedure(pid(), binary()) -> ok | {error, term()}.

Unregister an RPC procedure (legacy API, use unadvertise/2 instead)

unsubscribe(Pid, SubRef)

-spec unsubscribe(pid(), reference()) -> ok | {error, term()}.

Unsubscribe from a topic