// -*- Mode: C++; -*-

// -------------------------------------------------------------- //
// See legacy_clients.h
// -------------------------------------------------------------- //

#include <iostream.h>
#include <iomanip.h>
#include "omnithread.h"

#include "CosNotifyShorthands.h"
#include "legacy_clients.h"

// ----------------- CosEvC::PushConsumer ---------------- //

LegacyPushConsumer_i::
LegacyPushConsumer_i(CosEA::ProxyPushSupplier_ptr proxy, CORBA::ULong max_events,
		     const char* objnm, consume_any_fn* consume_fn,
		     CORBA::ULong millisecs, CORBA::Boolean verbose) :
  _my_proxy(proxy), _obj_name(objnm), _consume_fn(consume_fn),
  _num_events(0), _max_events(max_events),
  _millisecs(millisecs), _verbose(verbose), _done(0), _com_err(0),
  _oplock(), _finish(&_oplock), _worker(0)
{
  // nothing else to do
}

LegacyPushConsumer_i*
LegacyPushConsumer_i::create(CosEA::EventChannel_ptr channel,
			     CORBA::ULong max_events,
			     const char* objnm,
			     consume_any_fn* consume_fn,
			     CORBA::ULong millisecs,
			     CORBA::Boolean verbose)
{
  // Obtain appropriate proxy object
  CosEA::ProxyPushSupplier_ptr proxy = CosEA::ProxyPushSupplier::_nil();
  CosEA::ConsumerAdmin_var admin;
  try {
    admin = channel->for_consumers();
    if ( CORBA::is_nil(admin) ) {
      cerr << "Failed to obtain admin" << endl;
      return 0; // failure
    }
  } catch (...) {
    cerr << "Failed to obtain admin" << endl;
    return 0;  // failure
  }
  if (verbose) cout << "Obtained admin from the channel" << endl;
  try {
    proxy = admin->obtain_push_supplier();
  } catch (...) {  }
  if ( CORBA::is_nil(proxy) ) {
    cerr << "Failed to obtain proxy" << endl;
    return 0;
  }
  if (verbose) cout << "Obtained proxy from admin" << endl;

    // Construct a client
  LegacyPushConsumer_i* client =
    new LegacyPushConsumer_i(proxy, max_events, objnm, consume_fn, millisecs, verbose);
  return client;
}

CORBA::Boolean LegacyPushConsumer_i::connect() {
  omni_mutex_lock l(_oplock);
  if (_done) return 0;
  _com_err = 0;
  try {
    _my_proxy->connect_push_consumer(_this());
  } catch (CORBA::BAD_PARAM& ex) {
    cerr << _obj_name << ": BAD_PARAM Exception while connecting" << endl;
    return 1; // error
  } catch (CosEA::AlreadyConnected& ex) {
    cerr << _obj_name << ": Already connected" << endl;
    return 1; // error
  } catch (...) {
    cerr << _obj_name << ": Failed to connect" << endl;
    return 1; // error
  }
  if (_verbose) cout << _obj_name << ": Connected to proxy, ready to consume events" << endl; 
  // if _millisecs is set, spawn a thread to ping the proxy
  if (_millisecs) 
    _worker = new GenericBoundWorkerThread(this);
  return 0; // OK
}

void LegacyPushConsumer_i::cleanup() {
  _oplock.lock();
  if (_worker || (!_done)) {
    cerr << "Coding error: only call c->cleanup() after c->wait_done()" << endl;
    _oplock.unlock();
    return;
  }
  // this method takes sole ownership of _my_proxy ref
  CosEA::ProxyPushSupplier_var proxy = _my_proxy;
  _my_proxy = CosEA::ProxyPushSupplier::_nil();
  // do not hold oplock while invoking disconnect
  _oplock.unlock();
  try {
    if ((!_com_err) && (!CORBA::is_nil(proxy)))
      proxy->disconnect_push_supplier();
  } catch(...) {}
}

