Index: source/ariba/utility/messages/Message.cpp =================================================================== --- source/ariba/utility/messages/Message.cpp (revision 10700) +++ source/ariba/utility/messages/Message.cpp (revision 12060) @@ -41,4 +41,6 @@ #include "ariba/utility/serialization/DataStream.hpp" +#include "ariba/utility/logging/Logging.h" + NAMESPACE_BEGIN @@ -80,15 +82,59 @@ } + +reboost::message_t Message::wrap_up_for_sending() +{ + assert( ! wrapped_up ); + wrapped_up = true; + + //// Adapt to new message system //// + Data data = data_serialize(this, DEFAULT_V); + reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8); + + newstyle_payload.push_front(buf); + + return newstyle_payload; +} + +reboost::shared_buffer_t Message::serialize_into_shared_buffer() +{ + assert ( newstyle_payload.length() == 0 ); + + //// Adapt to new message system //// + Data data = data_serialize(this, DEFAULT_V); + reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8); + + return buf; +} + + +reboost::shared_buffer_t Message::deserialize_from_shared_buffer(reboost::shared_buffer_t buff) +{ + // NOTE: legacy payload is not allowed when using shared buffers + this->legacy_payload_disabled = true; + + assert( buff.size() > 0 ); + + // const_cast is necessary here, but without legacy payload we should be save here (more or less) + Data dat(const_cast(buff.data()), buff.size() * 8); + + size_t len = this->SERIALIZATION_METHOD_NAME(DESERIALIZE, dat) / 8; + + // return remaining sub-buffer + return buff(len); +} + + + NAMESPACE_END std::ostream& operator<<(std::ostream& stream, const ariba::utility::Message& msg ) { using_serialization; - stream << "msg(type=" << typeid(msg).name(); + stream << "msg(type=" << typeid(msg).name() << ","; stream << "len=" << (data_length(&msg)/8) << ","; Data data = data_serialize(&msg); - stream << ",data=" << data; + stream << "data=" << data; data.release(); stream << ")"; return stream; } - Index: source/ariba/utility/messages/Message.h =================================================================== --- source/ariba/utility/messages/Message.h (revision 10700) +++ source/ariba/utility/messages/Message.h (revision 12060) @@ -62,4 +62,8 @@ #include "ariba/utility/serialization.h" +// reboost messages +#include "ariba/utility/transport/messages/message.hpp" + + std::ostream& operator<<(std::ostream& stream, const ariba::utility::Message& msg ); @@ -82,10 +86,12 @@ friend std::ostream& ::operator<<(std::ostream& stream, const ariba::utility::Message& msg ); - // root binary data - shared_array root; - // payload + bool legacy_payload_disabled; bool releasePayload; Data payload; //< messages binary data + + // XXX testing... + reboost::message_t newstyle_payload; + bool wrapped_up; // addresses and control info @@ -98,5 +104,6 @@ */ inline Message() : - root(), releasePayload(true), payload(), srcAddr(NULL),destAddr(NULL) { + legacy_payload_disabled(false), releasePayload(true), payload(), + newstyle_payload(), wrapped_up(false), srcAddr(NULL),destAddr(NULL) { } @@ -106,5 +113,6 @@ */ explicit inline Message( const Data& data ) : - releasePayload(true), srcAddr(NULL),destAddr(NULL) { + legacy_payload_disabled(false), releasePayload(true), + newstyle_payload(), wrapped_up(false), srcAddr(NULL),destAddr(NULL) { // FIXME newstyle_payload..? this->payload = data.clone(); // this->root = shared_array((uint8_t*)data.getBuffer()); @@ -225,4 +233,37 @@ return decapsulate(); } + + + // XXX testing + void set_payload_message(reboost::message_t msg) + { + newstyle_payload = msg; + } + + void append_buffer(reboost::shared_buffer_t buff) + { + newstyle_payload.push_back(buff); + } + + + // XXX testing... packs this message into the payload message (do not use twice!!) + virtual reboost::message_t wrap_up_for_sending(); + + + /** + * Uses the old serialization system to serialize itself into a (new style) shared buffer. + */ + virtual reboost::shared_buffer_t serialize_into_shared_buffer(); + + /* + * XXX experimental + * + * Uses the old serialization system to deserialize itself out of a (new style) shared buffer. + * @return remaining sub-buffer (the "payload") + * + * Note: This is some kind of a hack! handle with care. + */ + virtual reboost::shared_buffer_t deserialize_from_shared_buffer(reboost::shared_buffer_t buff); + protected: @@ -262,5 +303,8 @@ * @return A explicit payload serializer */ - finline PayloadSerializer Payload( size_t length = ~0 ) { + finline PayloadSerializer Payload( size_t length = ~0 ) + { +// assert( ! legacy_payload_disabled ); // FIXME aktuell + return PayloadSerializer( this, length ); } Index: source/ariba/utility/messages/MessageProvider.cpp =================================================================== --- source/ariba/utility/messages/MessageProvider.cpp (revision 10700) +++ source/ariba/utility/messages/MessageProvider.cpp (revision 12060) @@ -37,32 +37,32 @@ // [License] -#include "MessageProvider.h" - -NAMESPACE_BEGIN - -MessageProvider::MessageProvider() { -} - -MessageProvider::~MessageProvider() { -} - -bool MessageProvider::sendMessageToReceivers( const Message* message ) { - bool sent = false; - for (size_t i=0; ireceiveMessage(message, LinkID::UNSPECIFIED, NodeID::UNSPECIFIED)) sent = true; - return sent; -} - -void MessageProvider::addMessageReceiver( MessageReceiver* receiver ) { - receivers.push_back(receiver); -} - -void MessageProvider::removeMessageReceiver( MessageReceiver* receiver ) { - for (size_t i=0; ireceiveMessage(message, LinkID::UNSPECIFIED, NodeID::UNSPECIFIED)) sent = true; +// return sent; +//} +// +//void MessageProvider::addMessageReceiver( MessageReceiver* receiver ) { +// receivers.push_back(receiver); +//} +// +//void MessageProvider::removeMessageReceiver( MessageReceiver* receiver ) { +// for (size_t i=0; i - -using std::vector; -using ariba::utility::LinkID; -using ariba::utility::NodeID; - -NAMESPACE_BEGIN - - -/** - * This class defines an interface for message providers. - * Implementing classes must allow receivers to register themselves. - * - * @author Sebastian Mies - */ -class MessageProvider { -private: - vector receivers; - -protected: - bool sendMessageToReceivers( const Message* message ); - -public: - /** - * Constructor. - */ - MessageProvider(); - - /** - * Destructor. - */ - ~MessageProvider(); - - /** - * Adds a message receiver. - * - * @param receiver The receiver. - */ - void addMessageReceiver( MessageReceiver* receiver ); - - /** - * Removes a message receiver. - * - * @param receiver The receiver. - */ - void removeMessageReceiver( MessageReceiver* receiver ); -}; - -NAMESPACE_END +//#include "_namespace.h" +//#include "MessageReceiver.h" +//#include "ariba/utility/types/LinkID.h" +//#include "ariba/utility/types/NodeID.h" +//#include +// +//using std::vector; +//using ariba::utility::LinkID; +//using ariba::utility::NodeID; +// +//NAMESPACE_BEGIN +// +// +///** +// * This class defines an interface for message providers. +// * Implementing classes must allow receivers to register themselves. +// * +// * @author Sebastian Mies +// */ +//class MessageProvider { +//private: +// vector receivers; +// +//protected: +// bool sendMessageToReceivers( const Message* message ); +// +//public: +// /** +// * Constructor. +// */ +// MessageProvider(); +// +// /** +// * Destructor. +// */ +// ~MessageProvider(); +// +// /** +// * Adds a message receiver. +// * +// * @param receiver The receiver. +// */ +// void addMessageReceiver( MessageReceiver* receiver ); +// +// /** +// * Removes a message receiver. +// * +// * @param receiver The receiver. +// */ +// void removeMessageReceiver( MessageReceiver* receiver ); +//}; +// +//NAMESPACE_END #endif /* MESSAGEPROVIDER_H_ */ Index: source/ariba/utility/messages/MessageReceiver.cpp =================================================================== --- source/ariba/utility/messages/MessageReceiver.cpp (revision 10700) +++ source/ariba/utility/messages/MessageReceiver.cpp (revision 12060) @@ -49,8 +49,8 @@ } -bool MessageReceiver::receiveMessage( const Message* message, const LinkID& link, const NodeID& node ) { - //std::cout << "UNIMPLEMENTED MessageReceiver got Message:" << (Message*)message << std::endl; - return false; -} +//bool MessageReceiver::receiveMessage( reboost::shared_buffer_t message, const LinkID& link, const NodeID& node ) { +// //std::cout << "UNIMPLEMENTED MessageReceiver got Message:" << (Message*)message << std::endl; +// return false; +//} NAMESPACE_END Index: source/ariba/utility/messages/MessageReceiver.h =================================================================== --- source/ariba/utility/messages/MessageReceiver.h (revision 10700) +++ source/ariba/utility/messages/MessageReceiver.h (revision 12060) @@ -40,5 +40,7 @@ #define MESSAGERECEIVER_H__ -#include "ariba/utility/messages/Message.h" +//#include "ariba/utility/messages/Message.h" +// reboost messages +#include "ariba/utility/transport/messages/message.hpp" #include "ariba/utility/types/LinkID.h" #include "ariba/utility/types/NodeID.h" @@ -73,5 +75,8 @@ * @return True, when the message has been accepted. */ - virtual bool receiveMessage( const Message* message, const LinkID& link, const NodeID& node ); + virtual bool receiveMessage( reboost::shared_buffer_t message, + const LinkID& link, + const NodeID& node, + bool bypass_overlay ) = 0; };