Changeset 10653 for source/ariba/communication/BaseCommunication.cpp

Timestamp:

Jul 25, 2012, 11:41:36 AM (12 years ago)

Author:

Michael Tänzer

Message:

Merge the ASIO branch back into trunk

File:

• source/ariba/communication/BaseCommunication.cpp (modified) (10 diffs)

source/ariba/communication/BaseCommunication.cpp

@@ -41 +41 @@
 #include "networkinfo/AddressDiscovery.h"
 #include "ariba/utility/types/PeerID.h"
+#include <boost/function.hpp>
 
 #ifdef UNDERLAY_OMNET
@@ -283 +284 @@
 SystemEventType MessageDispatchEvent("MessageDispatchEvent", TransportEvent );
 
-class DispatchMsg {
-public:
-        DispatchMsg() : local(NULL), remote(NULL), message(NULL) {}
-        address_v* local;
-        address_v* remote;
-        Message* message;
-};
-
 /// called when a system event is emitted by system queue
 void BaseCommunication::handleSystemEvent(const SystemEvent& event) {
@@ -297 +290 @@
         if ( event.getType() == MessageDispatchEvent ){
                 logging_debug( "Forwarding message receiver" );
-                DispatchMsg* dmsg = event.getData<DispatchMsg>();
-                Message* msg = dmsg->message;
-                receiveMessage(msg, dmsg->local, dmsg->remote);
-                msg->dropPayload();
-                delete dmsg->local;
-                delete dmsg->remote;
-                delete msg;
-                delete dmsg;
-        }
-}
-
-/// called when a message is received from transport_peer
-void BaseCommunication::receive_message(transport_protocol* transport,
-        const address_vf local, const address_vf remote, const uint8_t* data,
-        size_t size) {
-
-//      logging_debug( "Dispatching message" );
-
-        // convert data
-        Data data_( const_cast<uint8_t*>(data), size * 8 );
-        DispatchMsg* dmsg = new DispatchMsg();
-
-        Message* msg = new Message(data_);
-        dmsg->local = local->clone();
-        dmsg->remote = remote->clone();
-        dmsg->message = msg;
-
-        SystemQueue::instance().scheduleEvent(
-                SystemEvent( this, MessageDispatchEvent, dmsg )
-        );
-}
-
-/// handles a message from the underlay transport
-void BaseCommunication::receiveMessage(const Message* message,
-        const address_v* local, const address_v* remote ){
-
+                boost::function0<void>* handler = event.getData< boost::function0<void> >();
+                (*handler)();
+                delete handler;
+        }
+}
+
+/**
+ * called within the ASIO thread
+ * when a message is received from underlay transport
+ */ 
+void BaseCommunication::receive_message(transport_connection::sptr connection,
+        reboost::message_t msg) {
+
+        logging_debug( "Dispatching message" );
+        
+    boost::function0<void>* handler = new boost::function0<void>(
+            boost::bind(
+                    &BaseCommunication::receiveMessage,
+                    this,
+                    connection,
+                    msg)
+    );
+    
+    SystemQueue::instance().scheduleEvent(
+        SystemEvent(this, MessageDispatchEvent, handler)
+    );
+}
+
+/**
+ * called within the ARIBA thread (System Queue)
+ * when a message is received from underlay transport
+ */ 
+void BaseCommunication::receiveMessage(transport_connection::sptr connection,
+        reboost::message_t message)
+{
+    
+    //// Adapt to old message system ////
+    // Copy data
+    size_t bytes_len = message.size();
+    uint8_t* bytes = new uint8_t[bytes_len];
+    message.read(bytes, 0, bytes_len);
+    
+    Data data(bytes, bytes_len * 8);
+    
+    Message legacy_message;
+    legacy_message.setPayload(data);
+    
+    
+    
         /// decapsulate message
-        AribaBaseMsg* msg = ((Message*)message)->decapsulate<AribaBaseMsg>();
+        AribaBaseMsg* msg = legacy_message.decapsulate<AribaBaseMsg>();
         logging_debug( "Receiving message of type " << msg->getTypeString() );
 
@@ -379 +385 @@
                         LinkID localLink  = LinkID::create();
                         LinkID remoteLink = msg->getLocalLink();
-                        logging_debug( "local=" << local->to_string()
-                                << " remote=" << remote->to_string()
+                        logging_debug(
+                                "local=" << connection->getLocalEndpoint()->to_string()
+                                << " remote=" << connection->getRemoteEndpoint()->to_string()
                         );
 
@@ -386 +393 @@
                         bool allowlink = true;
                         BOOST_FOREACH( CommunicationEvents* i, eventListener ){
-                                allowlink &= i->onLinkRequest( localLink, local, remote );
+                                allowlink &= i->onLinkRequest( localLink,
+                                        connection->getLocalEndpoint(),
+                                        connection->getRemoteEndpoint());
                         }
 
@@ -400 +409 @@
                         ld->localLink = localLink;
                         ld->remoteLink = remoteLink;
-                        ld->localLocator = local->clone();
-                        ld->remoteLocator = remote->clone();
+                        ld->localLocator = connection->getLocalEndpoint()->clone();
+                        ld->remoteLocator = connection->getRemoteEndpoint()->clone();
+                        ld->connection = connection;
                         ld->remoteEndpoint = msg->getLocalDescriptor();
                         add_endpoint(ld->remoteLocator);
@@ -409 +419 @@
                                 ld->remoteLocator, endpoint_set::Layer1_3 | endpoint_set::NoLoopback);
                         localDescriptor.getEndpoints().add(
-                                local, endpoint_set::Layer1_3 | endpoint_set::NoLoopback);
+                                connection->getLocalEndpoint(),
+                                endpoint_set::Layer1_3 | endpoint_set::NoLoopback);
 
                         // link is now up-> add it
