jc

%%%-----------------------------------------------------------------------------
%%% @author Jim Rosenblum
%%% @copyright (C) 2011 - 2015, Jim Rosenblum
%%% @doc jc_psub 1) allows clients to subscribe to j_cache changes; 2) notifies 
%%% clients of changes to j_cache items; and 3) allows clients to subscribe to,
%%% and broadcast, topic events.
%%%
%%% Subscriptions are a local afair so if a node goes down, clients that were
%%% using that node's jc_psub will need to re-subscribe.
%%% 
%%% Mnesia LOCAL tables, ps_sub and ps_client, are used:
%%%
%%% ps_sub maps a subscription to a list of listening Pids
%%% ps_client maps a unique, subscriber's pid to a reference or link.
%%%
%%% The server subscribes to mnesia key_to_value, table changes thereby allowing
%%% it to notice cache changes, construct the corresponding subscription
%%% message and look for relevant subscribers to whom to send a message.
%%%
%%% @end
%%% Created : 16 Sep 2011 by Jim Rosenblum
%%%-----------------------------------------------------------------------------
-module(jc_psub).

-behaviour(gen_server).


% Module API for starting and stopping the change_relay server.
-export([start_link/0]).


% Pub/Sub API.
-export([map_subscribe/4,
	 map_unsubscribe/4,
	 topic_event/2,
	 topic_subscribe/3, 
	 topic_unsubscribe/3]).

% Fns returning various statistics.
-export([client_count/0, load/0]).

% Gen_server callbacks.
-export([init/1, 
	 handle_call/3, handle_cast/2, handle_info/2, 
	 terminate/2, 
	 code_change/3]).


-include("../include/records.hrl").


-define(SERVER,       ?MODULE). 
-define(DEADBEATS_MS, 3600000).       % Length of time to check for deadbeats.


% Records used to represent map and topic subscriptions.
-record(map_sub, {map              :: map_name(),
		  key = any        :: key() | any,
		  operation = any  :: write|delete|any}).

-record(topic_sub, {topic          :: atom(),
		    value = any    :: value() | any}).


% Server state. Monitored is the list of nodes at which jc_sup is being 
% monitored - don't want multiple monitors so we maintain a list of them.
-record(jc_psub_state, {evict_deadbeats_ms = ?DEADBEATS_MS :: non_neg_integer(),
			nodes              = []            :: list(node()),
			monitored          = []            :: list(node())}).



%%% ============================================================================
%%% Module API 
%%% ============================================================================

%% -----------------------------------------------------------------------------
%% @doc Starts the jc_psub server and links the caller to it.
%% 
-spec start_link() -> {ok, pid()} | ignore | {error, any()}.

start_link() ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).



%%% ============================================================================
%%% Pub/Sub API
%%% ============================================================================


%% -----------------------------------------------------------------------------
%% @doc Adds a subscription on behalf of the process indicated by the supplied 
%% Pid. 
%%
-spec map_subscribe(pid(), map_name(), key()|any, write | delete | any) ->  
			                                   ok | {error, badarg}.

map_subscribe(Client, Map, Key, Ops) when is_pid(Client)->
    SubSpec = #map_sub{map = Map, key = Key, operation = Ops},
    gen_server:call(?SERVER, {subscribe, Client, SubSpec});

map_subscribe(_C, _M, _K, _O) ->
    {error, badarg}.


%% -----------------------------------------------------------------------------
%% @doc Client unsubcribes from notifiction.
%% 
-spec map_unsubscribe(pid(), map_name(),key()|any, write | delete | any) ->
			                                   ok | {error, badarg}.
			 
map_unsubscribe(Client, Map, Key, Ops) when is_pid(Client) ->
    SubSpec = #map_sub{map = Map, key = Key, operation = Ops},
    gen_server:call(?SERVER, {unsubscribe, Client, SubSpec});

map_unsubscribe(_C, _M, _k, _O) ->
    {error, badarg}.



