ecg_cdr_message_receiver.cpp

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

  if ( this->check_crc_ && header.crc != crc)
    {
      static unsigned int err_count = 0;
      ACE_ERROR ((LM_DEBUG,
                  "******************************\n"));

      ACE_ERROR ((LM_DEBUG,
                  "ERROR DETECTED \n"));

      if (crc == 0)
        {
          ACE_ERROR ((LM_DEBUG,
                      "Sending process may not have computed CRC \n"));
        }
      else
        {
          ACE_ERROR ((LM_DEBUG,
                      " NETWORK CRC CHECKSUM FAILED\n"));
        }

      ACE_ERROR ((LM_ERROR,
                  "Message was received from [%s:%s:%d] \n",
                  from.get_host_name (),
                  from.get_host_addr (),
                  from.get_port_number()));

      ACE_ERROR ((LM_ERROR,
                  "Num errors = %d \n",
                  ++err_count));
      ACE_ERROR ((LM_ERROR,
                  "This is a bad thing. Attempting to ignore ..\n"));

      return 0;
    }

  // Process received data.
  if (header.fragment_count == 1)
    {
      // Update <request_map_> to mark this request as completed. (Not
      // needed if we don't care about duplicates.)
      int result = this->mark_received (from, header.request_id);
      if (result != 1)
        return result;

      TAO_InputCDR cdr (data_buf, header.request_size, header.byte_order);
      if (cdr_processor->decode (cdr) == -1)
        return -1;
      else
        return 1;
    }

  else
    {
      return this->process_fragment (from, header, data_buf, cdr_processor);
    }
}

int
TAO_ECG_CDR_Message_Receiver::mark_received (const ACE_INET_Addr &from,
                                             CORBA::ULong request_id)
{
  // ACE_GUARD_RETURN (ACE_Lock, guard, *this->lock_, -1);

  Request_Map::ENTRY * entry = this->get_source_entry (from);
  if (!entry)
    return -1;

  TAO_ECG_UDP_Request_Entry ** request =
    entry->int_id_->get_request (request_id);

  if (request == 0)
    {
      ACE_DEBUG ((LM_WARNING, "Received mcast request with sequence"
                              "below currently expected range."));
      return 0;
    }
  if (*request == &Request_Completed_)
    {
      ACE_DEBUG ((LM_INFO, "Received duplicate mcast fragment. "
                           "(Request already complete)."));
      return 0;
    }
  if (*request != 0)
    {
      ACE_ERROR_RETURN ((LM_ERROR, "Inconsistent fragments for "
                                   "mcast request."),
                         -1);
    }

  *request = &Request_Completed_;
  return 1;
}

int
TAO_ECG_CDR_Message_Receiver::process_fragment (
                                   const ACE_INET_Addr &from,
                                   const Mcast_Header &header,
                                   char * data_buf,
                                   TAO_ECG_CDR_Processor *cdr_processor)
{
  // ACE_GUARD_RETURN (ACE_Lock, guard, *this->lock_, -1);

  Request_Map::ENTRY * source_entry = this->get_source_entry (from);
  if (!source_entry)
    return -1;

  TAO_ECG_UDP_Request_Entry ** request =
    source_entry->int_id_->get_request (header.request_id);

  if (request == 0)
    {
      ACE_DEBUG ((LM_WARNING, "Received mcast request with sequence "
                              "below currently expected range."));
      return 0;
    }
  if (*request == &Request_Completed_)
    {
      ACE_DEBUG ((LM_INFO, "Received duplicate mcast fragment. "
                           "(Request already complete)."));
      return 0;
    }
  if (*request == 0)
    // Entry for this request has not yet been allocated.
    {
      ACE_NEW_RETURN (*request,
                      TAO_ECG_UDP_Request_Entry (header.byte_order,
                                                 header.request_id,
                                                 header.request_size,
                                                 header.fragment_count),
                      -1);
    }

  // Validate the fragment.
  if ((*request)->validate_fragment (header.byte_order,
                                     header.request_size,
                                     header.fragment_size,
                                     header.fragment_offset,
                                     header.fragment_id,
                                     header.fragment_count) == 0)
    {
      ACE_ERROR_RETURN ((LM_ERROR,
                         "Received invalid mcast fragment."),
                        -1);
    }

  // Check whether this fragment was already received.
  if ((*request)->test_received (header.fragment_id) == 1)
    {
      ACE_DEBUG ((LM_INFO, "Received duplicate mcast fragment."));
      return 0;
    }

  // Add the fragment to the request entry.
  (*request)->mark_received (header.fragment_id);
  ACE_OS::memcpy ((*request)->fragment_buffer (header.fragment_offset),
                  data_buf,
                  header.fragment_size);

  // The request is not yet complete.
  if (!(*request)->complete ())
    {
      return 0;
    }

  // The request is complete - decode it.
  TAO_InputCDR cdr ((*request)->fragment_buffer (0),
                    header.request_size,
                    header.byte_order);

  if (cdr_processor->decode (cdr) == -1)
    return -1;

  delete *request;
  *request = &Request_Completed_;
  return 1;
}

