pullerconsumer_impl.erl

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

    {ok, ?get_InitState(MyType, MyAdmin, MyAdminPid, InitQoS, 
			LQS, MyChannel, Secs, Operator, GCTime, 
			GCLimit, TimeRef)}.

terminate(_Reason, State) when ?is_UnConnected(State) ->
    %% We are currently not connected to a client. Hence, no need for sending
    %% a disconnect request.
    stop_timer(State),
    ok;
terminate(_Reason, State) when ?is_ANY(State) ->
    stop_timer(State),
    'CosNotification_Common':disconnect('CosEventComm_PullSupplier', 
					disconnect_pull_supplier, 
					?get_Client(State));
terminate(_Reason, State) when ?is_SEQUENCE(State) ->
    stop_timer(State),
    'CosNotification_Common':disconnect('CosNotifyComm_SequencePullSupplier',
					disconnect_sequence_pull_supplier,
					?get_Client(State));
terminate(_Reason, State) when ?is_STRUCTURED(State) ->
    stop_timer(State),
    'CosNotification_Common':disconnect('CosNotifyComm_StructuredPullSupplier', 
					disconnect_structured_pull_supplier, 
					?get_Client(State)).

%%-----------------------------------------------------------
%%----- CosNotifyChannelAdmin_ProxyConsumer 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}.

