Udp.cpp

Go to the documentation of this file.

#include "Udp.h"

#include <boost/thread/xtime.hpp>

namespace castor { namespace net { namespace channels {

        Udp::Udp(asio::io_service &service) :
                CastorChannel(service), socket(service), endpoint(), remoteEndpoint(),
                cond(), monitor(), localMcastAddress(), localMcastAddressHack(false)
        {
        }

        Udp::Udp(asio::io_service &service, const NetAddress &a) :
                CastorChannel(service), socket(service), endpoint(a.getAddress(), a.getPort()),
                remoteEndpoint(), cond(), monitor(), localMcastAddress(), localMcastAddressHack(false)
        {
                this->address = a;
//              this->socket.open(this->endpoint.protocol());
        }

        Udp::~Udp() {
        }

        NetAddress Udp::getLocalAddress() {
                if (this->address.isMulticast()) {

                        if (!this->localMcastAddress) {

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

                                enableLoopback(true);

                                this->localMcastAddressHack = true;

                                std::cout << "HACK: Determining local mcast address" << std::endl;
                                send("", 0);

                                boost::xtime xt;
                                boost::xtime_get(&xt, boost::TIME_UTC);
                                xt.nsec += 10000000;

                                while ((!this->localMcastAddress) && (this->cond.timed_wait(lock, xt))) {
                                        std::cout << "HACK retry" << std::endl;
                                        send("", 0);
                                }
                                std::cout << "HACK: Determining done" << std::endl;
                                enableLoopback(false);
                        }

                        return *this->localMcastAddress;
                }

                return NetAddress(this->address.getSpicaSpecific(), this->socket.local_endpoint());
        }

        void Udp::setAsync(bool async) {
                this->async = async;

                if ((async) && (isOpen())) {
                        this->socket.async_receive_from(getBuffer(), this->remoteEndpoint,
                                        this->strand.wrap(boost::bind(&Udp::handleReceive, this,
                                                        asio::placeholders::error,
                                                        asio::placeholders::bytes_transferred)));
                }
        }

        void Udp::open() {
                this->socket.open(this->endpoint.protocol());
        }

        void Udp::open(const NetAddress &a) {
                this->address = a;
                this->endpoint = asio::ip::udp::endpoint(a.getAddress(), a.getPort());
                this->socket.open(this->endpoint.protocol());
        }

        void Udp::bind() {
                if (this->address.isMulticast()) {
                        if (this->address.getAddress().is_v6()) {
                                this->socket.bind(asio::ip::udp::endpoint(asio::ip::address_v6::any(), this->address.getPort()));
                        } else {
                                this->socket.bind(asio::ip::udp::endpoint(asio::ip::address_v4::any(), this->address.getPort()));
                        }
                } else {
                        this->socket.bind(this->endpoint);
                }
        }

        void Udp::reuseAddress(bool on) {
                this->socket.set_option(asio::socket_base::reuse_address(on));
        }

        bool Udp::reuseAddress() {
                asio::socket_base::reuse_address result;
                this->socket.get_option(result);
                return result.value();
        }

        void Udp::enableLoopback(bool on) {
                this->socket.set_option(asio::ip::multicast::enable_loopback(on));
        }

        bool Udp::enableLoopback() {
                asio::ip::multicast::enable_loopback result;
                this->socket.get_option(result);
                return result.value();
        }

        void Udp::enableBroadcast(bool on) {
                this->socket.set_option(asio::socket_base::broadcast(on));
        }

        bool Udp::enableBroadcast() {
                asio::socket_base::broadcast result;
                this->socket.get_option(result);
                return result.value();
        }


        void Udp::multicastHops(int hops) {
                this->socket.set_option(asio::ip::multicast::hops(hops));
        }

        int Udp::multicastHops() {
                asio::ip::multicast::hops result;
                this->socket.get_option(result);
                return result.value();
        }

        void Udp::joinGroup(const asio::ip::address &address) {
//              std::cout << "Joining group " << address.to_string() << std::endl;
//              this->socket.set_option(asio::ip::multicast::leave_group(address));
                this->socket.set_option(asio::ip::multicast::join_group(address));
                this->address.setAddress(address);
                this->endpoint.address(address);
        }