TAO_ECG_CDR_Message_Receiver::Request_Map::ENTRY*
TAO_ECG_CDR_Message_Receiver::get_source_entry (const ACE_INET_Addr &from)
{
  // Get the entry for <from> from the <request_map_>.
  Request_Map::ENTRY * entry;

  if (this->request_map_.find (from, entry) == -1)
    {
      // Create an entry if one doesn't exist.
      Requests *requests = 0;
      ACE_NEW_RETURN (requests,
                      Requests,
                      0);
      auto_ptr<Requests> requests_aptr (requests);

      if (requests->init (this->max_requests_, this->min_purge_count_) == -1
          || this->request_map_.bind (from, requests, entry) == -1)
        {
          ACE_ERROR_RETURN ((LM_ERROR, "Unable to create hash map "
                                       "entry for a new request."),
                            0);
        }
      requests_aptr.release ();
    }

  return entry;
}
// ****************************************************************
int
TAO_ECG_CDR_Message_Receiver::Mcast_Header::read (char *header,
                                                  size_t bytes_received,
                                                  CORBA::Boolean checkcrc)
{
  // Decode.
  this->byte_order = header[0];
  if(this->byte_order != 0 && this->byte_order != 1)
    {
      ACE_ERROR_RETURN ((LM_ERROR, "Reading mcast packet header: byte "
                                   "order is neither 0 nor 1, it is %d.",
                         this->byte_order),
                        -1);
    }

  TAO_InputCDR header_cdr (header,
                           TAO_ECG_CDR_Message_Sender::ECG_HEADER_SIZE,
                           byte_order);
  CORBA::Boolean unused;
  CORBA::Octet a, b, c;
  if (!header_cdr.read_boolean (unused)
      || !header_cdr.read_octet (a)
      || !header_cdr.read_octet (b)
      || !header_cdr.read_octet (c)
      || a != 'A' || b != 'B' || c != 'C')
    {
      ACE_ERROR_RETURN ((LM_ERROR, "Error reading magic bytes "
                                   "in mcast packet header."),
                        -1);
    }

  if (!header_cdr.read_ulong (this->request_id)
      || !header_cdr.read_ulong (this->request_size)
      || !header_cdr.read_ulong (this->fragment_size)
      || !header_cdr.read_ulong (this->fragment_offset)
      || !header_cdr.read_ulong (this->fragment_id)
      || !header_cdr.read_ulong (this->fragment_count))
    {
      ACE_ERROR_RETURN ((LM_ERROR,
                        "Error decoding mcast packet header."),
                        -1);
    }

  if (checkcrc)
    {
      CORBA::Octet padding[4];
      header_cdr.read_octet_array (padding, 4);

      unsigned char *crcparts = (unsigned char *)(&this->crc);

      for (int cnt=0; cnt != 4; ++cnt)
        {
          crcparts[cnt] = padding[cnt];
        }

      this->crc = ntohl (this->crc);
    }

  // Validate.
  size_t data_bytes_received =
    bytes_received - TAO_ECG_CDR_Message_Sender::ECG_HEADER_SIZE;

  if (this->request_size < this->fragment_size
      || this->fragment_offset >= this->request_size
      || this->fragment_id >= this->fragment_count
      || (this->fragment_count == 1
          && (this->fragment_size != this->request_size
              || this->request_size != data_bytes_received)))
    {
      ACE_ERROR_RETURN ((LM_ERROR,
                        "Invalid mcast fragment: "
                        "inconsistent header fields."),
                        -1);
    }

  return 0;
}

