// -*- Mode: C++; -*-
//                              File      : CosEventProxy.cc
//                              Package   : omniNotify-Library
//                              Created on: 1-Jan-1998
//                              Authors   : gruber&panagos
//
//    Copyright (C) 1998-2001 AT&T Laboratories -- Research
//
//    This file is part of the omniNotify library
//    and is distributed with the omniNotify release.
//
//    The omniNotify library is free software; you can redistribute it and/or
//    modify it under the terms of the GNU Library General Public
//    License as published by the Free Software Foundation; either
//    version 2 of the License, or (at your option) any later version.
//
//    This library is distributed in the hope that it will be useful,
//    but WITHOUT ANY WARRANTY; without even the implied warranty of
//    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
//    Library General Public License for more details.
//
//    You should have received a copy of the GNU Library General Public
//    License along with this library; if not, write to the Free
//    Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
//    02111-1307, USA
//
//
// Description:
//    Implementation of CORBA Event Service Proxies
//
 
/*
$Log: CosEventProxy.cc,v $
Revision 1.27  2001/10/06 00:44:35  alcfp
moved some EventChannel_i state to private

Revision 1.26  2001/09/19 21:10:03  alcfp
Added cleanup support to interactive api

Revision 1.25  2001/08/03 17:54:16  alcfp
added support for AttNotification

Revision 1.24  2001/06/26 20:01:15  alcfp
updated copyright notices, added support for omniORB4, switched default to POA

Revision 1.23  2001/06/22 07:00:31  alcfp
moved to new logging scheme

Revision 1.22  2001/06/12 17:43:13  alcfp
switch to RDITimeWrappers

Revision 1.21  2001/05/07 16:15:08  alcfp
+ Added support for OrderPolicy and DiscardPolicy
+ Took some initial steps towards new time value handling approach
+ Took some initial steps towards a new logging approach
+ Factored out common code in proxy add_event handling

Revision 1.20  2000/11/19 19:02:30  alcfp
removed unnecessary debug output

Revision 1.19  2000/11/16 05:32:14  alcfp
fixes for aCC

Revision 1.18  2000/11/15 21:17:29  alcfp
large number of changes to switch to use of RDIOplocks for safe object disposal support.  also reduced code duplication a little, and tried hard to make all the proxy code consistent

Revision 1.17  2000/11/05 04:48:09  alcfp
changed in defaults, env variable overrride, try_pull variants

Revision 1.16  2000/10/30 04:39:28  alcfp
extensive changes in preparation for 1.1 release.  will add notes about changes to update.log

Revision 1.15  2000/10/01 13:39:52  alcfp
Removed sleep() calls used for synchronization with unbound threads. Counters and flags are used instead

Revision 1.14.2.1  2000/09/30 15:39:51  alcfp
Removed sleep() calls used for synchronization with unbound threads. Counters and flags are used instead

Revision 1.14  2000/08/22 18:23:53  alcfp
added description to each file

Revision 1.13  2000/08/16 20:19:25  alcfp
Added licensing notice to each .h and .cc file where library files get GLPL notice and daemon file gets GPL notice -- examples do not claim any license but point out that the library and daemon code does have a license notice

*/
 
#include "CosNotifyShorthands.h"
#include "CosNfyUtils.h"
#include "RDI.h"
#include "RDIStringDefs.h"
#include "RDIOplocksMacros.h"
#include "CosEventChannelAdmin_i.h"
#include "CosNotifyChannelAdmin_i.h"

// ------------------------------------------------------------ //
// EventProxyPushConsumer_i                                     //
// ------------------------------------------------------------ //

EventProxyPushConsumer_i::EventProxyPushConsumer_i(SupplierAdmin_i* admin, 
					 	   EventChannel_i*  evchn,
						   const CosNA::ProxyID&  prxid) :
  _oplockptr(0), _my_name(admin->L_my_name()),
  _channel(evchn), _myadmin(admin), 
  _nevents(0), _pxstate(RDI_NotConnected), _pserial(prxid)
{
  RDI_OPLOCK_INIT;
  char buf[20];
  sprintf(buf, "proxy%ld", prxid);
  _my_name.length(_my_name.length()+1);
  _my_name[_my_name.length()-1] = (const char*)buf;
  _supplier = CosEvC::PushSupplier::_nil();
  WRAPPED_REGISTER_IMPL2(this, &_my_name);
}

EventProxyPushConsumer_i::~EventProxyPushConsumer_i() {
  RDI_OPLOCKS_DESTROY_CHECK("EventProxyPushConsumer_i");
}

void EventProxyPushConsumer_i::connect_push_supplier
(CosEvC::PushSupplier_ptr supplier WRAPPED_IMPLARG)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( CORBA::is_nil(supplier) ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::BAD_PARAM(0, CORBA::COMPLETED_NO);
  }
  if ( _pxstate != RDI_NotConnected ) {
    RDI_OPLOCK_RELEASE;
    throw CosEvCA::AlreadyConnected();
  }
  _pxstate  = RDI_Connected;
  _supplier = CosEvC::PushSupplier::_duplicate(supplier);
  RDI_OPLOCK_RELEASE;
}

void EventProxyPushConsumer_i::push(const CORBA::Any& data WRAPPED_IMPLARG)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( _pxstate != RDI_Connected ) {
    RDI_OPLOCK_RELEASE;
    throw CosEvC::Disconnected();
  }
  if ( _channel->new_any_event(data) ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::IMP_LIMIT(0, CORBA::COMPLETED_NO);
  }
  _nevents += 1;
  RDI_OPLOCK_RELEASE;
}

void EventProxyPushConsumer_i::disconnect_push_consumer(WRAPPED_IMPLARG_VOID)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  _disconnect_client_and_dispose(1);
}

void EventProxyPushConsumer_i::disconnect_client_and_dispose
(CORBA::Boolean remove_proxy_from_admin) {
  RDI_OPLOCK_ACQUIRE(return);
  _disconnect_client_and_dispose(remove_proxy_from_admin);
}

