vos/corelibs/servicediscovery/servicedirectory.cc

Go to the documentation of this file.

/* $Id: servicedirectory.cc,v 1.26 2003/08/03 17:02:40 reed Exp $

    Copyright (C) 2002 Peter Amstutz <tetron@interreality.org>

    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2 of the License, or (at your option) any later version.

    This library is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA

*/

// XXX TODO
/// @todo remove code duplication (use some common function call for
/// constructing & sending messages to udp, tcp relays, etc

#include "config.h"

#include <stdexcept>
#include <set>
#include <string>
#include <iostream>
#include <map>
#include <deque>
using namespace std;

#include <sys/types.h>
#include <sys/time.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/select.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <netdb.h>  // for getthostbyname()
#include <stdio.h>
#include <errno.h>
#include <arpa/inet.h> // for inet_ntoa
#include <math.h>

#if defined(HAVE_BOOST_REGEX_H) && defined(HAVE_LIBBOOST)
# include <boost/regex.h>
#else
  extern "C" {
#   include <regex.h>
  }
#endif

/* socklen_t is not always defined (MacOSX) */
#if !defined(HAVE_SOCKLEN_T) && !defined(socklen_t)
typedef int socklen_t;
#endif


#include "servicedirectory.hh"
using namespace LSD;

#define WARNING(s)  { cerr << "libservicediscovery: Warning: " << s << endl; }

#ifdef DEBUG_MODE
#  define DEBUG(s)    { cerr << "libservicediscovery: " << s << endl; }
#else
#  define DEBUG(s)
#endif

extern "C" {
#include "packet.h"
}

void respondToMessage(
    set<Service*>& services,
    map<int, ServiceDirectory::Query>& queries,
    pack_ubyte* buf, unsigned int buf_len,
    int fd, sockaddr_in* skaddr, socklen_t sa_len
);

class ServiceMatch {
private:
    regex_t typereg;
    regex_t titlereg;
    regex_t descreg;
    regex_t urlreg;
    bool typeisdotstar;
public:
    ServiceMatch(const std::string& typepattern,
                 const std::string& titlepattern,
                 const std::string& descpattern,
                 const std::string& urlpattern);
    ~ServiceMatch();
    bool match(const Service& s);
};

ServiceMatch::ServiceMatch(const string& typepattern,
                           const string& titlepattern,
                           const string& descpattern,
                           const string& urlpattern)
{
    if(typepattern == "") {
        regcomp(&typereg, ".*", REG_EXTENDED | REG_NOSUB);
        typeisdotstar = true;
    }
    else {
        typeisdotstar = false;
        regcomp(&typereg, typepattern.c_str(), REG_EXTENDED | REG_NOSUB);
    }

    if(titlepattern == "") regcomp(&titlereg, ".*", REG_EXTENDED | REG_NOSUB);
    else regcomp(&titlereg, titlepattern.c_str(), REG_EXTENDED | REG_NOSUB);

    if(descpattern == "") regcomp(&descreg, ".*", REG_EXTENDED | REG_NOSUB);
    else regcomp(&descreg, descpattern.c_str(), REG_EXTENDED | REG_NOSUB);

    if(urlpattern == "") regcomp(&urlreg, ".*", REG_EXTENDED | REG_NOSUB);
    else regcomp(&urlreg, urlpattern.c_str(), REG_EXTENDED | REG_NOSUB);
}

ServiceMatch::~ServiceMatch()
{
    regfree(&typereg);
    regfree(&titlereg);
    regfree(&descreg);
    regfree(&urlreg);
}

bool ServiceMatch::match(const Service& sv)
{
    bool matchtype = typeisdotstar;
    for(set<string>::iterator ti = sv.types.begin(); ti != sv.types.end(); ti++) {
        if(regexec(&typereg, (*ti).c_str(), 0, 0, 0) == 0) {
            matchtype=true;
            break;
        }
    }
    return (matchtype
            && (sv.title.size() == 0 || regexec(&titlereg, sv.title.c_str(), 0, 0, 0) == 0)
            && (sv.description.size() == 0 || regexec(&descreg, sv.description.c_str(), 0, 0, 0) == 0)
            && (sv.url.size() == 0 || regexec(&urlreg, sv.url.c_str(), 0, 0, 0) == 0));
}



