<commit_before>%% Copyright 2012 PEAK6 Investments, L.P.
%%
%% Licensed 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.
%%%-------------------------------------------------------------------
%%% @author <EMAIL>
%%% @copyright (C) 2012, PEAK6 Investments, L.P.
%%% @doc
%%%
%%% @end
%%% Created : Mon May 21 09:23:59 CDT 2011
%%%-------------------------------------------------------------------
-module(sub_chans).

-export([start_link/0]).
-export([xformEntryAdd/1]).

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

-include_lib("p6core/include/logger.hrl").
-include("mmd.hrl").
-include_lib("p6core/include/dmap.hrl").

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

xformEntryAdd(E) -> E.

init([]) ->
    p6dmap:new(subs, ?MODULE),
    services:regLocal('sub.chans'),
    timer:send_interval(100, update_subs),
    {ok, channel_mgr:new()}.

subs(Topic) ->
    lists:foldl(
      fun ([_ChanPid, {_SubPid, ChanId}], ChanIds) ->
	      [?uuid(ChanId) | ChanIds]
      end,
      [], p6dmap:get(subs, Topic)).

handle_call({mmd, From, Msg}, _From, Chans) ->
    {reply, ok,
     case channel_mgr:process_remote(Chans, From, Msg) of
         {NewChans, CC=#channel_create{type=call, body=Raw}} ->
	     {NewChans2, []} =
		 channel_mgr:process_local(
		   NewChans,
		   case mmd_decode:decode(Raw) of
		       {<<Topic/binary>>, _} ->
			   mmd_msg:mkReply(CC, subs(Topic));
		       _ ->
			   mmd_msg:mkError(
			     CC,
			     ?INVALID_REQUEST,
			     <<"'subs' must be called with a topic to "
			       "list subscribers for">>)
		   end),
	     NewChans2;

         {NewChans, CC=#channel_create{type=sub, body=Raw}} ->
	     {NewChans2, []} =
		 case mmd_decode:decode(Raw) of
		       {<<Topic/binary>>, _} ->
			 Subs = subs(Topic),
			 channel_mgr:process_local_set_data(
			   NewChans,
			   mmd_msg:mkReply(CC, Subs),
			   {Topic, Subs});
		       _ ->
			 channel_mgr:process_local(
			   NewChans,
			   mmd_msg:mkError(
			     CC,
			     ?INVALID_REQUEST,
			     <<"'subs' must be subscribed with a topic to "
			       "list subscribers for">>))
		   end,
	     NewChans2;
         {NewChans, CM=#channel_message{}} ->
             {Chans2, _M} =
                 channel_mgr:process_local(
                   NewChans,
                   mmd_msg:mkError(CM,
                                   ?INVALID_REQUEST,
                                   <<"'sub' doesn't accept channel message">>)),
             Chans2;
         {NewChans, #channel_close{}} ->
             NewChans
     end};
handle_call(Request, From, State) ->
    ?lwarn("Unexpected handle_call(~p, ~p, ~p)",[Request,From,State]),
    {reply, ok, State}.

handle_cast({pub, Msg}, Chans) ->
    channel_mgr:sendAll(Chans, Msg),
    {noreply, Chans};
handle_cast(Msg, State) -><commit_msg>sub.chans: initial reply wasn't a in the {'added': ...} form<commit_after>%% Copyright 2012 PEAK6 Investments, L.P.
%%
%% Licensed 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.
%%%-------------------------------------------------------------------
%%% @author <EMAIL>
%%% @copyright (C) 2012, PEAK6 Investments, L.P.
%%% @doc
%%%
%%% @end
%%% Created : Mon May 21 09:23:59 CDT 2011
%%%-------------------------------------------------------------------
-module(sub_chans).

-export([start_link/0]).
-export([xformEntryAdd/1]).

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

-include_lib("p6core/include/logger.hrl").
-include("mmd.hrl").
-include_lib("p6core/include/dmap.hrl").

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

xformEntryAdd(E) -> E.

init([]) ->
    p6dmap:new(subs, ?MODULE),
    services:regLocal('sub.chans'),
    timer:send_interval(100, update_subs),
    {ok, channel_mgr:new()}.

subs(Topic) ->
    lists:foldl(
      fun ([_ChanPid, {_SubPid, ChanId}], ChanIds) ->
	      [?uuid(ChanId) | ChanIds]
      end,
      [], p6dmap:get(subs, Topic)).

handle_call({mmd, From, Msg}, _From, Chans) ->
    {reply, ok,
     case channel_mgr:process_remote(Chans, From, Msg) of
         {NewChans, CC=#channel_create{type=call, body=Raw}} ->
	     {NewChans2, []} =
		 channel_mgr:process_local(
		   NewChans,
		   case mmd_decode:decode(Raw) of
		       {<<Topic/binary>>, _} ->
			   mmd_msg:mkReply(CC, subs(Topic));
		       _ ->
			   mmd_msg:mkError(
			     CC,
			     ?INVALID_REQUEST,
			     <<"'subs' must be called with a topic to "
			       "list subscribers for">>)
		   end),
	     NewChans2;

         {NewChans, CC=#channel_create{type=sub, body=Raw}} ->
	     {NewChans2, []} =
		 case mmd_decode:decode(Raw) of
		       {<<Topic/binary>>, _} ->
			 Subs = subs(Topic),
			 channel_mgr:process_local_set_data(
			   NewChans,
			   mmd_msg:mkReply(CC, ?map([{added, ?array(Subs)}])),
			   {Topic, Subs});
		       _ ->
			 channel_mgr:process_local(
			   NewChans,
			   mmd_msg:mkError(
			     CC,
			     ?INVALID_REQUEST,
			     <<"'subs' must be subscribed with a topic to "
			       "list subscribers for">>))
		   end,
	     NewChans2;
         {NewChans, CM=#channel_message{}} ->
             {Chans2, _M} =
                 channel_mgr:process_local(
                   NewChans,
                   mmd_msg:mkError(CM,
                                   ?INVALID_REQUEST,
                                   <<"'sub' doesn't accept channel message">>)),
             Chans2;
         {NewChans, #channel_close{}} ->
             NewChans
     end};
handle_call(Request, From, State) ->
    ?lwarn("Unexpected handle_call(~p, ~p, ~p)",[Request,From,State]),
    {reply, ok, State}.

handle_cast({pub, Msg}, Chans) ->
    channel_mgr:sendAll(Chans, Msg),
    {noreply, Chans};
handle_cast(Msg, State) ->
{'fraction_non_alphanumeric': 0.18681965283907032, 'fraction_numerical': 0.013827596351868197, 'mean_word_length': 2.318204001952172, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '575996', 'n_tokens_mistral': 2454, 'n_tokens_neox': 2218, 'n_words': 509}