source: source/ariba/communication/modules/transport/tcp/TCPTransport.cpp@ 4591

// [License]

// The Ariba-Underlay Copyright

//

// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH)

//

// Institute of Telematics

// UniversitÀt Karlsruhe (TH)

// Zirkel 2, 76128 Karlsruhe

// Germany

//

// Redistribution and use in source and binary forms, with or without

// modification, are permitted provided that the following conditions are

// met:

//

// 1. Redistributions of source code must retain the above copyright

// notice, this list of conditions and the following disclaimer.

// 2. Redistributions in binary form must reproduce the above copyright

// notice, this list of conditions and the following disclaimer in the

// documentation and/or other materials provided with the distribution.

//

// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND

// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE

// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR

// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,

// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,

// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR

// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF

// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING

// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

//

// The views and conclusions contained in the software and documentation

// are those of the authors and should not be interpreted as representing

// official policies, either expressed or implied, of the Institute of

// Telematics.

// [License]



#include "TCPTransport.h"



#define _NO_LOGGING



// std includes

#include <unistd.h>

#include <iostream>

#include <string>

#include <sstream>



// protlib includes

#include "../protlib/network_message.h"

#include "../protlib/tp_over_tcp.h"

#include "../protlib/logfile.h"

#include "../protlib/queuemanager.h"

#include "../protlib/threadsafe_db.h"

#include "../protlib/setuid.h"



// spovnet includes

#include "ariba/utility/serialization.h"

#include "ariba/utility/system/SystemQueue.h"

#include "ariba/utility/system/SystemEvent.h"

#include "ariba/utility/system/SystemEventType.h"

#include "ariba/communication/modules/network/ip/IPv4Locator.h"



// protlib namespaces

using namespace protlib;

using namespace protlib::log;



// spovnet namespaces

using ariba::utility::SystemQueue;

using ariba::utility::SystemEvent;

using ariba::utility::SystemEventType;

using ariba::utility::MessageProvider;

using ariba::utility::TextMessage;

using ariba::utility::MessageReceiver;

using ariba::communication::IPv4Locator;



using_serialization;



logfile commonlog;

protlib::log::logfile& protlib::log::DefaultLog(commonlog);



#include "ariba/communication/modules/_namespace.h"

NAMESPACE_BEGIN;



SystemEventType TCPTransportEvent("TCPTransport");

SystemEventType TCPMessageDispatchEvent("MessageDispatchEvent", TCPTransportEvent );

SystemEventType TCPTransportTestEvent("Test", TCPTransportEvent );



use_logging_cpp(TCPTransport);



TCPTransport::TCPTransport( port_t port ) {

        this->running = false;

        this->done = false;

        this->port = port;



        logging_debug( "creating tcp transport module" );

}



TCPTransport::~TCPTransport() {

        logging_debug( "deleting tcp transport module" );

}



void TCPTransport::start() {



        logging_info( "starting tcp transport module ..." );



        // initalize netdb and setuid

        protlib::tsdb::init();

        protlib::setuid::init();



        /* set tcp parameters */

        port_t port = this->port; // port

        TPoverTCPParam tppar(4, TCPTransport::getMessageLength, port);



        /* create receiver thread */

        FastQueue* tpchecker_fq = new FastQueue("TCPTransport", true);

        QueueManager::instance()->register_queue(tpchecker_fq,

                        message::qaddr_signaling);



        /* start thread */

        pthread_create( &tpreceivethread, NULL, TCPTransport::receiverThread, this );

        tpthread = new ThreadStarter<TPoverTCP, TPoverTCPParam> ( 1, tppar );

        tpthread->start_processing();



        logging_info( "tcp transport module started" );

}



void TCPTransport::stop() {



        logging_info( "stopping tcp transport module ..." );



        // stop receiver thread

        done = true;



        // stop TPoverTCP

        tpthread->stop_processing();

        tpthread->abort_processing(true);

        tpthread->wait_until_stopped();



        // unregister TPoverTCP

        QueueManager::instance()->unregister_queue( message::qaddr_signaling );



        // destroy QueueManager

        QueueManager::clear();



        // de-initalize netdb and setuid

        protlib::setuid::end();

        protlib::tsdb::end();



        logging_info( "tcp transport module stopped" );

}



bool TCPTransport::getMessageLength( NetMsg& m, uint32& clen_bytes ) {

        clen_bytes = m.decode32();

        m.set_pos_r(-4);

        return true;

}