short ServiceDirectory::LISTEN_BCAST = 0x1;
short ServiceDirectory::LISTEN_TCP = 0x2;
short ServiceDirectory::RELAY_TCP_TO_BCAST = 0x4;
short ServiceDirectory::RELAY_BCAST_TO_TCP = 0x8;
short ServiceDirectory::RELAY_TCP_TO_REMOTE = 0x16;


ServiceDirectory::ServiceDirectory(short opt) {
    DEBUG("new service directory (opt=" << opt << ")");
    udp_port = 4230;
    if(opt & LISTEN_BCAST)
        listenUDP();
    else
        udp_fd = -1;
    if(opt & LISTEN_TCP)
        listenTCP();
    else
        tcp_fd = -1;
    relay_tcp_local = (opt & RELAY_TCP_TO_BCAST);
    relay_udp_remote = (opt & RELAY_BCAST_TO_TCP);
    relay_tcp_remote = (opt & RELAY_TCP_TO_REMOTE);
}

ServiceDirectory::ServiceDirectory() {
    DEBUG("new service directory (default options)");
    udp_port = 4230;
    listenUDP();
    tcp_fd = -1;
    relay_tcp_local = true;
    relay_udp_remote = true;
    relay_tcp_remote = false;
}




void ServiceDirectory::listenUDP(int port) {

    if(port == -1 && udp_fd != -1) {
        shutdown(udp_fd, SHUT_RDWR);
        udp_fd = -1;
        return;
    }

    udp_fd = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
    struct sockaddr_in skaddr;
    int opt_bcast, opt_report, opt_readdr;
    opt_bcast = opt_report = opt_readdr = 1;
    setsockopt(udp_fd, SOL_SOCKET, SO_BROADCAST, &opt_bcast, sizeof(int));
#ifdef SO_REUSEPORT
    setsockopt(udp_fd, SOL_SOCKET, SO_REUSEPORT, &opt_report, sizeof(int));
#else
#warning This platform is missing SO_REUSEPORT. Will just SO_REUSEADDR work?
#endif
    setsockopt(udp_fd, SOL_SOCKET, SO_REUSEADDR, &opt_readdr, sizeof(int));
    skaddr.sin_family = AF_INET;
    skaddr.sin_port = htons(port);
    skaddr.sin_addr.s_addr = 0;
    bind(udp_fd, (struct sockaddr*)&skaddr, sizeof(struct sockaddr_in));
    udp_port = port;
    DEBUG("listening on udp port " << port);
}

ServiceDirectory::~ServiceDirectory() {
    if(udp_fd != -1)
        shutdown(udp_fd, SHUT_RDWR);
    for( map<pair<char*, int>, int>::const_iterator i = tcp_relays.begin();
            i != tcp_relays.end();
            i++)
    {
        shutdown(i->second, SHUT_RDWR);
        tcp_relays.erase(i->first);
    }
    for(deque<int>::const_iterator i = incoming_tcp_fds.begin();
                i != incoming_tcp_fds.end(); i++) {
        shutdown(*i, SHUT_RDWR);
    }
    if(tcp_fd != -1)
        shutdown(tcp_fd, SHUT_RDWR);
}

int ServiceDirectory::query(const string& typepattern,
                            const string& titlepattern,
                            const string& descpattern,
                            const string& urlpattern,
                            ServiceAdvertismentListener* cb)
{
    /* Construct message packet */
    int r;
    do {
        r = rand();
    } while(queries.count(r));
    unsigned char pkt[2048];
    int pkt_len;
    pkt_len = pack(pkt, sizeof(pkt), "cclSSSS",
        0, 'Q', r,
        typepattern.c_str(), typepattern.size(),
        titlepattern.c_str(), titlepattern.size(),
        descpattern.c_str(), descpattern.size(),
        urlpattern.c_str(), urlpattern.size()
    );

    /* Broadcast UDP packet */
    if(udp_fd != -1) {
        struct sockaddr_in skaddr;
        skaddr.sin_family = AF_INET;
        skaddr.sin_port = htons(udp_port);
        skaddr.sin_addr.s_addr = INADDR_BROADCAST;
        DEBUG("sending query message to udp bcast...");
        sendto(udp_fd, pkt, pkt_len, 0, (struct sockaddr*)&skaddr, sizeof(struct sockaddr_in));
    }

    /* Send to TCP relays */
    for( map<pair<char*, int>, int>::const_iterator i = tcp_relays.begin();
            i != tcp_relays.end();
            i++)
    {
        int fd = i->second;
        DEBUG("sending query to tcp socket " << fd);
        if( send(fd, pkt, pkt_len, 0) < 0 ) {
            removeRelay(i->first.first, i->first.second);
        }
    }


    /* Save query */
    Query q;
    q.typepattern = typepattern;
    q.titlepattern = titlepattern;
    q.descpattern = descpattern;
    q.urlpattern = urlpattern;
    q.listener = cb;
    queries[r] = q;

    return r;
}