// ****************************************************************
#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
template class ACE_Hash_Map_Manager<ACE_INET_Addr,TAO_ECG_CDR_Message_Receiver::Requests*,ACE_Null_Mutex>;
template class ACE_Hash_Map_Manager_Ex<ACE_INET_Addr, TAO_ECG_CDR_Message_Receiver::Requests*, ACE_Hash<ACE_INET_Addr>, ACE_Equal_To<ACE_INET_Addr>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Entry<ACE_INET_Addr,TAO_ECG_CDR_Message_Receiver::Requests*>;
template class ACE_Hash_Map_Iterator_Base_Ex<ACE_INET_Addr, TAO_ECG_CDR_Message_Receiver::Requests*, ACE_Hash<ACE_INET_Addr>, ACE_Equal_To<ACE_INET_Addr>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator<ACE_INET_Addr,TAO_ECG_CDR_Message_Receiver::Requests*,ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator_Ex<ACE_INET_Addr, TAO_ECG_CDR_Message_Receiver::Requests*, ACE_Hash<ACE_INET_Addr>, ACE_Equal_To<ACE_INET_Addr>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Reverse_Iterator<ACE_INET_Addr,TAO_ECG_CDR_Message_Receiver::Requests*,ACE_Null_Mutex>;
template class ACE_Hash_Map_Reverse_Iterator_Ex<ACE_INET_Addr, TAO_ECG_CDR_Message_Receiver::Requests*, ACE_Hash<ACE_INET_Addr>, ACE_Equal_To<ACE_INET_Addr>, ACE_Null_Mutex>;

#elif defined(ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)
#pragma instantiate ACE_Hash_Map_Manager<ACE_INET_Addr,TAO_ECG_CDR_Message_Receiver::Requests*,ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Manager_Ex<ACE_INET_Addr, TAO_ECG_CDR_Message_Receiver::Requests*, ACE_Hash<ACE_INET_Addr>, ACE_Equal_To<ACE_INET_Addr>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Entry<ACE_INET_Addr,TAO_ECG_CDR_Message_Receiver::Requests*>
#pragma instantiate ACE_Hash_Map_Iterator_Base_Ex<ACE_INET_Addr, TAO_ECG_CDR_Message_Receiver::Requests*, ACE_Hash<ACE_INET_Addr>, ACE_Equal_To<ACE_INET_Addr>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator<ACE_INET_Addr,TAO_ECG_CDR_Message_Receiver::Requests*,ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator_Ex<ACE_INET_Addr, TAO_ECG_CDR_Message_Receiver::Requests*, ACE_Hash<ACE_INET_Addr>, ACE_Equal_To<ACE_INET_Addr>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Reverse_Iterator<ACE_INET_Addr,TAO_ECG_CDR_Message_Receiver::Requests*,ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Reverse_Iterator_Ex<ACE_INET_Addr, TAO_ECG_CDR_Message_Receiver::Requests*, ACE_Hash<ACE_INET_Addr>, ACE_Equal_To<ACE_INET_Addr>, ACE_Null_Mutex>

#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */