Index: source/ariba/communication/BaseCommunication.cpp =================================================================== --- source/ariba/communication/BaseCommunication.cpp (revision 9694) +++ source/ariba/communication/BaseCommunication.cpp (revision 10653) @@ -41,4 +41,5 @@ #include "networkinfo/AddressDiscovery.h" #include "ariba/utility/types/PeerID.h" +#include #ifdef UNDERLAY_OMNET @@ -283,12 +284,4 @@ 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,42 +290,55 @@ if ( event.getType() == MessageDispatchEvent ){ logging_debug( "Forwarding message receiver" ); - DispatchMsg* dmsg = event.getData(); - 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(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* handler = event.getData< boost::function0 >(); + (*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* handler = new boost::function0( + 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* msg = legacy_message.decapsulate(); logging_debug( "Receiving message of type " << msg->getTypeString() ); @@ -379,6 +385,7 @@ 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,5 +393,7 @@ bool allowlink = true; BOOST_FOREACH( CommunicationEvents* i, eventListener ){ - allowlink &= i->onLinkRequest( localLink, local, remote ); + allowlink &= i->onLinkRequest( localLink, + connection->getLocalEndpoint(), + connection->getRemoteEndpoint()); } @@ -400,6 +409,7 @@ 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,5 +419,6 @@ 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,7 +470,10 @@ } + // 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,5 +550,5 @@ // 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,16 +761,31 @@ /// 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); } Index: source/ariba/communication/BaseCommunication.h =================================================================== --- source/ariba/communication/BaseCommunication.h (revision 9694) +++ source/ariba/communication/BaseCommunication.h (revision 10653) @@ -60,4 +60,5 @@ #include "ariba/utility/addressing/addressing.hpp" #include "ariba/utility/transport/transport.hpp" +#include "ariba/utility/transport/transport_connection.hpp" // communication @@ -190,14 +191,19 @@ virtual void handleSystemEvent(const SystemEvent& event); - /// called when a message is received form transport_peer - virtual void receive_message(transport_protocol* transport, - const address_vf local, const address_vf remote, const uint8_t* data, - size_t size); + /** + * called within the ASIO thread + * when a message is received from underlay transport + */ + virtual void receive_message(transport_connection::sptr connection, + reboost::message_t msg); protected: - /// handle received message from a transport module - void receiveMessage(const Message* message, - const address_v* local, const address_v* remote ); + /** + * called within the ARIBA thread (System Queue) + * when a message is received from underlay transport + */ + void receiveMessage(transport_connection::sptr connection, + reboost::message_t msg); /// called when a network interface change happens @@ -250,4 +256,7 @@ /// flag, whether this link is up bool up; + + /// connection if link is up + transport_connection::sptr connection; }; Index: source/ariba/communication/networkinfo/AddressDiscovery.cpp =================================================================== --- source/ariba/communication/networkinfo/AddressDiscovery.cpp (revision 9694) +++ source/ariba/communication/networkinfo/AddressDiscovery.cpp (revision 10653) @@ -129,5 +129,5 @@ ip_address ip = straddr; if (ip.is_loopback()) continue; - if (ip.is_link_local()) continue; +// if (ip.is_link_local()) continue; address_vf vf = ip; endpoints.add( vf ); Index: source/ariba/communication/networkinfo/NetworkInformation.h =================================================================== --- source/ariba/communication/networkinfo/NetworkInformation.h (revision 9694) +++ source/ariba/communication/networkinfo/NetworkInformation.h (revision 10653) @@ -40,4 +40,5 @@ #define __NETWORK_INFORMATION_H +#include #include #include