The R13B03 release.OTP_R13B03

1 files changed, 914 insertions, 0 deletions

diff --git a/lib/cosNotification/src/PullerSupplier_impl.erl b/lib/cosNotification/src/PullerSupplier_impl.erl
new file mode 100644
index 0000000000..9f12f9c742
--- /dev/null
+++ b/lib/cosNotification/src/PullerSupplier_impl.erl
@@ -0,0 +1,914 @@
+%%--------------------------------------------------------------------
+%%
+%% %CopyrightBegin%
+%% 
+%% Copyright Ericsson AB 1999-2009. All Rights Reserved.
+%% 
+%% 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 online 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.
+%% 
+%% %CopyrightEnd%
+%%
+%%
+%%----------------------------------------------------------------------
+%% File    : PullerSupplier_impl.erl
+%% Purpose : 
+%%----------------------------------------------------------------------
+
+-module('PullerSupplier_impl').
+
+%%--------------- INCLUDES -----------------------------------
+-include_lib("orber/include/corba.hrl").
+-include_lib("orber/include/ifr_types.hrl").
+%% cosEvent files.
+-include_lib("cosEvent/include/CosEventChannelAdmin.hrl").
+%% Application files
+-include("CosNotification.hrl").
+-include("CosNotifyChannelAdmin.hrl").
+-include("CosNotifyComm.hrl").
+-include("CosNotifyFilter.hrl").
+
+-include("CosNotification_Definitions.hrl").
+
+%%--------------- EXPORTS ------------------------------------
+%%--------------- External -----------------------------------
+%%----- CosNotifyChannelAdmin::ProxyPullSupplier -------------
+-export([connect_any_pull_consumer/4]).
+
+%%----- CosNotifyChannelAdmin::SequenceProxyPullSupplier -----
+-export([connect_sequence_pull_consumer/4]).
+
+%%----- CosNotifyChannelAdmin::StructuredProxyPullSupplier ---
+-export([connect_structured_pull_consumer/4]).
+
+%%----- 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::PullSupplier -------------
+-export([pull/3, 
+	 try_pull/3, 
+	 disconnect_pull_supplier/3]).
+
+%%----- Inherit from CosNotifyComm::SequencePullSupplier --
+-export([pull_structured_events/4, 
+	 try_pull_structured_events/4, 
+	 disconnect_sequence_pull_supplier/3]).
+
+%%----- Inherit from CosNotifyComm::StructuredPullSupplier --
+-export([pull_structured_event/3, 
+	 try_pull_structured_event/3, 
+	 disconnect_structured_pull_supplier/3]).
+
+%%----- Inherit from CosEventChannelAdmin::ProxyPullSupplier
+-export([connect_pull_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,
+		pacingTimer,
+		respondTo,
+		subscribeType = false,
+		subscribeData = true,
+		etsR,
+		eventDB}).
+
+%% Data structures constructors
+-define(get_InitState(_MyT, _MyA, _MyAP, _QS, _LQS, _Ch, _MyOp, _GT, _GL, _TR), 
+	#state{myType    = _MyT,
+	       myAdmin   = _MyA,
+	       myAdminPid= _MyAP,
+	       myChannel = _Ch,
+	       myOperator= _MyOp,
+	       qosGlobal = _QS,
+	       qosLocal  = _LQS,
+	       etsR      = ets:new(oe_ets, [set, protected]),
+	       eventDB   = cosNotification_eventDB:create_db(_LQS, _GT, _GL, _TR)}).
+
+
+%% Data structures selectors
+%% Attributes
+-define(get_MyType(S),           S#state.myType).
+-define(get_MyAdmin(S),          S#state.myAdmin).
+-define(get_MyAdminPid(S),       S#state.myAdmin).
+-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)).
+%% Event
+-define(get_Event(S),            cosNotification_eventDB:get_event(S#state.eventDB)).
+-define(get_Events(S,M),         cosNotification_eventDB:get_events(S#state.eventDB, M)).
+-define(get_RespondTo(S),        S#state.respondTo).
+%% 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).
+
+%% 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}).
+%% 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}).
+-define(update_EventDB(S,Q),     S#state{eventDB=
+					 cosNotification_eventDB:update(S#state.eventDB, Q)}).
+%% 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=[]}).
+-define(set_Unconnected(S),      S#state{client=undefined}).
+-define(reset_RespondTo(S),      S#state{respondTo=undefined}).
+-define(set_RespondTo(S,F),      S#state{respondTo=F}).
+%% Event
+-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)).
+%% Admin
+-define(set_PacingTimer(S,T),    S#state{pacingTimer=T}).
+%% 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)).
+
+%% MISC
+-define(is_ANY(S),               S#state.myType == 'PULL_ANY').
+-define(is_STRUCTURED(S),        S#state.myType == 'PULL_STRUCTURED').
+-define(is_SEQUENCE(S),          S#state.myType == 'PULL_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_Waiting(S),           S#state.respondTo =/= undefined).
+-define(is_SubscribedFor(S,K),   ets:lookup(S#state.etsR, K) =/= []).
+-define(is_BatchLimitReached(S,M), cosNotification_eventDB:
+	status(S#state.eventDB, {batchLimit, 
+				 ?not_GetMaximumBatchSize((S#state.qosLocal)), M})).
+
+%%----------------------------------------------------------%
+%% 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) ->
+    ?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, TS} when ?is_Waiting(State) ->
+	    case ?get_PacingTimer(State) of
+		{_, TS} ->
+		    ?DBG("PULL SUPPLIER PACING LIMIT REACHED~n",[]),
+		    {RespondTo, Max} = ?get_RespondTo(State),
+		    {EventSeq, _} = ?get_Events(State, Max),
+		    corba:reply(RespondTo, EventSeq),
+		    {noreply, ?reset_RespondTo(State)};
+		_ ->
+		    %% Must have been an old timer event, i.e., we reached the
+		    %% Batch Limit before Pace limit and we were not able
+		    %% to stop the timer before it triggered an event.
+		    ?DBG("PULL SUPPLIER OLD PACING LIMIT REACHED~n",[]),
+		    {noreply, State}
+	    end;
+	{pacing, _} ->
+	    ?DBG("PULL SUPPLIER PACING LIMIT REACHED BUT NO CLIENT; IMPOSSIBLE!!!~n",[]),
+	    {noreply, 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(gcLimit, Options, 
+						  ?not_DEFAULT_SETTINGS),
+    TimeRef = 'CosNotification_Common':get_option(timeService, Options, 
+						  ?not_DEFAULT_SETTINGS),
+    {ok, ?get_InitState(MyType, MyAdmin, MyAdminPid, InitQoS, LQS, MyChannel, 
+			Operator, GCTime, GCLimit, TimeRef)}.
+
+terminate(_Reason, State) when ?is_UnConnected(State) ->
+    ok;
+terminate(_Reason, State) ->
+    Client = ?get_Client(State),
+    case catch corba_object:is_nil(Client) of
+	false when ?is_ANY(State) ->
+	    'CosNotification_Common':disconnect('CosEventComm_PullConsumer', 
+						disconnect_pull_consumer,
+						Client);
+	false when ?is_SEQUENCE(State) ->
+	    'CosNotification_Common':disconnect('CosNotifyComm_SequencePullConsumer',
+						disconnect_sequence_pull_consumer,
+						Client);
+	false when ?is_STRUCTURED(State) ->
+	    'CosNotification_Common':disconnect('CosNotifyComm_StructuredPullConsumer',
+						disconnect_structured_pull_consumer,
+						Client);
+	_ ->
+	    ok
+    end.
+
+%%-----------------------------------------------------------
+%%----- 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) ->
+    {reply, ok, ?set_PrioFil(State, PrioF)}.
+
+
+%%----------------------------------------------------------%
+%% Attribute: '_*et_lifetime_filter'
+%% Type     : read/write
+%% Returns  : 
+%%-----------------------------------------------------------
+'_get_lifetime_filter'(_OE_THIS, _OE_FROM, State) ->
+    {reply, ?get_LifeTFil(State), State}.
+'_set_lifetime_filter'(_OE_THIS, _OE_FROM, State, LifeTF) ->
+    {reply, ok, ?set_LifeTFil(State, LifeTF)}.
+
+%%-----------------------------------------------------------
+%%------- Exported external functions -----------------------
+%%-----------------------------------------------------------
+%%----- CosEventChannelAdmin::ProxyPullSupplier -------------
+%%----------------------------------------------------------%
+%% function : connect_pull_consumer
+%% Arguments: Client - CosEventComm::PullConsumer
+%% Returns  :  ok | {'EXCEPTION', #'AlreadyConnected'{}} |
+%%            {'EXCEPTION', #'TypeError'{}} |
+%%            {'EXCEPTION', #'BAD_OPERATION'{}}
+%%            Both exceptions from CosEventChannelAdmin!!!
+%%-----------------------------------------------------------
+connect_pull_consumer(OE_THIS, OE_FROM, State, Client) ->
+    connect_any_pull_consumer(OE_THIS, OE_FROM, State, Client).
+
+%%----- CosNotifyChannelAdmin::ProxyPullSupplier ------------
+%%----------------------------------------------------------%
+%% function : connect_any_pull_consumer
+%% Arguments: Client - CosEventComm::PullConsumer
+%% Returns  :  ok | {'EXCEPTION', #'AlreadyConnected'{}} |
+%%            {'EXCEPTION', #'TypeError'{}} |
+%%            {'EXCEPTION', #'BAD_OPERATION'{}}
+%%            Both exceptions from CosEventChannelAdmin!!!
+%%-----------------------------------------------------------
+connect_any_pull_consumer(_OE_THIS, _OE_FROM, State, Client) when ?is_ANY(State) ->
+    ?not_TypeCheck(Client, 'CosEventComm_PullConsumer'),
+    if
+	?is_Connected(State) ->
+	    corba:raise(#'CosEventChannelAdmin_AlreadyConnected'{});
+	true ->
+	    {reply, ok, ?set_Client(State, Client)}
+    end;
+connect_any_pull_consumer(_, _, _, _) ->
+      corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).
+  
+%%----- CosNotifyChannelAdmin::SequenceProxyPullSupplier ----
+%%----------------------------------------------------------%
+%% function : connect_sequence_pull_consumer
+%% Arguments: Client - CosNotifyComm::SequencePullConsumer
+%% Returns  :  ok | {'EXCEPTION', #'AlreadyConnected'{}} |
+%%            {'EXCEPTION', #'TypeError'{}} |
+%%            {'EXCEPTION', #'BAD_OPERATION'{}}
+%%-----------------------------------------------------------
+connect_sequence_pull_consumer(_OE_THIS, _OE_FROM, State, Client) when ?is_SEQUENCE(State) ->
+    ?not_TypeCheck(Client, 'CosNotifyComm_SequencePullConsumer'),
+    if
+	?is_Connected(State) ->
+	    corba:raise(#'CosEventChannelAdmin_AlreadyConnected'{});
+	true ->
+	    {reply, ok, ?set_Client(State, Client)}
+    end;
+connect_sequence_pull_consumer(_, _, _, _) ->
+    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).
+
+%%----- CosNotifyChannelAdmin::StructuredProxyPullSupplier --
+%%----------------------------------------------------------%
+%% function : connect_structured_pull_consumer
+%% Arguments: Client - CosNotifyComm::StructuredPullConsumer
+%% Returns  :  ok | {'EXCEPTION', #'AlreadyConnected'{}} |
+%%            {'EXCEPTION', #'TypeError'{}} |
+%%            {'EXCEPTION', #'BAD_OPERATION'{}}
+%%-----------------------------------------------------------
+connect_structured_pull_consumer(_OE_THIS, _OE_FROM, State, Client) when ?is_STRUCTURED(State) ->
+    ?not_TypeCheck(Client, 'CosNotifyComm_StructuredPullConsumer'),
+    if
+	?is_Connected(State) ->
+	    corba:raise(#'CosEventChannelAdmin_AlreadyConnected'{});
+	true ->
+	    {reply, ok, ?set_Client(State, Client)}
+    end;
+connect_structured_pull_consumer(_, _, _, _) ->
+    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).
+
+%%----- Inherit from CosNotifyChannelAdmin::ProxySupplier ---
+%%----------------------------------------------------------%
+%% function : obtain_offered_types
+%% Arguments: Mode - enum 'ObtainInfoMode' (CosNotifyChannelAdmin)
+%% Returns  : CosNotification::EventTypeSeq
+%%-----------------------------------------------------------
+obtain_offered_types(_OE_THIS, _OE_FROM, State, 'ALL_NOW_UPDATES_OFF') ->
+    {reply, ?get_AllSubscribe(State), ?set_SubscribeType(State, false)};
+obtain_offered_types(_OE_THIS, _OE_FROM, State, 'ALL_NOW_UPDATES_ON') ->
+    {reply, ?get_AllSubscribe(State), ?set_SubscribeType(State, true)};
+obtain_offered_types(_OE_THIS, _OE_FROM, State, 'NONE_NOW_UPDATES_OFF') ->
+    {reply, [], ?set_SubscribeType(State, false)};
+obtain_offered_types(_OE_THIS, _OE_FROM, State, 'NONE_NOW_UPDATES_ON') ->
+    {reply, [], ?set_SubscribeType(State, true)};
+obtain_offered_types(_,_,_,What) ->
+    orber:dbg("[~p] PullerSupplier:obtain_offered_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::NotifySubscribe ---------
+%%----------------------------------------------------------%
+%% function : subscription_change
+%% Arguments: Added - Removed - CosNotification::EventTypeSeq
+%% Returns  : ok
+%%-----------------------------------------------------------
+subscription_change(_OE_THIS, _OE_FROM, State, Added, Removed) ->
+    cosNotification_Filter:validate_types(Added), 
+    cosNotification_Filter:validate_types(Removed),
+    %% On this "side", we care about which type of events the client 
+    %% will require, since the client (or an agent) clearly stated
+    %% that it's only interested in these types of events.
+    %% Also see PusherConsumer- and PullerConsumer-'offer_change'.
+    update_subscribe(remove, State, Removed),
+    CurrentSub = ?get_AllSubscribe(State),
+    NewState = 
+	case cosNotification_Filter:check_types(Added++CurrentSub) of
+	    true ->
+		%% Types supplied does in some way cause all events to be valid.
+		%% Smart? Would have been better to not supply any at all.
+		?set_SubscribeData(State, true);
+	    {ok, Which, WC} ->
+		?set_SubscribeData(State, {Which, WC})
+    end,
+    update_subscribe(add, NewState, Added),
+    case ?get_SubscribeType(NewState) of
+	true ->
+	    %% Perhaps we should handle exception here?!
+	    %% Probably not. Better to stay "on-line".
+	    catch 'CosNotifyComm_NotifyPublish':
+		offer_change(?get_Client(NewState), Added, Removed),
+	    ok;
+	_->
+	    ok
+    end,	
+    {reply, ok, NewState}.
+
+update_subscribe(_, _, [])->
+    ok;
+update_subscribe(add, State, [H|T]) ->
+    ?add_Subscribe(State, H),
+    update_subscribe(add, State, T);
+update_subscribe(remove, State, [H|T]) ->
+    ?del_Subscribe(State, H),
+    update_subscribe(remove, State, T).
+
+%%----- 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 is_integer(FilterID) ->
+    {reply, ok, ?del_Filter(State, FilterID)};
+remove_filter(_,_,_,What) ->
+    orber:dbg("[~p] PullerSupplier:remove_filter(~p); Not an integer", 
+	      [?LINE, What], ?DEBUG_LEVEL),
+    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 is_integer(FilterID) ->
+    {reply, ?get_Filter(State, FilterID), State};
+get_filter(_,_,_,What) ->
+    orber:dbg("[~p] PullerSupplier:get_filter(~p); Not an integer", 
+	      [?LINE, What], ?DEBUG_LEVEL),
+    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 CosEventComm::PullSupplier -------------
+%%----------------------------------------------------------%
+%% function : disconnect_pull_supplier
+%% Arguments: -
+%% Returns  : ok
+%%-----------------------------------------------------------
+disconnect_pull_supplier(_OE_THIS, _OE_FROM, State) ->
+    {stop, normal, ok, ?set_Unconnected(State)}.
+
+%%----------------------------------------------------------%
+%% function : pull
+%% Arguments: -
+%% Returns  : any - CORBA::ANY
+%%-----------------------------------------------------------
+pull(_OE_THIS, OE_FROM, State) when ?is_ANY(State) ->
+    case ?get_Event(State) of
+	{[], _} ->
+	    {noreply, ?set_RespondTo(State, OE_FROM)};
+	{Event,_} ->
+	    {reply, Event, State}
+    end;
+pull(_,_,_) ->
+    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).
+
+%%----------------------------------------------------------%
+%% function : try_pull
+%% Arguments: -
+%% Returns  : any - CORBA::ANY
+%%            HasEvent - boolean (out-type)
+%%-----------------------------------------------------------
+try_pull(_OE_THIS, _OE_FROM, State) when ?is_ANY(State) ->
+    case ?get_Event(State) of
+	{[], _} ->
+	    {reply, {any:create(orber_tc:null(), null), false}, State};
+	{Event, Bool} ->
+	    {reply, {Event, Bool}, State}
+    end;
+try_pull(_,_,_) ->
+    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).
+
+%%----- Inherit from CosNotifyComm::SequencePullSupplier ----
+%%----------------------------------------------------------%
+%% function : disconnect_sequence_pull_supplier
+%% Arguments: -
+%% Returns  : ok
+%%-----------------------------------------------------------
+disconnect_sequence_pull_supplier(_OE_THIS, _OE_FROM, State) ->
+    {stop, normal, ok, ?set_Unconnected(State)}.
+
+%%----------------------------------------------------------%
+%% function : pull_structured_events
+%% Arguments: Max - long()
+%% Returns  : [StructuredEvent, ..]
+%%-----------------------------------------------------------
+pull_structured_events(_OE_THIS, OE_FROM, State, Max) when ?is_SEQUENCE(State) ->
+    case ?is_BatchLimitReached(State, Max) of
+	true ->
+	    %% This test is not fool-proof; if Events have been stored
+	    %% using StartTime they will still be there but we cannot
+	    %% deliver them anyway. To solve this "problem" would cost!
+	    %% Hence, since it works fine otherwise it will do.
+	    case ?get_Events(State, Max) of
+		{[], false} ->
+		    NewState = start_timer(State),
+		    {noreply, ?set_RespondTo(NewState, {OE_FROM, Max})};
+		{Event,_} ->
+		    {reply, Event, State}
+	    end;
+	_->
+	    NewState = start_timer(State),
+	    {noreply, ?set_RespondTo(NewState, {OE_FROM, Max})}
+    end;
+pull_structured_events(_,_,_,_) ->
+    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).
+
+%%----------------------------------------------------------%
+%% function : try_pull_structured_events
+%% Arguments: Max - long()
+%% Returns  : [StructuredEvent, ..]
+%%            HasEvent - Boolean()
+%%-----------------------------------------------------------
+try_pull_structured_events(_OE_THIS, _OE_FROM, State, Max) when ?is_SEQUENCE(State) ->
+    {reply, ?get_Events(State, Max), State};
+try_pull_structured_events(_,_,_,_) ->
+    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).
+
+%%----- Inherit from CosNotifyComm::StructuredPullSupplier --
+%%----------------------------------------------------------%
+%% function : disconnect_structured_pull_supplier
+%% Arguments: -
+%% Returns  : ok
+%%-----------------------------------------------------------
+disconnect_structured_pull_supplier(_OE_THIS, _OE_FROM, State) ->
+    {stop, normal, ok, ?set_Unconnected(State)}.
+
+%%----------------------------------------------------------%
+%% function : pull_structured_event
+%% Arguments: -
+%% Returns  : 
+%%-----------------------------------------------------------
+pull_structured_event(_OE_THIS, OE_FROM, State) when ?is_STRUCTURED(State) ->
+    case ?get_Event(State) of
+	{[], _} ->
+	    {noreply, ?set_RespondTo(State, OE_FROM)};
+	{Event,_} ->
+	    {reply, Event, State}
+    end;
+pull_structured_event(_,_,_) ->
+    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).
+
+%%----------------------------------------------------------%
+%% function : try_pull_structured_event
+%% Arguments: -
+%% Returns  : 
+%%-----------------------------------------------------------
+try_pull_structured_event(_OE_THIS, _OE_FROM, State) when ?is_STRUCTURED(State) ->
+    case ?get_Event(State) of
+	{[], _} ->
+	    {reply, 
+	     {?not_CreateSE("","","",[],[],any:create(orber_tc:null(), null)), false}, 
+	     State};
+	{Event, Bool} ->
+	    {reply, {Event, Bool}, State}
+    end;
+try_pull_structured_event(_,_,_) ->
+    corba:raise(#'BAD_OPERATION'{completion_status=?COMPLETED_NO}).
+
+
+%%--------------- LOCAL FUNCTIONS ----------------------------
+find_obj({value, {_, Obj}}) -> Obj;
+find_obj(_) -> {'EXCEPTION', #'CosNotifyFilter_FilterNotFound'{}}.
+
+find_ids(List) ->           find_ids(List, []).
+find_ids([], Acc) ->        Acc;
+find_ids([{I,_}|T], Acc) -> find_ids(T, [I|Acc]);
+find_ids(_, _) -> corba:raise(#'INTERNAL'{completion_status=?COMPLETED_NO}).
+
+%% Delete a single object.
+%% The list do not differ, i.e., no filter removed, raise exception.
+delete_obj(List,List) -> corba:raise(#'CosNotifyFilter_FilterNotFound'{});
+delete_obj(List,_) -> List.
+
+
+%%-----------------------------------------------------------
+%% function : callSeq
+%% Arguments: 
+%% Returns  : 
+%%-----------------------------------------------------------
+callSeq(_OE_THIS, OE_FROM, State, EventsIn, Status) ->
+    %% We should do something here, i.e., see what QoS this Object offers and
+    %% act accordingly.
+    corba:reply(OE_FROM, ok),
+    case cosNotification_eventDB:validate_event(?get_SubscribeData(State), EventsIn,
+						?get_AllFilter(State),
+						?get_SubscribeDB(State),
+						Status) of
+	{[], _} ->
+	    ?DBG("PROXY NOSUBSCRIPTION SEQUENCE/STRUCTURED: ~p~n",[EventsIn]),
+	    {noreply, State};
+	%% Just one event and we got a client waiting => there is no need to store
+	%% the event, just transform it and pass it on.
+	{[Event], _} when ?is_ANY(State), ?is_Waiting(State) ->
+	    ?DBG("PROXY RECEIVED SEQUENCE[1]==>ANY: ~p~n",[Event]),
+	    AnyEvent = any:create('CosNotification_StructuredEvent':tc(),Event),
+	    case ?addAndGet_Event(State, AnyEvent) of
+		{[], _} ->
+		    ?DBG("PROXY RECEIVED UNDELIVERABLE SEQUENCE[1]: ~p~n",
+				 [Event]),
+		    %% Cannot deliver the event at the moment; perhaps Starttime
+		    %% set or Deadline passed.
+		    {noreply, State};
+		{PossiblyOtherEvent, _} ->
+		    ?DBG("PROXY RECEIVED SEQUENCE[1] ~p; DELIVER: ~p~n",
+				 [Event, PossiblyOtherEvent]),
+		    corba:reply(?get_RespondTo(State), PossiblyOtherEvent),
+		    {noreply, ?reset_RespondTo(State)}
+	    end;
+	{[Event],_} when ?is_STRUCTURED(State), ?is_Waiting(State) ->
+	    case ?addAndGet_Event(State, Event) of
+		{[], _} ->
+		    ?DBG("PROXY RECEIVED UNDELIVERABLE SEQUENCE[1]: ~p~n",
+				 [Event]),
+		    %% Cannot deliver the event at the moment; perhaps Starttime
+		    %% set or Deadline passed.
+		    {noreply, State};
+		{PossiblyOtherEvent, _} ->
+		    ?DBG("PROXY RECEIVED SEQUENCE[1] ~p; DELIVER: ~p~n",
+				 [Event, PossiblyOtherEvent]),
+		    corba:reply(?get_RespondTo(State), PossiblyOtherEvent),
+		    {noreply, ?reset_RespondTo(State)}
+	    end;
+	%% A sequence of events => store them and extract the first (according to QoS)
+	%% event and forward it.
+	{Events,_} when ?is_ANY(State), ?is_Waiting(State) ->
+	    ?DBG("PROXY RECEIVED SEQUENCE==>ANY: ~p~n",[Events]),
+	    store_events(State, Events),
+	    case ?get_Event(State) of
+		{[], _} ->
+		    {noreply, State};
+		{AnyEvent, _} ->
+		    corba:reply(?get_RespondTo(State), AnyEvent),
+		    {noreply, ?reset_RespondTo(State)}
+	    end;
+	{Events, _} when ?is_STRUCTURED(State), ?is_Waiting(State) ->
+	    ?DBG("PROXY RECEIVED SEQUENCE: ~p~n",[Events]),
+	    store_events(State, Events),
+	    case ?get_Event(State) of
+		{[], _} ->
+		    {noreply, State};
+		{_StrEvent, _} ->
+		    corba:reply(?get_RespondTo(State), Events),
+		    {noreply, ?reset_RespondTo(State)}
+	    end;
+	{Events, _} when ?is_SEQUENCE(State), ?is_Waiting(State) ->
+	    ?DBG("PROXY RECEIVED SEQUENCE: ~p~n",[Events]),
+	    %% Store them first and extract Max events in QoS order.
+	    store_events(State, Events),
+	    {RespondTo, Max} = ?get_RespondTo(State),
+	    case ?is_BatchLimitReached(State, Max) of
+		true ->
+		    {EventSeq, _} = ?get_Events(State, Max),
+		    corba:reply(RespondTo, EventSeq),
+		    stop_timer(State),
+		    {noreply, ?reset_RespondTo(State)};
+		_->
+		   {noreply, State}
+	    end;
+	%% No client waiting. Store the event(s).
+	{Events, _} ->
+	    ?DBG("PROXY RECEIVED SEQUENCE: ~p~n",[Events]),
+	    store_events(State, Events),
+	    {noreply, State}
+    end.
+
+store_events(_State, []) ->
+    ok;
+store_events(State, [Event|Rest]) when ?is_ANY(State) ->
+    AnyEvent = any:create('CosNotification_StructuredEvent':tc(),Event),
+    ?add_Event(State,AnyEvent),
+    store_events(State, Rest);
+store_events(State, [Event|Rest]) ->
+    ?add_Event(State,Event),
+    store_events(State, Rest).
+
+%%-----------------------------------------------------------
+%% function : callAny
+%% Arguments: 
+%% Returns  : 
+%%-----------------------------------------------------------
+callAny(_OE_THIS, OE_FROM, State, EventIn, Status) ->
+    corba:reply(OE_FROM, ok),
+    case cosNotification_eventDB:validate_event(?get_SubscribeData(State), EventIn,
+						?get_AllFilter(State),
+						?get_SubscribeDB(State),
+						Status) of
+	{[],_} ->
+	    ?DBG("PROXY NOSUBSCRIPTION ANY: ~p~n",[EventIn]),
+	    {noreply, State};
+	{Event,_} when ?is_ANY(State), ?is_Waiting(State) ->
+	    ?DBG("PROXY RECEIVED ANY: ~p~n",[Event]),
+	    case ?addAndGet_Event(State, Event) of
+		{[],_} ->
+		    %% Unable to deliver the event (Starttime, Deadline etc).
+		    {noreply, State};
+		{MaybeOtherEvent , _} ->
+		    corba:reply(?get_RespondTo(State), MaybeOtherEvent),
+		    {noreply, ?reset_RespondTo(State)}
+	    end;
+	{Event,_} when ?is_ANY(State) ->
+	    ?DBG("PROXY RECEIVED ANY: ~p~n",[Event]),
+	    ?add_Event(State,Event),
+	    {noreply, State};
+	{Event,_} when ?is_STRUCTURED(State), ?is_Waiting(State) ->
+	    ?DBG("PROXY RECEIVED ANY==>STRUCTURED: ~p~n",[Event]),
+	    case ?addAndGet_Event(State, ?not_CreateSE("","%ANY","",[],[],Event)) of
+		{[],_} ->
+		    %% Unable to deliver the event (Starttime, Deadline etc).
+		    {noreply, State};
+		{MaybeOtherEvent , _} ->
+		    corba:reply(?get_RespondTo(State), MaybeOtherEvent),
+		    {noreply, ?reset_RespondTo(State)}
+	    end;
+	{Event,_} when ?is_SEQUENCE(State), ?is_Waiting(State) ->
+	    ?DBG("PROXY RECEIVED ANY==>SEQUENCE[1]: ~p~n",[Event]),
+	    ?add_Event(State,?not_CreateSE("","%ANY","",[],[],Event)),
+	    {RespondTo, Max} = ?get_RespondTo(State),
+	    case ?is_BatchLimitReached(State, Max) of
+		true ->
+		    {EventSeq, _} = ?get_Events(State, Max),
+		    corba:reply(RespondTo, EventSeq),
+		    stop_timer(State),
+		    {noreply, ?reset_RespondTo(State)};
+		_ ->
+		    {noreply, State}
+	    end;
+	{Event,_} ->
+	    ?DBG("PROXY RECEIVED ANY==>STRUCTURED/SEQUENCE: ~p~n",[Event]),
+	    ?add_Event(State,?not_CreateSE("","%ANY","",[],[],Event)),
+	    {noreply, State}
+    end.
+
+
+	    
+%% Start timers which send a message each time we should push events. Only used
+%% when this objects is defined to supply sequences.
+start_timer(State) ->
+    TS = now(),
+    case catch timer:send_after(timer:seconds(?get_PacingInterval(State)), 
+				{pacing, TS}) of
+	{ok,PacTRef} ->
+	    ?DBG("PULL SUPPLIER STARTED TIMER, BATCH LIMIT: ~p~n",
+			 [?get_BatchLimit(State)]),
+	    ?set_PacingTimer(State, {PacTRef, TS});
+	What ->
+	    orber:dbg("[~p] PullerSupplier:start_timer();~n"
+		      "Unable to invoke timer:send_interval/2: ~p", 
+		      [?LINE, What], ?DEBUG_LEVEL),
+	    corba:raise(#'INTERNAL'{completion_status=?COMPLETED_NO})
+    end.
+
+stop_timer(State) ->
+    case ?get_PacingTimer(State) of
+	undefined ->
+	    ok;
+	{Timer, _} ->
+	    ?DBG("PULL SUPPLIER STOPPED TIMER~n",[]),
+	    timer:cancel(Timer)
+    end.
+
+%%--------------- MISC FUNCTIONS, E.G. DEBUGGING -------------
+%%--------------- END OF MODULE ------------------------------