@@ -459 +470 @@
                         }
 
+                        // store the connection
+                        ld.connection = connection;
+                        
                         // set remote locator and link id
                         ld.remoteLink = msg->getLocalLink();
-                        ld.remoteLocator = remote->clone();
+                        ld.remoteLocator = connection->getRemoteEndpoint()->clone();
                         ld.remoteEndpoint.getEndpoints().add(
                                                         msg->getLocalDescriptor().getEndpoints(),
@@ -536 +550 @@
                         // update the remote locator
                         const address_v* oldremote = linkDesc.remoteLocator;
-                        linkDesc.remoteLocator = remote->clone();
+                        linkDesc.remoteLocator = connection->getRemoteEndpoint()->clone();
 
                         // inform the listeners (local link has _not_ changed!)
@@ -747 +761 @@
 
 /// sends a message to all end-points in the end-point descriptor
-void BaseCommunication::send(Message* message, const EndpointDescriptor& endpoint) {
-        Data data = data_serialize( message, DEFAULT_V );
-        transport->send( endpoint.getEndpoints(), data.getBuffer(), data.getLength() / 8);
-        data.release();
+void BaseCommunication::send(Message* legacy_message, const EndpointDescriptor& endpoint) {
+        Data data = data_serialize(legacy_message, DEFAULT_V);
+        
+        //// Adapt to new message system ////
+        // transfer data buffer ownership to the shared_buffer
+    reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8);
+        
+        reboost::message_t message;
+        message.push_back(buf);
+        
+        transport->send(endpoint.getEndpoints(), message);
 }
 
 /// sends a message to the remote locator inside the link descriptor
-void BaseCommunication::send(Message* message, const LinkDescriptor& desc) {
+void BaseCommunication::send(Message* legacy_message, const LinkDescriptor& desc) {
         if (desc.remoteLocator==NULL) return;
-        Data data = data_serialize( message, DEFAULT_V );
-        transport->send( desc.remoteLocator, data.getBuffer(), data.getLength() / 8);
-        data.release();
+        
+        Data data = data_serialize(legacy_message, DEFAULT_V);
+    
+    //// Adapt to new message system ////
+    // transfer data buffer ownership to the shared_buffer
+    reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8);
+    
+    reboost::message_t message;
+    message.push_back(buf);
+    
+        desc.connection->send(message);
 }