pushersupplier_impl.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 1,052 行 · 第 1/3 页

%%--------------------------------------------------------------------
%% ``The contents of this file are subject to the Erlang Public License,
%% Version 1.1, (the "License"); you may not use this file except in
%% compliance with the License. You should have received a copy of the
%% Erlang Public License along with this software. If not, it can be
%% retrieved via the world wide web at http://www.erlang.org/.
%% 
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and limitations
%% under the License.
%% 
%% The Initial Developer of the Original Code is Ericsson Utvecklings AB.
%% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
%% AB. All Rights Reserved.''
%% 
%%     $Id$
%%
%%----------------------------------------------------------------------
%% File    : PusherSupplier_impl.erl
%% Purpose : 
%%% Created : 20 Oct 1999
%%%----------------------------------------------------------------------

-module('PusherSupplier_impl').


%%--------------- INCLUDES -----------------------------------
-include_lib("orber/include/corba.hrl").
-include_lib("orber/include/ifr_types.hrl").
%% cosEvent files.
-include_lib("cosEvent/include/CosEventChannelAdmin.hrl").
-include_lib("cosEvent/include/CosEventComm.hrl").
%% Application files
-include("CosNotification.hrl").
-include("CosNotifyChannelAdmin.hrl").
-include("CosNotifyComm.hrl").
-include("CosNotifyFilter.hrl").

-include("CosNotification_Definitions.hrl").

%%--------------- EXPORTS ------------------------------------
%%--------------- External -----------------------------------
%%----- CosNotifyChannelAdmin::ProxyPushSupplier -------------
-export([connect_any_push_consumer/4]).

%%----- CosNotifyChannelAdmin::StructuredProxyPushSupplier ---
-export([connect_structured_push_consumer/4]).

%%----- CosNotifyChannelAdmin::SequenceProxyPushSupplier -----
-export([connect_sequence_push_consumer/4]).

%%----- CosNotifyChannelAdmin::*ProxyPushSupplier ------------
-export([suspend_connection/3, 
	 resume_connection/3]).

%%----- Inherit from CosNotifyChannelAdmin::ProxySupplier ----
-export([obtain_offered_types/4,
	 validate_event_qos/4]).

%%----- Inherit from CosNotification::QoSAdmin ---------------
-export([get_qos/3,
	 set_qos/4,
	 validate_qos/4]).

%%----- Inherit from CosNotifyComm::NotifySubscribe ----------
-export([subscription_change/5]).

%%----- Inherit from CosNotifyFilter::FilterAdmin ------------
-export([add_filter/4, 
	 remove_filter/4, 
	 get_filter/4,
	 get_all_filters/3, 
	 remove_all_filters/3]).

%%----- Inherit from CosEventComm::PushSupplier -------------
-export([disconnect_push_supplier/3]).

%%----- Inherit from CosNotifyComm::StructuredPushSupplier --
-export([disconnect_structured_push_supplier/3]).

%%----- Inherit from CosNotifyComm::SequencePushSupplier ----
-export([disconnect_sequence_push_supplier/3]).

%%----- Inherit from CosEventChannelAdmin::ProxyPushSupplier 
-export([connect_push_consumer/4]).

%% Attributes (external) CosNotifyChannelAdmin::ProxySupplier
-export(['_get_MyType'/3, 
	 '_get_MyAdmin'/3, 
	 '_get_priority_filter'/3,
	 '_set_priority_filter'/4,
	 '_get_lifetime_filter'/3,
	 '_set_lifetime_filter'/4]).

%%--------------- Internal -----------------------------------
%%----- Inherit from cosNotificationComm ---------------------
-export([callAny/5,
	 callSeq/5]).

%%--------------- gen_server specific exports ----------------
-export([handle_info/2, code_change/3]).
-export([init/1, terminate/2]).

