// -*- Mode: C++; -*-
//                              File      : ChannelAdmin_i.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 ConsumerAdmin_i and SupplierAdmin_i
//
 
/*
$Log: ChannelAdmin_i.cc,v $
Revision 1.79  2001/10/06 00:44:22  alcfp
moved some EventChannel_i state to private

Revision 1.78  2001/10/05 02:18:16  alcfp
small fix

Revision 1.77  2001/10/03 05:07:05  alcfp
further support for set and config commands at all levels

Revision 1.76  2001/09/28 02:10:48  alcfp
FilterFactory_i uses oplockptr

Revision 1.75  2001/09/26 20:41:23  alcfp
set command now supported

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

Revision 1.73  2001/09/05 18:02:43  alcfp
more interactive commands working

Revision 1.72  2001/08/26 16:09:30  alcfp
more interactive stuff working

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

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

Revision 1.69  2001/06/22 07:00:30  alcfp
moved to new logging scheme

Revision 1.68  2001/06/14 03:34:01  alcfp
fixed deadlock case - holding lock on admin while removing it from its admin group proved a bad idea, was unnecessary in any case

Revision 1.67  2001/06/12 17:42:44  alcfp
switch to RDITimeWrappers

Revision 1.66  2001/05/29 20:37:24  alcfp
Fixed bug: proper cleanup of proxies on admin destruct.
Use batch size for pulling rather than just using default of 5.
push_event, pull_event: set invalid to 1 if status is Disconnected or Exception

Revision 1.65  2001/05/08 20:47:20  alcfp
Fixed reported bug: adding revocation of outstanding offers when supplier disconnects

Revision 1.64  2001/05/07 16:15:07  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.63  2000/12/02 16:40:54  alcfp
fix for HP aCC

Revision 1.62  2000/12/01 23:01:09  alcfp
fixed 2 bugs, 1 in admin-level subscription_change, 1 in changepool _next_available method

Revision 1.61  2000/11/17 21:35:54  alcfp
fixed some cases where _fa_helper should have been used

Revision 1.60  2000/11/16 05:32:13  alcfp
fixes for aCC

Revision 1.59  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.58  2000/11/05 04:48:09  alcfp
changed in defaults, env variable overrride, try_pull variants

Revision 1.57  2000/10/30 05:41:46  alcfp
renamed CosNotify.h to CosNotifyShorthands.h and placed CosNotifyShorthands.h in omniNotify/include rather than src/services/include

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

Revision 1.55  2000/08/22 18:23:52  alcfp
added description to each file

Revision 1.54  2000/08/16 20:19:20  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 "corba_wrappers.h"
#include "RDICatchMacros.h"
#include "RDIStringDefs.h"
#include "RDI.h"
#include "CosNotifyShorthands.h"
#include "CosNfyUtils.h"
#include "CosNotification_i.h"
#include "RDITypeMap.h"
#include "RDIOplocksMacros.h"
#include "CosEventChannelAdmin_i.h"
#include "CosNotifyChannelAdmin_i.h"
#include "RDIInteractive.h"

// ------------------------------------------------------------- //
// ConsumerAdmin_i implementation                                //
// ------------------------------------------------------------- //

ConsumerAdmin_i::ConsumerAdmin_i(EventChannel_i* channel,
				 CosNA::InterFilterGroupOperator op,
				 const CosNA::AdminID& serial) :
  RDINotifySubscribe(),
  _oplockptr(0), _my_name(channel->L_my_name()), _disposed(0),
  _fa_helper(), _channel(channel), _qosprop(0),
  _serial(serial), _and_or_oper(op), _rqstypes(), _prx_serial(1),
  _num_proxies(0), _cosevent_push(), _cosevent_pull(),
  _prx_any_push(RDI_ULongHash, RDI_ULongRank),
  _prx_any_pull(RDI_ULongHash, RDI_ULongRank),
  _prx_struc_push(RDI_ULongHash, RDI_ULongRank),
  _prx_struc_pull(RDI_ULongHash, RDI_ULongRank),
  _prx_batch_push(RDI_ULongHash, RDI_ULongRank),
  _prx_batch_pull(RDI_ULongHash, RDI_ULongRank)
{
  RDI_OPLOCK_INIT;
  _qosprop = new RDI_NotifQoS(_channel->qos_properties());
  RDI_AssertAllocThrowNo(_qosprop, "Memory allocation failure - RDI_NotifQoS\n");
  _rqstypes.length(0);
  _prio_filter = CosNF::MappingFilter::_nil();
  _life_filter = CosNF::MappingFilter::_nil();
  _time_born.set_curtime();
  _prxy_exit   = _time_born;
  char buf[20];
  sprintf(buf, "cadmin%ld", serial);
  _my_name.length(_my_name.length()+1);
  _my_name[_my_name.length()-1] = (const char*)buf;
  WRAPPED_REGISTER_IMPL2(this, &_my_name);
}

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

// EventChannel_i admin + proxy dispatch routines invoke lock/unlock:
// lock returns true if lock acquired, else false (false means admin was disposed)
CORBA::Boolean ConsumerAdmin_i::lock()   { RDI_OPLOCK_ACQUIRE_BUMP(return 0); return 1; }
void           ConsumerAdmin_i::unlock() { RDI_OPLOCK_DEBUMP_RELEASE; }


CosNA::AdminID
ConsumerAdmin_i::MyID( WRAPPED_IMPLARG_VOID ) 
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNA::AdminID res = _serial;
  RDI_OPLOCK_RELEASE;
  return res;
}

CosNA::EventChannel_ptr
ConsumerAdmin_i::MyChannel( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNA::EventChannel_ptr res = WRAPPED_IMPL2OREF(CosNA::EventChannel, _channel);
  RDI_OPLOCK_RELEASE;
  return res;
}

CosNA::InterFilterGroupOperator
ConsumerAdmin_i::MyOperator( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNA::InterFilterGroupOperator res = _and_or_oper;
  RDI_OPLOCK_RELEASE;
  return res;
}

CosNF::MappingFilter_ptr
ConsumerAdmin_i::priority_filter( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNF::MappingFilter_ptr res = CosNF::MappingFilter::_nil();
  if ( !CORBA::is_nil(_prio_filter) ) {
    res = CosNF::MappingFilter::_duplicate(_prio_filter);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}
 
void
ConsumerAdmin_i::priority_filter(CosNF::MappingFilter_ptr prio_filter   
				 WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  if ( CORBA::is_nil(prio_filter) ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::BAD_PARAM(0, CORBA::COMPLETED_NO);
  }
  _prio_filter = CosNF::MappingFilter::_duplicate(prio_filter);
  RDI_OPLOCK_RELEASE;
}

CosNF::MappingFilter_ptr
ConsumerAdmin_i::lifetime_filter( WRAPPED_IMPLARG_VOID )
{ 
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNF::MappingFilter_ptr res = CosNF::MappingFilter::_nil();
  if ( ! CORBA::is_nil(_life_filter) ) {
    res = CosNF::MappingFilter::_duplicate(_life_filter); 
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

void
ConsumerAdmin_i::lifetime_filter(CosNF::MappingFilter_ptr life_filter   
				 WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  if ( CORBA::is_nil(life_filter) ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::BAD_PARAM(0, CORBA::COMPLETED_NO);
  }
  _life_filter = CosNF::MappingFilter::_duplicate(life_filter);
  RDI_OPLOCK_RELEASE;
}

CosEvCA::ProxyPushSupplier_ptr 
ConsumerAdmin_i::obtain_push_supplier( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosEvCA::ProxyPushSupplier_ptr res = 
    CosEvCA::ProxyPushSupplier::_nil();
  EventProxyPushSupplier_i* prx=0;
  
  RDIDbgCAdmLog("CosEvent ProxyPushSupplier creation requested\n");
  if ( ! _channel->incr_consumers() ) {
    // return null res
  } else if ( ! (prx = new EventProxyPushSupplier_i(this, _channel, _prx_serial)) ) {
    _channel->decr_consumers();
    // return null res
  } else if ( _cosevent_push.insert_tail(prx) != 0 ) {
    _channel->decr_consumers();
    prx->disconnect_client_and_dispose(0);
    // return null res
  } else { // return valid res
    _num_proxies++;
    _prx_serial++;
    RDI_NotifyConsumer* cpc = _channel->push_consumer();
    if ( cpc ) {        // Register Push Proxy
      cpc->insert_proxy(prx);
    }
    RDIDbgCAdmLog("CosEvent ProxyPushSupplier creation completed\n");
    res = WRAPPED_IMPL2OREF(CosEvCA::ProxyPushSupplier, prx);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

CosEvCA::ProxyPullSupplier_ptr
ConsumerAdmin_i::obtain_pull_supplier( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosEvCA::ProxyPullSupplier_ptr res = 
    CosEvCA::ProxyPullSupplier::_nil();
  EventProxyPullSupplier_i* prx=0;
  RDIDbgCAdmLog("CosEvent ProxyPullSupplier creation requested\n");
  if ( ! _channel->incr_consumers() ) {
    // return null res
  } else if ( ! (prx = new EventProxyPullSupplier_i(this, _channel, _prx_serial)) ) {
    _channel->decr_consumers();
    // return null res
  } else if ( _cosevent_pull.insert_tail(prx) != 0 ) {
    _channel->decr_consumers();
    prx->disconnect_client_and_dispose(0);
    // return null res
  } else { // return valid res
    _prx_serial++;
    _num_proxies++;
    RDIDbgCAdmLog("CosEvent ProxyPullSupplier creation completed\n");
    res = WRAPPED_IMPL2OREF(CosEvCA::ProxyPullSupplier, prx);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

// called externally by consumers
void
ConsumerAdmin_i::subscription_change(const CosN::EventTypeSeq& added,
				     const CosN::EventTypeSeq& deled   
				     WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed

  // A consumer can invoke this operation as long as filters have
  // not been placed at this admin
  if ( _fa_helper.has_filters() ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::BAD_OPERATION(0, CORBA::COMPLETED_NO);
  }

  CosN::EventTypeSeq added_copy = added;
  CosN::EventTypeSeq deled_copy = deled;
  CosN::EventTypeSeq nadded;
  CosN::EventTypeSeq ndeled;
  nadded.length(0);
  ndeled.length(0);
  CORBA::ULong   ix = 0;
  if ( ! RDI_EventType::valid_sequence(added_copy, ix) ) {
    RDI_OPLOCK_RELEASE;
    throw CosNC::InvalidEventType(added[ix]);
  }
  if ( ! RDI_EventType::valid_sequence(deled_copy, ix) ) {
    RDI_OPLOCK_RELEASE;
    throw CosNC::InvalidEventType(deled[ix]);
  }
  // If this is not the first time this operation gets invoked, we
  // need to compute the actual added and deleted event types that
  // will be used to update the TypeMap
  if ( _rqstypes.length() ) {
    RDI_EventType::compute_diff(_rqstypes, added_copy, deled_copy, nadded, ndeled);
    if ( nadded.length() || ndeled.length() ) {
      (void) _channel->update_mapping(nadded, ndeled, this, 0);
      RDI_EventType::update(_rqstypes, added_copy, deled_copy);
    }
  } else {
    ndeled.length(0);
    _rqstypes = added_copy; 
    RDI_EventType::update(_rqstypes, added_copy, deled_copy);
    (void) _channel->update_mapping(_rqstypes, ndeled, this, 0);
  }
  RDI_OPLOCK_RELEASE;
}

// called internally due to filter activity
void
ConsumerAdmin_i::propagate_subscription_change(const CosN::EventTypeSeq& added,
					       const CosN::EventTypeSeq& deled,
					       Filter_i*                flter)
{ (void) _channel->update_mapping(added, deled, this, flter); }

void
ConsumerAdmin_i::filter_destroy_i(Filter_i* filter)
{ _fa_helper.rem_filter_i(filter); }

CosNA::ProxyIDSeq*
ConsumerAdmin_i::pull_suppliers( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  RDI_HashCursor<CosNA::ProxyID, ProxyPullSupplier_i *>           ac;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullSupplier_i *> sc;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullSupplier_i * >  bc;
  unsigned long num;
  CosNA::ProxyIDSeq*  seq = new CosNA::ProxyIDSeq();
  if ( ! seq ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_MAYBE);
  }
  num = _prx_any_pull.length() +
    _prx_struc_pull.length() + _prx_batch_pull.length();
  seq->length(num);
  for ( num=0, ac = _prx_any_pull.cursor(); ac.is_valid(); ++ac, ++num ) {
    (*seq)[num] = ac.key();
  }
  for ( sc = _prx_struc_pull.cursor(); sc.is_valid(); ++sc, ++num ) {
    (*seq)[num] = sc.key();
  }
  for ( bc = _prx_batch_pull.cursor(); bc.is_valid(); ++bc, ++num ) {
    (*seq)[num] = bc.key();
  }
  RDI_OPLOCK_RELEASE;
  return seq;
}

CosNA::ProxyIDSeq*
ConsumerAdmin_i::push_suppliers( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  RDI_HashCursor<CosNA::ProxyID, ProxyPushSupplier_i *>           ac;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushSupplier_i *> sc;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushSupplier_i * >  bc;
  unsigned long num;
  CosNA::ProxyIDSeq*  seq = new CosNA::ProxyIDSeq();
  if ( ! seq ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_MAYBE);
  }
  num = _prx_any_push.length() +
    _prx_struc_push.length() + _prx_batch_push.length();
  seq->length(num);
  for ( num=0, ac = _prx_any_push.cursor(); ac.is_valid(); ++ac, ++num ) {
    (*seq)[num] = ac.key();
  }
  for ( sc = _prx_struc_push.cursor(); sc.is_valid(); ++sc, ++num ) {
    (*seq)[num] = sc.key();        
  }
  for ( bc = _prx_batch_push.cursor(); bc.is_valid(); ++bc, ++num ) {
    (*seq)[num] = bc.key();
  }
  RDI_OPLOCK_RELEASE;
  return seq;
}

CosNA::ProxySupplier_ptr 
ConsumerAdmin_i::get_proxy_supplier(CosNA::ProxyID proxy_id WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  ProxyPushSupplier_i*           apush=0;
  ProxyPullSupplier_i*           apull=0;
  StructuredProxyPushSupplier_i* spush=0;
  StructuredProxyPullSupplier_i* spull=0;
  SequenceProxyPushSupplier_i*   bpush=0;
  SequenceProxyPullSupplier_i*   bpull=0;

  CosNA::ProxySupplier_ptr res = CosNA::ProxySupplier::_nil();

  if ( proxy_id <= _prx_serial ) {
    if ( _prx_any_push.lookup(proxy_id, apush) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, apush);
    } else if ( _prx_any_pull.lookup(proxy_id, apull) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, apull);
    } else if ( _prx_struc_push.lookup(proxy_id, spush) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, spush);
    } else if ( _prx_struc_pull.lookup(proxy_id, spull) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, spull);
    } else if ( _prx_batch_push.lookup(proxy_id, bpush) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, bpush);
    } else if ( _prx_batch_pull.lookup(proxy_id, bpull) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, bpull);
    }
  }
  RDI_OPLOCK_RELEASE;
  if (CORBA::is_nil(res)) {
    throw CosNA::ProxyNotFound();
  }
  return res;
}

CosNA::ProxySupplier_ptr 
ConsumerAdmin_i::obtain_notification_pull_supplier(CosNA::ClientType ctype, 
						   CosNA::ProxyID&   proxy_id
						   WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed

  // Check with  the channel if we are about to exceed the maximum
  // number of consumers supported by the channel
  RDIDbgCAdmLog("Creation of a new ProxySupplier -- type " << ctype << '\n');
  if ( ! _channel->incr_consumers() ) {
    CosNA::AdminLimit limit;
    RDIDbgCAdmLog("Consumer limit has been reached creation failed\n");
    limit.name    = (const char *) "MaxConsumers";
    limit.value <<= _channel->max_consumers();
    RDI_OPLOCK_RELEASE;
    throw CosNA::AdminLimitExceeded(limit);
  }

  CosNA::ProxySupplier_ptr res = CosNA::ProxySupplier::_nil();
  if ( ctype == CosNA::ANY_EVENT ) {
    ProxyPullSupplier_i* prx = 0;
    if ( ! (prx = new ProxyPullSupplier_i(this, _channel, _prx_serial)) ) {
      _channel->decr_consumers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_any_pull.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_consumers(); 
      } else {
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, prx);
      }
    }
  } else if ( ctype == CosNA::STRUCTURED_EVENT ) {
    StructuredProxyPullSupplier_i* prx = 
      new StructuredProxyPullSupplier_i(this,_channel,_prx_serial);
    if ( ! prx ) {
      _channel->decr_consumers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_struc_pull.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_consumers();
      } else {
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, prx);
      }
    }
  } else if ( ctype == CosNA::SEQUENCE_EVENT ) {
    SequenceProxyPullSupplier_i* prx =
      new SequenceProxyPullSupplier_i(this,_channel,_prx_serial);
    if ( ! prx ) {
      _channel->decr_consumers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_batch_pull.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_consumers();
      } else {
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, prx);
      }
    }
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

CosNA::ProxySupplier_ptr 
ConsumerAdmin_i::obtain_notification_push_supplier(CosNA::ClientType ctype, 
						   CosNA::ProxyID&   proxy_id
						   WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed

  // Check with  the channel if we are about to exceed the maximum
  // number of consumers supported by the channel
  RDIDbgCAdmLog("Creation of a new ProxySupplier -- type " << ctype << '\n');
  if ( ! _channel->incr_consumers() ) {
    CosNA::AdminLimit limit;
    RDIDbgCAdmLog("Consumer limit has been reached creation failed\n");
    limit.name    = (const char *) "MaxConsumers";
    limit.value <<= _channel->max_consumers();
    RDI_OPLOCK_RELEASE;
    throw CosNA::AdminLimitExceeded(limit);
  }

  CosNA::ProxySupplier_ptr res = CosNA::ProxySupplier::_nil();
  if ( ctype == CosNA::ANY_EVENT ) {
    ProxyPushSupplier_i* prx = 0;
    if ( ! (prx = new ProxyPushSupplier_i(this, _channel, _prx_serial)) ) {
      _channel->decr_consumers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_any_push.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_consumers(); 
      } else {
	RDI_NotifyConsumer* cpc = _channel->push_consumer();
	if ( cpc ) {    // Register Push Proxy
	  cpc->insert_proxy(prx);
	}
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, prx);
      }
    }
  } else if ( ctype == CosNA::STRUCTURED_EVENT ) {
    StructuredProxyPushSupplier_i* prx = 
      new StructuredProxyPushSupplier_i(this,_channel,_prx_serial);
    if ( ! prx ) {
      RDIDbgCAdmLog("\tfailed to create proxy\n");
      _channel->decr_consumers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_struc_push.insert(proxy_id, prx) != 0 ) {
	RDIDbgCAdmLog("\tfailed to register proxy with id " << proxy_id << '\n');
	prx->disconnect_client_and_dispose(0);
	_channel->decr_consumers();
      } else {
	RDI_NotifyConsumer* cpc = _channel->push_consumer();
	if ( cpc ) {    // Register Push Proxy
	  cpc->insert_proxy(prx);
	}
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, prx);
      }
    }
  } else if ( ctype == CosNA::SEQUENCE_EVENT ) {
    SequenceProxyPushSupplier_i* prx = 
      new SequenceProxyPushSupplier_i(this,_channel,_prx_serial);
    if ( ! prx ) {
      _channel->decr_consumers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_batch_push.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_consumers();
      } else { 
	RDI_NotifyConsumer* cpc = _channel->push_consumer();
	if ( cpc ) {    // Register Push Proxy
	  cpc->insert_proxy(prx);
	}
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxySupplier, prx);
      }
    }
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

CORBA::Boolean
ConsumerAdmin_i::safe_cleanup( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( _serial == 0 ) { // Default ConsumerAdmin not a cleanup candidate
    RDI_OPLOCK_RELEASE;
    return 0;
  }
  if (_num_proxies > 0) { // Admin with proxies not a cleanup candidate
    RDI_OPLOCK_RELEASE;
    return 0;
  }
  _disconnect_clients_and_dispose(1); // 1 means update channel
  return 1;
}

void
ConsumerAdmin_i::destroy( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( _serial == 0 ) {
    RDIDbgCAdmLog("Default ConsumerAdmin cannot be destroyed!\n");
    RDI_OPLOCK_RELEASE;
    return;
  }
  _disconnect_clients_and_dispose(1); // 1 means update channel
}

CosN::QoSProperties*
ConsumerAdmin_i::get_qos( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  //RDI_NotifQoS*   qosp = qos_properties();
  //RDI_Assert(qosp, "Event Channel should have RDI_NotifQoS set\n");
  //CosN::QoSProperties* res = qosp->get_qos(RDI_C_ADMIN);
  CosN::QoSProperties* res = _qosprop->get_qos(RDI_C_ADMIN);
  RDI_OPLOCK_RELEASE;
  return res;
}

void
ConsumerAdmin_i::set_qos(const CosN::QoSProperties& r_qos  WRAPPED_IMPLARG )
{
  if (r_qos.length() == 0) return;
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosN::PropertyErrorSeq eseq;
  CosN::NamedPropertyRangeSeq rseq;
  CORBA::Boolean  subobjs = (_num_proxies ? 1 : 0);
  
  if (! RDI_NotifQoS::validate(r_qos, *_qosprop, RDI_C_ADMIN, eseq, rseq, subobjs)) {
    RDI_OPLOCK_RELEASE;
    throw CosN::UnsupportedQoS(eseq);
  }
  _qosprop->set_qos(r_qos); 
  if (RDIRptNotifQoS) {
    RDIRptLogger(l, RDIRptNotifQoS_nm);
    l.str << _my_name << ": NotifQoS param(s) modified as follows\n";
    for (unsigned int i = 0; i < r_qos.length(); i++) {
      l.str << "  " << r_qos[i].name << " set to "; RDI_pp_any(l.str, r_qos[i].value); l.str << '\n';
    }
    l.str << '\n';
  }
  _qos_changed(0);
  RDI_OPLOCK_RELEASE;
}

void
ConsumerAdmin_i::validate_qos(const CosN::QoSProperties& r_qos,
			      CosN_NamedPropertyRangeSeq_outarg rseq
			      WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosN::PropertyErrorSeq eseq;
  CORBA::Boolean  subobjs = (_num_proxies ? 1 : 0);

  rseq = new CosNotification::NamedPropertyRangeSeq();
  RDI_AssertAllocThrowNo(rseq, "Memory allocation failed - NamedPropertyRangeSeq\n");
  if (! RDI_NotifQoS::validate(r_qos, *_qosprop, RDI_C_ADMIN, eseq, *rseq, subobjs)) {
    RDI_OPLOCK_RELEASE;
    throw CosN::UnsupportedQoS(eseq);
  }
  RDI_OPLOCK_RELEASE;
}

CosNF::FilterID
ConsumerAdmin_i::add_filter(CosNF::Filter_ptr fltr WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  // If we have registered interest in specific event types, using
  // 'subscription_change()', we have to cancel it at this point..
  if ( _rqstypes.length() != 0 ) {
    CosN::EventTypeSeq added; added.length(0);
    (void) _channel->update_mapping(added, _rqstypes, this, 0);
    _rqstypes.length(0);
  }
  CosNF::FilterID res = _fa_helper.add_filter_i(fltr, (RDINotifySubscribe_ptr) this);
  RDI_OPLOCK_RELEASE;
  return res;
}

void
ConsumerAdmin_i::remove_filter(CosNF::FilterID fltrID WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  _fa_helper.remove_filter(fltrID);
  RDI_OPLOCK_RELEASE;
}

CosNF::Filter_ptr
ConsumerAdmin_i::get_filter(CosNF::FilterID fltrID WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNF::Filter_ptr res = _fa_helper.get_filter(fltrID);
  RDI_OPLOCK_RELEASE;
  return res;
}

CosNF::FilterIDSeq*
ConsumerAdmin_i::get_all_filters( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNF::FilterIDSeq* res = _fa_helper.get_all_filters();
  RDI_OPLOCK_RELEASE;
  return res;
}

void
ConsumerAdmin_i::remove_all_filters( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  _fa_helper.remove_all_filters();
  RDI_OPLOCK_RELEASE;
}


// -------------------- ** Local-only routines ** ----------------------------


void
ConsumerAdmin_i::_qos_changed(CORBA::Boolean getlock)
{
  if (getlock) RDI_OPLOCK_ACQUIRE(return);

  RDI_HashCursor<CosNA::ProxyID, ProxyPushSupplier_i *>           apushcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullSupplier_i *>           apullcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushSupplier_i *> spushcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullSupplier_i *> spullcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushSupplier_i *>   bpushcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullSupplier_i *>   bpullcur;

  for ( apushcur = _prx_any_push.cursor(); apushcur.is_valid(); ++apushcur ) {
    apushcur.val()->_qos_changed(1);
  }
  for ( apullcur = _prx_any_pull.cursor(); apullcur.is_valid(); ++apullcur ) {
    apullcur.val()->_qos_changed(1);
  }
  for ( spushcur = _prx_struc_push.cursor(); spushcur.is_valid(); ++spushcur ) {
    spushcur.val()->_qos_changed(1);
  }
  for ( spullcur = _prx_struc_pull.cursor(); spullcur.is_valid(); ++spullcur ) {
    spullcur.val()->_qos_changed(1);
  }
  for ( bpushcur = _prx_batch_push.cursor(); bpushcur.is_valid(); ++bpushcur ) {
    bpushcur.val()->_qos_changed(1);
  }
  for ( bpullcur = _prx_batch_pull.cursor(); bpullcur.is_valid(); ++bpullcur ) {
    bpullcur.val()->_qos_changed(1);
  }
  if (getlock) RDI_OPLOCK_RELEASE;
}

void
ConsumerAdmin_i::disconnect_clients_and_dispose()
{
  RDI_OPLOCK_ACQUIRE(return);
  _disconnect_clients_and_dispose(0);
}

// *** requires OPLOCK already held
//  If update_channel is true, the admin unregisters itself
void
ConsumerAdmin_i::_disconnect_clients_and_dispose(CORBA::Boolean update_channel)
{
  if (_disposed) {
    RDIDbgCAdmLog("ConsumerAdmin_i::_disconnect_clients_and_dispose() called more than once\n");
    RDI_OPLOCK_RELEASE;
    return;
  }
  _disposed = 1; // acts as guard: the following is executed by only one thread 
  if (update_channel) {
    // Unregister from the channel -- must be done first
    // to avoid deadlock when channel is being destroyed

    // never hold oplock when passing 'this' to a parent object
    RDI_OPLOCK_BUMP_RELEASE;
    _channel->unregister(this);
    RDI_OPLOCK_REACQUIRE_DEBUMP(RDI_METHOD_RETURN_NULL(RDIDbgCAdm, RDIDbgCAdm_nm, "ConsumerAdmin_i::_disconnect_client_and_dispose [**unexpected REACQUIRE failure**]\n"));
  }

  RDIDbgCAdmLog("Destruction of ConsumerAdmin " << (void*)this << " [" << _serial << "]\n");

  unsigned int i;
  RDI_ListCursor<EventProxyPushSupplier_i *>                cpushcur;
  RDI_ListCursor<EventProxyPullSupplier_i *>                cpullcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPushSupplier_i *>           apushcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullSupplier_i *>           apullcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushSupplier_i *> spushcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullSupplier_i *> spullcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushSupplier_i *>   bpushcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullSupplier_i *>   bpullcur;

  cpushcur = _cosevent_push.cursor();
  for (i=0; i < _cosevent_push.length(); ++i, ++cpushcur) {
    EventProxyPushSupplier_i* prx = *cpushcur;
    _removed_push_proxy(prx);
    prx->disconnect_client_and_dispose(0);
  }
  cpullcur = _cosevent_pull.cursor();
  for (i=0; i < _cosevent_pull.length(); ++i, ++cpullcur) {
    _removed_pull_proxy();
    (*cpullcur)->disconnect_client_and_dispose(0);
  }
  for ( apushcur = _prx_any_push.cursor(); apushcur.is_valid(); ++apushcur ) {
    ProxyPushSupplier_i* prx = apushcur.val();
    _removed_push_proxy(prx);
    prx->disconnect_client_and_dispose(0);
  }
  for ( apullcur = _prx_any_pull.cursor(); apullcur.is_valid(); ++apullcur ) {
    _removed_pull_proxy();
    apullcur.val()->disconnect_client_and_dispose(0);
  }
  for ( spushcur = _prx_struc_push.cursor(); spushcur.is_valid(); ++spushcur ) {
    StructuredProxyPushSupplier_i* prx = spushcur.val();
    _removed_push_proxy(prx);
    prx->disconnect_client_and_dispose(0);
  }
  for ( spullcur = _prx_struc_pull.cursor(); spullcur.is_valid(); ++spullcur ) {
    _removed_pull_proxy();
    spullcur.val()->disconnect_client_and_dispose(0);
  }
  for ( bpushcur = _prx_batch_push.cursor(); bpushcur.is_valid(); ++bpushcur ) {
    SequenceProxyPushSupplier_i* prx = bpushcur.val();
    _removed_push_proxy(prx);
    prx->disconnect_client_and_dispose(0);
  }
  for ( bpullcur = _prx_batch_pull.cursor(); bpullcur.is_valid(); ++bpullcur ) {
    _removed_pull_proxy();
    bpullcur.val()->disconnect_client_and_dispose(0);
  }
  _cosevent_push.drain();  _cosevent_pull.drain();
  _prx_any_push.clear();   _prx_any_pull.clear();
  _prx_struc_push.clear(); _prx_struc_pull.clear();
  _prx_batch_push.clear(); _prx_batch_pull.clear();

  if ( ! CORBA::is_nil(_prio_filter) ) CORBA::release(_prio_filter);
  if ( ! CORBA::is_nil(_life_filter) ) CORBA::release(_life_filter);
  if ( _rqstypes.length() == 0 ) {
    _fa_helper.remove_all_filters();
  } else {
    CosN::EventTypeSeq added; added.length(0);
    (void) _channel->update_mapping(added, _rqstypes, this, 0);
  }
  if ( _qosprop ) {
    delete _qosprop; _qosprop = 0;
  }
  RDIDbgCAdmLog("ConsumerAdmin @ " << (void*)this << " [" << _serial << "] destroyed\n");
  RDI_OPLOCK_DISPOSE_IMPL_AND_RELEASE;
}

// NB ** Assumes oplock is held **
void
ConsumerAdmin_i::_removed_push_proxy(RDIProxyPushSupplier* proxy)
{
  _num_proxies -= 1;
  _prxy_exit.set_curtime();
  _channel->decr_consumers();
  RDI_NotifyConsumer* cpc = _channel->push_consumer();
  if ( cpc ) {       // Deregister Proxy
    cpc->remove_proxy(proxy);
  }
}

// NB ** Assumes oplock is held **
void
ConsumerAdmin_i::_removed_pull_proxy()
{
  _num_proxies -= 1;
  _prxy_exit.set_curtime();
  _channel->decr_consumers();
}

void
ConsumerAdmin_i::remove_proxy(EventProxyPushSupplier_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDI_ListCursor<EventProxyPushSupplier_i *> lc;
  unsigned int    indx = 0;

  for ( lc = _cosevent_push.cursor(); indx < _cosevent_push.length(); indx++ ) {
    EventProxyPushSupplier_i* proxy = *lc;
    if ( proxy == prx ) {
      _cosevent_push.remove(lc);
      _removed_push_proxy(prx);
      break;
    }
    ++lc;
  }
  RDI_OPLOCK_RELEASE;
}

void
ConsumerAdmin_i::remove_proxy(EventProxyPullSupplier_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDI_ListCursor<EventProxyPullSupplier_i *> lc;
  unsigned int    indx = 0;

  for ( lc = _cosevent_pull.cursor(); indx < _cosevent_pull.length(); indx++ ) {
    EventProxyPullSupplier_i* proxy = *lc;
    if (  proxy == prx ) {
      _cosevent_pull.remove(lc);
      _removed_pull_proxy();
      break;
    }
    ++lc;
  }
  RDI_OPLOCK_RELEASE;
}

void
ConsumerAdmin_i::remove_proxy(ProxyPushSupplier_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgCAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_any_push.exists(prx->_proxy_id()) ) {
    _prx_any_push.remove(prx->_proxy_id());
    _removed_push_proxy(prx);
  } else {
    RDIDbgCAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
ConsumerAdmin_i::remove_proxy(ProxyPullSupplier_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgCAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_any_pull.exists(prx->_proxy_id()) ) {
    _prx_any_pull.remove(prx->_proxy_id());
    _removed_pull_proxy();
  } else {
    RDIDbgCAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
ConsumerAdmin_i::remove_proxy(StructuredProxyPushSupplier_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgCAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_struc_push.exists(prx->_proxy_id()) ) {
    _prx_struc_push.remove(prx->_proxy_id());
    _removed_push_proxy(prx);
  } else {
    RDIDbgCAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
ConsumerAdmin_i::remove_proxy(StructuredProxyPullSupplier_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgCAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_struc_pull.exists(prx->_proxy_id()) ) {
    _prx_struc_pull.remove(prx->_proxy_id());
    _removed_pull_proxy();
  } else {
    RDIDbgCAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
ConsumerAdmin_i::remove_proxy(SequenceProxyPushSupplier_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgCAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_batch_push.exists(prx->_proxy_id()) ) {
    _prx_batch_push.remove(prx->_proxy_id());
    _removed_push_proxy(prx);
  } else {
    RDIDbgCAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
ConsumerAdmin_i::remove_proxy(SequenceProxyPullSupplier_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgCAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_batch_pull.exists(prx->_proxy_id()) ) {
    _prx_batch_pull.remove(prx->_proxy_id());
    _removed_pull_proxy();
  } else {
    RDIDbgCAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

// OPLOCK is acquired/released by caller [EventChannel_i::admin_dispatch()]
void
ConsumerAdmin_i::dispatch_event(RDI_StructuredEvent* event)
{
  RDI_ListCursor<EventProxyPushSupplier_i *> psc;
  RDI_ListCursor<EventProxyPullSupplier_i *> plc;
  unsigned int i=0;
  for (i=0,psc=_cosevent_push.cursor(); i<_cosevent_push.length(); ++i,++psc) {
    (*psc)->add_event(event);
  }
  for (i=0,plc=_cosevent_pull.cursor(); i<_cosevent_pull.length(); ++i,++plc) {
    (*plc)->add_event(event);
  }
}

// OPLOCK is acquired/released by caller
//  [EventChannel_i::admin_dispatch() or EventChannel_i::proxy_dispatch()]
void
ConsumerAdmin_i::dispatch_event(RDI_StructuredEvent* event,
				RDI_FilterState_t    astat, 
				RDI_TypeMap*         tmap)
{
  RDI_HashCursor<CosNA::ProxyID, ProxyPushSupplier_i *>           apushcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullSupplier_i *>           apullcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushSupplier_i *> spushcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullSupplier_i *> spullcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushSupplier_i *>   bpushcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullSupplier_i *>   bpullcur;
  RDI_TypeMap::FNode_t* fnode=0;
  CORBA::Boolean match_found=0;
  RDI_TypeMap::FList_t flist;
 
  // When consumers use 'subscription_change()'  to register 
  // interest in specific event types, we have 'has_filters()' return true
  // so that we execute the filter evaluation logic. However, there are no
  // filters registered and, hence, we need the '!fnode->_fltr' below...

  if ( _num_proxies == 0 )
    return;

  const char* dname = event->get_domain_name();
  const char* tname = event->get_type_name();

  // Evaluate Filters for consumers using ProxyPushSupplier_i

  for (apushcur = _prx_any_push.cursor(); apushcur.is_valid(); ++apushcur ) {
    if ( astat == OrMatch ) {
      apushcur.val()->add_event(event);
    } else if ( ! apushcur.val()->has_filters() ) {
      if ( (astat == NoFilters) || (astat == AndMatch) ) {
	apushcur.val()->add_event(event);
      }
    } else {
      tmap->lookup_begin(dname, tname, apushcur.val(), flist);
      if ( ! flist._star_star && ! flist._domn_star &&
	   ! flist._star_type && ! flist._domn_type ) {
	if ( astat == OrMatch ) {
	  apushcur.val()->add_event(event);
	}
      } else {
	match_found = 0;
	for ( fnode=flist._star_star; fnode; fnode=fnode->_next ) {
	  if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	    match_found = 1;
	    break;
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_star; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._star_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}

	if ( match_found ) {
	  apushcur.val()->add_event(event);
	}
      }
      tmap->lookup_end();
    }
  }

  // Evaluate Filters for consumers using ProxyPullSupplier_i

  for (apullcur = _prx_any_pull.cursor(); apullcur.is_valid(); ++apullcur ) {
    if ( astat == OrMatch ) {
      apullcur.val()->add_event(event);
    } else if ( ! apullcur.val()->has_filters() ) {
      if ( (astat == NoFilters) || (astat == AndMatch) ) {
	apullcur.val()->add_event(event);
      }
    } else {
      tmap->lookup_begin(dname, tname, apullcur.val(), flist);
      if ( ! flist._star_star && ! flist._domn_star &&
	   ! flist._star_type && ! flist._domn_type ) {
	if ( astat == OrMatch ) {
	  apullcur.val()->add_event(event);
	}
      } else {
	match_found = 0;
	for ( fnode=flist._star_star; fnode; fnode=fnode->_next ) {
	  if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	    match_found = 1;
	    break;
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_star; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._star_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}

	if ( match_found ) {
	  apullcur.val()->add_event(event);
	}
      }
      tmap->lookup_end();
    }
  }

  // Evaluate Filters for consumers using StructuredProxyPushSupplier_i

  for ( spushcur=_prx_struc_push.cursor(); spushcur.is_valid(); ++spushcur ) {
    if ( astat == OrMatch ) {
      spushcur.val()->add_event(event);
    } else if ( ! spushcur.val()->has_filters() ) {
      if ( (astat == NoFilters) || (astat == AndMatch) ) {
	spushcur.val()->add_event(event);
      }
    } else {
      tmap->lookup_begin(dname, tname, spushcur.val(), flist);
      if ( ! flist._star_star && ! flist._domn_star &&
	   ! flist._star_type && ! flist._domn_type ) {
	if ( astat == OrMatch ) {
	  spushcur.val()->add_event(event);
	}
      } else {
	match_found = 0;
	for ( fnode=flist._star_star; fnode; fnode=fnode->_next ) {
	  if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	    match_found = 1;
	    break;
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_star; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._star_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}

	if ( match_found ) {
	  spushcur.val()->add_event(event);
	}
      }
      tmap->lookup_end();
    }
  }

  // Evaluate Filters for consumers using StructuredProxyPullSupplier_i

  for ( spullcur=_prx_struc_pull.cursor(); spullcur.is_valid(); ++spullcur ) {
    if ( astat == OrMatch ) {
      spullcur.val()->add_event(event);
    } else if ( ! spullcur.val()->has_filters() ) {
      if ( (astat == NoFilters) || (astat == AndMatch) ) {
	spullcur.val()->add_event(event);
      }
    } else {
      tmap->lookup_begin(dname, tname, spullcur.val(), flist);
      if ( ! flist._star_star && ! flist._domn_star &&
	   ! flist._star_type && ! flist._domn_type ) {
	if ( astat == OrMatch ) {
	  spullcur.val()->add_event(event);
	}
      } else {
	match_found = 0;
	for ( fnode=flist._star_star; fnode; fnode=fnode->_next ) {
	  if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	    match_found = 1;
	    break;
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_star; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._star_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}

	if ( match_found ) {
	  spullcur.val()->add_event(event);
	}
      }
      tmap->lookup_end();
    }
  }

  // Evaluate Filters for consumers using SequenceProxyPushSupplier_i

  for ( bpushcur=_prx_batch_push.cursor(); bpushcur.is_valid(); ++bpushcur ) {
    if ( astat == OrMatch ) {
      bpushcur.val()->add_event(event);
    } else if ( ! bpushcur.val()->has_filters() ) {
      if ( (astat == NoFilters) || (astat == AndMatch) ) {
	bpushcur.val()->add_event(event);
      }
    } else {
      tmap->lookup_begin(dname, tname, bpushcur.val(), flist);
      if ( ! flist._star_star && ! flist._domn_star &&
	   ! flist._star_type && ! flist._domn_type ) {
	if ( astat == OrMatch ) {
	  bpushcur.val()->add_event(event);
	}
      } else {
	match_found = 0;
	for ( fnode=flist._star_star; fnode; fnode=fnode->_next ) {
	  if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	    match_found = 1;
	    break;
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_star; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._star_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}

	if ( match_found ) {
	  bpushcur.val()->add_event(event);
	}
      }
      tmap->lookup_end();
    }
  }

  // Evaluate Filters for consumers using SequenceProxyPullSupplier_i

  for ( bpullcur=_prx_batch_pull.cursor(); bpullcur.is_valid(); ++bpullcur ) {
    if ( astat == OrMatch ) {
      bpullcur.val()->add_event(event);
    } else if ( ! bpullcur.val()->has_filters() ) {
      if ( (astat == NoFilters) || (astat == AndMatch) ) {
	bpullcur.val()->add_event(event);
      }
    } else {
      tmap->lookup_begin(dname, tname, bpullcur.val(), flist);
      if ( ! flist._star_star && ! flist._domn_star &&
	   ! flist._star_type && ! flist._domn_type ) {
	if ( astat == OrMatch ) {
	  bpullcur.val()->add_event(event);
	}
      } else {
	match_found = 0;
	for ( fnode=flist._star_star; fnode; fnode=fnode->_next ) {
	  if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	    match_found = 1;
	    break;
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_star; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._star_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}
	if ( ! match_found ) {
	  for ( fnode=flist._domn_type; fnode; fnode=fnode->_next ) {
	    if (!fnode->_fltr || fnode->_fltr->rdi_match(event,_channel)) {
	      match_found = 1;
	      break;
	    }
	  }
	}

	if ( match_found ) {
	  bpullcur.val()->add_event(event);
	}
      }
      tmap->lookup_end();
    }
  }
}

// ------------------------------------------------------------- //
// SupplierAdmin_i implementation                                //
// ------------------------------------------------------------- //

SupplierAdmin_i::SupplierAdmin_i(EventChannel_i* channel,
				 CosNA::InterFilterGroupOperator op,
				 const CosNA::AdminID& serial) :
  _oplockptr(0), _my_name(channel->L_my_name()), _disposed(0),
  _fa_helper(), _channel(channel), _qosprop(0), 
  _serial(serial), _and_or_oper(op), _prx_serial(1), 
  _num_proxies(0), 
  _evtypes(RDI_EventType::hash, RDI_EventType::rank),
  _cosevent_push(), _cosevent_pull(),
  _prx_any_push(RDI_ULongHash, RDI_ULongRank),
  _prx_any_pull(RDI_ULongHash, RDI_ULongRank),
  _prx_struc_push(RDI_ULongHash, RDI_ULongRank),
  _prx_struc_pull(RDI_ULongHash, RDI_ULongRank),
  _prx_batch_push(RDI_ULongHash, RDI_ULongRank),
  _prx_batch_pull(RDI_ULongHash, RDI_ULongRank)
{
  RDI_OPLOCK_INIT;
  _qosprop = new RDI_NotifQoS(_channel->qos_properties());
  RDI_AssertAllocThrowNo(_qosprop, "Memory allocation failure - RDI_NotifQoS\n");
  _time_born.set_curtime();
  _prxy_exit = _time_born;
  char buf[20];
  sprintf(buf, "sadmin%ld", serial);
  _my_name.length(_my_name.length()+1);
  _my_name[_my_name.length()-1] = (const char*)buf;
  WRAPPED_REGISTER_IMPL2(this, &_my_name);
}

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

// Not needed?  See comment for ConsumerAdmin_i::lock/unlock
CORBA::Boolean SupplierAdmin_i::lock()   { RDI_OPLOCK_ACQUIRE_BUMP(return 0); return 1; }
void           SupplierAdmin_i::unlock() { RDI_OPLOCK_DEBUMP_RELEASE; }

CosNA::AdminID
SupplierAdmin_i::MyID( WRAPPED_IMPLARG_VOID ) 
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNA::AdminID res = _serial;
  RDI_OPLOCK_RELEASE;
  return res;
}

CosNA::EventChannel_ptr
SupplierAdmin_i::MyChannel( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNA::EventChannel_ptr res = WRAPPED_IMPL2OREF(CosNA::EventChannel, _channel);
  RDI_OPLOCK_RELEASE;
  return res;
}

CosNA::InterFilterGroupOperator
SupplierAdmin_i::MyOperator( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNA::InterFilterGroupOperator res = _and_or_oper;
  RDI_OPLOCK_RELEASE;
  return res;
}

CosEvCA::ProxyPushConsumer_ptr 
SupplierAdmin_i::obtain_push_consumer( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosEvCA::ProxyPushConsumer_ptr res = CosEvCA::ProxyPushConsumer::_nil();
  EventProxyPushConsumer_i* prx=0;
  RDIDbgSAdmLog("CosEvent ProxyPushConsumer creation requested\n");
  if ( ! _channel->incr_suppliers() ) {
    // return null res
  } else if ( ! (prx = new EventProxyPushConsumer_i(this, _channel, _prx_serial)) ) {
    _channel->decr_suppliers();
    // return null res
  } else if ( _cosevent_push.insert_tail(prx) != 0 ) {
    _channel->decr_suppliers();
    prx->disconnect_client_and_dispose(0);
    // return null res
  } else { // return valid res
    _prx_serial++;
    _num_proxies++;
    res = WRAPPED_IMPL2OREF(CosEvCA::ProxyPushConsumer, prx);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

CosEvCA::ProxyPullConsumer_ptr
SupplierAdmin_i::obtain_pull_consumer( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosEvCA::ProxyPullConsumer_ptr res = 
    CosEvCA::ProxyPullConsumer::_nil();
  EventProxyPullConsumer_i* prx=0;
  RDIDbgSAdmLog("CosEvent ProxyPullConsumer creation requested\n");
  if ( ! _channel->incr_suppliers() ) {
    // return null res
  } else if ( ! (prx = new EventProxyPullConsumer_i(this, _channel, _prx_serial)) ) {
    _channel->decr_suppliers();
    // return null res
  } else if ( _cosevent_pull.insert_tail(prx) != 0 ) {
    _channel->decr_suppliers();
    prx->disconnect_client_and_dispose(0);
    // return null res
  } else { // return valid res
    _prx_serial++;
    _num_proxies++;
    RDI_PullSupplier* cps = _channel->pull_supplier();
    if ( cps ) {      // Register Pull Proxy
      cps->insert_proxy(prx);
    }
    res = WRAPPED_IMPL2OREF(CosEvCA::ProxyPullConsumer, prx);
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

// Called externally -- not clear that suppliers should be doing this
void
SupplierAdmin_i::offer_change(const CosN::EventTypeSeq& added,
			      const CosN::EventTypeSeq& deled   WRAPPED_IMPLARG ) {
  // this code assumes it is OK for suppliers to do this
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed

  CosN::EventTypeSeq added_copy = added;
  CosN::EventTypeSeq deled_copy = deled;

  // First, validate the entries in the provided lists to make sure
  // that correct values are provided for these event types .......

  CORBA::ULong vl=0;
  if ( ! RDI_EventType::valid_sequence(added_copy, vl) ) {
    RDI_OPLOCK_RELEASE;
    throw CosNC::InvalidEventType(added[vl]);
  }
  if ( ! RDI_EventType::valid_sequence(deled_copy, vl) ) {
    RDI_OPLOCK_RELEASE;
    throw CosNC::InvalidEventType(deled[vl]);
  }

  // The same call a proxy uses
  _propagate_offer_change(added_copy, deled_copy);
}

CosNA::ProxyIDSeq*
SupplierAdmin_i::pull_consumers( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  RDI_HashCursor<CosNA::ProxyID, ProxyPullConsumer_i *>           ac;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullConsumer_i *> sc;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullConsumer_i *>   bc;
  unsigned long num;
  CosNA::ProxyIDSeq*  seq = new CosNA::ProxyIDSeq();
  RDI_AssertAllocThrowNo(seq, "Memory allocation failed -- ProxyIDSeq\n");
  num = _prx_any_pull.length() +
    _prx_struc_pull.length() + _prx_batch_pull.length();
  seq->length(num);
  for ( num=0, ac = _prx_any_pull.cursor(); ac.is_valid(); ++ac, ++num ) {
    (*seq)[num] = ac.key();
  }
  for ( sc = _prx_struc_pull.cursor(); sc.is_valid(); ++sc, ++num ) {
    (*seq)[num] = sc.key();
  }
  for ( bc = _prx_batch_pull.cursor(); bc.is_valid(); ++bc, ++num ) {
    (*seq)[num] = bc.key();
  }
  RDI_OPLOCK_RELEASE;
  return seq;
}

CosNA::ProxyIDSeq*
SupplierAdmin_i::push_consumers( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  RDI_HashCursor<CosNA::ProxyID, ProxyPushConsumer_i *>           ac;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushConsumer_i *> sc;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushConsumer_i *>  bc;
  unsigned long num;
  CosNA::ProxyIDSeq*  seq = new CosNA::ProxyIDSeq();
  RDI_AssertAllocThrowNo(seq, "Memory allocation failed -- ProxyIDSeq\n");
  num = _prx_any_push.length() +
    _prx_struc_push.length() + _prx_batch_push.length();
  seq->length(num);
  for ( num=0, ac = _prx_any_push.cursor(); ac.is_valid(); ++ac, ++num ) {
    (*seq)[num] = ac.key();
  }
  for ( sc = _prx_struc_push.cursor(); sc.is_valid(); ++sc, ++num ) {
    (*seq)[num] = sc.key();        
  }
  for ( bc = _prx_batch_push.cursor(); bc.is_valid(); ++bc, ++num ) {
    (*seq)[num] = bc.key();
  }
  RDI_OPLOCK_RELEASE;
  return seq;
}

CosNA::ProxyConsumer_ptr
SupplierAdmin_i::get_proxy_consumer(CosNA::ProxyID proxy_id   WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  ProxyPushConsumer_i*           apush=0;
  ProxyPullConsumer_i*           apull=0;
  StructuredProxyPushConsumer_i* spush=0;
  SequenceProxyPushConsumer_i*   bpush=0;
  StructuredProxyPullConsumer_i* spull=0;
  SequenceProxyPullConsumer_i*   bpull=0;

  CosNA::ProxyConsumer_ptr res = CosNA::ProxyConsumer::_nil();
  if ( proxy_id <= _prx_serial ) {
    if ( _prx_any_push.lookup(proxy_id, apush) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, apush);
    } else if ( _prx_any_pull.lookup(proxy_id, apull) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, apull);
    } else if ( _prx_struc_push.lookup(proxy_id, spush) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, spush);
    } else if ( _prx_struc_pull.lookup(proxy_id, spull) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, spull);
    } else if ( _prx_batch_push.lookup(proxy_id, bpush) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, bpush);
    } else if ( _prx_batch_pull.lookup(proxy_id, bpull) ) {
      res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, bpull);
    }
  }
  RDI_OPLOCK_RELEASE;
  if (CORBA::is_nil(res)) {
    throw CosNA::ProxyNotFound();
  }
  return res;
}

CosNA::ProxyConsumer_ptr
SupplierAdmin_i::obtain_notification_pull_consumer
(CosNA::ClientType ctype, CosNA::ProxyID& proxy_id   WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed

  RDIDbgSAdmLog("Creation of a new ProxyConsumer -- type " << ctype << '\n');

  // Check with  the channel if we are about to exceed the maximum
  // number of suppliers supported by the channel
  if ( ! _channel->incr_suppliers() ) {
    CosNA::AdminLimit limit;
    RDIDbgSAdmLog("Supplier limit has been reached creation failed\n");
    limit.name    = (const char *) "MaxSuppliers";
    limit.value <<= _channel->max_suppliers();
    RDI_OPLOCK_RELEASE;
    throw CosNA::AdminLimitExceeded(limit);
  }

  CosNA::ProxyConsumer_ptr res = CosNA::ProxyConsumer::_nil();
  if ( ctype == CosNA::ANY_EVENT ) {
    ProxyPullConsumer_i* prx = 0;
    if ( ! (prx = new ProxyPullConsumer_i(this, _channel, _prx_serial)) ) {
      _channel->decr_suppliers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_any_pull.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_suppliers(); 
      } else {
	RDI_PullSupplier* cps = _channel->pull_supplier();
	if ( cps ) {   // Register Pull Proxy
	  cps->insert_proxy(prx);
	}
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, prx);
      }
    }
  } else if ( ctype == CosNA::STRUCTURED_EVENT ) {
    StructuredProxyPullConsumer_i* prx = 
      new StructuredProxyPullConsumer_i(this,_channel,_prx_serial);
    if ( ! prx ) {
      _channel->decr_suppliers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_struc_pull.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_suppliers();
      } else {
	RDI_PullSupplier* cps = _channel->pull_supplier();
	if ( cps ) {   // Register Pull Proxy
	  cps->insert_proxy(prx);
	}
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, prx);
      }
    }
  } else if ( ctype == CosNA::SEQUENCE_EVENT ) {
    SequenceProxyPullConsumer_i* prx =
      new SequenceProxyPullConsumer_i(this,_channel,_prx_serial);
    if ( ! prx ) {
      _channel->decr_suppliers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_batch_pull.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_suppliers();
      }  else {
	RDI_PullSupplier* cps = _channel->pull_supplier();
	if ( cps ) {     // Register Pull Proxy
	  cps->insert_proxy(prx);
	}
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, prx);
      }
    }
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

CosNA::ProxyConsumer_ptr
SupplierAdmin_i::obtain_notification_push_consumer
(CosNA::ClientType ctype, CosNA::ProxyID& proxy_id   WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed

  RDIDbgSAdmLog("Creation of a new ProxyConsumer -- type " << ctype << '\n');

  // Check with  the channel if we are about to exceed the maximum
  // number of suppliers supported by the channel
  if ( ! _channel->incr_suppliers() ) {
    CosNA::AdminLimit limit;
    RDIDbgSAdmLog("Supplier limit has been reached creation failed\n");
    limit.name    = (const char *) "MaxSuppliers";
    limit.value <<= _channel->max_suppliers();
    RDI_OPLOCK_RELEASE;
    throw CosNA::AdminLimitExceeded(limit);
  }

  CosNA::ProxyConsumer_ptr res = CosNA::ProxyConsumer::_nil();
  if ( ctype == CosNA::ANY_EVENT ) {
    ProxyPushConsumer_i* prx = 0;
    if ( ! (prx = new ProxyPushConsumer_i(this, _channel, _prx_serial)) ) {
      _channel->decr_suppliers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_any_push.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_suppliers(); 
      } else {
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, prx);
      }
    }
  } else if ( ctype == CosNA::STRUCTURED_EVENT ) {
    StructuredProxyPushConsumer_i* prx =
      new StructuredProxyPushConsumer_i(this,_channel,_prx_serial);
    if ( ! prx ) {
      _channel->decr_suppliers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_struc_push.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_suppliers();
      } else {
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, prx);
      }
    }
  } else if ( ctype == CosNA::SEQUENCE_EVENT ) {
    SequenceProxyPushConsumer_i* prx = 
      new SequenceProxyPushConsumer_i(this,_channel,_prx_serial);
    if ( ! prx ) {
      _channel->decr_suppliers();
    } else {
      proxy_id = _prx_serial++;
      if ( _prx_batch_push.insert(proxy_id, prx) != 0 ) {
	prx->disconnect_client_and_dispose(0);
	_channel->decr_suppliers();
      } else {
	_num_proxies += 1;
	res = WRAPPED_IMPL2OREF(CosNA::ProxyConsumer, prx);
      }
    }
  }
  RDI_OPLOCK_RELEASE;
  return res;
}

CORBA::Boolean
SupplierAdmin_i::safe_cleanup( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( _serial == 0 ) { // Default SupplierAdmin not a cleanup candidate
    RDI_OPLOCK_RELEASE;
    return 0;
  }
  if (_num_proxies > 0) { // Admin with proxies not a cleanup candidate
    RDI_OPLOCK_RELEASE;
    return 0;
  }
  _disconnect_clients_and_dispose(1); // 1 means update channel
  return 1;
}

void
SupplierAdmin_i::destroy( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if ( _serial == 0 ) {
    RDIDbgSAdmLog("Default SupplierAdmin is not allowed to be destroyed!\n");
    RDI_OPLOCK_RELEASE;
    return;
  }
  _disconnect_clients_and_dispose(1); // 1 means update channel
}

CosN::QoSProperties*
SupplierAdmin_i::get_qos( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  //RDI_NotifQoS*   qosp = qos_properties();
  //RDI_Assert(qosp, "Event Channel should have RDI_NotifQoS set\n");
  //CosN::QoSProperties* res = qosp->get_qos(RDI_C_ADMIN);
  CosN::QoSProperties* res = _qosprop->get_qos(RDI_C_ADMIN);
  RDI_OPLOCK_RELEASE;
  return res;
}

void
SupplierAdmin_i::set_qos(const CosN::QoSProperties& r_qos   WRAPPED_IMPLARG )
{
  if (r_qos.length() == 0) return;
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosN::PropertyErrorSeq eseq;
  CosN::NamedPropertyRangeSeq rseq;
  CORBA::Boolean  subobjs = (_num_proxies ? 1 : 0);
  
  if (! RDI_NotifQoS::validate(r_qos, *_qosprop, RDI_C_ADMIN, eseq, rseq, subobjs)) {
    RDI_OPLOCK_RELEASE;
    throw CosN::UnsupportedQoS(eseq);
  }
  _qosprop->set_qos(r_qos);
  if (RDIRptNotifQoS) {
    RDIRptLogger(l, RDIRptNotifQoS_nm);
    l.str << _my_name << ": NotifQoS param(s) modified as follows\n";
    for (unsigned int i = 0; i < r_qos.length(); i++) {
      l.str << "  " << r_qos[i].name << " set to "; RDI_pp_any(l.str, r_qos[i].value); l.str << '\n';
    }
    l.str << '\n';
  }
  RDI_OPLOCK_RELEASE;
}

void
SupplierAdmin_i::validate_qos(const CosN::QoSProperties& r_qos,
			      CosN_NamedPropertyRangeSeq_outarg rseq 
			      WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosN::PropertyErrorSeq eseq;
  CORBA::Boolean  subobjs = (_num_proxies ? 1 : 0);

  rseq = new CosNotification::NamedPropertyRangeSeq();
  RDI_AssertAllocThrowNo(rseq, "Memory allocation failed - NamedPropertyRangeSeq\n");
  if (! RDI_NotifQoS::validate(r_qos, *_qosprop, RDI_C_ADMIN, eseq, *rseq, subobjs)) {
    RDI_OPLOCK_RELEASE;
    throw CosN::UnsupportedQoS(eseq);
  }
  RDI_OPLOCK_RELEASE;
}

CosNF::FilterID
SupplierAdmin_i::add_filter(CosNF::Filter_ptr fltr WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNF::FilterID res = _fa_helper.add_filter_i(fltr);
  RDI_OPLOCK_RELEASE;
  return res;
}

void
SupplierAdmin_i::remove_filter(CosNF::FilterID fltrID WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  _fa_helper.remove_filter(fltrID);
  RDI_OPLOCK_RELEASE;
}

CosNF::Filter_ptr
SupplierAdmin_i::get_filter(CosNF::FilterID fltrID WRAPPED_IMPLARG )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNF::Filter_ptr res = _fa_helper.get_filter(fltrID);
  RDI_OPLOCK_RELEASE;
  return res;
}

CosNF::FilterIDSeq*
SupplierAdmin_i::get_all_filters( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  CosNF::FilterIDSeq* res = _fa_helper.get_all_filters();
  RDI_OPLOCK_RELEASE;
  return res;
}

void
SupplierAdmin_i::remove_all_filters( WRAPPED_IMPLARG_VOID )
{
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  _fa_helper.remove_all_filters();
  RDI_OPLOCK_RELEASE;
}

// -------------------- ** Local-only routines ** ----------------------------

void
SupplierAdmin_i::propagate_offer_change(const CosN::EventTypeSeq& added,
					const CosN::EventTypeSeq& deled)
{
  RDI_OPLOCK_ACQUIRE(return);
  _propagate_offer_change(added, deled);
}

// NB ** Assumes oplock is held **
void
SupplierAdmin_i::_removed_push_proxy()
{
  _num_proxies -= 1;
  _prxy_exit.set_curtime();
  _channel->decr_suppliers();
}

// NB ** Assumes oplock is held **
void
SupplierAdmin_i::_removed_pull_proxy(RDIProxyPullConsumer* prx)
{
  _num_proxies -= 1;
  _prxy_exit.set_curtime();
  _channel->decr_suppliers();
  RDI_PullSupplier* cps = _channel->pull_supplier();
  if ( cps ) {    // Deregister Pull Proxy
    cps->remove_proxy(prx);
  }
}

void
SupplierAdmin_i::remove_proxy(EventProxyPushConsumer_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDI_ListCursor<EventProxyPushConsumer_i *> lc;
  unsigned int    indx = 0;
 
  for ( lc = _cosevent_push.cursor(); indx < _cosevent_push.length(); indx++ ) {
    EventProxyPushConsumer_i* proxy = *lc;
    if ( proxy == prx ) {
      _cosevent_push.remove(lc);
      _removed_push_proxy();
      break;
    }
    ++lc;
  }
  RDI_OPLOCK_RELEASE;
}

void
SupplierAdmin_i::remove_proxy(EventProxyPullConsumer_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDI_ListCursor<EventProxyPullConsumer_i *> lc;
  unsigned int    indx = 0;
 
  for ( lc = _cosevent_pull.cursor(); indx < _cosevent_pull.length(); indx++ ) {
    EventProxyPullConsumer_i* proxy = *lc;
    if ( proxy == prx ) {
      _cosevent_pull.remove(lc);
      _removed_pull_proxy(prx);
      break;
    }
    ++lc;
  }
  RDI_OPLOCK_RELEASE;
}

void
SupplierAdmin_i::remove_proxy(ProxyPushConsumer_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgSAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_any_push.exists(prx->_proxy_id()) ) {
    _prx_any_push.remove(prx->_proxy_id());
    _removed_push_proxy();
  } else {
    RDIDbgSAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
SupplierAdmin_i::remove_proxy(ProxyPullConsumer_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgSAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_any_pull.exists(prx->_proxy_id()) ) {
    _prx_any_pull.remove(prx->_proxy_id());
    _removed_pull_proxy(prx);
  } else {
    RDIDbgSAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
SupplierAdmin_i::remove_proxy(StructuredProxyPushConsumer_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgSAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_struc_push.exists(prx->_proxy_id()) ) {
    _prx_struc_push.remove(prx->_proxy_id());
    _removed_push_proxy();
  } else {
    RDIDbgSAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
SupplierAdmin_i::remove_proxy(StructuredProxyPullConsumer_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgSAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_struc_pull.exists(prx->_proxy_id()) ) {
    _prx_struc_pull.remove(prx->_proxy_id());
    _removed_pull_proxy(prx);
  } else {
    RDIDbgSAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
SupplierAdmin_i::remove_proxy(SequenceProxyPushConsumer_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgSAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_batch_push.exists(prx->_proxy_id()) ) {
    _prx_batch_push.remove(prx->_proxy_id());
    _removed_push_proxy();
  } else {
    RDIDbgSAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

void
SupplierAdmin_i::remove_proxy(SequenceProxyPullConsumer_i* prx)
{
  RDI_OPLOCK_ACQUIRE(return);
  RDIDbgSAdmLog("Remove proxy [" << prx->_proxy_id() << "] from admin " << (void*)this << '\n');
  if ( _prx_batch_pull.exists(prx->_proxy_id()) ) {
    _prx_batch_pull.remove(prx->_proxy_id());
    _removed_pull_proxy(prx);
  } else {
    RDIDbgSAdmLog("Invalid proxy [" << prx->_proxy_id() << "] provided\n");
  }
  RDI_OPLOCK_RELEASE;
}

CORBA::Boolean
SupplierAdmin_i::match_event(const CORBA::Any& event)
{
  RDI_OPLOCK_ACQUIRE(return 0);
  RDIDbgSAdmLog("XXX SupplierAdmin_i::match_event(any) called\n");
  CORBA::Boolean matched = 1;
  if ( _fa_helper.has_filters() ) {
    CosNF::FilterIDSeq* filterseq = _fa_helper.get_all_filters();
    CosNF::Filter_ptr   filter = CosNF::Filter::_nil();
    Filter_i* rdfilter = 0;
    matched = 0;
    for (CORBA::ULong ix=0; ix < filterseq->length(); ix++) {
      filter = _fa_helper.get_filter((*filterseq)[ix]);
      rdfilter = Filter_i::Filter2Filter_i(filter);
      if (rdfilter) // XXX
	RDIDbgSAdmLog("XXX SupplierAdmin::match_event(any) calling rdfilter->match_chan\n");
      else // XXX
	RDIDbgSAdmLog("XXX SupplierAdmin::match_event(any) calling filter->match\n");
      if ( (rdfilter && rdfilter->match_chan(event, _channel)) ||
	   (!rdfilter && filter->match(event)) ) {
	matched = 1;
	break;
      }
    }
    delete filterseq;
  }
  RDI_OPLOCK_RELEASE;
  return matched;
}

CORBA::Boolean
SupplierAdmin_i::match_event(const CosN::StructuredEvent* event, RDI_StructuredEvent* sevnt)
{
  RDI_OPLOCK_ACQUIRE(return 0);
  RDIDbgSAdmLog("XXX SupplierAdmin_i::match_event(event, sevnt) called\n");
  CORBA::Boolean matched = 1;
  if ( _fa_helper.has_filters() ) {
    CosNF::FilterIDSeq* filterseq = _fa_helper.get_all_filters();
    CosNF::Filter_ptr filter = CosNF::Filter::_nil();
    Filter_i* rdfilter = 0;
    matched = 0;
    for (CORBA::ULong ix=0; ix < filterseq->length(); ix++) {
      filter = _fa_helper.get_filter((*filterseq)[ix]);
      rdfilter = Filter_i::Filter2Filter_i(filter);
      if (rdfilter) // XXX
	RDIDbgSAdmLog("XXX SupplierAdmin_i::match_event calling rdfilter->rdi_match\n");
      else // XXX
	RDIDbgSAdmLog("XXX SupplierAdmin_i::match_event calling filter->match_structured\n");
      if ( (rdfilter && rdfilter->rdi_match(sevnt, _channel)) ||
	   (!rdfilter && filter->match_structured(*event)) ) {
	matched = 1;
	break;
      }
    }
    delete filterseq;
  }
  RDI_OPLOCK_RELEASE;
  return matched;
}

// ---------------------- ** Private routines ** ------------------------------

void
SupplierAdmin_i::disconnect_clients_and_dispose()
{
  RDI_OPLOCK_ACQUIRE(return);
  _disconnect_clients_and_dispose(0);
}

// ** requires OPLOCK already held
//  If update_channel is true, the admin unregisters itself and also
//  does a propagate_offer_change on the channel if there are
//  event types registered with the admin.
void
SupplierAdmin_i::_disconnect_clients_and_dispose(CORBA::Boolean update_channel)
{
  if (_disposed) {
    RDIDbgSAdmLog("SupplierAdmin_i::_disconnect_clients_and_dispose() called more than once\n");
    RDI_OPLOCK_RELEASE;
    return;
  }
  _disposed = 1; // acts as guard: the following is executed by only one thread 

  if (update_channel) {

    // never hold oplock when passing 'this' to a parent object
    RDI_OPLOCK_BUMP_RELEASE;
    _channel->unregister(this);
    RDI_OPLOCK_REACQUIRE_DEBUMP(RDI_METHOD_RETURN_NULL(RDIDbgSAdm, RDIDbgSAdm_nm, "SupplierAdmin_i::_disconnect_client_and_dispose [**unexpected REACQUIRE failure**]\n"));

    if ( _evtypes.length() ) {
      // There are event types registered with this administrative
      // object; update the channel's aggregate information about them.
      // NOTE: we release the lock before we inkoke channel operations so
      //      that we avoid potential deadlocks when the channel is being
      //      destroyed at the same time
      CosN::EventTypeSeq added, deled;
      added.length(0);
      deled.length( _evtypes.length() );
      RDI_HashCursor<CosN::EventType, CORBA::ULong> curs;
      CORBA::ULong ix=0;
      for ( ix=0, curs = _evtypes.cursor(); curs.is_valid(); ++ix, ++curs ) {
	deled[ix] = curs.key();
      }
      EventChannel_i* chan = _channel;
      RDI_OPLOCK_RELEASE; // debump after invoking channel ops
      chan->propagate_offer_change(added, deled);
      RDI_OPLOCK_REACQUIRE(return);
    }
  }
  RDIDbgSAdmLog("Begin Destruction of SupplierAdmin @ " << (void*)this << " [" << _serial << "]\n");
  unsigned int i;
  RDI_ListCursor<EventProxyPushConsumer_i *>                cpushcur;
  RDI_ListCursor<EventProxyPullConsumer_i *>                cpullcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPushConsumer_i *>           apushcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullConsumer_i *>           apullcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushConsumer_i *> spushcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullConsumer_i *> spullcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushConsumer_i *>   bpushcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullConsumer_i *>   bpullcur;

  cpushcur = _cosevent_push.cursor();
  for (i=0; i < _cosevent_push.length(); ++i, ++cpushcur) {
    _removed_push_proxy();
    (*cpushcur)->disconnect_client_and_dispose(0);
  }
  cpullcur = _cosevent_pull.cursor();
  for (i=0; i < _cosevent_pull.length(); ++i, ++cpullcur) {
    EventProxyPullConsumer_i* prx = *cpullcur;
    _removed_pull_proxy(prx);
    prx->disconnect_client_and_dispose(0);
  }
  for ( apushcur = _prx_any_push.cursor(); apushcur.is_valid(); ++apushcur ) {
    _removed_push_proxy();
    apushcur.val()->disconnect_client_and_dispose(0);
  }
  for ( apullcur = _prx_any_pull.cursor(); apullcur.is_valid(); ++apullcur ) {
    ProxyPullConsumer_i* prx = apullcur.val();
    _removed_pull_proxy(prx);
    prx->disconnect_client_and_dispose(0);
  }
  for ( spushcur = _prx_struc_push.cursor(); spushcur.is_valid(); ++spushcur ) {
    _removed_push_proxy();
    spushcur.val()->disconnect_client_and_dispose(0);
  }
  for ( spullcur = _prx_struc_pull.cursor(); spullcur.is_valid(); ++spullcur ) {
    StructuredProxyPullConsumer_i* prx = spullcur.val();
    _removed_pull_proxy(prx);
    prx->disconnect_client_and_dispose(0);
  }
  for ( bpushcur = _prx_batch_push.cursor(); bpushcur.is_valid(); ++bpushcur ) {
    _removed_push_proxy();
    bpushcur.val()->disconnect_client_and_dispose(0);
  }
  for ( bpullcur = _prx_batch_pull.cursor(); bpullcur.is_valid(); ++bpullcur ) {
    SequenceProxyPullConsumer_i* prx = bpullcur.val();
    _removed_pull_proxy(prx);
    prx->disconnect_client_and_dispose(0);
  }
  _cosevent_push.drain();  _cosevent_pull.drain();
  _prx_any_push.clear();   _prx_any_pull.clear();
  _prx_struc_push.clear(); _prx_struc_pull.clear();
  _prx_batch_push.clear(); _prx_batch_pull.clear();

  if ( _fa_helper.has_filters() ) _fa_helper.remove_all_filters();
  if ( _qosprop ) {
    delete _qosprop; _qosprop = 0;
  }
  RDIDbgSAdmLog("SupplierAdmin @ " << (void*)this << " [" << _serial << "] destroyed\n");
  RDI_OPLOCK_DISPOSE_IMPL_AND_RELEASE;
}

// ** Requires OPLOCK is held
// ** Effect: Update the local hash table and if new entries are created or
//      deleted, update the channel as well, after releasing OPLOCK
//      to avoid deadlocks due to reverse order of lock acquisitions.
void
SupplierAdmin_i::_propagate_offer_change(const CosN::EventTypeSeq& added,
					 const CosN::EventTypeSeq& deled)
{
  CORBA::ULong ix=0, sz=0, vl=0;
  CosN::EventTypeSeq new_added;
  CosN::EventTypeSeq old_deled;
  new_added.length(0);
  old_deled.length(0);

  for (sz=0, ix = 0; ix < added.length(); ix++) {
    if ( _evtypes.lookup(added[ix], vl) ) {
      vl += 1; _evtypes.replace(added[ix], vl);
    } else {
      vl  = 1; _evtypes.insert(added[ix], vl);
      new_added.length(sz + 1);
      new_added[sz++] = added[ix];
    }
  }
  for (sz=0, ix = 0; ix < deled.length(); ix++) {
    if ( _evtypes.lookup(deled[ix], vl) ) {
      if ( vl == 1 ) {
	_evtypes.remove(deled[ix]);
	old_deled.length(sz + 1);
	old_deled[sz++] = deled[ix];
      } else {
	vl -= 1; _evtypes.replace(deled[ix], vl);
      }
    } else {
      RDIDbgSAdmLog("Invalid "<<deled[ix].domain_name<<"::"<<deled[ix].type_name << '\n');
    }
  }
  EventChannel_i* chan = _channel;
  RDI_OPLOCK_RELEASE;
  if ( new_added.length() || old_deled.length() )
    chan->propagate_offer_change(new_added, old_deled);
}

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

RDIstrstream& ConsumerAdmin_i::log_output(RDIstrstream& str)
{
  RDI_OPLOCK_ACQUIRE(return str);

  RDI_ListCursor<EventProxyPushSupplier_i *>                cpushcur;
  RDI_ListCursor<EventProxyPullSupplier_i *>                cpullcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPushSupplier_i *>           apushcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullSupplier_i *>           apullcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushSupplier_i *> spushcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullSupplier_i *> spullcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushSupplier_i *>   bpushcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullSupplier_i *>   bpullcur;
  unsigned int cnt = 0;

  str << "----------------------------------------------------------------------\n";
  str << "Consumer Admin " << _my_name;
  str << "\n----------------------------------------------------------------------\n";
  str << (void*)this <<" ID "; 
  str.setw(3); str << _serial;
  str.setw(15); str << "\n  Born ";
  _time_born.out_local(str);
  str.setw(15); str << "\n  Last Proxy ";
  _prxy_exit.out_local(str);
  str << '\n';
  if ( _rqstypes.length() != 0 ) {
    for (CORBA::ULong ix = 0; ix < _rqstypes.length(); ix++) {
      str << "\n\t" << (const char*)_rqstypes[ix].domain_name; 
      str << "::" << (const char*)_rqstypes[ix].type_name;
    }
  }
  out_config(str);
  cpushcur = _cosevent_push.cursor();
  for (cnt=0; cnt < _cosevent_push.length(); ++cnt, ++cpushcur) {
    str.setw(4); str << cnt << ": " << *(*cpushcur) << '\n';
  }
  cpullcur = _cosevent_pull.cursor();
  for (cnt=0; cnt < _cosevent_pull.length(); ++cnt, ++cpullcur) {
    str.setw(4); str << cnt << ": " << *(*cpullcur) << '\n';
  }
  cnt = 0;
  for (apushcur = _prx_any_push.cursor(); apushcur.is_valid(); ++apushcur) {
    str.setw(4); str << ++cnt << ": " << *(apushcur.val()) << '\n';
  }
  cnt = 0;
  for (apullcur = _prx_any_pull.cursor(); apullcur.is_valid(); ++apullcur) {
    str.setw(4); str << ++cnt << ": " << *(apullcur.val()) << '\n';
  }
  cnt = 0;
  for (spushcur = _prx_struc_push.cursor(); spushcur.is_valid(); ++spushcur) {
    str.setw(4); str << ++cnt << ": " << *(spushcur.val()) << '\n';
  }
  cnt = 0;
  for (spullcur = _prx_struc_pull.cursor(); spullcur.is_valid(); ++spullcur) {
    str.setw(4); str << ++cnt << ": " << *(spullcur.val()) << '\n';
  }
  cnt = 0;
  for (bpushcur = _prx_batch_push.cursor(); bpushcur.is_valid(); ++bpushcur) {
    str.setw(4); str << ++cnt << ": " << *(bpushcur.val()) << '\n';
  }
  cnt = 0;
  for (bpullcur = _prx_batch_pull.cursor(); bpullcur.is_valid(); ++bpullcur) {
    str.setw(4); str << ++cnt << ": " << *(bpullcur.val()) << '\n';
  }
  RDI_OPLOCK_RELEASE;
  return str;
}

RDIstrstream& SupplierAdmin_i::log_output(RDIstrstream& str)
{
  RDI_ListCursor<EventProxyPushConsumer_i *>                cpushcur;
  RDI_ListCursor<EventProxyPullConsumer_i *>                cpullcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPushConsumer_i *>           apushcur;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullConsumer_i *>           apullcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushConsumer_i *> spushcur;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullConsumer_i *> spullcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushConsumer_i *>   bpushcur;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullConsumer_i *>   bpullcur;
  unsigned int cnt = 0;

  RDI_OPLOCK_ACQUIRE(return str);
  str << "----------------------------------------------------------------------\n";
  str << "Supplier Admin " << _my_name;
  str << "\n----------------------------------------------------------------------\n";
  str << (void*)this <<" ID "; 
  str.setw(3); str << _serial;
  str.setw(15); str << "\n  Born ";
  _time_born.out_local(str);
  str.setw(15); str << "\n  Last Proxy ";
  _prxy_exit.out_local(str);
  str << '\n';
  out_config(str);
  cpushcur = _cosevent_push.cursor();
  for (cnt=0; cnt < _cosevent_push.length(); ++cnt, ++cpushcur) {
    str.setw(4); str << cnt << ": " << *(*cpushcur) << '\n';
  }
  cpullcur = _cosevent_pull.cursor();
  for (cnt=0; cnt < _cosevent_pull.length(); ++cnt, ++cpullcur) {
    str.setw(4); str << cnt << ": " << *(*cpullcur) << '\n';
  }
  cnt = 0;
  for (apushcur = _prx_any_push.cursor(); apushcur.is_valid(); ++apushcur) {
    str.setw(4); str << ++cnt << ": " << *(apushcur.val()) << '\n';
  }
  cnt = 0;
  for (apullcur = _prx_any_pull.cursor(); apullcur.is_valid(); ++apullcur) {
    str.setw(4); str << ++cnt << ": " << *(apullcur.val()) << '\n';
  }
  cnt = 0;
  for (spushcur = _prx_struc_push.cursor(); spushcur.is_valid(); ++spushcur) {
    str.setw(4); str << ++cnt << ": " << *(spushcur.val()) << '\n';
  }
  cnt = 0;
  for (spullcur = _prx_struc_pull.cursor(); spullcur.is_valid(); ++spullcur) {
    str.setw(4); str << ++cnt << ": " << *(spullcur.val()) << '\n';
  }
  cnt = 0;
  for (bpushcur = _prx_batch_push.cursor(); bpushcur.is_valid(); ++bpushcur) {
    str.setw(4); str << ++cnt << ": " << *(bpushcur.val()) << '\n';
  }
  cnt = 0;
  for (bpullcur = _prx_batch_pull.cursor(); bpullcur.is_valid(); ++bpullcur) {
    str.setw(4); str << ++cnt << ": " << *(bpullcur.val()) << '\n';
  }
  RDI_OPLOCK_RELEASE;
  return str;
}

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

AttN::NameSeq* ConsumerAdmin_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* ConsumerAdmin_i::child_names( WRAPPED_IMPLARG_VOID ) {
  char buf[20];
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  AttN::NameSeq* names = new AttN::NameSeq;
  if ( names == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  RDI_HashCursor<CosNA::ProxyID, ProxyPushSupplier_i *>           ac1;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushSupplier_i *> sc1;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushSupplier_i * >  bc1;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullSupplier_i *>           ac2;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullSupplier_i *> sc2;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullSupplier_i * >  bc2;
  unsigned long num = _prx_any_push.length() +
    _prx_struc_push.length() + _prx_batch_push.length() +
    _prx_any_pull.length() +
    _prx_struc_pull.length() + _prx_batch_pull.length();
  names->length(num);
  num = 0;
  for ( ac1 = _prx_any_push.cursor(); ac1.is_valid(); ++ac1, ++num ) {
    sprintf(buf, "proxy%ld", ac1.key());
    (*names)[num] = (const char*)buf;
  }
  for ( sc1 = _prx_struc_push.cursor(); sc1.is_valid(); ++sc1, ++num ) {
    sprintf(buf, "proxy%ld", sc1.key());
    (*names)[num] = (const char*)buf;
  }
  for ( bc1 = _prx_batch_push.cursor(); bc1.is_valid(); ++bc1, ++num ) {
    sprintf(buf, "proxy%ld", bc1.key());
    (*names)[num] = (const char*)buf;
  }
  for ( ac2 = _prx_any_pull.cursor(); ac2.is_valid(); ++ac2, ++num ) {
    sprintf(buf, "proxy%ld", ac2.key());
    (*names)[num] = (const char*)buf;
  }
  for ( sc2 = _prx_struc_pull.cursor(); sc2.is_valid(); ++sc2, ++num ) {
    sprintf(buf, "proxy%ld", sc2.key());
    (*names)[num] = (const char*)buf;
  }
  for ( bc2 = _prx_batch_pull.cursor(); bc2.is_valid(); ++bc2, ++num ) {
    sprintf(buf, "proxy%ld", bc2.key());
    (*names)[num] = (const char*)buf;
  }
  RDI_OPLOCK_RELEASE;
  return names;
}

void ConsumerAdmin_i::out_info_filters(RDIstrstream& str, CORBA::Boolean admin, CORBA::Boolean proxies) {
  RDI_OPLOCK_ACQUIRE(return);
  if (admin) {
    str << "----------------------------------------------------------------------\n";
    str << "Admin Filters attached to " << _my_name << '\n';
    str << "----------------------------------------------------------------------\n";
    _fa_helper.out_info_filters(str);
  }
  if (proxies) {
    RDI_HashCursor<CosNA::ProxyID, ProxyPushSupplier_i *>           ac1;
    RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushSupplier_i *> sc1;
    RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushSupplier_i * >  bc1;
    RDI_HashCursor<CosNA::ProxyID, ProxyPullSupplier_i *>           ac2;
    RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullSupplier_i *> sc2;
    RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullSupplier_i * >  bc2;
    for ( ac1 = _prx_any_push.cursor(); ac1.is_valid(); ++ac1 ) {
      ac1.val()->out_info_filters(str);
    }
    for ( sc1 = _prx_struc_push.cursor(); sc1.is_valid(); ++sc1 ) {
      sc1.val()->out_info_filters(str);
    }
    for ( bc1 = _prx_batch_push.cursor(); bc1.is_valid(); ++bc1 ) {
      bc1.val()->out_info_filters(str);
    }
    for ( ac2 = _prx_any_pull.cursor(); ac2.is_valid(); ++ac2 ) {
      ac2.val()->out_info_filters(str);
    }
    for ( sc2 = _prx_struc_pull.cursor(); sc2.is_valid(); ++sc2 ) {
      sc2.val()->out_info_filters(str);
    }
    for ( bc2 = _prx_batch_pull.cursor(); bc2.is_valid(); ++bc2 ) {
      bc2.val()->out_info_filters(str);
    }
  }
  RDI_OPLOCK_RELEASE;
}

// _children: fills in 'ren' sequence
// NB: REQUIRES oplock is held
void
ConsumerAdmin_i::_children(AttN::IactSeq& ren,
			   CORBA::Boolean only_cleanup_candidates) {
  // Unconnected proxies are cleanup candidates
  unsigned int idx = 0;
  // start with max length
  ren.length(_prx_any_push.length() +
	     _prx_struc_push.length() +
	     _prx_batch_push.length() +
	     _prx_any_pull.length() +
	     _prx_struc_pull.length() +
	     _prx_batch_pull.length());
  RDI_HashCursor<CosNA::ProxyID, ProxyPushSupplier_i *>           ac1;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushSupplier_i *> sc1;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushSupplier_i * >  bc1;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullSupplier_i *>           ac2;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullSupplier_i *> sc2;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullSupplier_i * >  bc2;
  for ( ac1 = _prx_any_push.cursor(); ac1.is_valid(); ++ac1 ) {
    if ((!only_cleanup_candidates) || (!ac1.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, ac1.val());
    }
  }
  for ( sc1 = _prx_struc_push.cursor(); sc1.is_valid(); ++sc1 ) {
    if ((!only_cleanup_candidates) || (!sc1.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, sc1.val());
    }
  }
  for ( bc1 = _prx_batch_push.cursor(); bc1.is_valid(); ++bc1 ) {
    if ((!only_cleanup_candidates) || (!bc1.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, bc1.val());
    }
  }
  for ( ac2 = _prx_any_pull.cursor(); ac2.is_valid(); ++ac2 ) {
    if ((!only_cleanup_candidates) || (!ac2.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, ac2.val());
    }
  }
  for ( sc2 = _prx_struc_pull.cursor(); sc2.is_valid(); ++sc2 ) {
    if ((!only_cleanup_candidates) || (!sc2.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, sc2.val());
    }
  }
  for ( bc2 = _prx_batch_pull.cursor(); bc2.is_valid(); ++bc2 ) {
    if ((!only_cleanup_candidates) || (!bc2.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, bc2.val());
    }
  }
  // trim to actual length
  ren.length(idx);
}

AttN::IactSeq*
ConsumerAdmin_i::children(CORBA::Boolean only_cleanup_candidates  WRAPPED_IMPLARG ) {
  // Unconnected proxies are cleanup candidates
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::IactSeq* ren = new AttN::IactSeq;
  if ( ren == (AttN::IactSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  _children(*ren, only_cleanup_candidates);
  RDI_OPLOCK_RELEASE;
  return ren;
}

void
ConsumerAdmin_i::cleanup(RDIstrstream& str, CORBA::Boolean admin, CORBA::Boolean proxies) {
  RDIInteractive::cleanup_admin(str, WRAPPED_IMPL2OREF(AttN::Interactive, this), admin, proxies);
}

CORBA::Boolean
ConsumerAdmin_i::do_set_command(RDIstrstream& str, RDIParseCmd& p) {
  CosN::QoSProperties    n_qos;
  CosN::AdminProperties  a_qos;
  AttN::ServerProperties s_qos;
  CORBA::Boolean success = 1;
  success = RDI_AllQoS::parse_set_command(str, p, RDI_C_ADMIN, n_qos, a_qos, s_qos);
  if (success && n_qos.length() > 0) {
    try {
      set_qos(n_qos);
    } catch (CosN::UnsupportedQoS& e) {
      str << "\nThe following NotifQoS Property Settings are invalid:\n";
      RDI_describe_prop_errors(str, e.qos_err);
      str << '\n';
      success = 0;
    } catch (CORBA::INV_OBJREF) { throw; }
    if (success) {
      str << '\n';
      for (unsigned int i = 0; i < n_qos.length(); i++) {
	str << n_qos[i].name << " set to "; RDI_pp_any(str, n_qos[i].value); str << '\n';
      }
      str << "\nSome properties updated successfully.  Current settings:\n\n";
      out_config(str);
    }
  }
  return success;
}

CORBA::Boolean
ConsumerAdmin_i::do_go_command(RDIstrstream& str, RDIParseCmd& p,
			       CORBA::Boolean& target_changed,
			       AttN_Interactive_outarg next_target) {
  CORBA::Boolean success = 1;
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  char* go_targ = CORBA_STRING_DUP(p.argv[1]);
  char* rest_go_targ = RDI_STRCHR(go_targ, '.');
  if (rest_go_targ) {
    *rest_go_targ = '\0';
    rest_go_targ++;
  }
  char buf[20];
  RDI_HashCursor<CosNA::ProxyID, ProxyPushSupplier_i *>           ac1;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushSupplier_i *> sc1;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushSupplier_i * >  bc1;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullSupplier_i *>           ac2;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullSupplier_i *> sc2;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullSupplier_i * >  bc2;
  for ( ac1 = _prx_any_push.cursor(); ac1.is_valid(); ++ac1 ) {
    sprintf(buf, "proxy%ld", ac1.key());
    if (RDI_STR_EQ_I(go_targ, buf)) {
      target_changed = 1;
      next_target = WRAPPED_IMPL2OREF(AttN::ProxyPushSupplier, ac1.val());
      break;
    }
  }
  if (target_changed == 0) {
    for ( sc1 = _prx_struc_push.cursor(); sc1.is_valid(); ++sc1 ) {
      sprintf(buf, "proxy%ld", sc1.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::StructuredProxyPushSupplier, sc1.val());
	break;
      }
    }
  }
  if (target_changed == 0) {
    for ( bc1 = _prx_batch_push.cursor(); bc1.is_valid(); ++bc1 ) {
      sprintf(buf, "proxy%ld", bc1.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::SequenceProxyPushSupplier, bc1.val());
	break;
      }
    }
  }
  if (target_changed == 0) {
    for ( ac2 = _prx_any_pull.cursor(); ac2.is_valid(); ++ac2 ) {
      sprintf(buf, "proxy%ld", ac2.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::ProxyPullSupplier, ac2.val());
	break;
      }
    }
  }
  if (target_changed == 0) {
    for ( sc2 = _prx_struc_pull.cursor(); sc2.is_valid(); ++sc2 ) {
      sprintf(buf, "proxy%ld", sc2.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::StructuredProxyPullSupplier, sc2.val());
	break;
      }
    }
  }
  if (target_changed == 0) {
    for ( bc2 = _prx_batch_pull.cursor(); bc2.is_valid(); ++bc2 ) {
      sprintf(buf, "proxy%ld", bc2.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::SequenceProxyPullSupplier, bc2.val());
	break;
      }
    }
  }
  RDI_OPLOCK_RELEASE;
  if (target_changed == 0) {
    str << "Invalid target " << p.argv[1] << " : " << go_targ << " is not a proxy name\n";
    str << "  (Use 'children' for list of valid proxy names)\n";
    success = 0;
  } else {
    str << "\nomniNotify: new target ==> " << go_targ << '\n';
    if (rest_go_targ && (RDI_STRLEN(rest_go_targ) > 0)) {
      CORBA::Boolean target_changed2 = 0;
      AttN::Interactive_ptr next_target2 = AttN::Interactive::_nil();
      CORBA::String_var cmdres;
      char* newcmd = CORBA_STRING_ALLOC(4 + RDI_STRLEN(rest_go_targ));
      sprintf(newcmd, "go %s", rest_go_targ);
      CORBA::Boolean docmd_prob = 0;
      try {
	cmdres = next_target->do_command(newcmd, success, target_changed2, next_target2);
      } CATCH_INVOKE_PROBLEM(docmd_prob);
      CORBA_STRING_FREE(newcmd);
      if (docmd_prob) {
	str << "The target " << rest_go_targ << " is not available\n";
      } else {
	str << cmdres.in();
	if (target_changed2) {
	  next_target = next_target2; // XXX duplicate not needed ???
	}
      }
    }
  }
  CORBA_STRING_FREE(go_targ);
  return success;
}

void ConsumerAdmin_i::out_config(RDIstrstream& str) {
  str << "----------------------------------------------------------------------\n";
  str << "NotifQos settings for " << _my_name << '\n';
  str << "----------------------------------------------------------------------\n";
  _qosprop->log_output(str);
  str << '\n';
}

void ConsumerAdmin_i::out_commands(RDIstrstream& str) {
  str << "omniNotify ConsumerAdmin commands:\n"
      << "  up                : change target to admin's channel\n"
      << "  go <proxy_name>   : change target to a specific proxy\n"
      << "                        ('children' lists proxy names)\n"
      << "  debug             : show debugging information for this admin + its proxies\n"
      << "  config            : show config information for this admin + its proxies\n"
      << "  info filters      : show filters attached to this admin and its proxies\n"
      << "  info afilters     : show filters attached to this admin\n"
      << "  info pfilters     : show filters attached to this admin's proxies\n"
      << "  cleanup proxies   : destroy all proxies of this admin that are not connected\n"
      << "  set <name1> <value1> [ <name2> <value2> ... ]\n"
      << "                    : set NotifQoS property name1 to value1, name2 to value2, etc.\n";
}

char*
ConsumerAdmin_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 if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "debug")) {
    log_output(str);
  } else if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "config")) {
    out_config(str);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "info") && RDI_STR_EQ_I(p.argv[1], "filters")) {
    out_info_filters(str, 1, 1);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "info") && RDI_STR_EQ_I(p.argv[1], "afilters")) {
    out_info_filters(str, 1, 0);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "info") && RDI_STR_EQ_I(p.argv[1], "pfilters")) {
    out_info_filters(str, 0, 1);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "cleanup") && RDI_STR_EQ_I(p.argv[1], "proxies")) {
    cleanup(str, 0, 1);
  } else if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "up")) {
    target_changed = 1;
    next_target = WRAPPED_IMPL2OREF(AttN::EventChannel, _channel);
    str << "\nomniNotify: new target ==> " << _my_name[_my_name.length()-2] << '\n';
  } else if ((p.argc >= 1) && RDI_STR_EQ_I(p.argv[0], "set")) {
    success = do_set_command(str, p);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "go")) {
    success = do_go_command(str, p, target_changed, next_target);
  } 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::NameSeq* SupplierAdmin_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* SupplierAdmin_i::child_names( WRAPPED_IMPLARG_VOID ) {
  char buf[20];
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  if (_disposed) { RDI_OPLOCK_RELEASE; RDI_THROW_INV_OBJREF; } // in process of being disposed
  AttN::NameSeq* names = new AttN::NameSeq;
  if ( names == (AttN::NameSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  RDI_HashCursor<CosNA::ProxyID, ProxyPushConsumer_i *>           ac1;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushConsumer_i *> sc1;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushConsumer_i *>   bc1;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullConsumer_i *>           ac2;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullConsumer_i *> sc2;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullConsumer_i *>   bc2;
  unsigned long num = _prx_any_push.length() +
    _prx_struc_push.length() + _prx_batch_push.length() +
    _prx_any_pull.length() +
    _prx_struc_pull.length() + _prx_batch_pull.length();
  names->length(num);
  num = 0;
  for ( ac1 = _prx_any_push.cursor(); ac1.is_valid(); ++ac1, ++num ) {
    sprintf(buf, "proxy%ld", ac1.key());
    (*names)[num] = (const char*)buf;
  }
  for ( sc1 = _prx_struc_push.cursor(); sc1.is_valid(); ++sc1, ++num ) {
    sprintf(buf, "proxy%ld", sc1.key());
    (*names)[num] = (const char*)buf;
  }
  for ( bc1 = _prx_batch_push.cursor(); bc1.is_valid(); ++bc1, ++num ) {
    sprintf(buf, "proxy%ld", bc1.key());
    (*names)[num] = (const char*)buf;
  }
  for ( ac2 = _prx_any_pull.cursor(); ac2.is_valid(); ++ac2, ++num ) {
    sprintf(buf, "proxy%ld", ac2.key());
    (*names)[num] = (const char*)buf;
  }
  for ( sc2 = _prx_struc_pull.cursor(); sc2.is_valid(); ++sc2, ++num ) {
    sprintf(buf, "proxy%ld", sc2.key());
    (*names)[num] = (const char*)buf;
  }
  for ( bc2 = _prx_batch_pull.cursor(); bc2.is_valid(); ++bc2, ++num ) {
    sprintf(buf, "proxy%ld", bc2.key());
    (*names)[num] = (const char*)buf;
  }
  RDI_OPLOCK_RELEASE;
  return names;
}

void SupplierAdmin_i::out_info_filters(RDIstrstream& str, CORBA::Boolean admin, CORBA::Boolean proxies) {
  RDI_OPLOCK_ACQUIRE(return);
  if (admin) {
    str << "----------------------------------------------------------------------\n";
    str << "Admin Filters attached to " << _my_name << '\n';
    str << "----------------------------------------------------------------------\n";
    _fa_helper.out_info_filters(str);
  }
  if (proxies) {
    RDI_HashCursor<CosNA::ProxyID, ProxyPushConsumer_i *>           ac1;
    RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushConsumer_i *> sc1;
    RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushConsumer_i *>   bc1;
    RDI_HashCursor<CosNA::ProxyID, ProxyPullConsumer_i *>           ac2;
    RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullConsumer_i *> sc2;
    RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullConsumer_i *>   bc2;
    for ( ac1 = _prx_any_push.cursor(); ac1.is_valid(); ++ac1 ) {
      ac1.val()->out_info_filters(str);
    }
    for ( sc1 = _prx_struc_push.cursor(); sc1.is_valid(); ++sc1 ) {
      sc1.val()->out_info_filters(str);
    }
    for ( bc1 = _prx_batch_push.cursor(); bc1.is_valid(); ++bc1 ) {
      bc1.val()->out_info_filters(str);
    }
    for ( ac2 = _prx_any_pull.cursor(); ac2.is_valid(); ++ac2 ) {
      ac2.val()->out_info_filters(str);
    }
    for ( sc2 = _prx_struc_pull.cursor(); sc2.is_valid(); ++sc2 ) {
      sc2.val()->out_info_filters(str);
    }
    for ( bc2 = _prx_batch_pull.cursor(); bc2.is_valid(); ++bc2 ) {
      bc2.val()->out_info_filters(str);
    }
  }
  RDI_OPLOCK_RELEASE;
}

// _children: fills in 'ren'
// NB: REQUIRES oplock is held
void
SupplierAdmin_i::_children(AttN::IactSeq& ren, CORBA::Boolean only_cleanup_candidates) {
  // Unconnected proxies are cleanup candidates
  unsigned int idx = 0;
  // start with max length
  ren.length(_prx_any_push.length() +
	     _prx_struc_push.length() +
	     _prx_batch_push.length() +
	     _prx_any_pull.length() +
	     _prx_struc_pull.length() +
	     _prx_batch_pull.length());
  RDI_HashCursor<CosNA::ProxyID, ProxyPushConsumer_i *>           ac1;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushConsumer_i *> sc1;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushConsumer_i * >  bc1;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullConsumer_i *>           ac2;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullConsumer_i *> sc2;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullConsumer_i * >  bc2;
  for ( ac1 = _prx_any_push.cursor(); ac1.is_valid(); ++ac1 ) {
    if ((!only_cleanup_candidates) || (!ac1.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, ac1.val());
    }
  }
  for ( sc1 = _prx_struc_push.cursor(); sc1.is_valid(); ++sc1 ) {
    if ((!only_cleanup_candidates) || (!sc1.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, sc1.val());
    }
  }
  for ( bc1 = _prx_batch_push.cursor(); bc1.is_valid(); ++bc1 ) {
    if ((!only_cleanup_candidates) || (!bc1.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, bc1.val());
    }
  }
  for ( ac2 = _prx_any_pull.cursor(); ac2.is_valid(); ++ac2 ) {
    if ((!only_cleanup_candidates) || (!ac2.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, ac2.val());
    }
  }
  for ( sc2 = _prx_struc_pull.cursor(); sc2.is_valid(); ++sc2 ) {
    if ((!only_cleanup_candidates) || (!sc2.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, sc2.val());
    }
  }
  for ( bc2 = _prx_batch_pull.cursor(); bc2.is_valid(); ++bc2 ) {
    if ((!only_cleanup_candidates) || (!bc2.val()->is_connected())) {
      ren[idx++] = WRAPPED_IMPL2OREF(AttN::Interactive, bc2.val());
    }
  }
  // trim to actual length
  ren.length(idx);
}

AttN::IactSeq*
SupplierAdmin_i::children(CORBA::Boolean only_cleanup_candidates  WRAPPED_IMPLARG ) {
  // Unconnected proxies are cleanup candidates
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  AttN::IactSeq* ren = new AttN::IactSeq;
  if ( ren == (AttN::IactSeq*) 0 ) {
    RDI_OPLOCK_RELEASE;
    throw CORBA::NO_MEMORY(0, CORBA::COMPLETED_NO);
  }
  _children(*ren, only_cleanup_candidates);
  RDI_OPLOCK_RELEASE;
  return ren;
}

void
SupplierAdmin_i::cleanup(RDIstrstream& str, CORBA::Boolean admin, CORBA::Boolean proxies) {
  RDIInteractive::cleanup_admin(str, WRAPPED_IMPL2OREF(AttN::Interactive, this), admin, proxies);
}

CORBA::Boolean
SupplierAdmin_i::do_set_command(RDIstrstream& str, RDIParseCmd& p) {
  CosN::QoSProperties    n_qos;
  CosN::AdminProperties  a_qos;
  AttN::ServerProperties s_qos;
  CORBA::Boolean success = 1;
  success = RDI_AllQoS::parse_set_command(str, p, RDI_S_ADMIN, n_qos, a_qos, s_qos);
  if (success && n_qos.length() > 0) {
    try {
      set_qos(n_qos);
    } catch (CosN::UnsupportedQoS& e) {
      str << "\nThe following NotifQoS Property Settings are invalid:\n";
      RDI_describe_prop_errors(str, e.qos_err);
      str << '\n';
      success = 0;
    } catch (CORBA::INV_OBJREF) { throw; }
    if (success) {
      str << '\n';
      for (unsigned int i = 0; i < n_qos.length(); i++) {
	str << n_qos[i].name << " set to "; RDI_pp_any(str, n_qos[i].value); str << '\n';
      }
      str << "\nSome properties updated successfully.  Current settings:\n\n";
      out_config(str);
    }
  }
  return success;
}

CORBA::Boolean
SupplierAdmin_i::do_go_command(RDIstrstream& str, RDIParseCmd& p,
			       CORBA::Boolean& target_changed,
			       AttN_Interactive_outarg next_target) {
  CORBA::Boolean success = 1;
  RDI_OPLOCK_ACQUIRE(RDI_THROW_INV_OBJREF);
  char* go_targ = CORBA_STRING_DUP(p.argv[1]);
  char* rest_go_targ = RDI_STRCHR(go_targ, '.');
  if (rest_go_targ) {
    *rest_go_targ = '\0';
    rest_go_targ++;
  }
  char buf[20];
  RDI_HashCursor<CosNA::ProxyID, ProxyPushConsumer_i *>           ac1;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPushConsumer_i *> sc1;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPushConsumer_i *>   bc1;
  RDI_HashCursor<CosNA::ProxyID, ProxyPullConsumer_i *>           ac2;
  RDI_HashCursor<CosNA::ProxyID, StructuredProxyPullConsumer_i *> sc2;
  RDI_HashCursor<CosNA::ProxyID, SequenceProxyPullConsumer_i *>   bc2;
  for ( ac1 = _prx_any_push.cursor(); ac1.is_valid(); ++ac1 ) {
    sprintf(buf, "proxy%ld", ac1.key());
    if (RDI_STR_EQ_I(go_targ, buf)) {
      target_changed = 1;
      next_target = WRAPPED_IMPL2OREF(AttN::ProxyPushConsumer, ac1.val());
      break;
    }
  }
  if (target_changed == 0) {
    for ( sc1 = _prx_struc_push.cursor(); sc1.is_valid(); ++sc1 ) {
      sprintf(buf, "proxy%ld", sc1.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::StructuredProxyPushConsumer, sc1.val());
	break;
      }
    }
  }
  if (target_changed == 0) {
    for ( bc1 = _prx_batch_push.cursor(); bc1.is_valid(); ++bc1 ) {
      sprintf(buf, "proxy%ld", bc1.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::SequenceProxyPushConsumer, bc1.val());
	break;
      }
    }
  }
  if (target_changed == 0) {
    for ( ac2 = _prx_any_pull.cursor(); ac2.is_valid(); ++ac2 ) {
      sprintf(buf, "proxy%ld", ac2.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::ProxyPullConsumer, ac2.val());
	break;
      }
    }
  }
  if (target_changed == 0) {
    for ( sc2 = _prx_struc_pull.cursor(); sc2.is_valid(); ++sc2 ) {
      sprintf(buf, "proxy%ld", sc2.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::StructuredProxyPullConsumer, sc2.val());
	break;
      }
    }
  }
  if (target_changed == 0) {
    for ( bc2 = _prx_batch_pull.cursor(); bc2.is_valid(); ++bc2 ) {
      sprintf(buf, "proxy%ld", bc2.key());
      if (RDI_STR_EQ_I(go_targ, buf)) {
	target_changed = 1;
	next_target = WRAPPED_IMPL2OREF(AttN::SequenceProxyPullConsumer, bc2.val());
	break;
      }
    }
  }
  RDI_OPLOCK_RELEASE;
  if (target_changed == 0) {
    str << "Invalid target " << p.argv[1] << " : " << go_targ << " is not a proxy name\n";
    str << "  (Use 'children' for list of valid proxy names)\n";
    success = 0;
  } else {
    str << "\nomniNotify: new target ==> " << p.argv[1] << '\n';
    if (rest_go_targ && (RDI_STRLEN(rest_go_targ) > 0)) {
      CORBA::Boolean target_changed2 = 0;
      AttN::Interactive_ptr next_target2 = AttN::Interactive::_nil();
      CORBA::String_var cmdres;
      char* newcmd = CORBA_STRING_ALLOC(4 + RDI_STRLEN(rest_go_targ));
      sprintf(newcmd, "go %s", rest_go_targ);
      CORBA::Boolean docmd_prob = 0;
      try {
	cmdres = next_target->do_command(newcmd, success, target_changed2, next_target2);
      } CATCH_INVOKE_PROBLEM(docmd_prob);
      CORBA_STRING_FREE(newcmd);
      if (docmd_prob) {
	str << "The target " << rest_go_targ << " is not available\n";
      } else {
	str << cmdres.in();
	if (target_changed2) {
	  next_target = next_target2; // XXX duplicate not needed ???
	}
      }
    }
  }
  CORBA_STRING_FREE(go_targ);
  return success;
}

void SupplierAdmin_i::out_config(RDIstrstream& str) {
  str << "----------------------------------------------------------------------\n";
  str << "NotifQos settings for " << _my_name << '\n';
  str << "----------------------------------------------------------------------\n";
  _qosprop->log_output(str);
  str << '\n';
}

void SupplierAdmin_i::out_commands(RDIstrstream& str) {
  str << "omniNotify SupplierAdmin commands:\n"
      << "  debug             : show debugging information for this admin + its proxies\n"
      << "  up                : change target to admin's channel\n"
      << "  go <proxy_name>   : change target to a specific proxy\n"
      << "                        ('children' lists proxy names)\n"
      << "  config            : show config information for this admin + its proxies\n"
      << "  info filters      : show filters attached to this admin and its proxies\n"
      << "  info afilters     : show filters attached to this admin\n"
      << "  info pfilters     : show filters attached to this admin's proxies\n"
      << "  cleanup proxies   : destroy all proxies of this admin that are not connected\n"
      << "  set <name1> <value1> [ <name2> <value2> ... ]\n"
      << "                    : set NotifQoS property name1 to value1, name2 to value2, etc.\n";
}

char*
SupplierAdmin_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 if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "debug")) {
    log_output(str);
  } else if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "config")) {
    out_config(str);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "info") && RDI_STR_EQ_I(p.argv[1], "filters")) {
    out_info_filters(str, 1, 1);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "info") && RDI_STR_EQ_I(p.argv[1], "afilters")) {
    out_info_filters(str, 1, 0);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "info") && RDI_STR_EQ_I(p.argv[1], "pfilters")) {
    out_info_filters(str, 0, 1);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "cleanup") && RDI_STR_EQ_I(p.argv[1], "proxies")) {
    cleanup(str, 0, 1);
  } else if ((p.argc == 1) && RDI_STR_EQ_I(p.argv[0], "up")) {
    target_changed = 1;
    next_target = WRAPPED_IMPL2OREF(AttN::EventChannel, _channel);
    str << "\nomniNotify: new target ==> " << _my_name[_my_name.length()-2] << '\n';
  } else if ((p.argc >= 1) && RDI_STR_EQ_I(p.argv[0], "set")) {
    success = do_set_command(str, p);
  } else if ((p.argc == 2) && RDI_STR_EQ_I(p.argv[0], "go")) {
    success = do_go_command(str, p, target_changed, next_target);
  } 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());
}