        void Udp::leaveGroup(const asio::ip::address &address) {
                this->socket.set_option(asio::ip::multicast::leave_group(address));
        }

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

                this->mode = mode_accept;

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

                if (this->async) {
                        this->socket.async_receive_from(getBuffer(), this->remoteEndpoint,
                                        this->strand.wrap(boost::bind(&Udp::handleReceive, this,
                                                        asio::placeholders::error,
                                                        asio::placeholders::bytes_transferred)));
                }
        }

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

                this->mode = mode_connect;

                asio::ip::udp::endpoint ep;
                if (this->address.getAddress().is_v4()) {
                        ep = asio::ip::udp::endpoint(asio::ip::address_v4::any(), 0);
                } else if (this->address.getAddress().is_v6()) {
                        ep = asio::ip::udp::endpoint(asio::ip::address_v6::any(), 0);
                }

                this->socket.bind(ep);

//              this->socket.open(this->endpoint.protocol());

                if (this->async) {
                        this->socket.async_receive_from(getBuffer(), this->remoteEndpoint,
                                        this->strand.wrap(boost::bind(&Udp::handleReceive, this,
                                                        asio::placeholders::error,
                                                        asio::placeholders::bytes_transferred)));
                }
        }

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

                if (this->closed) return;

                this->closed = true;

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

        void Udp::close(const NetAddress &) throw(NetException) {
        }

        bool Udp::isOpen() {
                return this->socket.is_open();
        }
/*
        size_t Udp::receive(char *data, size_t length) {
                return this->socket.receive(asio::buffer(data, length));
        }*/
/*
        size_t Udp::receive(char *data, size_t length, NetAddress &remote) {

                asio::ip::udp::endpoint ep;

                size_t count = this->socket.receive_from(asio::buffer(data, length), ep);

                remote = *NetAddress::create(this->address.getSpicaSpecific(), ep);

                return count;
        }
*/
/*      void Udp::send(const char *data, size_t length) {

                if ((this->mode == mode_connect) || (this->address.isMulticast())) {

                        try {
                                this->socket.send_to(asio::buffer(data, length), this->endpoint);
                        } catch (...) {
                                // Silently drop exception
                        }

                } else {
                        std::cerr << "Udp: unable to send data without remote endpoint in bind mode" << std::endl;
                }
        }
*/
/*      void Udp::send(const char *data, size_t length, const NetAddress &remote) {

                asio::ip::udp::endpoint ep(remote.getAddress(), remote.getPort());

                try {
                        this->socket.send_to(asio::buffer(data, length), ep);
                } catch (...) {
                        // Silently drop exception
                }
        }
*/
        void Udp::handleReceive(const asio::error_code& error, size_t count) {

//              std::cout << "geminga Udp handle receive" << std::endl;

                if ((error) && (count == 0) && (!this->localMcastAddress) && (this->localMcastAddressHack)) {

                        NetAddressPtr address = NetAddress::create(
                                        this->address.getSpicaSpecific(),
                                        this->remoteEndpoint);

                        std::cout << "HACK: Determined local mcast address " << *address << std::endl;
                        this->localMcastAddress = address;
                        this->cond.notify_one();

                        this->localMcastAddressHack = false;

                        if (this->async) {
                                this->socket.async_receive_from(
                                                asio::buffer(this->buffer, sizeof(this->buffer)), this->remoteEndpoint,
                                                this->strand.wrap(boost::bind(&Udp::handleReceive, this,
                                                                asio::placeholders::error,
                                                                asio::placeholders::bytes_transferred)));
                        }
                }

                if (!error) {

                        NetAddressPtr address = NetAddress::create(
                                        this->address.getSpicaSpecific(),
                                        this->remoteEndpoint);

                        if (count > 0) {
                                this->signalReceive(this->buffer, count, address);
                        }

                        if (this->async) {
                                this->socket.async_receive_from(
                                                asio::buffer(this->buffer, sizeof(this->buffer)), this->remoteEndpoint,
                                                this->strand.wrap(boost::bind(&Udp::handleReceive, this,
                                                                asio::placeholders::error,
                                                                asio::placeholders::bytes_transferred)));
                        }
                                
                        return;
                }

//              std::cerr << "Udp: error receiving message: " << error.message() << std::endl;
        }

} } }