// While _millisecs is positive, pings proxy every _millisecs milliseconds.
// Sets error code and sets _done to 1 if there is an error during ping.
void* LegacyPushConsumer_i::start_working(void *) {
  CosEvC::PushConsumer_var bump_my_refcount_during_outcalls = _this();

  _oplock.lock();
  if (_verbose && (!_done) && _millisecs) cout << _obj_name << ": Spawned thread entering ping loop" << endl;
  // invariant: _oplock held at top of loop
  while ( 1 ) {
    if (_done || (_millisecs == 0)) break;
    unsigned long t_secs = 0, t_nanosecs = 0;
    unsigned long d_secs = 0, d_nanosecs = 0;
    d_secs = _millisecs / 1000;
    d_nanosecs = (_millisecs % 1000) * 1000000;
    _oplock.unlock(); // do not hold oplock across ping
    try {
      _my_proxy->connect_push_consumer(_this());
    } catch (CosEA::AlreadyConnected& ex) {
      // this is the expected result -- continue pinging
    } catch (...) { // assume everything else is a communication error
      _oplock.lock();
      if (_done) break;
      if (_verbose) cout << _obj_name << ": communication error while pinging proxy using connect_push_consumer" << endl;
      _done = 1;
      _finish.broadcast();
      _com_err = 1;
      break; // break from while loop -- done
    }
    _oplock.lock();
    if (_done) break; // must have disconnected during pull
    // sleep for specified interval
    omni_thread::get_time(&t_secs, &t_nanosecs, d_secs, d_nanosecs);
    _finish.timedwait(t_secs, t_nanosecs); // this release _oplock during wait
    // continue ping loop
  }
  // done
  _oplock.unlock();
  return 0;
}

void LegacyPushConsumer_i::push(const CORBA::Any& data)
{
  if (_done) return; // see "Optimization Notes" at top of sample_clients.cc

  omni_mutex_lock l(_oplock);
  if (_done) return;
  _num_events++;
  if (_consume_fn)
    (*_consume_fn)(data, _obj_name, _num_events, _verbose);
  else if (_verbose) cout << _obj_name << ": event count = " << _num_events << endl;
  if (_max_events && (_num_events >= _max_events)) {
    if (_verbose) cout << _obj_name << ": DONE [max_events reached]" << endl;
    _done = 1;
    _finish.broadcast();
  }
}

void LegacyPushConsumer_i::disconnect_push_consumer()
{
  if (_done) return; // see "Optimization Notes" at top of sample_clients.cc

  omni_mutex_lock l(_oplock);
  if (_done) return;
  if (_verbose) cout << _obj_name << ": disconnected" << endl;
  _done = 1;
  _finish.broadcast();
}

CORBA::Boolean LegacyPushConsumer_i::wait_done() {
  _oplock.lock();
  while (!_done) {
    _finish.wait();
  }
  // using tmp_worker ensures only one wait_done call tries join()
  GenericBoundWorkerThread* tmp_worker = _worker;
  _worker = 0;
  _oplock.unlock();
  if (tmp_worker) {
    tmp_worker->join(0);
  }
  return _com_err;
}

// ----------------- CosEvC::PullConsumer ---------------- //

LegacyPullConsumer_i::
LegacyPullConsumer_i(CosEA::ProxyPullSupplier_ptr proxy, CORBA::ULong max_events, 
		     const char* objnm, consume_any_fn* consume_fn,
		     CORBA::ULong millisecs, CORBA::Boolean verbose) :
  _my_proxy(proxy), _obj_name(objnm), _consume_fn(consume_fn),
  _num_events(0), _max_events(max_events),
  _millisecs(millisecs), _verbose(verbose), _done(0), _com_err(0),
  _oplock(), _finish(&_oplock), _worker(0)
{
  // nothing else to do
}

LegacyPullConsumer_i*
LegacyPullConsumer_i::create(CosEA::EventChannel_ptr channel,
			     CORBA::ULong max_events,
			     const char* objnm,
			     consume_any_fn* consume_fn,
			     CORBA::ULong millisecs,
			     CORBA::Boolean verbose)
{
  // Obtain appropriate proxy object
  CosEA::ProxyPullSupplier_ptr proxy = CosEA::ProxyPullSupplier::_nil();
  CosEA::ConsumerAdmin_var admin;
  try {
    admin = channel->for_consumers();
    if ( CORBA::is_nil(admin) ) {
      cerr << "Failed to obtain admin" << endl;
      return 0; // failure
    }
  } catch (...) {
    cerr << "Failed to obtain admin" << endl;
    return 0;  // failure
  }
  if (verbose) cout << "Obtained admin from the channel" << endl;
  try {
    proxy = admin->obtain_pull_supplier();
  } catch (...) {  }
  if ( CORBA::is_nil(proxy) ) {
    cerr << "Failed to obtain proxy" << endl;
    return 0;
  }
  if (verbose) cout << "Obtained proxy from admin" << endl;

    // Construct a client
  LegacyPullConsumer_i* client =
    new LegacyPullConsumer_i(proxy, max_events, objnm, consume_fn, millisecs, verbose);
  return client;
}

