Tcp.cpp

Go to the documentation of this file.

#include "Tcp.h"

        namespace castor { namespace net { namespace channels {

                Tcp::Connection::Connection(asio::io_service &service, Tcp *tcp) :
                        address(), socket(service), channel(tcp), closed(false), buffer()
                {
                }

                Tcp::Connection::Connection(const Connection &other) :
                        address(other.address), socket(other.channel->getIoService()),
                        channel(other.channel), closed(false), buffer()
                {
                }

                Tcp::Connection::~Connection() {
                }

                void Tcp::Connection::handleRead(const asio::error_code& error, size_t count) {
//                      std::cout << "Read " << count << std::endl;

                        bool socketFailure = false;

                        if ((!error) && (!this->closed)) {

                                NetAddressPtr address;

                                try {

                                        address = NetAddress::create(
                                                        this->channel->getAddress().getSpicaSpecific(),
                                                        this->socket.remote_endpoint());

                                } catch (const std::exception &e) {
                                        std::cout << "Error: " << e.what() << " closing socket." << std::endl;
                                        socketFailure = true;
                                }

                                if (!socketFailure) {

        //std::cout << " XXXXX geminga: " << count << ": " << std::string(this->buffer, count) << " " << *address << std::endl;

                                this->channel->signalReceive(this->buffer, count, address);

                                if (this->channel->async) {
                                        this->socket.async_read_some(
                                                        asio::buffer(this->buffer, sizeof(this->buffer)),
                                                        this->channel->strand.wrap(
                                                                        boost::bind(&Connection::handleRead, this,
                                                                                        asio::placeholders::error,
                                                                                        asio::placeholders::bytes_transferred)));
                                }
                                
                                return;
                        }
                }

                if (this->channel->peerRemoved) {
//                      std::cout << error.message() << " " << this->closed << std::endl;
                        this->channel->peerRemoved(this->address);
                }

                this->channel->remove(this->address);
        }

        Tcp::Connection &Tcp::Connection::operator=(const Tcp::Connection &other) {
                this->address = other.address;
                this->channel = other.channel;
                return *this;
        }

        asio::ip::tcp::socket &Tcp::Connection::getSocket() {
                return this->socket;
        }

        void Tcp::Connection::start(const NetAddress &address) {

                this->address = address;
//              *NetAddress::create(
//                              this->channel->getAddress().getSpicaSpecific(),
//                              this->socket.remote_endpoint());

                this->socket.set_option(asio::ip::tcp::no_delay(true));

                if (this->channel->async) {
                        this->socket.async_read_some(asio::buffer(this->buffer, sizeof(this->buffer)),
                                                this->channel->strand.wrap(
                                                                boost::bind(&Connection::handleRead, this,
                                                                                asio::placeholders::error,
                                                                                asio::placeholders::bytes_transferred)));
                }
        }

        void Tcp::Connection::stop() {

                if (this->closed) return;

                this->socket.cancel();

                this->socket.close();

                this->closed = true;
        }

        Tcp::Tcp(asio::io_service &service) :
                CastorChannel(service), connections(), service(&service), acceptor(service),
                socket(service), endpoint(), mutex(), opened(false)
        {
        }

        Tcp::Tcp(asio::io_service &service, const NetAddress &a) :
                CastorChannel(service), connections(), service(&service), acceptor(service),
                socket(service), endpoint(a.getAddress(), a.getPort()), mutex(), opened(false)
        {
                this->address = a;
        }

        Tcp::~Tcp() {
        }

        void Tcp::setAsync(bool async) {
                this->async = async;
                if ((async) && (isOpen())) {
                        this->socket.async_read_some(asio::buffer(this->buffer, sizeof(this->buffer)),
                                        this->strand.wrap(
                                                        boost::bind(&Tcp::handleRead, this,
                                                                asio::placeholders::error,
                                                                asio::placeholders::bytes_transferred)));
                }
        }

        void Tcp::add(ConnectionPtr connection) {
                
                boost::mutex::scoped_lock lock(this->mutex);

                ConnectionMap::iterator itr = this->connections.find(connection->address);

                if (itr == this->connections.end()) {
                        this->connections[connection->address] = connection;
                }
        }

        void Tcp::remove(const NetAddress &address) {
                
                boost::mutex::scoped_lock lock(this->mutex);

                ConnectionMap::iterator itr = this->connections.find(address);

                if (itr != this->connections.end()) {
                        this->connections.erase(itr);
                }
        }

        NetAddress Tcp::getLocalAddress() {
                if (this->acceptor.is_open()) {
//std::cout << "XXXXY get local address " <<
//                      NetAddress(this->address.getSpicaSpecific(), this->acceptor.local_endpoint()) << std::endl;
                        return NetAddress(this->address.getSpicaSpecific(), this->acceptor.local_endpoint());

                } else if (this->socket.is_open()) {
//std::cout << "XXXXY get local address socket " << 
//                      NetAddress(this->address.getSpicaSpecific(), this->socket.local_endpoint()) << std::endl;
                        return NetAddress(this->address.getSpicaSpecific(), this->socket.local_endpoint());
                }

//std::cout << "XXXXY get local address not" << std::endl;
                return this->address;
        }

        void Tcp::open() {
                this->acceptor.open(this->endpoint.protocol());
        }

        void Tcp::open(const NetAddress &a) {
                this->address = a;
        }

        void Tcp::bind() {
                this->acceptor.bind(this->endpoint);
        }

        void Tcp::reuseAddress(bool on) {
                if (this->acceptor.is_open()) {
                        this->acceptor.set_option(asio::socket_base::reuse_address(on));

                } else if (this->socket.is_open()) {
                        this->socket.set_option(asio::socket_base::reuse_address(on));
                }
        }

        bool Tcp::reuseAddress() {
                asio::socket_base::reuse_address result;
                if (this->acceptor.is_open()) {
                        this->acceptor.get_option(result);
                } else if (this->socket.is_open()) {
                        this->socket.get_option(result);
                }
                return result.value();
        }

        void Tcp::enableLoopback(bool) {
        }

        bool Tcp::enableLoopback() {
                return false;
        }

        void Tcp::enableBroadcast(bool) {
        }

        bool Tcp::enableBroadcast() {
                return false;
        }

        void Tcp::multicastHops(int) {
        }

        int Tcp::multicastHops() {
                return 0;
        }

        void Tcp::joinGroup(const asio::ip::address &) {
        }

        void Tcp::leaveGroup(const asio::ip::address &) {
        }

        void Tcp::accept() throw(NetException) {

                this->acceptor.listen();

                if (this->address.getPort() == 0) {
                        if (this->acceptor.is_open()) {
                                this->address.setPort(this->acceptor.local_endpoint().port());
                        } else if (this->socket.is_open()) {
                                this->address.setPort(this->socket.local_endpoint().port());
                        }
                }

                this->mode = mode_accept;
                this->opened = true;

                ConnectionPtr connection = ConnectionPtr(new Connection(*this->service, this));

                this->acceptor.async_accept(connection->getSocket(), this->remoteEndpoint,
                                this->strand.wrap(boost::bind(&Tcp::handleAccept,
                                                this, connection, asio::placeholders::error)));
        }

        void Tcp::connect() throw(NetException) {

                this->mode = mode_connect;
                this->opened = true;

                this->socket.async_connect(this->endpoint,
                                this->strand.wrap(boost::bind(&Tcp::handleConnect,
                                                this, asio::placeholders::error)));
        }

        void Tcp::close() throw(NetException) {

                if (!this->opened) return;

                this->opened = false;

                boost::mutex::scoped_lock lock(this->mutex);

                if (this->mode == mode_accept) {

                        for (ConnectionMap::iterator itr = this->connections.begin();
                                 itr != this->connections.end(); itr++)
                        {
                                itr->second->stop();
                        }

                        this->acceptor.cancel();
                        this->acceptor.close();

                } else if (mode == mode_connect) {

                        this->socket.cancel();
                        this->socket.close();
                }
        }

        void Tcp::close(const NetAddress &address) throw(NetException) {

                ConnectionMap::iterator itr = this->connections.find(address);

                if (itr != this->connections.end()) {
                        itr->second->stop();
                }
        }

        bool Tcp::isOpen() {

                if (this->mode == mode_accept) {

                        return this->acceptor.is_open();

                } else if (this->mode == mode_connect) {
                        return this->socket.is_open();
                }

                return false;
        }

        size_t Tcp::receive(char *data, size_t length) {

                if (this->mode == mode_connect) {
                        return this->socket.read_some(asio::buffer(data, length));
                }

                return -1;
        }

        size_t Tcp::receive(char *data, size_t length, NetAddress &remote) {

                if (this->mode == mode_connect) {
                        remote = this->address;
                        return this->socket.read_some(asio::buffer(data, length));
                }

                return -1;
        }

        void Tcp::send(const char *data, size_t length) {

                if (this->mode == mode_connect) {

                        if (this->async) {
                                this->socket.async_send(asio::buffer(data, length),
                                                this->strand.wrap(
                                                                boost::bind(&Tcp::handleSend, this,
                                                                                asio::placeholders::error,
                                                                                asio::placeholders::bytes_transferred)));
                        } else {
                                this->socket.send(asio::buffer(data, length));
                        }
                }
        }

        void Tcp::send(const char *data, size_t length, const NetAddress &remote) {

                if (this->mode == mode_accept) {

                        boost::mutex::scoped_lock lock(mutex);

                        for (ConnectionMap::iterator itr = this->connections.begin();
                                 itr != this->connections.end(); itr++)
                        {
                                if (itr->first == remote) {
//std::cout << "send to " << remote << std::endl;
                                        if (this->async) {
                                                itr->second->socket.async_send(asio::buffer(data, length),
                                                                this->strand.wrap(
                                                                                boost::bind(&Tcp::handleSend, this,
                                                                                                asio::placeholders::error,
                                                                                                asio::placeholders::bytes_transferred)));
                                        } else {
//                                      std::cout << itr->second->socket.send(asio::buffer(data, length)) << std::endl;
                                        itr->second->socket.send(asio::buffer(data, length));
                                        }

                                        break;
                                }
                        }

                } else if (this->mode == mode_connect) {

                        if (this->async) {
                                this->socket.async_send(asio::buffer(data, length),
                                                this->strand.wrap(
                                                                boost::bind(&Tcp::handleSend, this,
                                                                                asio::placeholders::error,
                                                                                asio::placeholders::bytes_transferred)));
                        } else {
                                this->socket.send(asio::buffer(data, length));
                        }
                }
        }

        void Tcp::handleAccept(ConnectionPtr connection, const asio::error_code& error) {

                if (!error) {

                        connection->start(NetAddress(ss_none, this->remoteEndpoint));

                        add(connection);

                        if (this->peerAdded) {
                                this->peerAdded(connection->address);
                        }

                        ConnectionPtr connection = ConnectionPtr(new Connection(*this->service, this));

                        this->acceptor.async_accept(connection->getSocket(), this->remoteEndpoint,
                                        this->strand.wrap(
                                                        boost::bind(&Tcp::handleAccept, this,
                                                                        connection, asio::placeholders::error)));

                        return;
                }
        }

        void Tcp::handleConnect(const asio::error_code& error) {

                if (!error) {

                        this->socket.set_option(asio::ip::tcp::no_delay(true));

                        if (this->connectionEstablished) {
                                this->connectionEstablished(this->address);
                        }

                        if (this->async) {
                                this->socket.async_read_some(asio::buffer(this->buffer, sizeof(this->buffer)),
                                                this->strand.wrap(
                                                                boost::bind(&Tcp::handleRead, this,
                                                                        asio::placeholders::error,
                                                                        asio::placeholders::bytes_transferred)));
                        }
                                
                } else if (this->opened) {

                        asio::deadline_timer timer(*this->service, boost::posix_time::milliseconds(250));
                        timer.wait();

                        this->socket.async_connect(this->endpoint,
                                        this->strand.wrap(
                                                        boost::bind(&Tcp::handleConnect, this,
                                                                        asio::placeholders::error)));
                }
        }

        void Tcp::handleSend(const asio::error_code&, size_t) {

//              if (error) {
//                      std::cerr << "Tcp: unable to send data: " << error.message() << std::endl;
//              }
        }

        void Tcp::handleRead(const asio::error_code& error, size_t count) {
//              std::cout << "Read" << std::endl;

                bool socketFailure = false;
                
                if (!error) {

                        NetAddressPtr address;

                        try {

                                address = NetAddress::create(
                                                this->address.getSpicaSpecific(),
                                                this->socket.remote_endpoint());

                        } catch (const std::exception &e) {
                                std::cout << e.what() << std::endl;
                                socketFailure = true;
                        }

                        if (!socketFailure) {
//std::cout << " XXXXY geminga: " << count << ": " << std::string(this->buffer, count) << " " << *address << std::endl;

//try {

//std::cout << " XXXXY geminga: try " << std::endl;
                                this->signalReceive(this->buffer, count, address);
//std::cout << " XXXXY geminga: done " << std::endl;
//} catch (const std::exception &e) {
//std::cout << " XXXXY geminga: " << e.what() << std::endl;
//}

//std::cout << " XXXXY geminga: " << this->async << std::endl;
                                if (this->async) {
//std::cout << " XXXXY geminga: " << std::endl;
                                        this->socket.async_read_some(asio::buffer(this->buffer, sizeof(this->buffer)),
                                                        this->strand.wrap(
                                                                        boost::bind(&Tcp::handleRead, this,
                                                                                asio::placeholders::error,
                                                                                asio::placeholders::bytes_transferred)));
                                }
                                
                                return;
                        }
                }

                if (this->mode == mode_accept) {

//              std::cerr << "Tcp: error receiving from " << address << ": " << error.message() << std::endl;

                        close();

//              std::cerr << "Tcp: Connection closed" << std::endl;
                } else if (this->opened) {

                        if (this->connectionLost) {
                                this->connectionLost(this->address);
                        }

                        this->socket.close();

                        asio::deadline_timer timer(*this->service, boost::posix_time::milliseconds(250));
                        timer.wait();

                        this->socket.async_connect(this->endpoint,
                                        this->strand.wrap(
                                                        boost::bind(&Tcp::handleConnect, this,
                                                                        asio::placeholders::error)));
                }
        }

} } }