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

// [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;