CORBA::Boolean LegacyPullConsumer_i::connect() {
  omni_mutex_lock l(_oplock);
  if (_done) return 0;
  _com_err = 0;
  try {
    _my_proxy->connect_pull_consumer(_this());
  } catch (CORBA::BAD_PARAM& ex) {
    cerr << _obj_name << ": BAD_PARAM Exception while connecting" << endl;
    return 1; // error
  } catch (CosEA::AlreadyConnected& ex) {
    cerr << _obj_name << ": Already connected" << endl;
    return 1; // error
  } catch (...) {
    cerr << _obj_name << ": Failed to connect" << endl;
    return 1; // error
  }
  // spawn a thread to do pulling
  if (_verbose) cout << _obj_name << ": Connected to proxy, ready to consume events" << endl; 
  _worker = new GenericBoundWorkerThread(this);
  return 0; // OK
}

void LegacyPullConsumer_i::cleanup() {
  _oplock.lock();
  if (_worker || (!_done)) {
    cerr << "Coding error: only call c->cleanup() after c->wait_done()" << endl;
    _oplock.unlock();
    return;
  }
  // this method takes sole ownership of _my_proxy ref
  CosEA::ProxyPullSupplier_var proxy = _my_proxy;
  _my_proxy = CosEA::ProxyPullSupplier::_nil();
  // do not hold oplock while invoking disconnect
  _oplock.unlock();
  try {
    if ((!_com_err) && (!CORBA::is_nil(proxy)))
      proxy->disconnect_pull_supplier();
  } catch(...) {}
}

void* LegacyPullConsumer_i::start_working(void *) {
  CosEvC::PullConsumer_var bump_my_refcount_during_outcalls = _this();

  unsigned long t_secs = 0, t_nanosecs = 0;
  unsigned long d_secs = 0, d_nanosecs = 0;
  if (_millisecs) {
    d_secs = _millisecs / 1000;
    d_nanosecs = (_millisecs % 1000) * 1000000;
  }
  CORBA::Any* data = 0;
  // invariants: _oplock is held entering top of loop and 
  // also whenever we break out of loop
  // (it is *not* held during pull(), yield(), and timedwait() calls)
  _oplock.lock(); 
  if (_verbose && (!_done)) cout << _obj_name << ": Spawned thread entering main pull loop" << endl;
  while ( 1 ) {
    if (_done) break; // must have disconnected
    _oplock.unlock(); // do not hold oplock across pull
    try {
      data = _my_proxy->pull();
    } catch (...) {
      _oplock.lock();
      if (_done) break;
      if (_verbose) cout << _obj_name << ": communication error while calling pull()" << endl;
      _com_err = 1;
      break; // break from while loop -- done
    }
    _oplock.lock();
    if (_done) break; // must have disconnected during pull
    if (!data) {
      if (_verbose) cout << _obj_name << ": strange failure: pull() returned nil" << endl;
      _com_err = 1;
      break; // break from while loop -- done
    }
    // got an event
    _num_events++;
    if (_consume_fn)
      (*_consume_fn)(*data, _obj_name, _num_events, _verbose);
    else if (_verbose) cout << _obj_name << ": event count = " << _num_events << endl;
    delete data;
    data = 0;
    if (_max_events && (_num_events >= _max_events)) {
      if (_verbose) cout << _obj_name << ": DONE [max_events reached]" << endl;
      break; // done 
    }
    if (_millisecs) { // sleep for specified interval
      omni_thread::get_time(&t_secs, &t_nanosecs, d_secs, d_nanosecs);
      _finish.timedwait(t_secs, t_nanosecs); // this release _oplock during wait
    } else { // use yield to let other threads have a shot at _oplock
      _oplock.unlock();
      omni_thread::yield();
      _oplock.lock();
    }
    // continue pulling
  }
  // done
  delete data;
  if (!_done) {
    _done = 1;
    _finish.broadcast();
  }
  _oplock.unlock();
  return 0;
}


