reactive_io.cpp

ACE源码

  this->was_active_ = 1;

  ssize_t count = ACE::send ( handle
                            , this->mb_->rd_ptr ()
                            , this->mb_->length ()
                            );

  if (count <= 0 && this->bytes_ == 0)
    {
      JAWS_Event_Result io_result ( 0
                                  , JAWS_Event_Result::JE_ERROR
                                  , JAWS_Event_Result::JE_SEND_FAIL
                                  );
      this->io_result_ = io_result;
    }
  else if (count <= 0 && this->bytes_ > 0)
    {
      JAWS_Event_Result io_result ( this->bytes_
                                  , JAWS_Event_Result::JE_ERROR
                                  , JAWS_Event_Result::JE_SEND_SHORT
                                  );
      this->io_result_ = io_result;
    }
  else
    {
      if (count > 0)
        this->mb_->rd_ptr (count);

      this->bytes_ += count;

      JAWS_Event_Result io_result ( this->bytes_
                                  , JAWS_Event_Result::JE_OK
                                  , JAWS_Event_Result::JE_SEND_OK
                                  );
      this->io_result_ = io_result;
    }

  if (count <= 0 || this->mb_->length () == 0)
    return -1;

  // Not done yet, so stay registered.
  return 0;
}


int
JAWS_IO_Reactive_Recv::handle_input (ACE_HANDLE handle)
{
  ssize_t count = ACE::recv ( handle
                            , this->mb_->wr_ptr ()
                            , this->mb_->space ()
                            );

  if (count < 0)
    {
      JAWS_Event_Result io_result ( 0
                                  , JAWS_Event_Result::JE_ERROR
                                  , JAWS_Event_Result::JE_RECV_FAIL
                                  );
      this->io_result_ = io_result;
    }
  else
    {
      if (count > 0)
        this->mb_->wr_ptr (count);

      JAWS_Event_Result io_result ( count
                                  , JAWS_Event_Result::JE_OK
                                  , JAWS_Event_Result::JE_RECV_OK
                                  );
      this->io_result_ = io_result;
    }

  return -1;
}


int
JAWS_IO_Reactive_Transmit::handle_output (ACE_HANDLE handle)
{
  this->was_active_ = 1;

  if (this->header_ && this->header_->length () > 0)
    return this->handle_output_header (handle);
  else
    this->header_ = 0;

  if (this->source_ != ACE_INVALID_HANDLE)
    return this->handle_output_source (handle);

  if (this->trailer_ && this->trailer_->length () > 0)
    return this->handle_output_trailer (handle);
  else
    this->trailer_ = 0;

  JAWS_Event_Result io_result ( this->bytes_
                              , JAWS_Event_Result::JE_OK
                              , JAWS_Event_Result::JE_TRANSMIT_OK
                              );
  this->io_result_ = io_result;

  return -1;
}

int
JAWS_IO_Reactive_Transmit::handle_output_header (ACE_HANDLE handle)
{
  return this->handle_output_mb (handle, this->header_);
}

int
JAWS_IO_Reactive_Transmit::handle_output_source (ACE_HANDLE handle)
{
  ACE_Message_Block *mb = this->source_buf_;

  // Try to read data into the mb if data is still available.
  if (mb->space () && this->source_ != ACE_INVALID_HANDLE)
    {
      ssize_t count;
      count = ACE_OS::read (this->source_, mb->wr_ptr (), mb->space ());

      if (count < 0)
        {
          this->source_ = ACE_INVALID_HANDLE;
          this->source_buf_ = 0;

          if (this->bytes_ == 0)
            {
              JAWS_Event_Result io_result ( 0
                                          , JAWS_Event_Result::JE_ERROR
                                          , JAWS_Event_Result::JE_TRANSMIT_FAIL
                                          );
              this->io_result_ = io_result;
            }
          else if (this->bytes_ > 0)
            {
              JAWS_Event_Result io_result ( this->bytes_
                                          , JAWS_Event_Result::JE_ERROR
                                          , JAWS_Event_Result::JE_TRANSMIT_SHORT
                                          );
              this->io_result_ = io_result;
            }

          return -1;
        }
      else if (count == 0)
        this->source_ = ACE_INVALID_HANDLE;
      else
        mb->wr_ptr (count);
    }

  int result = 0;

  if (mb->length () > 0)
    result = this->handle_output_mb (handle, mb);

  if (result < 0)
    {
      this->source_ = ACE_INVALID_HANDLE;
      this->source_buf_ = 0;
    }
  else if (mb == 0 && this->source_ == ACE_INVALID_HANDLE)
    this->source_buf_ = 0;
  else
    this->source_buf_->crunch ();

  return result;
}

