Merge pull request #4120 from jrw972/async-tcp-on-start-callbacks

TcpDataLink fails to send association message

Author: jrw972

dds/DCPS/transport/framework/TransportClient.cpp

@@ -203,6 +203,7 @@ TransportClient::associate(const AssociationData& data, bool active)
   ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, lock_, false);
 
   repo_id_ = repo_id;
+  OPENDDS_ASSERT(repo_id_ != GUID_UNKNOWN);
 
   if (impls_.empty()) {
     if (DCPS_debug_level) {
@@ -248,6 +249,7 @@ TransportClient::associate(const AssociationData& data, bool active)
     pa->impls_.clear();
     pa->blob_index_ = 0;
     pa->data_ = data;
+    OPENDDS_ASSERT(repo_id_ != GUID_UNKNOWN);
     pa->attribs_.local_id_ = repo_id_;
     pa->attribs_.priority_ = get_priority_value(data);
     pa->attribs_.local_reliable_ = reliable_;
@@ -621,9 +623,9 @@ TransportClient::add_link(const DataLink_rch& link, const GUID_t& peer)
 
   TransportReceiveListener_rch trl = get_receive_listener();
 
+  OPENDDS_ASSERT(repo_id_ != GUID_UNKNOWN);
   if (trl) {
     link->make_reservation(peer, repo_id_, trl, reliable_);
-
   } else {
     link->make_reservation(peer, repo_id_, get_send_listener(), reliable_);
   }
@@ -745,6 +747,7 @@ TransportClient::disassociate(const GUID_t& peerId)
                link.in()));
   }
 
+  OPENDDS_ASSERT(repo_id_ != GUID_UNKNOWN);
   link->release_reservations(peerId, repo_id_, released);
 
   if (!released.empty()) {
@@ -777,6 +780,11 @@ void TransportClient::transport_stop()
   const GUID_t repo_id = repo_id_;
   guard.release();
 
+  if (repo_id == GUID_UNKNOWN) {
+    // Not associated so nothing to stop.
+    return;
+  }
+
   for (size_t i = 0; i < impls.size(); ++i) {
     const TransportImpl_rch impl = impls[i].lock();
     if (impl) {
@@ -1112,6 +1120,9 @@ SendControlStatus
 TransportClient::send_control(const DataSampleHeader& header,
                               Message_Block_Ptr msg)
 {
+  if (repo_id_ == GUID_UNKNOWN) {
+    return SEND_CONTROL_OK;
+  }
   return links_.send_control(repo_id_, get_send_listener(), header, move(msg));
 }
 
@@ -1120,6 +1131,10 @@ TransportClient::send_control_to(const DataSampleHeader& header,
                                  Message_Block_Ptr msg,
                                  const GUID_t& destination)
 {
+  if (repo_id_ == GUID_UNKNOWN) {
+    return SEND_CONTROL_OK;
+  }
+
   DataLinkSet singular;
   {
     ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, lock_, SEND_CONTROL_ERROR);
@@ -1143,6 +1158,9 @@ TransportClient::remove_sample(const DataSampleElement* sample)
 bool
 TransportClient::remove_all_msgs()
 {
+  if (repo_id_ == GUID_UNKNOWN) {
+    return true;
+  }
   return links_.remove_all_msgs(repo_id_);
 }
 

dds/DCPS/transport/framework/TransportControlElement.cpp

@@ -13,10 +13,11 @@
 #include "TransportControlElement.inl"
 #endif /* __ACE_INLINE__ */
 
-OpenDDS::DCPS::TransportControlElement::TransportControlElement(
-  Message_Block_Ptr msg_block
-) : TransportQueueElement(1),
-    msg_( msg_block.release())
+OpenDDS::DCPS::TransportControlElement::TransportControlElement(Message_Block_Ptr msg_block,
+                                                                const GUID_t& publication_id)
+  : TransportQueueElement(1)
+  , msg_( msg_block.release())
+  , publication_id_(publication_id)
 {
   DBG_ENTRY_LVL("TransportControlElement", "TransportControlElement", 6);
 }

dds/DCPS/transport/framework/TransportControlElement.h

@@ -31,7 +31,8 @@ class OpenDDS_Dcps_Export TransportControlElement
    * msg_block - chain of ACE_Message_Blocks containing the control
    *             sample held by this queue element, if any.
    */
-  explicit TransportControlElement(Message_Block_Ptr msg_block);
+  explicit TransportControlElement(Message_Block_Ptr msg_block,
+                                   const GUID_t& publication_id = GUID_UNKNOWN);
 
   virtual ~TransportControlElement();
 
@@ -57,6 +58,7 @@ class OpenDDS_Dcps_Export TransportControlElement
   TransportControlElement(const TransportControlElement&);
   /// The control message.
   Message_Block_Ptr msg_;
+  GUID_t publication_id_;
 };
 
 } // namespace DCPS

dds/DCPS/transport/framework/TransportControlElement.inl

@@ -26,7 +26,7 @@ ACE_INLINE
 OpenDDS::DCPS::GUID_t
 OpenDDS::DCPS::TransportControlElement::publication_id() const
 {
-  return GUID_UNKNOWN;
+  return publication_id_;
 }
 
 ACE_INLINE

dds/DCPS/transport/tcp/TcpDataLink.cpp

@@ -554,7 +554,7 @@ OpenDDS::DCPS::TcpDataLink::send_association_msg(const GUID_t& local, const GUID
   Serializer ser(message.get(), encoding_unaligned_native);
   ser << remote;
 
-  TransportControlElement* send_element = new TransportControlElement(move(message));
+  TransportControlElement* send_element = new TransportControlElement(move(message), local);
 
   this->send_i(send_element, false);
 }