void LegacyPullConsumer_i::disconnect_pull_consumer()
{
  if (_done) return; // see "Optimization Notes" at top of sample_clients.cc

  omni_mutex_lock l(_oplock);
  if (_done) return;
  if (_verbose) cout << _obj_name << ": disconnected" << endl;
  _done = 1;
  _finish.broadcast();
}

CORBA::Boolean LegacyPullConsumer_i::wait_done() {
  _oplock.lock();
  while (!_done) {
    _finish.wait();
  }
  // using tmp_worker ensures only one wait_done call tries join()
  GenericBoundWorkerThread* tmp_worker = _worker;
  _worker = 0;
  _oplock.unlock();
  if (tmp_worker) {
    tmp_worker->join(0);
  }
  return _com_err;
}

// ------------------ CosEvC::PushSupplier ---------------- //

LegacyPushSupplier_i::
LegacyPushSupplier_i(CosEA::ProxyPushConsumer_ptr proxy, CORBA::ULong max_events,
		     const char* objnm, supply_any_fn* supply_fn,
		     CORBA::ULong millisecs, CORBA::Boolean verbose) :
  _my_proxy(proxy), _obj_name(objnm), _supply_fn(supply_fn),
  _num_events(0), _max_events(max_events),
  _millisecs(millisecs), _verbose(verbose), _done(0), _com_err(0),
  _oplock(), _finish(&_oplock), _worker(0)
{
  // providing explict NULL for supply_fn is not OK -- must have a valid function
  if (!_supply_fn) _supply_fn = sample_supply_any_fn;
}

LegacyPushSupplier_i*
LegacyPushSupplier_i::create(CosEA::EventChannel_ptr channel,
			     CORBA::ULong max_events,
			     const char* objnm,
			     supply_any_fn* supply_fn,
			     CORBA::ULong millisecs,
			     CORBA::Boolean verbose)
{
  // Obtain appropriate proxy object
  CosEA::ProxyPushConsumer_ptr proxy = CosEA::ProxyPushConsumer::_nil();
  CosEA::SupplierAdmin_var admin;
  try {
    admin = channel->for_suppliers();
    if ( CORBA::is_nil(admin) ) {
      cerr << "Failed to obtain admin" << endl;
      return 0; // failure
    }
  } catch (...) {
    cerr << "Failed to obtain admin" << endl;
    return 0;  // failure
  }
  if (verbose) cout << "Obtained admin from the channel" << endl;
  try {
    proxy = admin->obtain_push_consumer();
  } catch (...) {  }
  if ( CORBA::is_nil(proxy) ) {
    cerr << "Failed to obtain proxy" << endl;
    return 0; // failure
  }
  if (verbose) cout << "Obtained proxy from admin" << endl;

    // Construct a client
  LegacyPushSupplier_i* client =
    new LegacyPushSupplier_i(proxy, max_events, objnm, supply_fn, millisecs, verbose);
  return client;
}

CORBA::Boolean LegacyPushSupplier_i::connect() {
  omni_mutex_lock l(_oplock);
  if (_done) return 0;
  _com_err = 0;
  try {
    _my_proxy->connect_push_supplier(_this());
  } catch (CORBA::BAD_PARAM& ex) {
    cerr << _obj_name << ": BAD_PARAM Exception while connecting" << endl;
    return 1; // error
  } catch (CosEA::AlreadyConnected& ex) {
    cerr << _obj_name << ": Already connected" << endl;
    return 1; // error
  } catch (...) {
    cerr << _obj_name << ": Failed to connect" << endl;
    return 1; // error
  }
  // spawn a thread to do pushing
  if (_verbose) cout << _obj_name << ": Connected to proxy, ready to supply events" << endl; 
  _worker = new GenericBoundWorkerThread(this);
  return 0; // OK
}