void ServiceDirectory::endQuery(int q)
{
    queries.erase(q);
}

void ServiceDirectory::handleIncoming(float timeout)
{
    DEBUG("handleIncoming...");
    fd_set readset;
    FD_ZERO(&readset);
    int max_fd = -1;
    if(udp_fd != -1) {
        FD_SET(udp_fd, &readset);
        max_fd = udp_fd;
    }
    if(tcp_fd != -1) {
        FD_SET(tcp_fd, &readset);
        max_fd = (tcp_fd > max_fd)?(tcp_fd):(max_fd);
    }
    for(deque<int>::const_iterator i = incoming_tcp_fds.begin();
                i != incoming_tcp_fds.end(); i++) {
        if(*i != -1) {
            FD_SET(*i, &readset);
            max_fd = (*i > max_fd)?(*i):(max_fd);
        }
    }

    if(max_fd == -1) {
        DEBUG("handleIncoming: no sockets; returning");
        return;
    }

    struct timeval tv = {0, 0};
    if(timeout > 0) {
        tv.tv_sec = (long)floor(timeout);
        tv.tv_usec = (long)( (timeout - tv.tv_sec) * 1000000.0 );
    }
    select( max_fd+1, &readset, 0, 0, ( (timeout >= 0) ? (&tv) : NULL ) );

    pack_ubyte buf[2048];
    struct sockaddr_in skaddr;
    socklen_t sa_len = sizeof(struct sockaddr_in);
    memset(&skaddr, 0, sa_len);

    /* Check UDP socket */
    if( (udp_fd != -1) && FD_ISSET(udp_fd, &readset)) {
        DEBUG("handleIncoming: udp socket has data.");
        int r = recvfrom(udp_fd, buf, sizeof(buf), 0, (struct sockaddr*)&skaddr, &sa_len);
        if(r > 0) {

            /* Respond */
            respondToMessage(services, queries, buf, r, udp_fd, &skaddr, sa_len);

            /* Relay to remote TCP sockets */
            if(relay_udp_remote) {
                for(map<pair<char*, int>, int>::const_iterator i = tcp_relays.begin();
                    i != tcp_relays.end();
                    i++)
                {
                    int fd = i->second;
                    DEBUG("relaying message to tcp socket " << fd);
                    if( send(fd, buf, r, 0) < 0 ) {
                        removeRelay(i->first.first, i->first.second);
                    }
                }
            }
        } else DEBUG("no data!");
    }

    /* Check TCP sockets */

    if( (tcp_fd != -1) && FD_ISSET(tcp_fd, &readset) ) {
        int fd = accept(tcp_fd, 0, 0);
        if(fd != -1) {
            DEBUG("accepted incoming socket " << fd);
            incoming_tcp_fds.push_back(fd);
        }
    }

    for(deque<int>::iterator i = incoming_tcp_fds.begin(); i!= incoming_tcp_fds.end(); i++) {
        if( (*i != -1) && FD_ISSET(*i, &readset)) {
            DEBUG("handleIncoming: tcp socket " << *i << " has data.");
            int r = recv(*i, buf, sizeof(buf), 0);
            if(r > 0) {

                /* Respond */
                respondToMessage(services, queries, buf, r, *i, NULL, 0);

                /* Relay message broadcast to local net */
                if(relay_tcp_local) {
                    skaddr.sin_family = AF_INET;
                    skaddr.sin_port = htons(udp_port);
                    skaddr.sin_addr.s_addr = INADDR_BROADCAST;
                    skaddr.sin_port = htons(udp_port);
                    DEBUG("relaying  message to udp bcast...");
                    int s = sendto(udp_fd, buf, r, 0, (struct sockaddr*)&skaddr, sizeof(struct sockaddr_in));
                    if(s < 0)
                        WARNING("Error sending UDP broadcast message.");
                }

                if(relay_udp_remote) {
                    for( map<pair<char*, int>, int>::const_iterator i = tcp_relays.begin();
                        i != tcp_relays.end();
                        i++)
                    {
                        int fd = i->second;
                        DEBUG("relaying message to tcp socket " << fd);
                        if( send(fd, buf, r, 0) < 0 ) {
                            removeRelay(i->first.first, i->first.second);
                        }
                    }
                }

            } else {
                WARNING("Error reading from incoming TCP-relay socket.");
                shutdown(*i, SHUT_RDWR);
                *i = -1;
                // XXX dangerous? incoming_tcp_fds.erase(i);
            }
        }
    }
}

