cosnotifychanneladmin_consumeradmin_impl.erl

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

		{'CosNotifyChannelAdmin_SequenceProxyPushSupplier', 'PUSH_SEQUENCE'};
	    _ ->
		orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:"
			  "obtain_notification_push_supplier(~p);~n"
			  "Incorrect enumerant", [?LINE, Ctype], ?DEBUG_LEVEL),
		corba:raise(#'BAD_PARAM'{completion_status=?COMPLETED_NO})
	end,
    SO = 'CosNotification_Common':get_option(server_options, ?get_Options(State), 
					     ?not_DEFAULT_SETTINGS),
    case Mod:oe_create_link([Type, OE_THIS, self(), ?get_GlobalQoS(State), 
			     ?get_LocalQoS(State), ?get_MyChannel(State),
			     ?get_Options(State), ?get_MyOperator(State)], 
			    [{sup_child, true}|SO]) of
	{ok, Pid, PrRef} ->
	    ProxyID = ?new_Id(State),
	    NewState = ?add_PushSupplier(State, ProxyID, PrRef, Pid),
	    {reply, {PrRef, ProxyID}, ?set_IdCounter(NewState, ProxyID)};
	What ->
	    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:obtain_notification_push_supplier();~n"
		      "Unable to create: ~p/~p~n"
		      "Reason: ~p", 
		      [?LINE, Mod, Type, What], ?DEBUG_LEVEL),
	    corba:raise(#'INTERNAL'{completion_status=?COMPLETED_NO})
    end.
    

%%----------------------------------------------------------%
%% function : destroy
%% Arguments: -
%% Returns  : ok
%%------------------------------------------------------------
destroy(_OE_THIS, _OE_FROM, State) ->
    {stop, normal, ok, State}.    

%%----- Inherit from CosNotification::QoSAdmin --------------
%%----------------------------------------------------------%
%% function : get_qos
%% Arguments: 
%% Returns  : 
%%-----------------------------------------------------------
get_qos(_OE_THIS, _OE_FROM, State) ->
    {reply, ?get_GlobalQoS(State), State}.    

%%----------------------------------------------------------%
%% function : set_qos
%% Arguments: QoS - CosNotification::QoSProperties, i.e.,
%%            [#'Property'{name, value}, ...] where name eq. string()
%%            and value eq. any().
%% Returns  : ok | {'EXCEPTION', CosNotification::UnsupportedQoS}
%%-----------------------------------------------------------
set_qos(_OE_THIS, _OE_FROM, State, QoS) ->
    {NewQoS, LQS} = 'CosNotification_Common':set_qos(QoS, ?get_BothQoS(State), 
						     admin, ?get_MyChannel(State), 
						     ?get_AllSupplierRefs(State)),
    {reply, ok, ?set_BothQoS(State, NewQoS, LQS)}.

%%----------------------------------------------------------%
%% function : validate_qos
%% Arguments: Required_qos - CosNotification::QoSProperties
%%            [#'Property'{name, value}, ...] where name eq. string()
%%            and value eq. any().
%% Returns  : {'EXCEPTION', CosNotification::UnsupportedQoS}
%%-----------------------------------------------------------
validate_qos(_OE_THIS, _OE_FROM, State, Required_qos) ->
    QoS = 'CosNotification_Common':validate_qos(Required_qos, ?get_BothQoS(State), 
						admin, ?get_MyChannel(State), 
						?get_AllSupplierRefs(State)),
    {reply, {ok, QoS}, State}.

%%----- Inherit from CosNotifyComm::NotifySubscribe ---------
%%----------------------------------------------------------*
%% function : subscription_change
%% Arguments: 
%% Returns  : ok | 
%%            {'EXCEPTION', #'CosNotifyComm_InvalidEventType'{type}}
%%-----------------------------------------------------------
subscription_change(_OE_THIS, _OE_FROM, State, _Added, _Removed) ->
    ?DBG("CALLBACK INFORMED:  ~p   ~p~n",[_Added, _Removed]),
    {reply, ok, State}.

%%----- Inherit from CosNotifyFilter::FilterAdmin -----------
%%----------------------------------------------------------%
%% function : add_filter
%% Arguments: Filter - CosNotifyFilter::Filter
%% Returns  : FilterID - long
%%-----------------------------------------------------------
add_filter(_OE_THIS, _OE_FROM, State, Filter) ->
    'CosNotification_Common':type_check(Filter, 'CosNotifyFilter_Filter'),
    FilterID = ?new_Id(State),
    NewState = ?set_IdCounter(State, FilterID),
    {reply, FilterID, ?add_Filter(NewState, FilterID, Filter)}.

%%----------------------------------------------------------%
%% function : remove_filter
%% Arguments: FilterID - long
%% Returns  : ok
%%-----------------------------------------------------------
remove_filter(_OE_THIS, _OE_FROM, State, FilterID) when integer(FilterID) ->
    {reply, ok, ?del_Filter(State, FilterID)};
remove_filter(_,_,_,_) ->
    corba:raise(#'BAD_PARAM'{completion_status=?COMPLETED_NO}).

%%----------------------------------------------------------%
%% function : get_filter
%% Arguments: FilterID - long
%% Returns  : Filter - CosNotifyFilter::Filter |
%%            {'EXCEPTION', #'CosNotifyFilter_FilterNotFound'{}}
%%-----------------------------------------------------------
get_filter(_OE_THIS, _OE_FROM, State, FilterID) when integer(FilterID) ->
    {reply, ?get_Filter(State, FilterID), State};
get_filter(_,_,_,_) ->
    corba:raise(#'BAD_PARAM'{completion_status=?COMPLETED_NO}).

%%----------------------------------------------------------%
%% function : get_all_filters
%% Arguments: -
%% Returns  : Filter - CosNotifyFilter::FilterIDSeq
%%-----------------------------------------------------------
get_all_filters(_OE_THIS, _OE_FROM, State) ->
    {reply, ?get_AllFilterID(State), State}.

%%----------------------------------------------------------%
%% function : remove_all_filters
%% Arguments: -
%% Returns  : ok
%%-----------------------------------------------------------
remove_all_filters(_OE_THIS, _OE_FROM, State) ->
    {reply, ok, ?del_AllFilter(State)}.

%%----- Inherit from CosEventChannelAdmin::ConsumerAdmin ----
%%----------------------------------------------------------%
%% function : obtain_push_supplier
%% Arguments: -
%% Returns  : ProxyPushSupplier
%%-----------------------------------------------------------
obtain_push_supplier(OE_THIS, _OE_FROM, State) ->
    SO = 'CosNotification_Common':get_option(server_options, ?get_Options(State), 
					     ?not_DEFAULT_SETTINGS),
    case 'CosNotifyChannelAdmin_ProxyPushSupplier':oe_create_link(['PUSH_ANY', OE_THIS, 
								   self(), 
								   ?get_GlobalQoS(State), 
								   ?get_LocalQoS(State),
								   ?get_MyChannel(State),
								   ?get_Options(State),
								   ?get_MyOperator(State)], 
								  [{sup_child, true}|SO]) of
	{ok, Pid, PrRef} ->
	    ProxyID = ?new_Id(State),
	    NewState = ?add_PushSupplier(State, ProxyID, PrRef, Pid),
	    {reply, PrRef, ?set_IdCounter(NewState, ProxyID)};
	What ->
	    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:obtain_push_supplier();~n"
		      "Unable to create: CosNotifyChannelAdmin_ProxyPushSupplier~n"
		      "Reason: ~p", [?LINE, What], ?DEBUG_LEVEL),
	    corba:raise(#'INTERNAL'{completion_status=?COMPLETED_NO})
    end.

%%----------------------------------------------------------%
%% function : obtain_pull_supplier
%% Arguments: -
%% Returns  : ProxyPullSupplier
%%-----------------------------------------------------------
obtain_pull_supplier(OE_THIS, _OE_FROM, State) ->
    SO = 'CosNotification_Common':get_option(server_options, ?get_Options(State), 
					     ?not_DEFAULT_SETTINGS),
    case 'CosNotifyChannelAdmin_ProxyPullSupplier':oe_create_link(['PULL_ANY', OE_THIS, 
								   self(), 
								   ?get_GlobalQoS(State), 
								   ?get_LocalQoS(State),
								   ?get_MyChannel(State),
								   ?get_Options(State), 
								   ?get_MyOperator(State)], 
								  [{sup_child, true}|SO]) of
	{ok, Pid, PrRef} ->
	    ProxyID = ?new_Id(State),
	    NewState = ?add_PullSupplier(State, ProxyID, PrRef, Pid),
	    {reply, PrRef, ?set_IdCounter(NewState, ProxyID)};
	What ->
	    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:obtain_pull_supplier();~n"
		      "Unable to create: CosNotifyChannelAdmin_ProxyPullSupplier~n"
		      "Reason: ~p", [?LINE, What], ?DEBUG_LEVEL),
	    corba:raise(#'INTERNAL'{completion_status=?COMPLETED_NO})
    end.

%%--------------- LOCAL FUNCTIONS ----------------------------
%% To match suppliers
find_obj({value, {_,Obj,_,_}},_) -> Obj;
%% To match filters
find_obj({value, {_,Obj}},_) -> Obj;
find_obj(_, supplier) -> {'EXCEPTION', #'CosNotifyChannelAdmin_ProxyNotFound'{}};
find_obj(_, filter) -> {'EXCEPTION', #'CosNotifyFilter_FilterNotFound'{}}.

find_ids(List) ->              
    find_ids(List, [], false).
find_ids(List, Type) ->        
    find_ids(List, [], Type).

find_ids([], Acc, _) ->        
    Acc;
find_ids([{I,_}|T], Acc, Type) -> 
    find_ids(T, [I|Acc], Type);
find_ids([{I,_,_,Type}|T], Acc, Type) -> 
    find_ids(T, [I|Acc], Type);
find_ids([{_I,_,_,_}|T], Acc, Type) -> 
    find_ids(T, Acc, Type);
find_ids(What, _, _) -> 
    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:find_ids();~n"
	      "Id corrupt: ~p", [?LINE, What], ?DEBUG_LEVEL),
    corba:raise(#'INTERNAL'{completion_status=?COMPLETED_NO}).

find_refs(List) ->              
    find_refs(List, []).

find_refs([], Acc) ->      
    Acc;
find_refs([{_,R,_,_}|T], Acc) -> 
    find_refs(T, [R|Acc]);
find_refs(What, _) -> 
    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:find_refs();~n"
	      "Reference corrupt: ~p", [?LINE, What], ?DEBUG_LEVEL),
    corba:raise(#'INTERNAL'{completion_status=?COMPLETED_NO}).

%% Delete a single filter.
%% The list do not differ, i.e., no filter removed, raise exception.
delete_filter(List,List) -> corba:raise(#'CosNotifyFilter_FilterNotFound'{});
delete_filter(List, _) -> List.


%%-----------------------------------------------------------
%% function : callSeq
%% Arguments: 
%% Returns  : 
%%-----------------------------------------------------------
callSeq(_OE_THIS, OE_FROM, State, Events, _Status) ->
    corba:reply(OE_FROM, ok),
    case cosNotification_eventDB:filter_events(Events, ?get_AllFilter(State)) of
	{[], _}  when ?is_ANDOP(State) ->
	    %% Since AND it doesn't matter what the proxies 'think'. Done.
	    {noreply, State};
	{[], Failed} ->
	    %% Is OR but the Proxy may allow the events; pass on.
	    forward(seq, State, Failed, 'MATCH');
	{Passed, _}  when ?is_ANDOP(State) ->
	    %% Since AND we only forward those who passed this objects filters.
	    forward(seq, State, Passed, 'MATCH');
	{Passed, []} ->
	    %% Since OR we forward and tell the proxy to do no filtering.
	    forward(seq, State, Passed, 'MATCHED');
	{Passed, Failed} ->
	    %% Since OR we forward both and instruct the proxy to check only
	    %% the ones that failed.
	    forward(seq, State, Passed, 'MATCHED'),
	    forward(seq, State, Failed, 'MATCH')
    end.


%%-----------------------------------------------------------
%% function : callAny
%% Arguments: 
%% Returns  : 
%%-----------------------------------------------------------
callAny(_OE_THIS, OE_FROM, State, Event, _Status) ->
    corba:reply(OE_FROM, ok),
    case cosNotification_eventDB:filter_events([Event], ?get_AllFilter(State)) of
	{[], _}  when ?is_ANDOP(State) ->
	    %% Since AND it doesn't matter what the proxies 'think'. Done.
	    {noreply, State};
	{[], [Failed]} ->
	    %% Is OR but the Proxy may allow the event; pass on.
	    forward(any, State, Failed, 'MATCH');
	{[Passed], _}  when ?is_ANDOP(State) ->
	    %% Since AND we only forward those who passed this objects filters.
	    forward(any, State, Passed, 'MATCH');
	{[Passed], _} ->
	    %% Since OR we forward and instruct the proxy to do no checks.
	    forward(any, State, Passed, 'MATCHED')
    end.
 
    

%% Forward events
forward(Type, State, Event, Status) ->
    forward(Type, ?get_AllSuppliers(State), State, Event, Status).
forward(_, [], State, _, _) ->
    {noreply, State};
forward(any, [{_,H,_,_}|T], State, Event, Status) ->
    case catch oe_CosNotificationComm_Event:callAny(H, Event, Status) of
	ok ->
	    ?DBG("CONSUMERADM FORWARD ANY: ~p~n",[Event]),
	    forward(any, T, State, Event, Status);
	{'EXCEPTION', E} when record(E, 'OBJECT_NOT_EXIST') ->
	    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:forward();~n"
		      "Proxy no longer exists; dropping it: ~p", 
		      [?LINE, H], ?DEBUG_LEVEL),
	    NewState = ?del_SupplierRef(State,H),
	    forward(any, T, NewState, Event, Status);
	R when ?is_PersistentConnection(State) ->
	    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:forward();~n"
		      "Proxy behaves badly: ~p/~p", 
		      [?LINE, R, H], ?DEBUG_LEVEL),
	    forward(any, T, State, Event, Status);
	R ->
	    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:forward();~n"
		      "Proxy behaves badly: ~p~n"
		      "Dropping it: ~p", [?LINE, R, H], ?DEBUG_LEVEL),
	    NewState = ?del_SupplierRef(State, H),
	    forward(any, T, NewState, Event, Status)
    end;
forward(seq, [{_,H,_,_}|T], State, Event, Status) ->
    case catch oe_CosNotificationComm_Event:callSeq(H, Event, Status) of
	ok ->
	    ?DBG("CONSUMERADM FORWARD SEQUENCE: ~p~n",[Event]),
	    forward(seq, T, State, Event, Status);
	{'EXCEPTION', E} when record(E, 'OBJECT_NOT_EXIST') ->
	    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:forward();~n"
		      "Proxy no longer exists; dropping it: ~p", 
		      [?LINE, H], ?DEBUG_LEVEL),
	    NewState = ?del_SupplierRef(State,H),
	    forward(seq, T, NewState, Event, Status);
	R when ?is_PersistentConnection(State) ->
	    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:forward();~n"
		      "Proxy behaves badly: ~p/~p", [?LINE, R, H], ?DEBUG_LEVEL),
	    forward(seq, T, State, Event, Status);
	R ->
	    orber:dbg("[~p] CosNotifyChannelAdmin_ConsumerAdmin:forward();~n"
		      "Proxy behaves badly: ~p~n"
		      "Dropping it: ~p", [?LINE, R, H], ?DEBUG_LEVEL),
	    NewState = ?del_SupplierRef(State, H),
	    forward(seq, T, NewState, Event, Status)
    end.


%%--------------- MISC FUNCTIONS, E.G. DEBUGGING -------------
%%--------------- END OF MODULE ------------------------------