// *** lock must be held
// Effect: initiate impl dispose + release lock
void EventProxyPushConsumer_i::_disconnect_client_and_dispose
(CORBA::Boolean remove_proxy_from_admin)
{
  if (_pxstate == RDI_Disconnected) {
    RDIDbgCosCPxyLog("EventProxyPushConsumer_i::disconnect_client_and_dispose called twice on same proxy!\n");
    RDI_OPLOCK_RELEASE;
    return;
  }
  _pxstate = RDI_Disconnected; // acts as guard -- only one thread does the following
  // This proxy does not need to worry about inuse count
  // because it is never bumped above zero.
  if (remove_proxy_from_admin) {
    // do not hold OPLOCK across upcalls involving this
    RDI_OPLOCK_BUMP_RELEASE;
    _myadmin->remove_proxy(this);
    RDI_OPLOCK_REACQUIRE_DEBUMP(RDI_METHOD_RETURN_NULL(RDIDbgCosCPxy, RDIDbgCosCPxy_nm, "EventProxyPushConsumer_i::_disconnect_client_and_dispose [**unexpected REACQUIRE failure**]\n"));
  }
  _supplier = CosEvC::PushSupplier::_nil();
  RDI_OPLOCK_DISPOSE_IMPL_AND_RELEASE;
  _oplockptr = 0; // speeds up acquire check
}

// ------------------------------------------------------------ //
// EventProxyPullConsumer_i                                     //
// ------------------------------------------------------------ //

/** We use unbound threads here for we assume that in the majority
  * of configuations a thread pool will be used for pulling events
  * from suppliers and pushing events to consumers.
  */

class EventPullWorker : public omni_thread {
public:
  typedef void (EventProxyPullConsumer_i::*Method)(void);
  EventPullWorker(EventProxyPullConsumer_i* p, Method m) :
    omni_thread(NULL,PRIORITY_NORMAL), _proxy(p), _method(m) {;}
  void run(void *)      { (_proxy->*_method)(); }
private:
  EventProxyPullConsumer_i* _proxy;
  Method                    _method;
  EventPullWorker()  {;}
};

// ------------------------------------------------------------ //

EventProxyPullConsumer_i::EventProxyPullConsumer_i(SupplierAdmin_i* admin, 
                                                   EventChannel_i*  evchn,
						   const CosNA::ProxyID&  prxid) :
  _oplockptr(0), _my_name(admin->L_my_name()),
  _channel(evchn), _myadmin(admin), 
  _pworker(0), _thrdone(0), _nevents(0), 
  _pxstate(RDI_NotConnected), _pserial(prxid),_timeout_s(0), _timeout_n(0)
{
  RDI_OPLOCK_INIT;
  char buf[20];
  sprintf(buf, "proxy%ld", prxid);
  _my_name.length(_my_name.length()+1);
  _my_name[_my_name.length()-1] = (const char*)buf;
  _supplier = CosEvC::PullSupplier::_nil();
  // When the number of pull threads allocated at the channel level is
  // 0, each proxy uses its own thread to pull events from its supplier
  if ( _channel->pull_threads() == 0 ) {
    _pworker = new EventPullWorker(this, &EventProxyPullConsumer_i::_pull);
    RDI_AssertAllocThrowNo(_pworker, "Memory allocation failed -- omni_thread\n");
    _pworker->start();
    _thrdone = 0;
    RDIDbgCosCPxyLog("Pull thread for proxy " << (void*)this << " -- " << _pworker->id() << '\n');
  }
  WRAPPED_REGISTER_IMPL2(this, &_my_name);
}

EventProxyPullConsumer_i::~EventProxyPullConsumer_i() {
  RDI_OPLOCKS_DESTROY_CHECK("EventProxyPullConsumer_i");
}

void EventProxyPullConsumer_i::connect_pull_supplier
(CosEvC::PullSupplier_ptr supplier WRAPPED_IMPLARG)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( CORBA::is_nil(supplier) ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::BAD_PARAM(0, CORBA::COMPLETED_NO);
  }
  if ( _pxstate != RDI_NotConnected ) {
    RDI_OPLOCK_RELEASE;
    throw CosEvCA::AlreadyConnected();
  }
  _pxstate  = RDI_Connected;
  _supplier = CosEvC::PullSupplier::_duplicate(supplier);
  if ( _pworker ) {
    RDI_OPLOCK_SIGNAL;
  }
  RDI_PullSupplier* cps = _channel->pull_supplier();
  if ( cps ) {
    cps->signal_pull_threads();
  }
  RDI_OPLOCK_RELEASE;
}

// is_available determines whether a pull should occur.
// in this case the requirement is that either the pull interval is zero
// or an interval timeout has occurred
CORBA::Boolean EventProxyPullConsumer_i::is_available(unsigned long* wait_s, unsigned long* wait_n) {
  RDI_OPLOCK_ACQUIRE(return 0);
  if ( _pxstate != RDI_Connected ) {
    RDI_OPLOCK_RELEASE;
    return 0;
  }

  unsigned long pull_interval_s, pull_interval_n;
  unsigned long time_s, time_n;

  _channel->pull_period_s_n(pull_interval_s, pull_interval_n);
  if ( (pull_interval_s == 0) && (pull_interval_n == 0) ) {
    _timeout_s = 0; _timeout_n = 0;
    RDI_OPLOCK_RELEASE;
    return 1; // pull as fast as possible
  }

  omni_thread::get_time(&time_s, &time_n);

  if ( (_timeout_s == 0) && (_timeout_n == 0) ) {
    // proxy has not established timeout yet : set it now
    omni_thread::get_time(&_timeout_s, &_timeout_n, pull_interval_s, pull_interval_n);
  }
  if ( (time_s > _timeout_s) || ((time_s == _timeout_s) && (time_n > _timeout_n)) ) {
    RDI_OPLOCK_RELEASE;
    return 1; // interval timeout occurred for this proxy
  }
  // update wait_s/_n to reflect the future timeout point of this proxy
  if ( ( ((*wait_s) == 0) && ((*wait_n) == 0)                   ) ||
       ( ((*wait_s) > _timeout_s)                               ) ||
       ( ((*wait_s) == _timeout_s) && ((*wait_n) > _timeout_n)) ) {
    (*wait_s) = _timeout_s;
    (*wait_n) = _timeout_n;
  }
  RDI_OPLOCK_RELEASE;
  return 0;
}