void LegacyPushSupplier_i::cleanup() {
  _oplock.lock();
  if (_worker || (!_done)) {
    cerr << "Coding error: only call c->cleanup() after c->wait_done()" << endl;
    _oplock.unlock();
    return;
  }
  // this method takes sole ownership of _my_proxy ref
  CosEA::ProxyPushConsumer_var proxy = _my_proxy;
  _my_proxy = CosEA::ProxyPushConsumer::_nil();
  // do not hold oplock while invoking disconnect
  _oplock.unlock();
  try {
    if ((!_com_err) && (!CORBA::is_nil(proxy)))
      proxy->disconnect_push_consumer();
  } catch(...) {}
}

void* LegacyPushSupplier_i::start_working(void *) {
  CosEvC::PushSupplier_var bump_my_refcount_during_outcalls = _this();

  unsigned long t_secs = 0, t_nanosecs = 0;
  unsigned long d_secs = 0, d_nanosecs = 0;
  if (_millisecs) {
    d_secs = _millisecs / 1000;
    d_nanosecs = (_millisecs % 1000) * 1000000;
  }
  CORBA::Any* data = new CORBA::Any;
  _oplock.lock();
  if (_verbose && (!_done)) cout << _obj_name << ": Spawned thread entering main push loop" << endl;
  // invariants: _oplock is held entering top of loop and 
  // also whenever we break out of loop
  // (it is *not* held during push(), yield() and timedwait() calls)
  while ( 1 ) {
    if (_done) break; // must have disconnected
    if ((*_supply_fn)(*data, _obj_name, _num_events+1, _verbose)) {
      _oplock.unlock(); // do not hold oplock across push
      try {
	_my_proxy->push(*data);
      } catch (...) {
	_oplock.lock();
	if (_done) break; // must have disconnected during push
	if (_verbose) cout << _obj_name << ": communication error while calling push()" << endl;
	_com_err = 1;
	break; // break from while loop -- done
      }
      _oplock.lock();
      if (_done) break; // must have disconnected during push
      _num_events++;
      // [rather than doing the following we let supply_fn report if it wants to]
      // if (_verbose) cout << _obj_name << ": Pushed an event, event count = " << _num_events << endl;
      if (_max_events && (_num_events >= _max_events)) {
	if (_verbose) cout << _obj_name << ": DONE [max_events reached]" << endl;
	break; // done
      }
    }
    // wait or yield then try again
    if (_millisecs) { // sleep for specified interval
      omni_thread::get_time(&t_secs, &t_nanosecs, d_secs, d_nanosecs);
      _finish.timedwait(t_secs, t_nanosecs); // this release _oplock during wait
    } else { // use yield to let other threads have a shot at _oplock
      _oplock.unlock();
      omni_thread::yield();
      _oplock.lock();
    }
    // continue pushing
  }
  // done
  delete data;
  if (!_done) {
    _done = 1;
    _finish.broadcast();
  }
  _oplock.unlock();
  return 0;
}

void LegacyPushSupplier_i::disconnect_push_supplier()
{
  if (_done) return; // see "Optimization Notes" at top of sample_clients.cc

  omni_mutex_lock l(_oplock);
  if (_done) return;
  if (_verbose) cout << _obj_name << ": disconnected" << endl;
  _done = 1;
  _finish.broadcast();
}

CORBA::Boolean LegacyPushSupplier_i::wait_done() {
  _oplock.lock();
  while (!_done) {
    _finish.wait();
  }
  // using tmp_worker ensures only one wait_done call tries join()
  GenericBoundWorkerThread* tmp_worker = _worker;
  _worker = 0;
  _oplock.unlock();
  if (tmp_worker) {
    tmp_worker->join(0);
  }
  return _com_err;
}

// ------------------- CosEvC::PullSupplier ---------------- //

LegacyPullSupplier_i::
LegacyPullSupplier_i(CosEA::ProxyPullConsumer_ptr proxy, CORBA::ULong max_events,
		     const char* objnm, supply_any_fn* supply_fn,
		     CORBA::ULong millisecs, CORBA::Boolean verbose) :
  _my_proxy(proxy), _obj_name(objnm), _supply_fn(supply_fn),
  _num_events(0), _max_events(max_events),
  _millisecs(millisecs), _verbose(verbose), _done(0), _com_err(0),
  _oplock(), _finish(&_oplock), _worker(0)
{
  // providing explict NULL for supply_fn is not OK -- must have a valid function
  if (!_supply_fn) _supply_fn = sample_supply_any_fn;
}