%% -----------------------------------------------------------------------------
%% @doc Push a topic message to the server for broadcast to its local 
%% subscribers. If the event is a node event, then the other jc_psubs will 
%% know about it via their monitoring of foreign jc_sup's, so there is no need 
%% to propagate the event to other nodes' jc_psub servers; else propagate it.
%%
-spec topic_event(Topic::atom(), Value::value()) -> ok.

topic_event(jc_node_events, Value)->
    gen_server:cast(?SERVER, {topic_event, jc_node_events, Value});

topic_event(Topic, Value)->
    gen_server:abcast([node()|nodes()], ?SERVER, {topic_event, Topic, Value}),
    ok.


%% -----------------------------------------------------------------------------
%% @doc Client subscribes to notifications regarding the given Topic.
%%
-spec topic_subscribe(Client::pid(), Topic::atom(), Value::value() | any) -> 
  			                                    ok| {error, badarg}.

topic_subscribe(Client, Topic, Value) when is_pid(Client)->
    TopicSpec = #topic_sub{topic = Topic, value = Value},
    gen_server:call(?SERVER, {subscribe, Client, TopicSpec});

topic_subscribe(_C, _T, _V) ->
    {error, badarg}.


%% -----------------------------------------------------------------------------
%% @doc Client unsubscribes to notifications regarding the given Topic.
%%
-spec topic_unsubscribe(Client::pid(), Topic::atom(), Value::value() | any) ->
			                                   ok | {error, badarg}.

topic_unsubscribe(Client, Topic, Value) when is_pid(Client) ->
    TopicSpec = #topic_sub{topic = Topic, value = Value},
    gen_server:call(?SERVER, {unsubscribe, Client, TopicSpec});

topic_unsubscribe(_C, _T, _V) ->
    {error, badarg}.


%% -----------------------------------------------------------------------------
%% Various statistics
%% -----------------------------------------------------------------------------


%% -----------------------------------------------------------------------------
%% @doc Return an estimate of how much load the server is under. It counts the
%% number of subscriptions + the number of Pids that are currenlty subscribed
%% to anything. Under extreme load this can time-out resulting in an error.
%% 
-spec load() -> non_neg_integer() | {error, any()}.

load() ->
    try gen_server:call(?SERVER, load)
    catch
	_Error:Reason ->
	    lager:warning("~p: load/0 error. ~p.", [?SERVER, Reason]),
	    {error, Reason}
    end.


%% -----------------------------------------------------------------------------
%% @doc Return the number of distinct, subscribing Pids being served by the
%% server. Under extreme load this function can time-out.
%% 
-spec client_count() -> non_neg_integer() | {error, any()}.

client_count() ->
    try gen_server:call(?SERVER, client_count)
    catch
	_Error: Reason ->
	    lager:warning("~p: client_count/0 error. ~p", [?SERVER, Reason]),
	    {error, Reason}
    end.



%%% ============================================================================
%%% gen_server callbacks
%%% ============================================================================


%% -----------------------------------------------------------------------------
%% @private
%% Initializes the server. Set process_flag to trap linked subscriber exits,
%% monitor any jc_sups at already up nodes, schedule the deadbeat_eviction job 
%% and subscribe to relevant mnesia changes to table key_to_value.
%%
-spec init([]) -> {'ok', #jc_psub_state{}}.


init([]) ->
    process_flag(trap_exit, true), 

    Nodes = application:get_env(jc, cache_nodes,[node()]),
    DBMs  = application:get_env(jc, evict_deadbeats_ms, ?DEADBEATS_MS),

    MNodes = monitor_sups(Nodes),
    lager:info("~p: found and monitoring existing supervisors at ~p.", 
	       [?SERVER, MNodes]),

    schedule_deadbeat_eviction(DBMs),    

    mnesia:subscribe({table, key_to_value, detailed}),

    lager:info("~p: up.", [?SERVER]),
    {ok, #jc_psub_state{evict_deadbeats_ms = DBMs, 
			nodes = Nodes, 
			monitored = MNodes}}.