// REQUIREMENT: A thread from pull pool only calls this method
// if is_available() is true and it thinks state is connected
// (For safety we verify that these conditions actually hold.)
void EventProxyPullConsumer_i::pull_event(CORBA::Boolean& invalid)
{
  RDI_OPLOCK_ACQUIRE_BUMP(return); // debump on method exit
  if ( _pxstate != RDI_Connected ) {
    RDI_OPLOCK_DEBUMP_RELEASE;
    return;
  }

  unsigned long pull_interval_s, pull_interval_n;
  unsigned long time_s, time_n;
  _channel->pull_period_s_n(pull_interval_s, pull_interval_n);

  CORBA::Any*     event = 0;
  CORBA::Boolean  hasev = 0;
  invalid = 0;

  // sanity check: if pull interval is positive, we should only be called if a timeout occurred
  if (pull_interval_s || pull_interval_n) {
    omni_thread::get_time(&time_s, &time_n);
    if ((_timeout_s == 0) && (_timeout_n == 0)) { // timeout not set; set it now
      omni_thread::get_time(&_timeout_s, &_timeout_n, pull_interval_s, pull_interval_n);
    } 
    if ( (time_s < _timeout_s) || ((time_s == _timeout_s) && (time_n < _timeout_n)) ) {
      RDIDbgCosCPxyLog("** INTERNAL ERROR: ProxyPullConsumer_i::pull_event called too soon\n");
      RDI_OPLOCK_DEBUMP_RELEASE;
      return;
    }
  }

  // update timeout before releasing OPLOCK -- this means we do the update
  // before doing the try_pull (seems OK)
  if (pull_interval_s || pull_interval_n) {
    omni_thread::get_time(&_timeout_s, &_timeout_n, pull_interval_s, pull_interval_n);
  } else {
    _timeout_s = 0; _timeout_n = 0;
  }

  // do not hold OPLOCK across try_pull call
  RDI_OPLOCK_RELEASE;

  CORBA::Boolean outcall_worked = 0;
  try {
    event = _supplier->try_pull(hasev);
    outcall_worked = 1;
  } catch ( CORBA::INV_OBJREF& e ) {
    RDIDbgCosCPxyLog("Pull pool calling pull_event for proxy " << (void*)this << " - Invalid object reference\n");
  } catch ( CORBA::OBJECT_NOT_EXIST& e ) {
    RDIDbgCosCPxyLog("Pull pool calling pull_event for proxy " << (void*)this << " - Consumer object does not exist\n");
  } catch ( CORBA::COMM_FAILURE& e ) {
    RDIDbgCosCPxyLog("Pull pool calling pull_event for proxy " << (void*)this << " - Communication Failure\n");
  } catch (...) {
    RDIDbgCosCPxyLog("Pull pool calling pull_event for proxy " << (void*)this << " - Exception while pushing event to consumer\n");
  }

  RDI_OPLOCK_REACQUIRE(return);

  // add new event to channel even if we have been disconnected in the meantime
  if (outcall_worked && hasev && event ) {
    _nevents += 1;
    _channel->new_any_event(*event);
  }
  // new_any_event copies event, so we delete regardless
  if ( event ) {
    delete event; event = 0;
  }

  if (_pxstate != RDI_Connected) {
    // disconnected/disposed during try_pull -- do nothing else with my state
  } else {
    if (!outcall_worked) { // exception
      _pxstate = RDI_Exception;
      invalid  = 1;
    }
  }

  RDI_OPLOCK_DEBUMP_RELEASE;
}

void EventProxyPullConsumer_i::_pull()
{
  RDI_OPLOCK_ACQUIRE_BUMP(return); // debump on thread exit

  CORBA::Any*    event = 0;
  CORBA::Boolean hasev = 0, update_timeout = 0;
  CORBA::Boolean do_yield = 0;
  unsigned long pull_interval_s, pull_interval_n;
  unsigned long time_s, time_n;

  // invariant: oplock is held at top of loop
  while ( 1 ) {
    do_yield = 1;
    while ( 1 ) {
      // must recompute these here because they can change across a wait/timedwait
      _channel->pull_period_s_n(pull_interval_s, pull_interval_n);
      if (pull_interval_s || pull_interval_n) {
	// if timeout not set, or if update_timeout true due to pull, recompute timeout
	if ( update_timeout || ((_timeout_s == 0) && (_timeout_n == 0)) ) {
	  omni_thread::get_time(&_timeout_s, &_timeout_n, pull_interval_s, pull_interval_n);
	}
      } else {
	_timeout_s = 0; _timeout_n = 0;
      }
      update_timeout = 0;
      // breaking from inner loop requires a pull interval timeout (if set)
      // and state must be RDI_Connected
      if ((_pxstate!=RDI_NotConnected) && (_pxstate!=RDI_Connected)) {
	break; // also break on exceptional state
      }
      if (_pxstate==RDI_Connected) {
	if ((_timeout_s == 0) && (_timeout_n == 0)) {
	  break; // pulling as fast as possible so pull now
	}
	omni_thread::get_time(&time_s, &time_n);
	if ( (time_s > _timeout_s) || ((time_s == _timeout_s) && (time_n > _timeout_n)) ) {
	  break; // pull interval has passed so pull now
	}
	do_yield = 0;
	RDI_OPLOCK_TIMEDWAIT(_timeout_s, _timeout_n); // must wait for pull interval
      } else {
	do_yield = 0;
	RDI_OPLOCK_WAIT; // must wait for _pxstate to change
      }
    }
    if ( _pxstate != RDI_Connected ) {
      RDI_OPLOCK_DEBUMP_RELEASE;
      RDI_THREAD_EXIT(RDIDbgCosCPxy, RDIDbgCosCPxy_nm, "pull thread\n");
    }

    // do not hold OPLOCK across try_pull call
    RDI_OPLOCK_RELEASE;

    // if we did not wait, yield here (wait or yield each time through outer loop)
    if (do_yield) {
      omni_thread::yield();
    }

    CORBA::Boolean outcall_worked = 0;
    try {
      event = _supplier->try_pull(hasev);
      outcall_worked = 1;
    } catch ( CORBA::INV_OBJREF& e ) {
      RDIDbgCosCPxyLog("Pull pool calling pull_event for proxy " << (void*)this << " - Invalid object reference\n");
    } catch ( CORBA::OBJECT_NOT_EXIST& e ) {
      RDIDbgCosCPxyLog("Pull pool calling pull_event for proxy " << (void*)this << " - Consumer object does not exist\n");
    } catch ( CORBA::COMM_FAILURE& e ) {
      RDIDbgCosCPxyLog("Pull pool calling pull_event for proxy " << (void*)this << " - Communication Failure\n");
    } catch (...) {
      RDIDbgCosCPxyLog("Pull pool calling pull_event for proxy " << (void*)this <<
	       " - Exception while pushing event to consumer\n");
    }

    RDI_OPLOCK_REACQUIRE(RDI_THREAD_EXIT(RDIDbgCosCPxy, RDIDbgCosCPxy_nm, "pull thread [**unexpected REACQUIRE failure**]\n"));

    update_timeout = 1; // force update of timeout at top of loop

    // add new event to channel even if we have been disconnected in the meantime
    if (outcall_worked && hasev && event) {
      _nevents += 1;
      _channel->new_any_event(*event);
    }
    // new_any_event copies event so we can delete it regardless
    if ( event ) {
      delete event; event = 0;
    }

    if (_pxstate != RDI_Connected) {
      // disconnected/disposed during try_pull -- do nothing else with my state
    } else {
      if (!outcall_worked) { // exception
	_pxstate = RDI_Exception;
      }
    }
  }
}

