io.cpp

ACE自适配通信环境（ADAPTIVE Communication Environment）是可以自由使用、开放源码的面向对象（OO）框架（Framework）

// IO.cpp,v 1.10 2006/01/20 22:58:08 spaunov Exp

#include "ace/Message_Block.h"
#include "ace/SOCK_Stream.h"
#include "ace/Filecache.h"
#include "ace/OS_NS_string.h"
#include "ace/OS_NS_unistd.h"
#include "ace/OS_NS_sys_uio.h"
#include "ace/OS_NS_sys_socket.h"
#include "ace/Min_Max.h"

#include "JAWS/JAWS.h"
#include "JAWS/Data_Block.h"
#include "JAWS/Policy.h"
#include "JAWS/IO.h"
#include "JAWS/IO_Handler.h"
#include "JAWS/IO_Acceptor.h"
#include "JAWS/Filecache.h"

#include "ace/Asynch_IO.h"  //for ACE_Asynch_Write_Stream

// #include "HTTP_Helpers.h"

ACE_RCSID(JAWS, IO, "IO.cpp,v 1.10 2006/01/20 22:58:08 spaunov Exp")

JAWS_IO::JAWS_IO (void)
  : handle_ (ACE_INVALID_HANDLE),
    handler_ (0),
    inet_addr_ (0),
    acceptor_ (0)
{
}

JAWS_IO::~JAWS_IO (void)
{
}

#if 0
ACE_HANDLE
JAWS_IO::handle (void)
{
  return this->handle_;
}

void
JAWS_IO::handle (ACE_HANDLE handle)
{
  this->handle_ = handle;
}

void
JAWS_IO::handler (JAWS_IO_Handler *handler)
{
  this->handler_ = handler;
}

void
JAWS_IO::acceptor (JAWS_IO_Acceptor *acceptor)
{
  this->acceptor_ = acceptor;
}
#endif /* 0 */

JAWS_Synch_IO::JAWS_Synch_IO (void)
{
  this->acceptor_ = JAWS_IO_Synch_Acceptor_Singleton::instance ();
}

JAWS_Synch_IO::~JAWS_Synch_IO (void)
{
  if (this->handle_ != ACE_INVALID_HANDLE)
    ACE_OS::closesocket (this->handle_);
}

void
JAWS_Synch_IO::accept (JAWS_IO_Handler *ioh,
                       ACE_Message_Block *,
                       unsigned int)
{
  ACE_SOCK_Stream new_stream;
  new_stream.set_handle (ACE_INVALID_HANDLE);
  if (this->acceptor_->accept (new_stream) == -1)
    ioh->accept_error ();
  else
    ioh->accept_complete (new_stream.get_handle ());
}

void
JAWS_Synch_IO::read (JAWS_IO_Handler *ioh,
                     ACE_Message_Block *mb,
                     unsigned int size)
{
  JAWS_TRACE ("JAWS_Synch_IO::read");

  ACE_SOCK_Stream stream;

  stream.set_handle (ioh->handle ());
  int result = stream.recv (mb->wr_ptr (), size);

  if (result <= 0)
    ioh->read_error ();
  else
    {
      JAWS_TRACE ("JAWS_Synch_IO::read success");
      mb->wr_ptr (result);
      ioh->read_complete (mb);
    }
}

void
JAWS_Synch_IO::receive_file (JAWS_IO_Handler *ioh,
                             const char *filename,
                             void *initial_data,
                             unsigned int initial_data_length,
                             unsigned int entire_length)
{
  ACE_Filecache_Handle handle (filename,
                               (int) entire_length);

  int result = handle.error ();

  if (result == ACE_Filecache_Handle::ACE_SUCCESS)
    {
      ACE_SOCK_Stream stream;
      stream.set_handle (ioh->handle ());

      int bytes_to_memcpy = ACE_MIN (entire_length, initial_data_length);
      ACE_OS::memcpy (handle.address (), initial_data, bytes_to_memcpy);

      int bytes_to_read = entire_length - bytes_to_memcpy;

      int bytes = stream.recv_n ((char *)
                                 handle.address () + initial_data_length,
                                 bytes_to_read);
      if (bytes == bytes_to_read)
        ioh->receive_file_complete ();
      else
        result = -1;
    }

  if (result != ACE_Filecache_Handle::ACE_SUCCESS)
    ioh->receive_file_error (result);
}