% Monitor existing jc_sup processes at remote nodes, return list of those nodes.
monitor_sups(Nodes) ->
    F = fun(Node, Acc) -> 
		case rpc:call(Node, erlang, whereis, [jc_bridge], 500) of
		    undefined                    -> Acc;
		    {badrpc, _}                  -> Acc;
		    Pid when node(Pid) == node() ->  
			% shouldn't happen because jc_bridge isn't up yet.
                        % regardless, don't monitor yourself
			Acc;
		    Pid when Pid /= self()       -> 
			erlang:monitor(process, {jc_sup, Node}),
			[Node | Acc]
		end
	end,
    lists:foldl(F, [], Nodes).


%% -----------------------------------------------------------------------------
%% @private Handling call messages: subscribe, unsubscribe, load, client_clount.
%%
-spec handle_call(term(), {pid(), _}, #jc_psub_state{}) -> 
			 {reply, ok, #jc_psub_state{}}.

handle_call({subscribe, Client, Subscription}, _From, State) ->
    lager:debug("~p: adding subscription: ~p.",[?SERVER, Subscription]),
    subscribe(Client, Subscription),
    {reply, ok, State};

handle_call({unsubscribe, Client, Subscription}, _From, State) ->
    lager:debug("~p: deleting subscription: ~p.",[?SERVER, Subscription]),
    unsubscribe(Client, Subscription),
    {reply, ok, State};

handle_call(load,  _From, State) ->
    lager:debug("~p: calculating load.",[?SERVER]),
    {reply, calculate_load(), State};

handle_call(client_count,  _From, State) ->
    lager:debug("~p: calculating client_count.",[?SERVER]),
    {reply, mnesia:table_info(ps_client, size), State};

handle_call(Request,  _From, State) ->
    lager:warning("~p: unexpected call request: ~p.",[?SERVER, Request]),
    {reply, ok, State}.


%% -----------------------------------------------------------------------------
%% @private Handling cast messages: topic events and foreign jc_bridge up.
%%
-spec handle_cast(any(), #jc_psub_state{}) -> {noreply, #jc_psub_state{}}.

handle_cast({topic_event, Topic, Value} = Message, State) ->
    lager:debug("~p: topic_event: ~p.",[?SERVER, Message]),
    broadcast_topic_event(Topic, Value),
    {noreply, State};

% after jc_bridge started, it notifies all jc_psubs that it is up, monitor it.
handle_cast({jc_bridge_up, UpNode}, #jc_psub_state{nodes = Nodes, monitored=Ms}=S)->
    case (not lists:member(UpNode, Ms)) andalso lists:member(UpNode, Nodes) of
	true ->
	    lager:info("~p: jc_bridge is up at ~p.", [?SERVER, UpNode]),
	    erlang:monitor(process, {jc_sup, UpNode}),
	    signal_nodeup(UpNode),
	    {noreply, S#jc_psub_state{monitored = [UpNode | Ms]}};
	false ->
	    {noreply, S}
    end;

handle_cast(Msg, State) ->
    lager:warning("~p: unexpected cast message: ~p.",[?SERVER, Msg]),
    {noreply, State}.


%% -----------------------------------------------------------------------------
%% @private Handle info messages for: subscribing client or Node down; a 
%% foreign jc_sup goes down, evicting deadbeats and key_to_value table changes.
%%
-spec handle_info(any(), #jc_psub_state{}) -> {noreply, #jc_psub_state{}}.

% Monitored jc_sup went down on some node
handle_info({'DOWN', _MonitorRef, _Type, {jc_sup, Node}, _Info}, 
	    #jc_psub_state{monitored = Ms} = S) ->

    lager:warning("~p: jc_sup down at ~p.", [?SERVER, Node]),
    signal_nodedown(Node),
    {noreply, S#jc_psub_state{monitored = lists:delete(Node, Ms)}};

% Subscribers that are monitored, produce a Down signal.
handle_info({'DOWN', _MonitorRef, _Type, Object, _Info}, State) ->
    lager:warning("~p: monitored client (~p) down.", [?SERVER, Object]),
    unsubscribe_all(Object),
    {noreply, State};

% Subscribers that are linked, produce an Exit signal.
handle_info({'EXIT', From, Reason}, State) ->
    lager:warning("~p: linked client (~p) down: ~p.", [?SERVER, From, Reason]),
    unsubscribe_all(From),
    {noreply, State};

% Time to remove any deadbeats -- clients (Pids) with no active subscriptions.
handle_info({evict_deadbeats, IntMs}, State) ->
    lager:debug("~p: evicting deadbeats.",[?SERVER]),
    remove_deadbeats(),
    schedule_deadbeat_eviction(IntMs),
    {noreply, State};

% mnesia write, record replaced: broadcast that change to subscribers.
handle_info({mnesia_table_event, {write,key_to_value,Rec,[Old],_Trx}}, State) ->

    notify_of_evict(Old),
    notify_of_write(Rec),

    {noreply, State};


% mnesia write, nothing replaced.
handle_info({mnesia_table_event,{write, key_to_value, Rec, [], _Trx}}, State) ->

    notify_of_write(Rec),

    {noreply, State};


handle_info({mnesia_table_event,{delete,key_to_value,_What,[Rec],_Tx}},State) ->

    notify_of_delete(Rec),
    {noreply, State};


handle_info({mnesia_table_event, {delete, key_to_value,_What,[],_Trx}},State) ->
    {noreply, State};


% flush delete's the table form the scheema
handle_info({mnesia_table_event,{delete,schema,{schema,_},_What,_Trx}},State) ->
    {noreply, State};

% flush adds the table to the sceema
handle_info({mnesia_table_event,{write,schema,{schema,_,_Wht},_Rc,_T}},State) ->

    {noreply, State};


handle_info(Info, State) ->
    lager:warning("~p: unexpected info message: ~p.",[?SERVER, Info]),
    {noreply, State}.


%% -----------------------------------------------------------------------------
%% @private
%%
-spec terminate(any(), #jc_psub_state{}) -> any().

terminate(_Reason, _State) ->
    ok.


%% -----------------------------------------------------------------------------
%% @private Convert process state when code is changed
%%
-spec code_change(term(), #jc_psub_state{}, any()) -> {ok, #jc_psub_state{}}.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.



%%% ============================================================================
%%%  Internal functions
%%% ============================================================================


%% ----------------------------------------------------------------------------
%% @private Construct a nodeup topic event and send it.
%%
signal_nodeup(UpNode) ->
    {{active, A}, {configured, C}} = jc:cache_nodes(),
    topic_event(jc_node_events, {nodeup, UpNode, A, C}).


%% ----------------------------------------------------------------------------
%% @private Construct a nodedown topic event and send it.
%%
signal_nodedown(DownNode) ->
    {{active, A}, {configured, C}} = jc:cache_nodes(),
    % Mnesia might not know that the node is down yet, so remove
    % it explicitly from the list.
    topic_event(jc_node_events, {nodedown, 
				 DownNode, 
				 lists:delete(DownNode, A), 
				 C}).


%% ----------------------------------------------------------------------------
%% @private Notify subscribers of evicts, writes and deletes.
%%

notify_of_evict(#key_to_value{map = M, key = K}) ->
    lager:debug("~p: broadcast evict: {~p, ~p}.", [?SERVER, M, K]),
    broadcast_map_event({M, K, delete}, {M, K, delete}).


notify_of_write(#key_to_value{map = M, key = K, value = V}) ->
    lager:debug("~p: broadcast write: {~p, ~p}.", [?SERVER, M, K]),
    broadcast_map_event({M, K, write}, {M, K, write, V}).


notify_of_delete(#key_to_value{map = Map, key = Key}) ->
    lager:debug("~p: mnesia delete activity: {~p, ~p}.",
		[?SERVER, Map, Key]),
    broadcast_map_event({Map, Key, delete}, {Map, Key, delete}).


    
%% -----------------------------------------------------------------------------
%% @private Number of subscriptions + Pids listening as a proxy for load.
%% 
calculate_load()->
    F = fun() -> mnesia:foldl(fun(#ps_sub{clients = Clients}, Load) ->
				      Load + sets:size(Clients) + 1
			      end, 0, ps_sub)
	end,
    mnesia:async_dirty(F).
    

%% -----------------------------------------------------------------------------
%% @private Monitor a client (Pid), add its Pid to the client table.
%%
-spec monitor_client(pid()) -> ok.

monitor_client(Pid) ->
    lager:debug("~p: monitoring client: ~p:", [?SERVER, Pid]),
    
    F = fun() ->
		case mnesia:read(ps_client, Pid) of
		    [] ->
			Result = monitor_as_possible(Pid),
			mnesia:write(#ps_client{pid = Pid, m_type = Result});
		    [#ps_client{}] ->
			% process already being monitored
			ok
		end
	end,
    mnesia:transaction(F),
    ok.


%% -----------------------------------------------------------------------------
%% @private Demonitor a client and remove its Pid from the client table.
%%
-spec demonitor_client(Client::pid()) -> true | false.

demonitor_client(Pid) ->
    lager:debug("~p: demonitoring client: ~p.", [?SERVER, Pid]),

    F = fun() ->
		Result = case mnesia:read(ps_client, Pid) of
			     [#ps_client{m_type = {ref, Ref}}] ->
				 true = erlang:demonitor(Ref, [flush]);
			     [#ps_client{m_type = {link, _}}] -> 
				 true = erlang:unlink(Pid);
			     [] -> 
				 false
			 end,
		mnesia:delete({ps_client, Pid}),
		Result 
	end,
    mnesia:sync_dirty(F).


%% -----------------------------------------------------------------------------
%% Pids that cannot be monitored will be liked.
%%
monitor_as_possible(Pid) ->
    try erlang:monitor(process, Pid) of
	Ref -> {ref, Ref}
   catch
	error:_ ->
	    {link, erlang:link(Pid)}
    end.
    

%% -----------------------------------------------------------------------------
%% @private If the subscription does not exist, add it and set the client (Pid)
%% as its only subscriber. If the subscription exists, add the client (Pid)
%% to the set of subscribers. Monitor the client (Pid) so that if it goes away,
%% we can removie it as a subscriber.
%%
subscribe(Pid, Subscription ) ->
    lager:debug("~p: adding subscription: ~p for client: ~p.", 
		[?SERVER, Subscription, Pid]),

    monitor_client(Pid),

    F = fun() ->
		case mnesia:read(ps_sub, Subscription) of
		    []->
			mnesia:write(#ps_sub{subscription = Subscription, 
					     clients = sets:from_list([Pid])});
		    [#ps_sub{clients = Pids} = Rec]->
			PidSet = sets:add_element(Pid, Pids),
			mnesia:write(Rec#ps_sub{clients = PidSet})
		end
	end,
    mnesia:transaction(F).
		

%% -----------------------------------------------------------------------------
%% @private Assuming the subscription exists, remove the client (Pid) from the
%% set of listeners to that subscription.
%%
unsubscribe(Pid, Subscription) ->
    lager:debug("~p: deleting subscription: ~p for client: ~p.", 
		[?SERVER, Subscription, Pid]),
    F = fun() ->
		case mnesia:read(ps_sub, Subscription) of
		    [#ps_sub{clients = Pids} = Rec] ->
			NewSetOfPids = sets:del_element(Pid, Pids),
			case sets:size(NewSetOfPids) of
			    0 ->
				mnesia:delete_object(Rec);
			    _N ->
				mnesia:write(Rec#ps_sub{clients = NewSetOfPids})
			end;
		    []->
			ok
		end
	end,
    mnesia:transaction(F).


%% -----------------------------------------------------------------------------
%% @private Remove subscriber (Pid) from all subscriptions.
%%
unsubscribe_all(Pid) when is_pid(Pid) ->
    case demonitor_client(Pid) of
	true ->
	    F = fun(#ps_sub{clients = Pids} = Rec, _)->
			NewSetOfPids = sets:del_element(Pid, Pids),
			case sets:size(NewSetOfPids) of
			    0 ->
				mnesia:delete_object(Rec);
			    _N ->
				mnesia:write(Rec#ps_sub{clients = NewSetOfPids})
			end
		end,
	    Fld = fun() -> mnesia:foldl(F, [], ps_sub) end,
	    mnesia:transaction(Fld);
	false ->
	    ok
    end.


%% -----------------------------------------------------------------------------
%% Broadcast changes to subscribing pids. An Operation on a key could 
%% particiapte in 'any' subscriptions with respect to Keys or Operations. Ditto
%% for topics.
%%
broadcast_topic_event(Topic, Value)->
    lager:debug("~p: broadcasting topic event.", [?SERVER]),
    Variations = topic_variations(Topic, Value),
    [broadcast_change(Variation, {topic_event,{Topic, Value}}) 
     || Variation <- Variations].

topic_variations(Topic, Value)->
    [#topic_sub{topic = Topic, value = Value},
     #topic_sub{topic = Topic, value = any}].


broadcast_map_event(Event, Payload)->
    lager:debug("~p: broadcasting map event.", [?SERVER]),
    Variations = map_variations(Event),
    [broadcast_change(Variation, {map_event, Payload}) || Variation <- Variations].


map_variations({Map, Key, Operation}) ->
    [#map_sub{map = Map, key = Key, operation = Operation},
     #map_sub{map = Map, key = Key, operation = any},
     #map_sub{map = Map, key = any, operation = Operation},
     #map_sub{map = Map, key = any, operation = any}].


broadcast_change(Subscription, Message) ->
    F = fun() ->
		case mnesia:read(ps_sub, Subscription) of
		    [#ps_sub{clients = Set}] ->
			send_to_pids(Set, Message);
		    _else ->
			ok
		end
	end,
    mnesia:async_dirty(F).


%% -----------------------------------------------------------------------------
%% Send the Message to all of the Pids in the list Pids.
%%
send_to_pids(Pids, Message) ->
    sets:fold(fun(P, _) -> P ! Message, [] end, [], Pids).



%% -----------------------------------------------------------------------------
%% @private Create the message that gets sent back to jc_psub to trigger an 
%% eviction of any clients that have no active subscriptions.
%%
-spec schedule_deadbeat_eviction(non_neg_integer()) -> ok.

schedule_deadbeat_eviction(IntMs)->
    erlang:send_after(IntMs, ?SERVER, {evict_deadbeats, IntMs}),
    ok.


%% -----------------------------------------------------------------------------
%% Collect all Pids that are not in a subscription, demonitor and remove.
%%
-spec remove_deadbeats() -> ok.

remove_deadbeats() ->
    F = fun() ->
		PidsWithSubs = 
		    mnesia:foldl(fun(#ps_sub{clients=Set}, Acc) -> 
					 sets:union(Set, Acc) 
				 end,
				 sets:new(),
				 ps_sub),

		F2 = fun(#ps_client{pid=Pid}, Acc) ->
			     case sets:is_element(Pid, PidsWithSubs) of
				 true  -> Acc;
				 false -> [Pid|Acc]
			     end
		     end,
		DeadBeats = mnesia:foldl(F2, [], ps_client),
		[demonitor_client(Pid) || Pid<-DeadBeats]
	end,
    mnesia:transaction(F),
    ok.