void EventProxyPullConsumer_i::disconnect_pull_consumer(WRAPPED_IMPLARG_VOID)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  _disconnect_client_and_dispose(1);
}

void EventProxyPullConsumer_i::disconnect_client_and_dispose
(CORBA::Boolean remove_proxy_from_admin)
{
  RDI_OPLOCK_ACQUIRE(return);
  _disconnect_client_and_dispose(remove_proxy_from_admin);
}

// *** lock must be held
// Effect: initiate impl dispose + release lock
void EventProxyPullConsumer_i::_disconnect_client_and_dispose
(CORBA::Boolean remove_proxy_from_admin)
{
  if (_pxstate == RDI_Disconnected) {
    RDIDbgCosCPxyLog("EventProxyPullConsumer_i::disconnect_client_and_dispose called twice on same proxy!\n");
    RDI_OPLOCK_RELEASE;
    return;
  }
  _pxstate = RDI_Disconnected; // acts as guard; following only entered by 1 thread
  // Wait for inuse to drop to zero.  This is not strictly necessary, but it
  // ensures that any outstanding try_pull call has a chance to complete cleanly.
  while (_oplockptr->inuse() > 0) {
    RDI_OPLOCK_BROADCAST;
    RDI_OPLOCK_INUSEZERO_WAIT;
  }
  if (remove_proxy_from_admin) {
    // do not hold OPLOCK across upcalls involving this
    RDI_OPLOCK_BUMP_RELEASE;
    _myadmin->remove_proxy(this);
    RDI_OPLOCK_REACQUIRE_DEBUMP(RDI_METHOD_RETURN_NULL(RDIDbgCosCPxy, RDIDbgCosCPxy_nm, "EventProxyPullConsumer_i::_disconnect_client_and_dispose [**unexpected REACQUIRE failure**]\n"));
  }
  _supplier = CosEvC::PullSupplier::_nil();
  RDI_OPLOCK_DISPOSE_IMPL_AND_RELEASE;
  _oplockptr = 0; // speeds up acquire check
}

// ------------------------------------------------------------ //
// EventProxyPushSupplier_i                                     //
// ------------------------------------------------------------ //

/** We use unbound threads here for we assume that in the majority
  * of configuations a thread pool will be used for pulling events
  * from suppliers and pushing events to consumers.
  */

class EventPushWorker : public omni_thread {
public:
  typedef void (EventProxyPushSupplier_i::*Method)(void);
  EventPushWorker(EventProxyPushSupplier_i* p, Method m) :
    omni_thread(NULL,PRIORITY_NORMAL), _proxy(p), _method(m) {;}
  void run(void *)      { (_proxy->*_method)(); }
private:
  EventProxyPushSupplier_i* _proxy;
  Method                    _method;
  EventPushWorker()  {;}
};

// ------------------------------------------------------------ //

EventProxyPushSupplier_i::EventProxyPushSupplier_i(ConsumerAdmin_i* admin, 
						   EventChannel_i*  evchn,
						   const CosNA::ProxyID&  prxid) :
  _oplockptr(0), _my_name(admin->L_my_name()),
  _channel(evchn), _myadmin(admin), 
  _pworker(0),  _thrdone(0), _nevents(0), 
  _pxstate(RDI_NotConnected), _pserial(prxid), _cnfqueue()
{
  RDI_OPLOCK_INIT;
  char buf[20];
  sprintf(buf, "proxy%ld", prxid);
  _my_name.length(_my_name.length()+1);
  _my_name[_my_name.length()-1] = (const char*)buf;
  _consumer = CosEvC::PushConsumer::_nil();
  // When the number of push threads allocated at the channel level is
  // 0, each proxy uses its own thread to pull events from its supplier
  if ( _channel->push_threads() == 0 ) {
    _pworker = new EventPushWorker(this, &EventProxyPushSupplier_i::_push);
    RDI_AssertAllocThrowNo(_pworker, "Memory allocation failed -- omni_thread\n");
    _pworker->start();
    _thrdone = 0;
    RDIDbgCosSPxyLog("Pull thread for proxy " << (void*)this << " -- " << _pworker->id() << '\n');
  }
  WRAPPED_REGISTER_IMPL2(this, &_my_name);
}

EventProxyPushSupplier_i::~EventProxyPushSupplier_i() {
  RDI_OPLOCKS_DESTROY_CHECK("EventProxyPushSupplier_i");
}

void EventProxyPushSupplier_i::connect_push_consumer
(CosEvC::PushConsumer_ptr consumer WRAPPED_IMPLARG)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( CORBA::is_nil(consumer) ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::BAD_PARAM(0, CORBA::COMPLETED_NO);
  }
  if ( _pxstate != RDI_NotConnected ) {
    RDI_OPLOCK_RELEASE;
    throw CosEvCA::AlreadyConnected();
  }
  _pxstate  = RDI_Connected;
  _consumer = CosEvC::PushConsumer::_duplicate(consumer);
  if ( _pworker ) {
    RDI_OPLOCK_SIGNAL;
  }
  RDI_NotifyConsumer* cpc = _channel->push_consumer();
  if ( cpc ) {
    cpc->signal_push_threads();
  }
  RDI_OPLOCK_RELEASE;
}

