Packages
nova
0.9.4
0.15.1
0.14.3
0.14.1
0.13.7
0.13.1
0.13.0
0.12.1
0.12.0
0.11.0
0.10.4
0.10.3
0.10.2
0.10.1
0.10.0
0.9.24
0.9.23
0.9.22
0.9.21
0.9.20
0.9.19
0.9.18
0.9.17
0.9.16
0.9.15
0.9.11
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.2
0.7.1
0.7.0
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.4.4
0.4.3
0.4.2
0.4.1
0.2.2
0.1.0
0.0.1
0.0.0
Nova is a web application framework
Current section
Files
Jump to
Current section
Files
src/nova_channel.erl
%%%-------------------------------------------------------------------
%%% @author Niclas Axelsson <niclas@burbas.se>
%%% @doc
%%% Channels is a pubsub system for Nova. It uses the `pg` module.
%%% @end
%%% Created : 8 Apr 2022 by Niclas Axelsson <niclas@burbas.se>
%%%-------------------------------------------------------------------
-module(nova_channel).
-export([
start/0,
join/1,
join/2,
leave/1,
leave/2,
broadcast/3,
local_broadcast/3,
get_members/1,
get_local_members/1
]).
-define(SCOPE, nova_scope).
-include("../include/nova_channel.hrl").
%% Start scope
-if(?OTP_RELEASE >= 23).
-spec start() -> ok.
start() ->
pg:start(?SCOPE),
ok.
-else.
start() ->
pg2:start(),
ok.
-endif.
%% Joining a channel
-spec join(Channel :: atom()) -> ok.
join(Channel) ->
join(Channel, self()).
-spec leave(Channel :: atom()) -> ok | not_joined.
leave(Channel) ->
leave(Channel, self()).
-spec join(Channel :: atom(), Pid :: pid()) -> ok.
-if(?OTP_RELEASE >= 23).
join(Channel, Pid) when is_pid(Pid) ->
pg:join(?SCOPE, Channel, Pid).
-else.
join(Channel, Pid) when is_pid(Pid) ->
pg2:create({?SCOPE, Channel}), %% Create this to ensure it exists
pg2:join({?SCOPE, Channel}, Pid).
-endif.
-spec leave(Channel :: atom(), Pid :: pid()) -> ok | not_joined.
-if(?OTP_RELEASE >= 23).
leave(Channel, Pid) ->
pg:leave(?SCOPE, Channel, Pid).
-else.
leave(Channel, Pid) ->
case pg2:leave({?SCOPE, Channel}, Pid) of
{error, _} ->
not_joined;
_ ->
ok
end.
-endif.
%% Broadcast to members
-spec broadcast(Channel :: atom(), Topic :: list() | binary(), Message :: any()) -> ok.
broadcast(Channel, Topic, Message) ->
Members = get_members(Channel),
Envelope = create_envelope(Channel, self(), Topic, Message),
[ Receiver ! Envelope || Receiver <- Members ],
ok.
%% Broadcast to processes located on the same node
-spec local_broadcast(Channel :: atom(), Topic :: list() | binary(), Message :: any()) -> ok.
local_broadcast(Channel, Topic, Message) ->
Members = get_local_members(Channel),
Envelope = create_envelope(Channel, self(), Topic, Message),
[ Receiver ! Envelope || Receiver <- Members ],
ok.
-spec get_members(Channel :: atom()) -> [pid()].
-if(?OTP_RELEASE >= 23).
get_members(Channel) ->
pg:get_members(?SCOPE, Channel).
-else.
get_members(Channel) ->
case pg2:get_members({?SCOPE, Channel}) of
{error, _} ->
[];
Pidlist ->
Pidlist
end.
-endif.
-spec get_local_members(Channel :: atom()) -> [pid()].
-if(?OTP_RELEASE >= 23).
get_local_members(Channel) ->
pg:get_local_members(?SCOPE, Channel).
-else.
get_local_members(Channel) ->
case pg2:get_local_members({?SCOPE, Channel}) of
{error, _} ->
[];
Pidlist ->
Pidlist
end.
-endif.
create_envelope(Channel, Sender, Topic, Payload) ->
#nova_channel{channel = Channel,
sender = Sender,
topic = Topic,
payload = Payload}.