%% Copyright (c) 2016 Contributors as noted in the AUTHORS file %% %% This file is part teleport %% %% This Source Code Form is subject to the terms of the Mozilla Public %% License, v. 2.0. If a copy of the MPL was not distributed with this %% file, You can obtain one at http://mozilla.org/MPL/2.0/. -module(teleport). -author("Benoit Chesneau"). %% API -export([ start_server/2, stop_server/1, server_uri/1, connect/2, disconnect/1, incoming_conns/0, outgoing_conns/0, call/4, call/5, cast/4, blocking_call/4, blocking_call/5, abcast/3, sbcast/3, monitor_link/1, demonitor_link/1, monitor_links/1, new_channel/1, close_channel/1, send_channel/3, send_channel_sync/3, send_channel_sync/4, recv_channel/1, recv_channel/2, register_channel/2, unregister_channel/2 ]). -include("teleport.hrl"). -define(DEFAULT_STRATEGY, random). -type host() :: inet:socket_address() | inet:hostname(). -type uri() :: string(). -type pool_options() :: #{ host := host(), port => inet:port_number(), num_connections => non_neg_integer(), connect_timeout => non_neg_integer(), sup_intensity => non_neg_integer(), sup_period => non_neg_integer() }. -type server_config() :: #{ host := host(), port => inet:port_number(), %% default is 0 transport => tcp | ssl, num_acceptors => non_neg_integer() }. -export_types([ host/0, uri/0, connect_options/0, client_config/0, server_config/0 ]). %% @doc start a system. A system is a hierarchical group of processes which %% share common configuration. It is also the entry point for registering or %% looking up proceses. -spec start_server(atom(), server_config()) -> {ok, pid()} |{error, term()}. start_server(Name, Config) -> teleport_server_sup:start_server(Name, Config). %% @doc stop a system -spec stop_server(atom()) -> ok. stop_server(Name) -> teleport_server_sup:stop_server(Name). %% @doc get the server uri that can be used to connect from a client -spec server_uri(atom()) -> uri(). server_uri(Name) -> teleport_server_sup:get_uri(Name). %% @doc connect to a server using an uri -spec connect(Uri::uri(), pool_options()) -> boolean() ; (Name::atom(), pool_options()) -> boolean(). connect(Name, Uri) when is_atom(Name), is_list(Uri) -> {ok, Config} = teleport_uri:config_from_uri(Uri), connect(Name, Config); connect(Name, Config) when is_atom(Name), is_map(Config) -> case supervisor:start_child(teleport_link_sup, [Name, Config]) of {ok, _Pid} -> true; {error, {already_started, _Pid}} -> false; Error -> Error end. %% @doc disconnect a link -spec disconnect(atom() | pid()) -> ok. disconnect(Pid) when is_pid(Pid) -> teleport_link_sup:stop_child(Pid); disconnect(Name) -> case whereis(Name) of undefined -> ok; Pid -> teleport_link_sup:stop_child(Pid) end. %% @doc on a server node list incoming connections incoming_conns() -> lists:usort( [Node || {_, _, Node} <- ets:tab2list(teleport_incoming_conns)]). %% @doc on a client node list outgoing connections outgoing_conns() -> lists:usort( [Node || {_, _, Node} <- ets:tab2list(teleport_outgoing_conns)]). %% MONITOR API %% @doc monitor all links -spec monitor_links(boolean()) -> ok. monitor_links(true) -> teleport_monitor:monitor_link('$all_links'); monitor_links(false) -> teleport_monitor:monitor_link('$all_links'). %% @doc monitor a link monitor_link(Name) -> teleport_monitor:monitor_link(Name). demonitor_link(Name) -> teleport_monitor:demonitor_link(Name). %% RPC API call(Name, M, F, A) -> call(Name, M, F, A, 5000). call(Name, M, F, A, Timeout) -> teleport_link:call(Name, M, F, A, Timeout). cast(Name, M, F, A) -> teleport_link:cast(Name, M, F, A). blocking_call(Name, M, F, A) -> blocking_call(Name, M, F, A, 5000). blocking_call(Name, M, F, A, Timeout) -> teleport_link:blocking_call(Name, M, F, A, Timeout). abcast(Names, ProcName, Msg) -> teleport_link:abcast(Names, ProcName, Msg). sbcast(Names, ProcName, Msg) -> teleport_link:sbcast(Names, ProcName, Msg). %% Channel new_channel(LinkId) -> teleport_link:new_channel(LinkId). close_channel(Channel) -> teleport_link:close_channel(Channel). send_channel(Channel, To, Msg) -> teleport_link:send_channel(Channel, To, Msg). send_channel_sync(Channel, To, Msg) -> send_channel_sync(Channel, To, Msg, 5000). send_channel_sync(Channel, To, Msg, Timeout) -> ok = teleport_link:send_channel(Channel, To, {'$channel_req', Msg}), recv_channel(Channel, Timeout). recv_channel(Channel) -> recv_channel(Channel, 5000). recv_channel(Channel, Timeout) -> teleport_link:recv_channel(Channel, Timeout). register_channel(Channel, To) -> teleport_link:send_channel(Channel, To, '$channel_register'). unregister_channel(Channel, To) -> teleport_link:send_channel(Channel, To, '$channel_unregister').