void EventProxyPushSupplier_i::add_event(RDI_StructuredEvent* event)
{
  RDI_OPLOCK_ACQUIRE(return);
  if ( (_pxstate == RDI_Connected) && event ) {
    event->incr_ref_counter( RDI_LOCK_EVENT );
    _cnfqueue.insert_tail(event);
    if ( _pworker ) {
      RDI_OPLOCK_SIGNAL;
    }
    RDI_NotifyConsumer* cpc = _channel->push_consumer();
    if ( cpc ) {
      cpc->signal_push_threads();
    }
  }
  RDI_OPLOCK_RELEASE;
}

void EventProxyPushSupplier_i::push_event(CORBA::Boolean& invalid)
{
  RDI_OPLOCK_ACQUIRE_BUMP(return); // debump on method exit

  RDI_StructuredEvent* event = 0;
  unsigned int         qsize = 0;

  invalid = 0;
  if ( (_pxstate != RDI_Connected) || (_cnfqueue.length() == 0) ) {
    RDI_OPLOCK_DEBUMP_RELEASE;
    return;
  }
  event = _cnfqueue.get_head();
  qsize = _cnfqueue.length() - 1;
  _cnfqueue.remove_head();
  _nevents += 1;

  // do not hold OPLOCK across push call
  RDI_OPLOCK_RELEASE;

  CORBA::Boolean outcall_worked = 0;
  if (  RDI_LOCK_EVENT  ) event->mylock().lock();
  try {
    const CosN::StructuredEvent& cosev = event->get_cos_event();
    if ( RDI_STR_EQ(event->get_type_name(), "%ANY") ) {
      _consumer->push(cosev.remainder_of_body);
    } else {
      CORBA::Any anyev;
      anyev <<= cosev;
      _consumer->push(anyev);
    }
    outcall_worked = 1;
  } catch ( CORBA::INV_OBJREF& e ) {
    RDIDbgCosSPxyLog("Push pool calling push_event for proxy " << (void*)this << " - Invalid object reference\n");
  } catch ( CORBA::OBJECT_NOT_EXIST& e ) {
    RDIDbgCosSPxyLog("Push pool calling push_event for proxy " << (void*)this << " - Consumer object does not exist\n");
  } catch ( CORBA::COMM_FAILURE& e ) {
    RDIDbgCosSPxyLog("Push pool calling push_event for proxy " << (void*)this << " - Communication Failure\n");
  } catch (...) {
    RDIDbgCosSPxyLog("Push pool calling push_event for proxy " << (void*)this << " - Exception while pushing event to consumer\n");
  }

  RDI_OPLOCK_REACQUIRE(return);

  if (  RDI_LOCK_EVENT  ) event->mylock().unlock();
  event->decr_ref_counter( RDI_LOCK_EVENT );

  if (outcall_worked) {
    _channel->incr_num_notifications(qsize);
  }

  if (_pxstate != RDI_Connected) {
    // disconnected/disposed during push -- do nothing else with my state
  } else {
    if (!outcall_worked) { // exception
      _pxstate = RDI_Exception;
      invalid  = 1;
    }
  }
  RDI_OPLOCK_DEBUMP_RELEASE;
}

void EventProxyPushSupplier_i::_push()
{
  RDI_OPLOCK_ACQUIRE_BUMP(return); // debump on thread exit

  RDI_StructuredEvent* event = 0;
  unsigned int         qsize = 0;
  CORBA::Boolean       do_yield = 0, outcall_worked = 0;

  // invariant : OPLOCK held at top of loop
  while ( 1 ) {
    do_yield = 1;
    while ( ((_pxstate == RDI_Connected) && (_cnfqueue.length() == 0)) ||
	    (_pxstate == RDI_NotConnected) ) {
      do_yield = 0;
      RDI_OPLOCK_WAIT;
    }
    if ( _pxstate != RDI_Connected ) {
      RDI_OPLOCK_DEBUMP_RELEASE;
      RDI_THREAD_EXIT(RDIDbgCosSPxy, RDIDbgCosCPxy_nm, "push thread\n");
    }
    event = _cnfqueue.get_head();
    qsize = _cnfqueue.length() - 1;
    _cnfqueue.remove_head();
    _nevents += 1;

    // do not hold OPLOCK across push call
    RDI_OPLOCK_RELEASE;

    // if we did not wait, yield here (wait or yield each time through outer loop)
    if (do_yield) {
      omni_thread::yield();
    }

    if (  RDI_LOCK_EVENT  ) event->mylock().lock();
    outcall_worked = 0;
    try {
      const CosN::StructuredEvent& cosev = 
	event->get_cos_event();
      if ( RDI_STR_EQ(event->get_type_name(), "%ANY") ) {
	_consumer->push(cosev.remainder_of_body);
      } else {
	CORBA::Any anyev;
	anyev <<= cosev;
	_consumer->push(anyev);
      }
      outcall_worked = 1;
    } catch ( CORBA::INV_OBJREF& e ) {
      RDIDbgCosSPxyLog("Push pool calling push_event for proxy " << (void*)this << " - Invalid object reference\n");
    } catch ( CORBA::OBJECT_NOT_EXIST& e ) {
      RDIDbgCosSPxyLog("Push pool calling push_event for proxy " << (void*)this << " - Consumer object does not exist\n");
    } catch ( CORBA::COMM_FAILURE& e ) {
      RDIDbgCosSPxyLog("Push pool calling push_event for proxy " << (void*)this << " - Communication Failure\n");
    } catch (...) {
      RDIDbgCosSPxyLog("Push pool calling push_event for proxy " << (void*)this << " - Exception while pushing event to consumer\n");
    }

    RDI_OPLOCK_REACQUIRE(RDI_THREAD_EXIT(RDIDbgCosSPxy, RDIDbgCosSPxy_nm, "push thread [**unexpected REACQUIRE failure**]\n"));

    if (  RDI_LOCK_EVENT  ) event->mylock().unlock();
    event->decr_ref_counter( RDI_LOCK_EVENT );

    if (outcall_worked) {
      _channel->incr_num_notifications(qsize);
    }

    if (_pxstate != RDI_Connected) {
      // disconnected/disposed during push -- do nothing else with my state
    } else {
      if (!outcall_worked) { // exception
	_pxstate = RDI_Exception;
      }
    }
  }
}

void EventProxyPushSupplier_i::disconnect_push_supplier(WRAPPED_IMPLARG_VOID)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  _disconnect_client_and_dispose(1);
}