int
JAWS_IO_Reactive_Transmit::handle_output_trailer (ACE_HANDLE handle)
{
  int result = this->handle_output_mb (handle, this->trailer_);

  if (result == 0 && this->trailer_ == 0)
    {
      JAWS_Event_Result io_result ( this->bytes_
                                  , JAWS_Event_Result::JE_ERROR
                                  , JAWS_Event_Result::JE_TRANSMIT_SHORT
                                  );
      this->io_result_ = io_result;
      return -1;
    }

  return result;
}

int
JAWS_IO_Reactive_Transmit::handle_output_mb ( ACE_HANDLE handle
                                            , ACE_Message_Block *&mb
                                            )
{
  ssize_t count = ACE::send (handle, mb->rd_ptr (), mb->length ());

  if (count <= 0 && this->bytes_ == 0)
    {
      JAWS_Event_Result io_result ( 0
                                  , JAWS_Event_Result::JE_ERROR
                                  , JAWS_Event_Result::JE_TRANSMIT_FAIL
                                  );
      this->io_result_ = io_result;
    }
  else if (count <= 0 && this->bytes_ > 0)
    {
      JAWS_Event_Result io_result ( this->bytes_
                                  , JAWS_Event_Result::JE_ERROR
                                  , JAWS_Event_Result::JE_TRANSMIT_SHORT
                                  );
      this->io_result_ = io_result;
    }
  else
    {
      mb->rd_ptr (count);
      this->bytes_ += count;
    }

  if (count <= 0)
    return -1;

  if (mb->length () == 0)
    mb = 0;

  return 0;
}

void
JAWS_IO_Reactive_Transmit::close (int result)
{
  if (result < 0)
    {
      JAWS_Event_Result io_result ( 0
                                  , JAWS_Event_Result::JE_ERROR
                                  , JAWS_Event_Result::JE_TRANSMIT_FAIL
                                  );
      this->io_result_ = io_result;

      this->handle_close (this->handle_, this->mask_);
    }
}

int
JAWS_IO_Reactive_Transmit::handle_timeout (const ACE_Time_Value &, const void *)
{
  if (this->was_active_)
    {
      this->was_active_ = 0;

      this->timer_id_ =
        ACE_Reactor::instance ()->schedule_timer (this, 0, this->tv_);

      return 0;
    }

  ACE_Reactor::instance ()
  ->remove_handler ( this
                   , ACE_Event_Handler::RWE_MASK|ACE_Event_Handler::DONT_CALL
                   );

  this->timer_id_ = -1;

  JAWS_Event_Result io_result ( 0
                              , JAWS_Event_Result::JE_ERROR
                              , JAWS_Event_Result::JE_TRANSMIT_TIMEOUT
                              , ETIME
                              );

  this->io_result_ = io_result;

  return -1;
}

int
JAWS_IO_Reactive_Transmit::handle_exception (ACE_HANDLE handle)
{
  if (handle == ACE_INVALID_HANDLE)
    {
      // We are being called back from a notify call.
      // This is our cue to register ourselves with the Reactor.

      int result;
      result =
        ACE_Reactor::instance ()
        ->register_handler (this, this->mask_|ACE_Event_Handler::EXCEPT_MASK);

      if (result < 0)
        this->close (result);

      return 0;
    }

  // back to our regularly scheduled except mask handling.

  JAWS_Event_Result io_result ( this->bytes_
                              , JAWS_Event_Result::JE_ERROR
                              , JAWS_Event_Result::JE_TRANSMIT_SHORT
                              );
  this->io_result_ = io_result;

  return -1;
}