LegacyPullSupplier_i*
LegacyPullSupplier_i::create(CosEA::EventChannel_ptr channel,
			     CORBA::ULong max_events,
			     const char* objnm,
			     supply_any_fn* supply_fn,
			     CORBA::ULong millisecs,
			     CORBA::Boolean verbose)
{
  // Obtain appropriate proxy object
  CosEA::ProxyPullConsumer_ptr proxy = CosEA::ProxyPullConsumer::_nil();
  CosEA::SupplierAdmin_var admin;
  try {
    admin = channel->for_suppliers();
    if ( CORBA::is_nil(admin) ) {
      cerr << "Failed to obtain admin" << endl;
      return 0; // failure
    }
  } catch (...) {
    cerr << "Failed to obtain admin" << endl;
    return 0;  // failure
  }
  if (verbose) cout << "Obtained admin from the channel" << endl;
  try {
    proxy = admin->obtain_pull_consumer();
  } catch (...) {  }
  if ( CORBA::is_nil(proxy) ) {
    cerr << "Failed to obtain proxy" << endl;
    return 0; // failure
  }
  if (verbose) cout << "Obtained proxy from admin" << endl;

    // Construct a client
  LegacyPullSupplier_i* client =
    new LegacyPullSupplier_i(proxy, max_events, objnm, supply_fn, millisecs, verbose);
  return client;
}

CORBA::Boolean LegacyPullSupplier_i::connect() {
  omni_mutex_lock l(_oplock);
  if (_done) return 0;
  _com_err = 0;
  try {
    _my_proxy->connect_pull_supplier(_this());
  } catch (CORBA::BAD_PARAM& ex) {
    cerr << _obj_name << ": BAD_PARAM Exception while connecting" << endl;
    return 1; // error
  } catch (CosEA::AlreadyConnected& ex) {
    cerr << _obj_name << ": Already connected" << endl;
    return 1; // error
  } catch (...) {
    cerr << _obj_name << ": Failed to connect" << endl;
    return 1; // error
  }
  if (_verbose) cout << _obj_name << ": Connected to proxy, ready to supply events" << endl; 
  // if _millisecs is set, spawn a thread to ping the proxy
  if (_millisecs)
    _worker = new GenericBoundWorkerThread(this);
  return 0; // OK
}

void LegacyPullSupplier_i::cleanup() {
  _oplock.lock();
  if (_worker || (!_done)) {
    cerr << "Coding error: only call c->cleanup() after c->wait_done()" << endl;
    _oplock.unlock();
    return;
  }
  // this method takes sole ownership of _my_proxy ref
  CosEA::ProxyPullConsumer_var proxy = _my_proxy;
  _my_proxy = CosEA::ProxyPullConsumer::_nil();
  // do not hold oplock while invoking disconnect
  _oplock.unlock();
  try {
    if ((!_com_err) && (!CORBA::is_nil(proxy)))
      proxy->disconnect_pull_consumer();
  } catch(...) {}
}

// While _millisecs is positive, pings proxy every _millisecs milliseconds.
// Sets error code and sets _done to 1 if there is an error during ping.
void* LegacyPullSupplier_i::start_working(void *) {
  CosEvC::PullSupplier_var bump_my_refcount_during_outcalls = _this();

  _oplock.lock();
  if (_verbose && (!_done) && _millisecs) cout << _obj_name << ": Spawned thread entering ping loop" << endl;
  // invariant: _oplock held at top of loop
  while ( 1 ) {
    if (_done || (_millisecs == 0)) break;
    unsigned long t_secs = 0, t_nanosecs = 0;
    unsigned long d_secs = 0, d_nanosecs = 0;
    d_secs = _millisecs / 1000;
    d_nanosecs = (_millisecs % 1000) * 1000000;
    _oplock.unlock(); // do not hold oplock across ping
    try {
      _my_proxy->connect_pull_supplier(_this());
    } catch (CosEA::AlreadyConnected& ex) {
      // this is the expected result -- continue pinging
    } catch (...) { // assume everything else is a communication error
      _oplock.lock();
      if (_done) break;
      if (_verbose) cout << _obj_name << ": communication error while pinging proxy using connect_pull_supplier" << endl;
      _done = 1;
      _finish.broadcast();
      _com_err = 1;
      break; // break from while loop -- done
    }
    _oplock.lock();
    if (_done) break; // must have disconnected during pull
    // sleep for specified interval
    omni_thread::get_time(&t_secs, &t_nanosecs, d_secs, d_nanosecs);
    _finish.timedwait(t_secs, t_nanosecs); // this release _oplock during wait
    // continue ping loop
  }
  // done
  _oplock.unlock();
  return 0;
}