void EventProxyPushSupplier_i::disconnect_client_and_dispose
(CORBA::Boolean remove_proxy_from_admin)
{
  RDI_OPLOCK_ACQUIRE(return);
  _disconnect_client_and_dispose(remove_proxy_from_admin);
}

// *** lock must be held
// Effect: initiate impl dispose + release lock
void EventProxyPushSupplier_i::_disconnect_client_and_dispose
(CORBA::Boolean remove_proxy_from_admin)
{
  if (_pxstate == RDI_Disconnected) {
    RDIDbgCosSPxyLog("EventProxyPushSupplier_i::disconnect_client_and_dispose called twice on same proxy!\n");
    RDI_OPLOCK_RELEASE;
    return;
  }
  _pxstate = RDI_Disconnected; // acts as guard; following only entered by 1 thread
  // Wait for inuse to drop to zero.  This is not strictly necessary, but it
  // ensures that any outstanding push call has a chance to complete cleanly.
  while (_oplockptr->inuse() > 0) {
    RDI_OPLOCK_BROADCAST;
    RDI_OPLOCK_INUSEZERO_WAIT;
  }
  if (remove_proxy_from_admin) {
    // do not hold OPLOCK across upcalls involving this
    RDI_OPLOCK_BUMP_RELEASE;
    _myadmin->remove_proxy(this);
    RDI_OPLOCK_REACQUIRE_DEBUMP(RDI_METHOD_RETURN_NULL(RDIDbgCosSPxy, RDIDbgCosSPxy_nm, "EventProxyPushSupplier_i::_disconnect_client_and_dispose [**unexpected REACQUIRE failure**]\n"));
  }
  _consumer = CosEvC::PushConsumer::_nil();
  RDI_StructuredEvent* event = 0;
  while ( (event = _cnfqueue.get_head()) ) {    // Remove all events
    _cnfqueue.remove_head();
    event->decr_ref_counter( RDI_LOCK_EVENT );
  }
  RDI_OPLOCK_DISPOSE_IMPL_AND_RELEASE;
  _oplockptr = 0; // speeds up acquire check
}

// ------------------------------------------------------------ //
// EventProxyPullSupplier_i                                     //
// ------------------------------------------------------------ //

EventProxyPullSupplier_i::EventProxyPullSupplier_i(ConsumerAdmin_i* admin, 
						   EventChannel_i*  evchn,
						   const CosNA::ProxyID&  prxid) :
  _oplockptr(0), _my_name(admin->L_my_name()),
  _channel(evchn), _myadmin(admin), _nevents(0), 
  _pxstate(RDI_NotConnected), _pserial(prxid), _cnfqueue()
{
  RDI_OPLOCK_INIT;
  char buf[20];
  sprintf(buf, "proxy%ld", prxid);
  _my_name.length(_my_name.length()+1);
  _my_name[_my_name.length()-1] = (const char*)buf;
  _consumer = CosEvC::PullConsumer::_nil();
  WRAPPED_REGISTER_IMPL2(this, &_my_name);
}

EventProxyPullSupplier_i:: ~EventProxyPullSupplier_i() {
  RDI_OPLOCKS_DESTROY_CHECK("EventProxyPullSupplier_i");
}

void EventProxyPullSupplier_i::connect_pull_consumer
(CosEvC::PullConsumer_ptr consumer WRAPPED_IMPLARG)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  RDIDbgCosSPxyLog("CosEvent consumer to connect to " << (void*)this << '\n');
  if ( CORBA::is_nil(consumer) ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::BAD_PARAM(0, CORBA::COMPLETED_NO);
  }
  if ( _pxstate != RDI_NotConnected ) {
    RDI_OPLOCK_RELEASE;
    throw CosEvCA::AlreadyConnected();
  }
  _pxstate  = RDI_Connected;
  _consumer = CosEvC::PullConsumer::_duplicate(consumer);
  RDIDbgCosSPxyLog("CosEvent consumer connected to " << (void*)this << '\n');
  RDI_OPLOCK_RELEASE;
}

void EventProxyPullSupplier_i::add_event(RDI_StructuredEvent* event)
{
  RDI_OPLOCK_ACQUIRE(return);
  if ( (_pxstate == RDI_Connected) && event ) {
    event->incr_ref_counter( RDI_LOCK_EVENT );
    _cnfqueue.insert_tail(event);
    RDI_OPLOCK_SIGNAL;
  }
  RDI_OPLOCK_RELEASE;
}

CORBA::Any* EventProxyPullSupplier_i::pull(WRAPPED_IMPLARG_VOID)
{
  RDI_OPLOCK_ACQUIRE_BUMP(RDI_THROW_INV_OBJREF); // debump on method exit
  if ( _pxstate == RDI_NotConnected ) {
    RDI_OPLOCK_DEBUMP_RELEASE;
    throw CosEvC::Disconnected();
  }
  CORBA::ULong qsize = 0;
  while ( (_pxstate == RDI_Connected) && ((qsize = _cnfqueue.length()) == 0) ) {
    RDI_OPLOCK_WAIT;
  }
  if ( _pxstate != RDI_Connected ) {
    RDI_OPLOCK_DEBUMP_RELEASE;
    throw CosEvC::Disconnected();
  }
  RDI_StructuredEvent* event = _cnfqueue.get_head();
  const CosN::StructuredEvent& cosev = event->get_cos_event();
  _cnfqueue.remove_head();
  _nevents += 1;
  RDIDbgCosSPxyLog("pull event " << event << " from queue of " << (void*)this << '\n');

  CORBA::Any* res = 0;
  if (  RDI_LOCK_EVENT  ) event->mylock().lock();
  if ( RDI_STR_EQ(event->get_type_name(), "%ANY") ) {
    res = new CORBA::Any(cosev.remainder_of_body);
  } else {
    res = new CORBA::Any;
    (*res) <<= cosev;
  }
  if (  RDI_LOCK_EVENT  ) event->mylock().unlock();
  event->decr_ref_counter( RDI_LOCK_EVENT );

  _channel->incr_num_notifications(qsize);
  RDI_OPLOCK_DEBUMP_RELEASE;
  return res;
}