void
JAWS_Synch_IO::transmit_file (JAWS_IO_Handler *ioh,
                              ACE_HANDLE handle,
                              const char *header,
                              unsigned int header_size,
                              const char *trailer,
                              unsigned int trailer_size)
{
  int result = 0;

  if (handle != ACE_INVALID_HANDLE)
    {
      ACE_SOCK_Stream stream;
      stream.set_handle (ioh->handle ());

      if ((unsigned long) stream.send_n (header, header_size) < header_size)
        {
          result = -1;
        }
      else
        {
          int count;
          char buf[BUFSIZ];

          do
            {
              count = ACE_OS::read (handle, buf, sizeof (buf));
              if (count <= 0)
                break;

              if (stream.send_n (buf, count) < count)
                {
                result = -1;
                }
            }
          while (result == 0);

          if ((unsigned long) stream.send_n (trailer, trailer_size)
              < trailer_size)
            {
            result = -1;
            }
        }
    }

  if (result == 0)
    ioh->transmit_file_complete ();
  else
    ioh->transmit_file_error (result);
}

void
JAWS_Synch_IO::transmit_file (JAWS_IO_Handler *ioh,
                              const char *filename,
                              const char *header,
                              unsigned int header_size,
                              const char *trailer,
                              unsigned int trailer_size)
{
  int result = 0;

  if (filename == 0)
    {
      ioh->transmit_file_error (-1);
      return;
    }

  JAWS_Cached_FILE cf (filename);

  if (cf.file ()->get_handle () != ACE_INVALID_HANDLE
      && cf.mmap () != 0)
    {
#if defined (ACE_JAWS_BASELINE) || defined (ACE_WIN32)
      ACE_FILE_Info info;
      cf.file ()->get_info (info);

      if (cf.file ()->get_info (info) == 0 && info.size_ > 0)
        {
          ACE_SOCK_Stream stream;
          stream.set_handle (ioh->handle ());
          if (((u_long) stream.send_n (header, header_size) == header_size)
              && (stream.send_n (cf.mmap ()->addr (), info.size_)
                  == info.size_)
              && ((u_long) stream.send_n (trailer, trailer_size)
                  == trailer_size))
            {
              ioh->transmit_file_complete ();
              return;
            }
          else
            {
              result = -1;
            }
        }
      else
        {
          result = -1;
        }
#else
      // Attempting to use writev
      // Is this faster?
      iovec iov[3];
      int iovcnt = 0;
      if (header_size > 0)
        {
          iov[iovcnt].iov_base = const_cast<char*> (header);
          iov[iovcnt].iov_len =  header_size;
          iovcnt++;
        }

      ACE_FILE_Info info;

      if (cf.file ()->get_info (info) == 0 && info.size_ > 0)
        {
          iov[iovcnt].iov_base = (char *) cf.mmap ()->addr ();
          iov[iovcnt].iov_len = info.size_;
          iovcnt++;
        }
      if (trailer_size > 0)
        {
          iov[iovcnt].iov_base = const_cast<char*> (trailer);
          iov[iovcnt].iov_len = trailer_size;
          iovcnt++;
        }
      if (ACE_OS::writev (ioh->handle (), iov, iovcnt) < 0)
        {
          result = -1;
        }
      else
        {
          ioh->transmit_file_complete ();
          return;
        }
#endif /* ACE_JAWS_BASELINE */
    }
  else if (cf.file ()->get_handle () != ACE_INVALID_HANDLE
           && cf.mmap () == 0)
    {
      this->transmit_file (ioh,
                           cf.file ()->get_handle (),
                           header, header_size,
                           trailer, trailer_size);
      return;
    }
  else
    {
      result = -1;
    }

  if (result != 0)
    {
      ioh->transmit_file_error (result);
    }
}

void
JAWS_Synch_IO::send_confirmation_message (JAWS_IO_Handler *ioh,
                                          const char *buffer,
                                          unsigned int length)
{
  this->send_message (ioh, buffer, length);
  ioh->confirmation_message_complete ();