void respondToMessage(set<Service*>& services, map<int, ServiceDirectory::Query>& queries, pack_ubyte* buf, unsigned int buf_len, int fd, sockaddr_in* skaddr, socklen_t sa_len ) {
    DEBUG("respondToMessage: got message (length " << buf_len << ")");

    pack_ubyte reply_buf[2048];
    char version, msgtype;
    int nonce;
    int p = unpack(buf, buf_len, "ccl", &version, &msgtype, &nonce);
    if(msgtype == 'Q') {
        char *typepattern, *titlepattern, *descpattern, *urlpattern;
        int tylen, tilen, dlen, ulen;

        unpack(buf + 6, buf_len - 6, "SSSS",
               &typepattern, &tylen,
               &titlepattern, &tilen,
               &descpattern, &dlen,
               &urlpattern, &ulen);

        DEBUG("respondToMessage: got Query. type=" << typepattern << " title=" << titlepattern << " desc=" << descpattern << " url=" << urlpattern);

        /* Check matches */
        ServiceMatch sm(string(typepattern, tylen),
                        string(titlepattern, tilen),
                        string(descpattern, dlen),
                        string(urlpattern, ulen));
        for(set<Service*>::iterator si = services.begin(); si != services.end(); si++) {
            if(sm.match(**si)) {

                /* Construct 'A' reply */
                int p = pack(reply_buf, sizeof(reply_buf), "ccls", 0, 'A', nonce, (*si)->types.size());
                for(set<string>::iterator ti = (*si)->types.begin(); ti != (*si)->types.end(); ti++) {
                    p += pack(reply_buf + p, sizeof(reply_buf) - p, "S", (*ti).c_str(), (*ti).size());
                }
                p += pack(reply_buf + p, sizeof(reply_buf) - p, "SSS",
                          (*si)->title.c_str(), (*si)->title.size(),
                          (*si)->description.c_str(), (*si)->description.size(),
                          (*si)->url.c_str(), (*si)->url.size());
                DEBUG("Built a reply to the query, with URL " << (*si)->url.c_str() << " (length " << (*si)->url.size() << ")");


                /* Send message */
                if(skaddr) {
                    DEBUG("respondToMessage: sending query-reply announcement message to " <<  fd << " - addr: " << inet_ntoa(skaddr->sin_addr) << " - msg(" << p << ")" );
                    sendto(fd, reply_buf, p, 0, (struct sockaddr*)skaddr, sa_len);
                } else {
                    DEBUG("respondToMessage: sending query-reply announcement to " << fd);
                    send(fd, reply_buf, p, 0);
                }

            }
        }

        free(typepattern);
        free(titlepattern);
        free(descpattern);
        free(urlpattern);

    } else if(msgtype == 'A') {
        DEBUG("respondToMessage: got Advertisment");
        Service sv;

        pack_uint16 ntypes;
        p += unpack(buf + p, buf_len - p, "s", &ntypes);
        int i, len;
        for(i = 0; i < ntypes; i++) {
            char* s;
            p += unpack(buf + p, buf_len - p, "S", &s, &len);
            sv.types.insert(string(s, len));
        }

        char *title, *desc, *url;
        int tlen, dlen, ulen;
        unpack(buf + p, buf_len - p, "SSS",
               &title, &tlen,
               &desc, &dlen,
               &url, &ulen);
        sv.title = string(title, tlen);
        sv.description = string(desc, dlen);
DEBUG("unpacking advertisement message... URL length is " << ulen);
        sv.url = string(url, ulen);
DEBUG("URL is " << sv.url);

        if(nonce == 0) {
            for(map<int, ServiceDirectory::Query>::iterator i = queries.begin(); i != queries.end(); i++) {
                ServiceMatch sm((*i).second.typepattern,
                                (*i).second.titlepattern,
                                (*i).second.descpattern,
                                (*i).second.urlpattern);
                if(sm.match(sv)) {
                    (*i).second.listener->notifyNewService((*i).first, sv);
                }
            }
        } else if(queries.count(nonce)) queries[nonce].listener->notifyNewService(nonce, sv);
    }
}