CORBA::Any* EventProxyPullSupplier_i::try_pull
(CORBA::Boolean& has_event WRAPPED_IMPLARG)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( _pxstate != RDI_Connected ) {
    RDI_OPLOCK_RELEASE;
    throw CosEvC::Disconnected();
  }
  CORBA::ULong qsize = _cnfqueue.length();
  if (qsize == 0 ) {
    RDI_OPLOCK_RELEASE;
    has_event = 0;
    return new CORBA::Any();
  }
  RDI_StructuredEvent* event = _cnfqueue.get_head();
  const CosN::StructuredEvent& cosev = event->get_cos_event();
  _cnfqueue.remove_head();
  _nevents += 1;
  RDIDbgCosSPxyLog("try_pull event " << event << " from queue of " << (void*)this << '\n');

  if (  RDI_LOCK_EVENT  ) event->mylock().lock();
  CORBA::Any* res = 0;
  has_event = 1;
  if ( RDI_STR_EQ(event->get_type_name(), "%ANY") ) {
    res = new CORBA::Any(cosev.remainder_of_body);
  } else {
    res = new CORBA::Any;
    (*res) <<= cosev;
  }
  if (  RDI_LOCK_EVENT  ) event->mylock().unlock();
  event->decr_ref_counter( RDI_LOCK_EVENT );

  _channel->incr_num_notifications(qsize);
  RDI_OPLOCK_RELEASE;
  return res;
}

void EventProxyPullSupplier_i::disconnect_pull_supplier(WRAPPED_IMPLARG_VOID)
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  _disconnect_client_and_dispose(1);
}

void EventProxyPullSupplier_i::disconnect_client_and_dispose(CORBA::Boolean remove_proxy_from_admin)
{
  RDI_OPLOCK_ACQUIRE(return);
  _disconnect_client_and_dispose(remove_proxy_from_admin);
}

// *** lock must be held
// Effect: initiate impl dispose + release lock
void EventProxyPullSupplier_i::_disconnect_client_and_dispose(CORBA::Boolean remove_proxy_from_admin)
{
  if (_pxstate == RDI_Disconnected) {
    RDIDbgCosSPxyLog("EventProxyPullSupplier_i::disconnect_client_and_dispose called twice on same proxy!\n");
    RDI_OPLOCK_RELEASE;
    return;
  }
  _pxstate = RDI_Disconnected; // acts as guard -- only one thread does the following
  // Wait for inuse to drop to zero.  This ensures that any outstanding blocked pull
  // call has a chance to complete cleanly.
  while (_oplockptr->inuse() > 0) {
    RDI_OPLOCK_BROADCAST;
    RDI_OPLOCK_INUSEZERO_WAIT;
  }
  if (remove_proxy_from_admin) {
    // do not hold OPLOCK across upcalls involving this
    RDI_OPLOCK_BUMP_RELEASE;
    _myadmin->remove_proxy(this);
    RDI_OPLOCK_REACQUIRE_DEBUMP(RDI_METHOD_RETURN_NULL(RDIDbgCosSPxy, RDIDbgCosSPxy_nm, "EventProxyPullSupplier_i::_disconnect_client_and_dispose [**unexpected REACQUIRE failure**]\n"));
  }
  _consumer = CosEvC::PullConsumer::_nil();
  RDI_StructuredEvent* event = 0;
  while ( (event = _cnfqueue.get_head()) ) {    // Remove all events
    _cnfqueue.remove_head();
    event->decr_ref_counter( RDI_LOCK_EVENT );
  }
  RDI_OPLOCK_DISPOSE_IMPL_AND_RELEASE;
  _oplockptr = 0; // speeds up acquire check
}

////////////////////////////////////////
// Logging

RDIstrstream& EventProxyPushConsumer_i::log_output(RDIstrstream& str) const
{
  str << (void*)this << " CosEvent ProxyPushConsumer";
  str << _pxstate;
  return str << " #Push "<< _nevents;
}

RDIstrstream& EventProxyPullConsumer_i::log_output(RDIstrstream& str) const
{
  str << (void*)this << " CosEvent ProxyPullConsumer";
  str << _pxstate;
  return str << "#Push "<< _nevents;
}

RDIstrstream& EventProxyPushSupplier_i::log_output(RDIstrstream& str) const
{
  str << (void*)this << " CosEvent ProxyPushSupplier";
  str << _pxstate;
  return str << "QSize "<< _cnfqueue.length() << " #Push "<< _nevents;
}

RDIstrstream& EventProxyPullSupplier_i::log_output(RDIstrstream& str) const
{
  str << (void*)this << " CosEvent ProxyPullSupplier";
  str << _pxstate;
  return str << "QSize "<< _cnfqueue.length() << " #Push "<< _nevents;
}

////////////////////////////////////////
// Interactive

AttN::IactSeq* EventProxyPushConsumer_i::children(CORBA::Boolean only_cleanup_candidates) {
  AttN::IactSeq* ren = new AttN::IactSeq;
  if ( ren == (AttN::IactSeq*) 0 ) {
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  ren->length(0);
  return ren;
}

CORBA::Boolean EventProxyPushConsumer_i::safe_cleanup()
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_pxstate == RDI_Disconnected) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // already in process of being disposed
  if (_pxstate == RDI_Connected) { RDI_OPLOCK_RELEASE; return 0; } // do not destroy connected proxy
  _disconnect_client_and_dispose(1);
  return 1;
}

AttN::NameSeq* EventProxyPushConsumer_i::my_name( WRAPPED_IMPLARG_VOID ) {
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::NameSeq* res = new AttN::NameSeq(_my_name);
  if ( res == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

AttN::NameSeq* EventProxyPushConsumer_i::child_names( WRAPPED_IMPLARG_VOID ) {
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::NameSeq* names = new AttN::NameSeq;
  if ( names == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  names->length(0);
  RDI_OPLOCK_RELEASE;
  return names;
}

void EventProxyPushConsumer_i::out_commands(RDIstrstream& str) {
  str << "CosEvent ProxyPushConsumer: No commands supported yet.\n";
}

char*
EventProxyPushConsumer_i::do_command(const char* cmnd, CORBA::Boolean& success,
				     CORBA::Boolean& target_changed,
				     AttN_Interactive_outarg next_target  WRAPPED_IMPLARG ) {
  RDIParseCmd p(cmnd);
  success = 1;
  target_changed = 0;
  if (p.argc == 0) { return CORBA_STRING_DUP("");  }

  RDIstrstream str;
  if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "help")) {
    out_commands(str);
  } else {
    str << "Invalid command: " << cmnd << "\n";
    success = 0;
  }
  RDIRptForceLog(_my_name << " received command: " << cmnd << "\nResult:\n" << str.buf());
  // this is the only safe way to return a string?
  return CORBA_STRING_DUP(str.buf());
}

AttN::IactSeq* EventProxyPushSupplier_i::children(CORBA::Boolean only_cleanup_candidates) {
  AttN::IactSeq* ren = new AttN::IactSeq;
  if ( ren == (AttN::IactSeq*) 0 ) {
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  ren->length(0);
  return ren;
}

CORBA::Boolean EventProxyPushSupplier_i::safe_cleanup()
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_pxstate == RDI_Disconnected) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // already in process of being disposed
  if (_pxstate == RDI_Connected) { RDI_OPLOCK_RELEASE; return 0; } // do not destroy connected proxy
  _disconnect_client_and_dispose(1);
  return 1;
}

