%% ------------------------------------------------------------------- %% %% Copyright (c) 2016 Christopher Meiklejohn. All Rights Reserved. %% %% This file is provided to you under the Apache License, %% Version 2.0 (the "License"); you may not use this file %% except in compliance with the License. You may obtain %% a copy of the License at %% %% http://www.apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, %% software distributed under the License is distributed on an %% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY %% KIND, either express or implied. See the License for the %% specific language governing permissions and limitations %% under the License. %% %% ------------------------------------------------------------------- -module(partisan_peer_service_manager). -author("Christopher S. Meiklejohn "). -include("partisan_logger.hrl"). -include("partisan.hrl"). -type server_ref() :: partisan_remote_ref:p() | partisan_remote_ref:encoded_pid() | partisan_remote_ref:n() | partisan_remote_ref:encoded_name() | Name :: atom() | {Name :: atom(), node()} | {global, atom()} | {via, module(), ViaName :: atom()} | pid(). -type forward_opts() :: #{ ack => boolean(), causal_label => atom(), channel => channel(), clock => any(), partition_key => non_neg_integer(), transitive => boolean() } | [ {ack, boolean()} | {causal_label, atom()} | {channel, channel()} | {clock, any()} | {partition_key, non_neg_integer()} | {transitive, boolean()} ]. -export_type([server_ref/0]). -export_type([forward_opts/0]). -export([mynode/0]). -export([myself/0]). -export([process_forward/2]). -export([send_message/2]). %% ============================================================================= %% BEHAVIOUR CALLBACKS %% ============================================================================= -callback start_link() -> {ok, pid()} | ignore | {error, term()}. -callback members() -> [node()]. %% TODO: Deprecate me. -callback members_for_orchestration() -> [node_spec()]. -callback myself() -> node_spec(). -callback update_members([node()]) -> ok | {error, not_implemented}. -callback get_local_state() -> term(). -callback on_down(node(), function()) -> ok | {error, not_implemented}. -callback on_up(node(), function()) -> ok | {error, not_implemented}. -callback join(node_spec()) -> ok. -callback sync_join(node_spec()) -> ok | {error, not_implemented}. -callback leave() -> ok. -callback leave(node_spec()) -> ok. -callback send_message(node(), message()) -> ok. -callback receive_message(node(), message()) -> ok. -callback cast_message( ServerRef :: server_ref(), Msg :: message()) -> ok. -callback cast_message( ServerRef :: server_ref(), Msg :: message(), Opts :: forward_opts()) -> ok. -callback cast_message( Node :: node(), ServerRef :: server_ref(), Msg :: message(), Opts :: forward_opts()) -> ok. -callback forward_message( ServerRef :: server_ref(), Msg :: message()) -> ok. -callback forward_message( ServerRef :: server_ref(), Msg :: message(), Opts :: forward_opts()) -> ok. -callback forward_message( Node :: node(), ServerRef :: server_ref(), Msg :: message(), Opts :: forward_opts()) -> ok. -callback decode(term()) -> term(). -callback reserve(atom()) -> ok | {error, no_available_slots}. -callback partitions() -> {ok, partitions()} | {error, not_implemented}. -callback inject_partition(node_spec(), ttl()) -> {ok, reference()} | {error, not_implemented}. -callback resolve_partition(reference()) -> ok | {error, not_implemented}. %% ============================================================================= %% API %% ============================================================================= %% ----------------------------------------------------------------------------- %% @doc Send a message to a remote peer_service_manager. %% @end %% ----------------------------------------------------------------------------- -spec send_message(node(), message()) -> ok. send_message(Node, Message) -> (?PEER_SERVICE_MANAGER):send_message(Node, Message). %% ----------------------------------------------------------------------------- %% @doc Internal function used by peer_service manager implementations to %% forward a message to a process identified by `ServerRef' that is either %% local or located at remote process when the remote node is connected via %% disterl. %% Trying to send a message to a remote server reference when the process is %% located at a node connected with Partisan will return `ok' but will not %% succeed. %% @end %% ----------------------------------------------------------------------------- -spec process_forward(ServerRef :: server_ref(), Msg :: any()) -> ok. process_forward(ServerRef, Msg) -> try do_process_forward(ServerRef, Msg) catch Class:Reason:Stacktrace -> ?LOG_DEBUG(#{ description => "Error forwarding message", message => Msg, destination => ServerRef, class => Class, reason => Reason, stacktrace => Stacktrace }), ok end. %% @deprecated use {@link partisan:node_spec/0} instead -spec myself() -> node_spec(). myself() -> partisan:node_spec(). %% @deprecated use {@link partisan:node/0} instead -spec mynode() -> atom(). mynode() -> partisan:node(). %% ============================================================================= %% PRIVATE %% ============================================================================= %% @private do_process_forward({global, Name}, Message) -> Pid = global:whereis_name(Name), Pid ! Message, ok; do_process_forward({via, Module, Name}, Message) -> Pid = Module:whereis_name(Name), Pid ! Message, ok; do_process_forward(ServerRef, Message) when is_pid(ServerRef) orelse is_atom(ServerRef) -> ServerRef ! Message, Trace = ( is_pid(ServerRef) andalso not is_process_alive(ServerRef) ) orelse ( not is_pid(ServerRef) andalso ( whereis(ServerRef) == undefined orelse not is_process_alive(whereis(ServerRef)) ) ), ?LOG_TRACE_IF( Trace, "Process ~p is NOT ALIVE.", [ServerRef] ), ok; do_process_forward(ServerRef, Message) when is_tuple(ServerRef) orelse is_binary(ServerRef) -> ?LOG_DEBUG( "node ~p recieved message ~p for ~p", [partisan:node(), Message, ServerRef] ), case partisan_remote_ref:to_term(ServerRef) of Pid when is_pid(Pid) -> ?LOG_TRACE_IF( not is_process_alive(Pid), "Process ~p is NOT ALIVE for message: ~p", [ServerRef, Message] ), Pid ! Message, ok; Name when is_atom(Name) -> Name ! Message, ok end.