Line data    Source code

       1             : /*
       2             :  *
       3             :  *
       4             :  * Distributed under the OpenDDS License.
       5             :  * See: http://www.opendds.org/license.html
       6             :  */
       7             : 
       8             : #include "ThreadSynch.h"
       9             : 
      10             : OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL
      11             : 
      12             : namespace OpenDDS {
      13             : namespace DCPS {
      14             : 
      15             : ACE_INLINE
      16             : TransportSendStrategy::SendMode TransportSendStrategy::mode() const
      17             : {
      18             :   DBG_ENTRY_LVL("TransportSendStrategy","mode",6);
      19             : 
      20             :   return mode_;
      21             : }
      22             : 
      23             : ACE_INLINE
      24             : ThreadSynch* TransportSendStrategy::synch() const
      25             : {
      26             :   DBG_ENTRY_LVL("TransportSendStrategy","synch",6);
      27             : 
      28             :   return synch_.get();
      29             : }
      30             : 
      31             : ACE_INLINE
      32             : void TransportSendStrategy::set_header_source(ACE_INT64 source)
      33             : {
      34             :   header_.source_ = source;
      35             : }
      36             : 
      37             : ACE_INLINE
      38           0 : void TransportSendStrategy::send_start()
      39             : {
      40             :   DBG_ENTRY_LVL("TransportSendStrategy","send_start",6);
      41             : 
      42           0 :   GuardType guard(this->lock_);
      43             : 
      44           0 :   if (!this->link_released_)
      45           0 :     ++this->start_counter_;
      46           0 : }
      47             : 
      48             : ACE_INLINE
      49           0 : void TransportSendStrategy::link_released(bool flag)
      50             : {
      51             :   DBG_ENTRY_LVL("TransportSendStrategy","link_released",6);
      52             : 
      53           0 :   GuardType guard(this->lock_);
      54           0 :   this->link_released_ = flag;
      55           0 : }
      56             : 
      57             : ACE_INLINE
      58           0 : void TransportSendStrategy::relink(bool)
      59             : {
      60             :   DBG_ENTRY_LVL("TransportSendStrategy","relink",6);
      61             :   // The subsclass needs implement this function for re-establishing
      62             :   // the link upon send failure.
      63           0 : }
      64             : 
      65             : ACE_INLINE
      66             : void TransportSendStrategy::suspend_send()
      67             : {
      68             :   DBG_ENTRY_LVL("TransportSendStrategy","suspend_send",6);
      69             :   GuardType guard(this->lock_);
      70             : 
      71             :   if (this->mode_ != MODE_TERMINATED && this->mode_ != MODE_SUSPEND) {
      72             :     this->mode_before_suspend_ = this->mode_;
      73             :     this->mode_ = MODE_SUSPEND;
      74             :   }
      75             : }
      76             : 
      77             : ACE_INLINE
      78           0 : void TransportSendStrategy::resume_send()
      79             : {
      80             :   DBG_ENTRY_LVL("TransportSendStrategy","resume_send",6);
      81           0 :   GuardType guard(this->lock_);
      82             : 
      83             :   // If this send strategy is reused when the connection is reestablished, then
      84             :   // we need re-initialize the mode_ and mode_before_suspend_.
      85           0 :   if (this->mode_ == MODE_TERMINATED) {
      86           0 :     this->header_.length_ = 0;
      87           0 :     this->pkt_chain_ = 0;
      88           0 :     this->header_complete_ = false;
      89           0 :     this->start_counter_ = 0;
      90           0 :     this->mode_ = MODE_DIRECT;
      91           0 :     this->mode_before_suspend_ = MODE_NOT_SET;
      92           0 :     this->delayed_delivered_notification_queue_.clear();
      93             : 
      94           0 :   } else if (this->mode_ == MODE_SUSPEND) {
      95           0 :     this->header_.length_ = 0;
      96           0 :     this->pkt_chain_ = 0;
      97           0 :     QueueType elems;
      98           0 :     elems.swap(this->elems_);
      99           0 :     this->mode_ = this->mode_before_suspend_;
     100           0 :     this->header_complete_ = false;
     101           0 :     this->mode_before_suspend_ = MODE_NOT_SET;
     102           0 :     if (this->queue_.size() > 0) {
     103           0 :       this->mode_ = MODE_QUEUE;
     104           0 :       this->synch_->work_available();
     105             :     }
     106             : 
     107           0 :   } else {
     108           0 :     ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: TransportSendStrategy::resume_send  The suspend or terminate"
     109             :                " is not called previously.\n"));
     110             :   }
     111           0 : }
     112             : 
     113             : ACE_INLINE
     114           0 : const char* TransportSendStrategy::mode_as_str(SendMode mode)
     115             : {
     116             :   static const char* SendModeStr[] = { "MODE_NOT_SET",
     117             :                                        "MODE_DIRECT",
     118             :                                        "MODE_QUEUE",
     119             :                                        "MODE_SUSPEND",
     120             :                                        "MODE_TERMINATED",
     121             :                                        "UNKNOWN"
     122             :                                      };
     123             : 
     124           0 :   return SendModeStr[mode];
     125             : }
     126             : 
     127             : ACE_INLINE
     128           0 : bool TransportSendStrategy::isDirectMode()
     129             : {
     130           0 :   return this->mode_ == MODE_DIRECT;
     131             : }
     132             : 
     133             : ACE_INLINE
     134           0 : ssize_t TransportSendStrategy::send_bytes(const iovec iov[], int n, int& /*bp*/)
     135             : {
     136           0 :   return send_bytes_i(iov, n);
     137             : }
     138             : 
     139             : ACE_INLINE
     140           0 : ACE_HANDLE TransportSendStrategy::get_handle()
     141             : {
     142           0 :   return ACE_INVALID_HANDLE;
     143             : }
     144             : 
     145             : 
     146             : ACE_INLINE
     147           0 : size_t TransportSendStrategy::max_message_size() const
     148             : {
     149           0 :   return 0;
     150             : }
     151             : 
     152             : ACE_INLINE
     153             : TransportQueueElement* TransportSendStrategy::current_packet_first_element() const
     154             : {
     155             :   return this->elems_.peek();
     156             : }
     157             : 
     158             : } // namespace DCPS
     159             : } // namespace OpenDDS
     160             : 
     161             : OPENDDS_END_VERSIONED_NAMESPACE_DECL