AttN::NameSeq* EventProxyPushSupplier_i::my_name( WRAPPED_IMPLARG_VOID ) {
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::NameSeq* res = new AttN::NameSeq(_my_name);
  if ( res == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

AttN::NameSeq* EventProxyPushSupplier_i::child_names( WRAPPED_IMPLARG_VOID ) {
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::NameSeq* names = new AttN::NameSeq;
  if ( names == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  names->length(0);
  RDI_OPLOCK_RELEASE;
  return names;
}

void EventProxyPushSupplier_i::out_commands(RDIstrstream& str) {
  str << "CosEvent ProxyPushSupplier: No commands supported yet.\n";
}

char*
EventProxyPushSupplier_i::do_command(const char* cmnd, CORBA::Boolean& success,
				     CORBA::Boolean& target_changed,
				     AttN_Interactive_outarg next_target  WRAPPED_IMPLARG ) {
  RDIParseCmd p(cmnd);
  success = 1;
  target_changed = 0;
  if (p.argc == 0) { return CORBA_STRING_DUP("");  }

  RDIstrstream str;
  if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "help")) {
    out_commands(str);
  } else {
    str << "Invalid command: " << cmnd << "\n";
    success = 0;
  }
  RDIRptForceLog(_my_name << " received command: " << cmnd << "\nResult:\n" << str.buf());
  // this is the only safe way to return a string?
  return CORBA_STRING_DUP(str.buf());
}

AttN::IactSeq* EventProxyPullConsumer_i::children(CORBA::Boolean only_cleanup_candidates) {
  AttN::IactSeq* ren = new AttN::IactSeq;
  if ( ren == (AttN::IactSeq*) 0 ) {
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  ren->length(0);
  return ren;
}

CORBA::Boolean EventProxyPullConsumer_i::safe_cleanup()
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_pxstate == RDI_Disconnected) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // already in process of being disposed
  if (_pxstate == RDI_Connected) { RDI_OPLOCK_RELEASE; return 0; } // do not destroy connected proxy
  _disconnect_client_and_dispose(1);
  return 1;
}

AttN::NameSeq* EventProxyPullConsumer_i::my_name( WRAPPED_IMPLARG_VOID ) {
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::NameSeq* res = new AttN::NameSeq(_my_name);
  if ( res == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

AttN::NameSeq* EventProxyPullConsumer_i::child_names( WRAPPED_IMPLARG_VOID ) {
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::NameSeq* names = new AttN::NameSeq;
  if ( names == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  names->length(0);
  RDI_OPLOCK_RELEASE;
  return names;
}

void EventProxyPullConsumer_i::out_commands(RDIstrstream& str) {
  str << "CosEvent ProxyPullConsumer: No commands supported yet.\n";
}

char*
EventProxyPullConsumer_i::do_command(const char* cmnd, CORBA::Boolean& success,
				     CORBA::Boolean& target_changed,
				     AttN_Interactive_outarg next_target  WRAPPED_IMPLARG ) {
  RDIParseCmd p(cmnd);
  success = 1;
  target_changed = 0;
  if (p.argc == 0) { return CORBA_STRING_DUP("");  }

  RDIstrstream str;
  if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "help")) {
    out_commands(str);
  } else {
    str << "Invalid command: " << cmnd << "\n";
    success = 0;
  }
  RDIRptForceLog(_my_name << " received command: " << cmnd << "\nResult:\n" << str.buf());
  // this is the only safe way to return a string?
  return CORBA_STRING_DUP(str.buf());
}

AttN::IactSeq* EventProxyPullSupplier_i::children(CORBA::Boolean only_cleanup_candidates) {
  AttN::IactSeq* ren = new AttN::IactSeq;
  if ( ren == (AttN::IactSeq*) 0 ) {
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  ren->length(0);
  return ren;
}

CORBA::Boolean EventProxyPullSupplier_i::safe_cleanup()
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_pxstate == RDI_Disconnected) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // already in process of being disposed
  if (_pxstate == RDI_Connected) { RDI_OPLOCK_RELEASE; return 0; } // do not destroy connected proxy
  _disconnect_client_and_dispose(1);
  return 1;
}

AttN::NameSeq* EventProxyPullSupplier_i::my_name( WRAPPED_IMPLARG_VOID ) {
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::NameSeq* res = new AttN::NameSeq(_my_name);
  if ( res == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

AttN::NameSeq* EventProxyPullSupplier_i::child_names( WRAPPED_IMPLARG_VOID ) {
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::NameSeq* names = new AttN::NameSeq;
  if ( names == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  names->length(0);
  RDI_OPLOCK_RELEASE;
  return names;
}

void EventProxyPullSupplier_i::out_commands(RDIstrstream& str) {
  str << "CosEvent ProxyPullSupplier: No commands supported yet.\n";
}

char*
EventProxyPullSupplier_i::do_command(const char* cmnd, CORBA::Boolean& success,
				     CORBA::Boolean& target_changed,
				     AttN_Interactive_outarg next_target  WRAPPED_IMPLARG ) {
  RDIParseCmd p(cmnd);
  success = 1;
  target_changed = 0;
  if (p.argc == 0) { return CORBA_STRING_DUP("");  }

  RDIstrstream str;
  if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "help")) {
    out_commands(str);
  } else {
    str << "Invalid command: " << cmnd << "\n";
    success = 0;
  }
  RDIRptForceLog(_my_name << " received command: " << cmnd << "\nResult:\n" << str.buf());
  // this is the only safe way to return a string?
  return CORBA_STRING_DUP(str.buf());
}