%%--------------- LOCAL DEFINITIONS --------------------------
%% Data structures
-record(state, {myType,
	        myAdmin,
	        myAdminPid,
		myChannel,
		myFilters = [],
		myOperator,
		idCounter = 0,
		prioFil,
		lifetFil,
		client,
		qosGlobal,
		qosLocal,
		suspended = false,
		pacingTimer,
		subscribeType = false,
		subscribeData = true,
		etsR,
		eventDB,
		this,
		maxCache,
		cacheTimeout,
		cacheInterval}).

%% Data structures constructors
-define(get_InitState(_MyT, _MyA, _MyAP, _QS, _LQS, _Ch, _MyOp, _GT, _GL, _TR), 
	#state{myType    = _MyT,
	       myAdmin   = _MyA,
	       myAdminPid= _MyAP,
	       qosGlobal = _QS,
	       qosLocal  = _LQS,
	       myChannel = _Ch,
	       myOperator=_MyOp,
	       etsR      = ets:new(oe_ets, [set, protected]),
	       eventDB   = cosNotification_eventDB:create_db(_LQS, _GT, _GL, _TR),
	       maxCache = cosNotificationApp:max_events()}).

%% Data structures selectors
%%-------------- Data structures selectors -----------------
%% Attributes
-define(get_MyType(S),           S#state.myType).
-define(get_MyAdmin(S),          S#state.myAdmin).
-define(get_MyAdminPid(S),       S#state.myAdminPid).
-define(get_MyChannel(S),        S#state.myChannel).
-define(get_MyOperator(S),       S#state.myOperator).
-define(get_PrioFil(S),          S#state.prioFil).
-define(get_LifeTFil(S),         S#state.lifetFil).
%% Client Object
-define(get_Client(S),           S#state.client).
%% QoS
-define(get_GlobalQoS(S),        S#state.qosGlobal).
-define(get_LocalQoS(S),         S#state.qosLocal).
-define(get_BothQoS(S),          {S#state.qosGlobal, S#state.qosLocal}).
%% Filters
-define(get_Filter(S, I),        find_obj(lists:keysearch(I, 1, S#state.myFilters))).
-define(get_AllFilter(S),        S#state.myFilters).
-define(get_AllFilterID(S),      find_ids(S#state.myFilters)).
%% Amin
-define(get_PacingTimer(S),      S#state.pacingTimer).
-define(get_PacingInterval(S),   round(?not_GetPacingInterval((S#state.qosLocal))/10000000)).
-define(get_BatchLimit(S),       ?not_GetMaximumBatchSize((S#state.qosLocal))).
%% Subscribe
-define(get_AllSubscribe(S),     lists:flatten(ets:match(S#state.etsR,
							 {'$1',subscribe}))).
-define(get_SubscribeType(S),    S#state.subscribeType).
-define(get_SubscribeData(S),    S#state.subscribeData).
%% ID
-define(get_IdCounter(S),        S#state.idCounter).
-define(get_SubscribeDB(S),      S#state.etsR).
%% Event
-define(is_PersistentConnection(S),
	(?not_GetConnectionReliability((S#state.qosLocal)) == ?not_Persistent)).
-define(is_PersistentEvent(S),
	(?not_GetEventReliability((S#state.qosLocal)) == ?not_Persistent)).

-define(get_Event(S),            cosNotification_eventDB:get_event(S#state.eventDB, 
								   false)).
%								   (not ?is_PersistentEvent(S)))).
-define(get_Events(S,M),         cosNotification_eventDB:get_events(S#state.eventDB, M, false)).
%								    (not ?is_PersistentEvent(S)))).

%%-------------- Data structures modifiers -----------------
%% Attributes
-define(set_PrioFil(S,D),        S#state{prioFil=D}).
-define(set_LifeTFil(S,D),       S#state{lifetFil=D}).
%% Client Object
-define(set_Client(S,D),         S#state{client=D}).
-define(del_Client(S),           S#state{client=undefined}).
-define(set_Unconnected(S),      S#state{client=undefined}).
-define(set_Suspended(S),        S#state{suspended=true}).
-define(set_NotSuspended(S),     S#state{suspended=false}).
%% QoS
-define(set_LocalQoS(S,D),       S#state{qosLocal=D}).
-define(set_GlobalQoS(S,D),      S#state{qosGlobal=D}).
-define(set_BothQoS(S,GD,LD),    S#state{qosGlobal=GD, qosLocal=LD}).
%% Filters
-define(add_Filter(S,I,O),       S#state{myFilters=[{I,O}|S#state.myFilters]}).
-define(del_Filter(S,I),         S#state{myFilters=
					 delete_obj(lists:keydelete(I, 1, S#state.myFilters),
						    S#state.myFilters)}).
-define(del_AllFilter(S),        S#state{myFilters=[]}).
%% Admin
-define(set_PacingTimer(S,T),    S#state{pacingTimer=T}).
%% Publish
%% Subscribe
-define(add_Subscribe(S,E),      ets:insert(S#state.etsR, {E, subscribe})).
-define(del_Subscribe(S,E),      ets:delete(S#state.etsR, E)).
-define(set_SubscribeType(S,T),  S#state{subscribeType=T}).
-define(set_SubscribeData(S,D),  S#state{subscribeData=D}).
%% ID
-define(set_IdCounter(S,V),      S#state{idCounter=V}).
-define(new_Id(S),               'CosNotification_Common':create_id(S#state.idCounter)).
%% Events
-define(add_Event(S,E),          catch cosNotification_eventDB:
	add_event(S#state.eventDB, E, S#state.lifetFil, S#state.prioFil)).
-define(addAndGet_Event(S,E),    catch cosNotification_eventDB:
	add_and_get_event(S#state.eventDB, E, S#state.lifetFil, S#state.prioFil,
			  false)).
%			  ?is_PersistentEvent(S))).
-define(update_EventDB(S,Q),     S#state{eventDB=
					 cosNotification_eventDB:update(S#state.eventDB, Q)}).


%%-------------- MISC ----------------------------------------
-define(is_ANY(S),               S#state.myType == 'PUSH_ANY').
-define(is_STRUCTURED(S),        S#state.myType == 'PUSH_STRUCTURED').
-define(is_SEQUENCE(S),          S#state.myType == 'PUSH_SEQUENCE').
-define(is_ANDOP(S),             S#state.myOperator == 'AND_OP').
-define(is_UnConnected(S),       S#state.client == undefined).
-define(is_Connected(S),         S#state.client =/= undefined).
-define(is_Suspended(S),         S#state.suspended == true).
-define(is_NotSuspended(S),      S#state.suspended == false).
-define(is_BatchLimitReached(S), cosNotification_eventDB:status(S#state.eventDB,
								{batchLimit, 
								 ?not_GetMaximumBatchSize((S#state.qosLocal))})).
-define(has_Filters(S),          S#state.myFilters =/= []).

%%----------------------------------------------------------%
%% function : handle_info, code_change
%% Arguments: 
%% Returns  : 
%% Effect   : Functions demanded by the gen_server module. 
%%-----------------------------------------------------------

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

handle_info(Info, #state{cacheTimeout = Timeout, 
			 cacheInterval = Interval} = State) ->
    ?DBG("INFO: ~p~n", [Info]),
    case Info of
        {'EXIT', Pid, Reason} when ?get_MyAdminPid(State)==Pid ->
            ?DBG("PARENT ADMIN: ~p  TERMINATED.~n",[Reason]),
	    {stop, Reason, State};
        {'EXIT', _Pid, _Reason} ->
            ?DBG("PROXYPUSHSUPPLIER: ~p  TERMINATED.~n",[_Reason]),
            {noreply, State};
        pacing ->
	    lookup_and_push(State, true);
        cacheInterval ->
	    lookup_and_push(State, true);
	cacheTimeout when Timeout == undefined, Interval == undefined ->
	    %% Late message, do not terminate
	    {noreply, State};
	cacheTimeout ->
	    'CosNotification_Common':notify([{proxy, State#state.this},
					     {client, State#state.client}, 
					     {reason, 
					      {timer, "Reached upper limit"}}]),
	    {stop, normal, State};
        _ ->
            {noreply, State}
    end.

%%----------------------------------------------------------%
%% function : init, terminate
%% Arguments: 
%%-----------------------------------------------------------

init([MyType, MyAdmin, MyAdminPid, InitQoS, LQS, MyChannel, Options, Operator]) ->
    process_flag(trap_exit, true),
    GCTime = 'CosNotification_Common':get_option(gcTime, Options, 
						 ?not_DEFAULT_SETTINGS),
    GCLimit = 'CosNotification_Common':get_option(gcTime, Options, 
						  ?not_DEFAULT_SETTINGS),
    TimeRef = 'CosNotification_Common':get_option(timeService, Options, 
						  ?not_DEFAULT_SETTINGS),
    timer:start(),
    {ok, ?get_InitState(MyType, MyAdmin, MyAdminPid, 
			InitQoS, LQS, MyChannel, Operator, GCTime, GCLimit, TimeRef)}.

terminate(_Reason, State) when ?is_UnConnected(State) ->
    stop_timer(State#state.cacheTimeout),
    stop_timer(State#state.cacheInterval),
    stop_timer(State#state.pacingTimer),
    %% We are not connected to a Client. Hence, no need to invoke disconnect.
    ok;
terminate(_Reason, State) when ?is_ANY(State) ->
    stop_timer(State#state.cacheTimeout),
    stop_timer(State#state.cacheInterval),
    stop_timer(State#state.pacingTimer),
    'CosNotification_Common':disconnect('CosEventComm_PushConsumer', 
					disconnect_push_consumer, 
					?get_Client(State));
terminate(_Reason, State) when ?is_SEQUENCE(State) ->
    stop_timer(State#state.cacheTimeout),
    stop_timer(State#state.cacheInterval),
    stop_timer(State#state.pacingTimer),
    'CosNotification_Common':disconnect('CosNotifyComm_SequencePushConsumer',
					disconnect_sequence_push_consumer,
					?get_Client(State));
terminate(_Reason, State) when ?is_STRUCTURED(State) ->
    stop_timer(State#state.cacheTimeout),
    stop_timer(State#state.cacheInterval),
    stop_timer(State#state.pacingTimer),
    'CosNotification_Common':disconnect('CosNotifyComm_StructuredPushConsumer', 
					disconnect_structured_push_consumer, 
					?get_Client(State)).

%%-----------------------------------------------------------
%%----- CosNotifyChannelAdmin_ProxySupplier attributes ------
%%-----------------------------------------------------------
%%----------------------------------------------------------%
%% Attribute: '_get_MyType'
%% Type     : readonly
%% Returns  : 
%%-----------------------------------------------------------
'_get_MyType'(_OE_THIS, _OE_FROM, State) ->
    {reply, ?get_MyType(State), State}.

%%----------------------------------------------------------%
%% Attribute: '_get_MyAdmin'
%% Type     : readonly
%% Returns  : 
%%-----------------------------------------------------------
'_get_MyAdmin'(_OE_THIS, _OE_FROM, State) ->
    {reply, ?get_MyAdmin(State), State}.

%%----------------------------------------------------------%
%% Attribute: '_*et_priority_filter'
%% Type     : read/write
%% Returns  : 
%%-----------------------------------------------------------
'_get_priority_filter'(_OE_THIS, _OE_FROM, State) ->
    {reply, ?get_PrioFil(State), State}.
'_set_priority_filter'(_OE_THIS, _OE_FROM, State, PrioF) ->