void* TCPTransport::receiverThread( void* ptp ) {



        logging_info( "running tcp transport receiver thread" );



        // get reference to transport object

        TCPTransport& tp = *((TCPTransport*)ptp);



        // get queue

        FastQueue* fq =

                QueueManager::instance()->get_queue(message::qaddr_signaling);



        // main processing loop

        tp.running = true;

        while (!tp.done) {



                // wait for new message to approach

                message* msg = fq->dequeue_timedwait(300);



                // handle message

                if (msg) {



                        logging_debug( "Received incoming message" );



                        // handle transport message

                        TPMsg* tpmsg = dynamic_cast<TPMsg*> (msg);

                        if (tpmsg) {

                                // evaluate TP message

                                const appladdress* peer = static_cast<const appladdress*>( tpmsg->get_peeraddress() );

                                NetMsg* datamsg = tpmsg->get_message();

                                TPError* err = tpmsg->get_error();



                                // get data

                                if (datamsg) {



                                        datamsg->set_pos(0);

                                        uint32_t msgLength = datamsg->decode32(true);

                                        uint16_t remotePort = datamsg->decode16(true);



                                        // convert data

                                        Data data(

                                                (uint8_t*)(datamsg->get_buffer()+6),

                                                (datamsg->get_size()-6)*8

                                        );



                                        // converting message

                                        logging_debug( "Converting message" );

                                        Message* msg = new Message(data);

                                        std::ostringstream o;

                                        o << (peer->get_ip_str() + 7) << ":" << remotePort;

                                        msg->setSourceAddress( new IPv4Locator(IPv4Locator::fromString(o.str())) );

                                        logging_debug( "> source address = " << o.str() );

                                        logging_debug( "> message = " << msg->toString() );



                                        // dispatching message

                                        logging_debug( "Dispatching message" );

                                        SystemQueue::instance().scheduleEvent(

                                                SystemEvent( &tp, TCPMessageDispatchEvent, msg )

                                        );

                                }



                                // check error

                                if (err)

                                        logging_error( "TCP transport error " + string(err->getstr()) );



                                logging_debug( "Message processed." );



                                tpmsg = NULL;

                        }

                        delete msg;

                }

        }



        // clean queue & stop

        fq->cleanup();

        tp.running = false;

        return NULL;

}



seqnum_t TCPTransport::sendMessage(const Message* message ) {



        Data data = data_serialize( message );

        const_cast<Message*>(message)->dropPayload();



        // prepare netmsg and send it

        NetMsg* datamsg = new NetMsg(data.getLength()/8+6);

        datamsg->encode32(data.getLength()/8+2, true);

        datamsg->encode16(this->port,true);

        for (int i=0; i<data.getLength()/8; i++)

                datamsg->encode8(data.getBuffer()[i],true);



        // send message

        const IPv4Locator* address = dynamic_cast<const IPv4Locator*>(message->getDestinationAddress());

        if( address == NULL) return 0;



        logging_debug( "sending message of size " << data.getLength() <<

                        " to address " + address->toString() <<

                        ": " + message->toString() );



        string s = address->toString();

        string::size_type i = s.find(':');

        string ip = address->toString().substr(0,i).c_str();

        logging_debug( "ip= " << ip << " port=" << address->getPort() );



        appladdress peer(ip.c_str(), "tcp", address->getPort() );

        tpthread->get_thread_object()->send(datamsg, peer, false);



        // release data

        data.release();

        logging_debug( "message sent!" );



        return 0;

}



void TCPTransport::terminate(const NetworkLocator* local, const NetworkLocator* remote){



        const IPv4Locator* address = dynamic_cast<const IPv4Locator*>(remote);

        if( address == NULL) return;



        string s = address->toString();

        string::size_type i = s.find(':');

        string ip = address->toString().substr(0,i).c_str();



        appladdress peer( ip.c_str(), "tcp", address->getPort() );

        peer.convert_to_ipv6();



        tpthread->get_thread_object()->terminate( peer );

}



TransportLocator::prot_t TCPTransport::getId() {

        return 6; // TCP

}







const vector<TransportLocator*> TCPTransport::getLocators() {

        return vector<TransportLocator*>();

}



/* system event handler */

void TCPTransport::handleSystemEvent( const SystemEvent& event ) {



        // dispatch received messages

        if ( event.getType() == TCPMessageDispatchEvent ){

                logging_debug( "forwarding message to local receivers" );

                Message* msg = event.getData<Message>();

                MessageProvider::sendMessageToReceivers( msg );

                msg->dropPayload();

//              delete msg->getSourceAddress();

                delete msg;

        }



        if ( event.getType() == TCPTransportTestEvent ) {



                // add listener

                addMessageReceiver( new MessageReceiver() );



                // send message

                //cout << "Sending message ..." << endl;

                sendMessage( new TextMessage( "Hello World!" ) );

                //cout << "Message sent ..." << endl;



        }

}



NAMESPACE_END;