CORBA::Any* LegacyPullSupplier_i::try_pull(CORBA::Boolean& has_event) {
  CORBA::Any* data = new CORBA::Any;
  has_event = 0;

  if (_done) return data; // see "Optimization Notes" at top of sample_clients.cc

  omni_mutex_lock l(_oplock);
  if (_done) return data;
  if (_verbose) cout << _obj_name << ": Channel called try_pull" << endl;

  // If we reached done point during last try_pull, it is broadcast now
  // and we return from this try_pull call without an event.
  // Waiting for one extra try_pull ensures the channel has successfully
  // processed the previous try_pull (the final supplied event) before
  // we disconnect this supplier.
  if (_max_events && (_num_events >= _max_events)) {
    if (_verbose) cout << _obj_name << ": DONE [max_events reached; we waited for 1 extra try_pull]" << endl;
    _done = 1;
    _finish.broadcast();
    if (_verbose) cout << _obj_name << ": NOT returning an event" << endl;
    return data;
  }

  if ((*_supply_fn)(*data, _obj_name, _num_events+1, _verbose)) {
    has_event = 1;
    _num_events++;
    // [rather than doing the following we let supply_fn report if it wants to]
    // if (_verbose) cout << _obj_name << ": Returning an event, event count = " << _num_events << endl;
  } else {
    if (_verbose) cout << _obj_name << ": NOT returning an event" << endl;
  }
  return data;
}

CORBA::Any* LegacyPullSupplier_i::pull() {
  CORBA::Any* data = new CORBA::Any;

  if (_done) return data; // see "Optimization Notes" at top of sample_clients.cc

  unsigned long t_secs = 0, t_nanosecs = 0;
  unsigned long d_secs = 0, d_nanosecs = 0;
  if (_millisecs) {
    d_secs = _millisecs / 1000;
    d_nanosecs = (_millisecs % 1000) * 1000000;
  }
  _oplock.lock();
  if (_verbose && (!_done)) cout << _obj_name << ": Channel called pull" << endl;
  // invariants: _oplock is held entering top of loop and 
  // also whenever we break out of loop
  // (it is *not* held during yield() and timedwait() calls)
  while ( 1 ) {
    if (_done) break; // must have disconnected
    if ((*_supply_fn)(*data, _obj_name, _num_events+1, _verbose)) {
      _num_events++;
      // [rather than doing the following we let supply_fn report if it wants to]
      // if (_verbose) cout << _obj_name << ": Returning an event, event count = " << _num_events << endl;
      if (_max_events && (_num_events >= _max_events)) {
	if (_verbose) cout << _obj_name << ": DONE [max_events reached]" << endl;
	_done = 1;
	_finish.broadcast();
      }
      break; // return data
    }
    // wait or yield then try again
    if (_millisecs) { // sleep for specified interval
      omni_thread::get_time(&t_secs, &t_nanosecs, d_secs, d_nanosecs);
      _finish.timedwait(t_secs, t_nanosecs); // this release _oplock during wait
    } else { // use yield to let other threads have a shot at _oplock
      _oplock.unlock();
      omni_thread::yield();
      _oplock.lock();
    }
  }
  _oplock.unlock();
  return data;
}

void LegacyPullSupplier_i::disconnect_pull_supplier()
{
  if (_done) return; // see "Optimization Notes" at top of sample_clients.cc

  omni_mutex_lock l(_oplock);
  if (_done) return;
  if (_verbose) cout << _obj_name << ": disconnected" << endl;
  _done = 1;
  _finish.broadcast();
}

CORBA::Boolean LegacyPullSupplier_i::wait_done() {
  _oplock.lock();
  while (!_done) {
    _finish.wait();
  }
  // using tmp_worker ensures only one wait_done call tries join()
  GenericBoundWorkerThread* tmp_worker = _worker;
  _worker = 0;
  _oplock.unlock();
  if (tmp_worker) {
    tmp_worker->join(0);
  }
  return _com_err;
}