%%-----------------------------------------------------------
%%------- Exported external functions -----------------------
%%-----------------------------------------------------------
%%----- CosEventChannelAdmin::ProxyPullConsumer -------------
%%----------------------------------------------------------%
%% function : connect_pull_supplier
%% Arguments: Client - CosEventComm::PullSupplier
%% Returns  :  ok | {'EXCEPTION', #'AlreadyConnected'{}} |
%%            {'EXCEPTION', #'TypeError'{}} |
%%            {'EXCEPTION', #'BAD_OPERATION'{}}
%%            Both exceptions from CosEventChannelAdmin!!!
%%-----------------------------------------------------------
connect_pull_supplier(OE_THIS, OE_FROM, State, Client) ->
    connect_any_pull_supplier(OE_THIS, OE_FROM, State, Client).

%%----- CosNotifyChannelAdmin::ProxyPullConsumer ------------
%%----------------------------------------------------------%
%% function : connect_any_pull_supplier
%% Arguments: Client - CosEventComm::PullSupplier
%% Returns  :  ok | {'EXCEPTION', #'AlreadyConnected'{}} |
%%            {'EXCEPTION', #'TypeError'{}} |
%%            {'EXCEPTION', #'BAD_OPERATION'{}}
%%            Both exceptions from CosEventChannelAdmin!!!
%%-----------------------------------------------------------
connect_any_pull_supplier(OE_THIS, _OE_FROM, State, Client) when ?is_ANY(State) ->
    'CosNotification_Common':type_check(Client, 'CosEventComm_PullSupplier'),
    if
	?is_Connected(State) ->
	    corba:raise(#'CosEventChannelAdmin_AlreadyConnected'{});
	true ->
	    NewState = start_timer(State),
	    {reply, ok, NewState#state{client = Client, this = OE_THIS}}
    end;
connect_any_pull_supplier(_, _, _,_) ->
    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).

%%----- CosNotifyChannelAdmin::SequenceProxyPullConsumer ----
%%----------------------------------------------------------%
%% function : connect_sequence_pull_supplier
%% Arguments: Client - CosNotifyComm::SequencePullSupplier
%% Returns  :  ok | {'EXCEPTION', #'AlreadyConnected'{}} |
%%            {'EXCEPTION', #'TypeError'{}} |
%%            {'EXCEPTION', #'BAD_OPERATION'{}}
%%-----------------------------------------------------------
connect_sequence_pull_supplier(OE_THIS, _OE_FROM, State, Client) when ?is_SEQUENCE(State) ->
    'CosNotification_Common':type_check(Client, 'CosNotifyComm_SequencePullSupplier'),
    if
	?is_Connected(State) ->
	    corba:raise(#'CosEventChannelAdmin_AlreadyConnected'{});
	true ->
	    NewState = start_timer(State),
	    {reply, ok, NewState#state{client = Client, this = OE_THIS}}
    end;
connect_sequence_pull_supplier(_, _, _, _) ->
    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).

%%----- CosNotifyChannelAdmin::StructuredProxyPullConsumer --
%%----------------------------------------------------------%
%% function : connect_structured_pull_supplier
%% Arguments: Client - CosNotifyComm::StructuredPullSupplier
%% Returns  :  ok | {'EXCEPTION', #'AlreadyConnected'{}} |
%%            {'EXCEPTION', #'TypeError'{}} |
%%            {'EXCEPTION', #'BAD_OPERATION'{}}
%%-----------------------------------------------------------
connect_structured_pull_supplier(OE_THIS, _OE_FROM, State, Client) when ?is_STRUCTURED(State) ->
    'CosNotification_Common':type_check(Client, 'CosNotifyComm_StructuredPullSupplier'),
    if
	?is_Connected(State) ->
	    corba:raise(#'CosEventChannelAdmin_AlreadyConnected'{});
	true ->
	    NewState = start_timer(State),
	    {reply, ok, NewState#state{client = Client, this = OE_THIS}}
    end;
connect_structured_pull_supplier(_, _, _, _)  ->
    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).

%%----- CosNotifyChannelAdmin::*ProxyPullConsumer -----------
%%----------------------------------------------------------%
%% function : suspend_connection
%% Arguments: 
%% Returns  : ok | {'EXCEPTION', #'ConnectionAlreadyInactive'{}} |
%%            {'EXCEPTION', #'NotConneced'{}}
%%-----------------------------------------------------------
suspend_connection(_OE_THIS, _OE_FROM, State) when ?is_Connected(State) ->
    if
	?is_Suspended(State) ->
	    corba:raise(#'CosNotifyChannelAdmin_ConnectionAlreadyInactive'{});
	true ->
	    stop_timer(State),
	    {reply, ok, ?set_Suspended(State)}
    end;
suspend_connection(_, _, _) ->
    corba:raise(#'CosNotifyChannelAdmin_NotConnected'{}).

%%----------------------------------------------------------%
%% function : resume_connection
%% Arguments: 
%% Returns  :  ok | {'EXCEPTION', #'ConnectionAlreadyActive'{}} |
%%            {'EXCEPTION', #'NotConneced'{}}
%%-----------------------------------------------------------
resume_connection(_OE_THIS, _OE_FROM, State) when ?is_Connected(State) ->
    if
	?is_NotSuspended(State) ->
	    corba:raise(#'CosNotifyChannelAdmin_ConnectionAlreadyActive'{});
	true ->
	    NewState = start_timer(State),
	    {reply, ok, ?set_NotSuspended(NewState)}
    end;
resume_connection(_, _, _) ->
    corba:raise(#'CosNotifyChannelAdmin_NotConnected'{}).

%%----- Inherit from CosNotifyChannelAdmin::ProxyConsumer ---
%%----------------------------------------------------------%
%% function : obtain_subscription_types
%% Arguments: Mode - enum 'ObtainInfoMode' (CosNotifyChannelAdmin)
%% Returns  : CosNotification::EventTypeSeq
%%-----------------------------------------------------------
obtain_subscription_types(_OE_THIS, _OE_FROM, State, 'ALL_NOW_UPDATES_OFF') ->
    {reply, ?get_AllPublish(State), ?set_PublishType(State, false)};
obtain_subscription_types(_OE_THIS, _OE_FROM, State, 'ALL_NOW_UPDATES_ON') ->
    {reply, ?get_AllPublish(State), ?set_PublishType(State, true)};
obtain_subscription_types(_OE_THIS, _OE_FROM, State, 'NONE_NOW_UPDATES_OFF') ->
    {reply, [], ?set_PublishType(State, false)};
obtain_subscription_types(_OE_THIS, _OE_FROM, State, 'NONE_NOW_UPDATES_ON') ->
    {reply, [], ?set_PublishType(State, true)};
obtain_subscription_types(_,_,_,What) ->
    orber:dbg("[~p] PullerConsumer:obtain_subscription_types(~p);~n"
	      "Incorrect enumerant", [?LINE, What], ?DEBUG_LEVEL),
    corba:raise(#'BAD_PARAM'{completion_status=?COMPLETED_NO}).

%%----------------------------------------------------------%
%% function : validate_event_qos
%% Arguments: RequiredQoS - CosNotification::QoSProperties
%% Returns  : ok | {'EXCEPTION', #'UnsupportedQoS'{}}
%%            AvilableQoS - CosNotification::NamedPropertyRangeSeq (out)
%%-----------------------------------------------------------
validate_event_qos(_OE_THIS, _OE_FROM, State, RequiredQoS) ->
    AvilableQoS = 'CosNotification_Common':validate_event_qos(RequiredQoS,
							      ?get_LocalQoS(State)),
    {reply, {ok, AvilableQoS}, 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), 
						     proxy, ?get_MyAdmin(State), 
						     false),
    NewState = ?update_EventDB(State, LQS),
    {reply, ok, ?set_BothQoS(NewState, 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}
%%            {ok, CosNotification::NamedPropertyRangeSeq}
%%-----------------------------------------------------------
validate_qos(_OE_THIS, _OE_FROM, State, Required_qos) ->
    QoS = 'CosNotification_Common':validate_qos(Required_qos, ?get_BothQoS(State), 
						proxy, ?get_MyAdmin(State), 
						false),
    {reply, {ok, QoS}, State}.

%%----- Inherit from CosNotifyComm::NotifyPublish -----------
%%----------------------------------------------------------%
%% function : offer_change
%% Arguments: Added - #'CosNotification_EventType'{}
%%            Removed - #'CosNotification_EventType'{}
%% Returns  : ok | 
%%            {'EXCEPTION', #'CosNotifyComm_InvalidEventType'{}}
%%-----------------------------------------------------------
offer_change(_OE_THIS, _OE_FROM, State, Added, Removed) ->
    cosNotification_Filter:validate_types(Added), 
    cosNotification_Filter:validate_types(Removed),
    %% On this "side" we don't really care about which
    %% type of events the client will supply. 
    %% Perhaps, later on, if we want to check this against Filters
    %% associated with this object we may change this approach, i.e., if
    %% the filter will not allow passing certain event types. But the
    %% user should see to that that situation never occurs. It would add
    %% extra overhead. Also see PusherSupplier- and PullerSuppler-
    %% 'subscription_change'.
    update_publish(add, State, Added),
    update_publish(remove, State, Removed),
    case ?get_PublishType(State) of
	true ->
	    %% Perhaps we should handle exception here?!
	    %% Probably not. Better to stay "on-line".
	    catch 'CosNotifyComm_NotifySubscribe':
		subscription_change(?get_Client(State), Added, Removed),
	    ok;
	_->
	    ok
    end,	
    {reply, ok, State}.

update_publish(_, _, [])->
    ok;
update_publish(add, State, [H|T]) ->