void ServiceDirectory::addService(Service* s)
{
    DEBUG("Advertising service. URL is " << s->url << " (length " << s->url.size() << ")");

    /* Construct 'A' message */
    pack_ubyte buf[2048];
    int p = pack(buf, sizeof(buf), "ccls", 0, 'A', 0, s->types.size());
    for(set<string>::iterator ti = s->types.begin(); ti != s->types.end(); ti++) {
        p += pack(buf + p, sizeof(buf) - p, "S", (*ti).c_str(), (*ti).size());
    }
    p += pack(buf + p, sizeof(buf) - p, "SSS",
              s->title.c_str(), s->title.size(),
              s->description.c_str(), s->description.size(),
              s->url.c_str(), s->url.size());

    /* Broadcast UDP packet */
    if(udp_fd != -1) {
        struct sockaddr_in skaddr;
        skaddr.sin_family = AF_INET;
        skaddr.sin_port = htons(udp_port);
        skaddr.sin_addr.s_addr = INADDR_BROADCAST;
        DEBUG("sending announcement message to udp bcast...");
        sendto(udp_fd, buf, p, 0, (struct sockaddr*)&skaddr, sizeof(struct sockaddr_in));
    }

    /* Send to TCP relays */
    for( map<pair<char*, int>, int>::const_iterator i = tcp_relays.begin();
            i != tcp_relays.end();
            i++)
    {
        int fd = i->second;
        DEBUG("sending announcement to tcp socket " << fd);
        if( send(fd, buf, p, 0) < 0 ) {
            removeRelay(i->first.first, i->first.second);
        }
    }

    /* Add to list */
    services.insert(s);
}

void ServiceDirectory::removeService(Service* s)
{
    services.erase(s);
}

void ServiceDirectory::listenTCP(int port) {
    if( port == -1 && tcp_fd != -1) {
        shutdown(tcp_fd, SHUT_RDWR);
        tcp_fd = -1;
        return;
    }
    tcp_fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = 0;
    bind(tcp_fd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in));
    if(tcp_fd < 0)
        throw runtime_error("Couldn't bind port (listenTCP): " + string(strerror(errno)));
    if( listen(tcp_fd, 5) < 0 )
        throw runtime_error("Couldn't listen on port (listenTCP): " + string(strerror(errno)));
    DEBUG("listening on tcp socket " << tcp_fd);
}

void ServiceDirectory::addRelay(char* hostname, int port) {
    int fd;
    fd = socket(PF_INET, SOCK_STREAM, 0);
    struct hostent* host;
    host = gethostbyname(hostname);
    if(!host)
        throw runtime_error("Error looking up hostname \"" + string(hostname) + "\": " + strerror(errno));
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr = *((struct in_addr*)host->h_addr);
    int c = connect(fd, (struct sockaddr*)&addr, sizeof(addr));
    if(c < 0)  {
        char ps[16];
        snprintf(ps, sizeof(ps), "%d", port);
        throw runtime_error("Error connecting to " + string(hostname) + ":" + ps + ": " + strerror(errno));
    }
    pair<char*, int> p;
    p.first = hostname;
    p.second = port;
    tcp_relays[p] = fd;
}

void ServiceDirectory::removeRelay(char* hostname, int port) {
    pair<char*, int> p;
    p.first = hostname;
    p.second = port;
    if( tcp_relays.find(p) == tcp_relays.end() )
        return;
    shutdown(tcp_relays[p], SHUT_RDWR);
    tcp_relays.erase(p);
}

deque<int> ServiceDirectory::getRelayFds() {
    deque<int> r;
    for( map<pair<char*, int>, int>::const_iterator i = tcp_relays.begin();
            i != tcp_relays.end();
            i++)
    {
        r.push_back(i->second);
    }
    return r;
}

deque<int> ServiceDirectory::getIncomingTCPFds() {
    return incoming_tcp_fds;
}

---