random.cpp

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

      return 1;
    }
  ACE_ENDTRY;
  return 0;
}

void
RND_Driver::timer (const RtecEventComm::Event &e
                   ACE_ENV_ARG_DECL)
{
  int r = ACE_OS::rand ();
  if (r < 0)
    r = -r;

  int n = r% 20;

  switch (n)
    {
    case 0:
    case 1:
      {
        // ACE_DEBUG ((LM_DEBUG, "Pushing an event\n"));
        if (e.header.source < this->max_recursion_)
          {
            RtecEventComm::EventSet event (1);
            event.length (1);
            event[0] = e;
            event[0].header.source ++;
            this->supplier_.push (event ACE_ENV_ARG_PARAMETER);
            ACE_CHECK;
          }
      }
      break;

    default:
    case 2:
    case 3:
    case 4:
    case 5:
      // ACE_DEBUG ((LM_DEBUG, "Received event\n"));
      break;

    case 6:
      {
        int n = ACE_OS::rand () % this->nsuppliers_;

        // ACE_DEBUG ((LM_DEBUG, "Connecting supplier %d\n", n));

        ACE_SupplierQOS_Factory qos;
        qos.insert (0, base_type, 0, 1);

        this->suppliers_[n]->connect (this->supplier_admin_.in (),
                                      qos.get_SupplierQOS ()
                                      ACE_ENV_ARG_PARAMETER);
        ACE_CHECK;
      }
      break;

    case 7:
      {
        int n = ACE_OS::rand () % this->nconsumers_;

        // ACE_DEBUG ((LM_DEBUG, "Connecting consumer %d\n", n));

        ACE_ConsumerQOS_Factory qos;
        qos.start_disjunction_group ();
        qos.insert_type (base_type, 0);

        this->consumers_[n]->connect (this->consumer_admin_.in (),
                                      qos.get_ConsumerQOS ()
                                      ACE_ENV_ARG_PARAMETER);
        ACE_CHECK;
      }
      break;

    case 8:
      {
        int n = ACE_OS::rand () % this->nsuppliers_;

        // ACE_DEBUG ((LM_DEBUG, "Disconnecting supplier %d\n", n));

        this->suppliers_[n]->disconnect (ACE_ENV_SINGLE_ARG_PARAMETER);
        ACE_CHECK;
      }
      break;

    case 9:
      {
        int n = ACE_OS::rand () % this->nconsumers_;

        // ACE_DEBUG ((LM_DEBUG, "Disconnecting consumer %d\n", n));

        this->consumers_[n]->disconnect (ACE_ENV_SINGLE_ARG_PARAMETER);
        ACE_CHECK;
      }
      break;
    }
}

void
RND_Driver::event (const RtecEventComm::Event &e
                   ACE_ENV_ARG_DECL)
{
  this->timer (e ACE_ENV_ARG_PARAMETER);
}

// ****************************************************************

void
RND_Timer::push (const RtecEventComm::EventSet &event
                 ACE_ENV_ARG_DECL)
    ACE_THROW_SPEC ((CORBA::SystemException))
{
  ACE_TRY
    {
      this->driver_->timer (event[0] ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
    }
  ACE_CATCHANY
    {
    }
  ACE_ENDTRY;
}

// ****************************************************************

void
RND_Consumer::connect (RtecEventChannelAdmin::ConsumerAdmin_ptr admin,
                       const RtecEventChannelAdmin::ConsumerQOS &qos
                       ACE_ENV_ARG_DECL)
{
  RtecEventChannelAdmin::ProxyPushSupplier_var proxy;
  {
    ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

    if (CORBA::is_nil (this->proxy_.in ()))
      {
        this->proxy_ = admin->obtain_push_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
        ACE_CHECK;
      }
    proxy =
      RtecEventChannelAdmin::ProxyPushSupplier::_duplicate(this->proxy_.in ());
  }
  RtecEventComm::PushConsumer_var me =
    this->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;
  proxy->connect_push_consumer (me.in (),
                                qos
                                ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
RND_Consumer::disconnect (ACE_ENV_SINGLE_ARG_DECL)
{
  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

  if (CORBA::is_nil (this->proxy_.in ()))
    return;
  this->proxy_->disconnect_push_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;
  this->proxy_ =
    RtecEventChannelAdmin::ProxyPushSupplier::_nil ();
}

void
RND_Consumer::push (const RtecEventComm::EventSet &event
                    ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  this->driver_->event (event[0] ACE_ENV_ARG_PARAMETER);
}

void
RND_Consumer::disconnect_push_consumer (ACE_ENV_SINGLE_ARG_DECL_NOT_USED)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
}

// ****************************************************************

void
RND_Supplier::connect (RtecEventChannelAdmin::SupplierAdmin_ptr admin,
                       const RtecEventChannelAdmin::SupplierQOS &qos
                       ACE_ENV_ARG_DECL)
{
  RtecEventChannelAdmin::ProxyPushConsumer_var proxy;
  {
    ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

    if (CORBA::is_nil (this->proxy_.in ()))
      {
        this->proxy_ = admin->obtain_push_consumer (ACE_ENV_SINGLE_ARG_PARAMETER);
        ACE_CHECK;
      }

    proxy =
      RtecEventChannelAdmin::ProxyPushConsumer::_duplicate(this->proxy_.in ());
  }
  RtecEventComm::PushSupplier_var me =
    this->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;
  proxy->connect_push_supplier (me.in (),
                                qos
                                ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
RND_Supplier::disconnect (ACE_ENV_SINGLE_ARG_DECL)
{
  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

  if (CORBA::is_nil (this->proxy_.in ()))
    return;
  this->proxy_->disconnect_push_consumer (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;
  this->proxy_ =
    RtecEventChannelAdmin::ProxyPushConsumer::_nil ();
}

void
RND_Supplier::push_new_event (ACE_ENV_SINGLE_ARG_DECL)
{
  RtecEventComm::EventSet event (1);
  event.length (1);
  event[0].header.type   = base_type;
  event[0].header.source = 0;

  this->push (event ACE_ENV_ARG_PARAMETER);
}

void
RND_Supplier::push (RtecEventComm::EventSet &event
                    ACE_ENV_ARG_DECL)
{
  RtecEventChannelAdmin::ProxyPushConsumer_var proxy;
  {
    ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

    if (CORBA::is_nil (this->proxy_.in ()))
      return;

    proxy =
      RtecEventChannelAdmin::ProxyPushConsumer::_duplicate(this->proxy_.in ());
  }

  proxy->push (event ACE_ENV_ARG_PARAMETER);
}

void
RND_Supplier::disconnect_push_supplier (ACE_ENV_SINGLE_ARG_DECL_NOT_USED)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
}

int
RND_Supplier::svc (void)
{
  ACE_DEBUG ((LM_DEBUG, "Thread %t started\n"));
  int percent = 10;
  int niterations = 5000;
  for (int i = 0; i != niterations; ++i)
    {
      ACE_DECLARE_NEW_CORBA_ENV;
      ACE_TRY
        {
          ACE_Time_Value tv (0, 10000);
          ACE_OS::sleep (tv);

          this->push_new_event (ACE_ENV_SINGLE_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
      ACE_CATCHANY
        {
        }
      ACE_ENDTRY;
      if (this->verbose_
          && i * 100 / niterations >= percent)
        {
          ACE_DEBUG ((LM_DEBUG, "Thread %t %d%%\n", percent));
          percent += 10;
        }
    }
  ACE_DEBUG ((LM_DEBUG, "Thread %t completed\n"));
  return 0;
}