vos/corelibs/vos/site.cc

Go to the documentation of this file.

/*
    This file is part of the Virtual Object System of
    the Interreality project (http://interreality.org).

    Copyright (C) 2001, 2002 Peter Amstutz

    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

    Peter Amstutz <tetron@interreality.org>
*/

#include "site.hh"
#include "remotesocketsite.hh"
#include "message.hh"
#include "messageblock.hh"

using namespace VOS;

#include <stdio.h>
#include <string>
#include <map>
#include <iterator>
#include <queue>
#include <algorithm>


#if !defined(HAVE_SOCKLEN_T) && !defined(__socklen_t_defined)
//#warning socklen_t not defined. Defining it as int.
typedef int socklen_t;
#define __socklen_t_defined 1
#endif


/** @file
    Implements Site.
*/

map<string, Site*> Site::siteTable;
boost::mutex Site::siteTable_mutex;
LocalSite* Site::defaultPeer = 0;
set<int> Site::allOpenSockets;

Site::Site()
    : MetaObject(0)
{
}

Site::~Site()
{
     for(map<string, MessageBlock*>::iterator i = messageBlockTable.begin();
        i != messageBlockTable.end();
        i++) {
        (*i).second->release(); // acquired by: addMessageBlock()
    }
    for(map<string, Message*>::iterator i = outgoingTraps.begin();
        i != outgoingTraps.end();
        i++) {
        (*i).second->release(); // acquired by: trapOutgoingReply()
    }
}

void Site::addHostAlias(const string& h)
{
    if(! hasHostAlias(h)) {
        boost::mutex::scoped_lock lock(hostnames_mutex);
        boost::mutex::scoped_lock lock2(siteTable_mutex);

        hostnames.insert(h);
        acquire(); // released by: removeHostAlias() or removeSite()
        siteTable[h]=this;
    }
    LOG("refcount", 5, "addHostAlias: count is " << getCount());
}

void Site::removeHostAlias(const string& h)
{
    if(hasHostAlias(h)) {
        boost::mutex::scoped_lock lock(hostnames_mutex);

        hostnames.erase(h);
        release(); // acquired by: addHostAlias() or addSite()
    }
    LOG("refcount", 5, "removeHostAlias: count is " << getCount());
}

bool Site::hasHostAlias(const string& h)
{
    boost::mutex::scoped_lock lock(hostnames_mutex);
    return (hostnames.count(h) > 0);
}

set<string>& Site::getHostAliases()
{
    boost::mutex::scoped_lock lock(hostnames_mutex);
    return hostnames;
}

void Site::addSite(Site* s)
{
    removeSite(*s);

    set<string>& v=s->getHostAliases();

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

    for(set<string>::iterator i=v.begin(); i != v.end(); i++) {
        LOG("addSite", 3, "adding " << *i << " to site table");
        s->acquire(); // released by: Site::removeSite() or removeHostAlias()
        siteTable[*i]=s;
    }
    LOG("refcount", 5, "addSite: count is " << s->getCount());
}

void Site::removeSite(Site& s)
{
    set<string>& v=s.getHostAliases();

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

    for(set<string>::iterator i=v.begin(); i != v.end(); i++) {
        if(siteTable.count(*i)) {
            siteTable.erase(*i);
            s.release(); // acquired by: Site::addSite or addHostAlias()
        }
    }
    LOG("refcount", 5, "removeSite: count is " << s.getCount());
}

struct hostent* Site::hostent_deepcopy(struct hostent* hp)
{
    struct hostent* ret = (struct hostent*)malloc(sizeof(struct hostent));

    ret->h_name = strdup(hp->h_name);
    int i = 0;
    while(hp->h_aliases[i]) i++;
    ret->h_aliases = (char**)malloc((i+1) * sizeof(char*));
    int n;
    for(n = 0; n < i; n++) {
        ret->h_aliases[n] = strdup(hp->h_aliases[n]);
    }
    ret->h_aliases[n] = 0;

    ret->h_addrtype = hp->h_addrtype;
    ret->h_length = hp->h_length;

    i=0;
    while(hp->h_addr_list[i]) i++;
    ret->h_addr_list = (char**)malloc((i+1) * sizeof(char*));
    for(n = 0; n < i; n++) {
        ret->h_addr_list[n] = (char*)malloc(hp->h_length);
        memcpy(ret->h_addr_list[n], hp->h_addr_list[n], hp->h_length);
    }
    ret->h_addr_list[n] = 0;

    return ret;
}

struct hostent* Site::gethostbyaddr_locked(const char *addr, int len, int type)
{
    static boost::mutex ghba_mutex;

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

    struct hostent* hp = gethostbyaddr(addr, len, type);

    if(! hp) return 0;

    return hostent_deepcopy(hp);
}

struct hostent* Site::gethostbyname_locked(const char *addr)
{
    static boost::mutex ghbn_mutex;

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

    struct hostent* hp = gethostbyname(addr);

    if(! hp) return 0;

    return hostent_deepcopy(hp);
}


void Site::freeHostEnt(struct hostent* hp)
{
    free(hp->h_name);
    for(int i = 0; hp->h_aliases[i]; i++) free(hp->h_aliases[i]);
    free(hp->h_aliases);
    for(int i = 0; hp->h_addr_list[i]; i++) free(hp->h_addr_list[i]);
    free(hp->h_addr_list);
    free(hp);
}

string Site::detectHostname(int fd, unsigned char ipaddr[4])
{
    struct sockaddr_in sa;
    socklen_t socklen = sizeof(struct sockaddr_in);
    getsockname(fd, (struct sockaddr *)&sa, &socklen);

    long addr = ntohl(sa.sin_addr.s_addr);
    char dottedquad[64];
    ipaddr[0] = (unsigned int)(addr >> 24) & 0xFF;
    ipaddr[1] = (unsigned int)(addr >> 16) & 0xFF;
    ipaddr[2] = (unsigned int)(addr >> 8) & 0xFF;
    ipaddr[3] = (unsigned int)(addr & 0xFF);
    sprintf(dottedquad, "%u.%u.%u.%u", (unsigned int)ipaddr[0], (unsigned int)ipaddr[1],
           (unsigned int)ipaddr[2], (unsigned int)ipaddr[3]);

    struct hostent* hp;
    int err = 0;
#ifdef HAVE_GETIPNODEBYNAME
    hp = getipnodebyaddr((const char*)&sa.sin_addr.s_addr, sizeof(addr), AF_INET, &err);
#else
    hp = gethostbyaddr_locked((const char*)&sa.sin_addr.s_addr, sizeof(addr), AF_INET);
        err = h_errno;
#endif
    if(hp) {
        struct hostent* hp2;
#ifdef HAVE_GETIPNODEBYNAME
        hp2 = getipnodebyname(hp->h_name, AF_INET, 0, &err);
#else
        hp2 = gethostbyname_locked(hp->h_name);
        err = h_errno;
#endif

#ifdef HAVE_GETIPNODEBYNAME
        freehostent(hp);
#else
        freeHostEnt(hp);
#endif
        string ret;
        if(hp2) {
            return ret = string(hp2->h_name);
        } else {
            LOG("site", 1, "Warning: gethostbyaddr failed for " << hp->h_name << ".  Using address " << dottedquad << " instead.");
            ret = dottedquad;
        }
#ifdef HAVE_GETIPNODEBYNAME
        freehostent(hp2);
#else
        freeHostEnt(hp2);
#endif
        return ret;
    } else {
        LOG("Site", 1, "Warning: gethostbyaddr failed for " << dottedquad << ".  Using that address as it is.");
        return dottedquad;
    }
}

Site& Site::findSite(const string& s) throw (NoSuchSiteError)
{
    map<string,Site*>::iterator i = siteTable.find(s);
    if(i == siteTable.end()) {
        char t[256];
        snprintf(t, sizeof(t), "%s:%i", s.c_str(), VOS_DEFAULT_PORT);
        i=siteTable.find(string(t));
        if(i == siteTable.end()) {
            try {
                unsigned int c = (unsigned int)s.find(":");
                if(c==string::npos) throw NoSuchSiteError("Bad site name: must be in the form \"host:port\"");
                string h=s.substr(0, c);
                int p=atoi(s.substr(c+1).c_str());

                RemoteSocketSite* rs = AsyncConnect::connect(h, p, defaultPeer, false);
                rs->getTypes();
                // pass reference through
                return *rs;
            } catch(RemoteSite::SiteConnectionError x) {
                LOG("site", 3, "cannot connect to site " << x.what());
                throw NoSuchSiteError(string("Cannot connect to site: ")+x.what());
            } catch(TimeoutError) {
                throw NoSuchSiteError(string("Site doesn't want to talk to us (timeout)"));
            } catch(runtime_error e) {
                LOG("site", 3, "cannot connect to site " << e.what());
                throw NoSuchSiteError(string("Unknown error: ") + e.what());
            }
        } else {
            ((*i).second)->acquire(); // released by: the caller of this method
            return *((*i).second);
        }
    } else {
        ((*i).second)->acquire(); // released by: the caller of this method
        return *((*i).second);
    }
}

LocalSite* Site::getDefaultPeer()
{
    if(defaultPeer) defaultPeer->acquire(); // released by: the caller of this method
    return defaultPeer;
}

void Site::setDefaultPeer(LocalSite* localsite)
{
    if(defaultPeer) defaultPeer->release(); // acquired by: setDefaultPeer()
    localsite->acquire(); // released by: setDefaultPeer()
    defaultPeer = localsite;
}

void Site::doSitePeering(LocalSite* ls, RemoteSite* rs, bool isspooftest, bool waitforhello)
{
    // This method assumes that the remote site is only owned by this thread,
    // so we only have to deal with locking the local site

    ls->acquire(); // released by: RemoteSite destructor
    rs->localpeer = ls;

    struct LocalSite::SiteTableEntry* st=new LocalSite::SiteTableEntry();
    st->partialMessage=0;
    rs->acquire(); // released by: LocalSite::removeRemotePeer()
    if(ls->iteratorsUsingPeerSites == 0 && waitforhello) {
        ls->peerSites[rs]=st;
    } else {
        boost::mutex::scoped_lock lock(ls->peerSitesBuffer_mutex);

        /* this is slightly inelegant, but putting it in a pair<> causes gcc to fail (bleh!) */
        ls->peerSitesBuffer_rs.push_back(rs);
        ls->peerSitesBuffer_st.push_back(st);
    }

    rs->defaultContext.setParameter("from", rs->getURL().getString());
    rs->defaultContext.setParameter("to", ls->getURL().getString());
    rs->defaultContext.setParameter("method", "");

    vRef<Message> m = new Message();
    m->setType("message");
    m->setMethod("core:hello");

    if(! isspooftest) {
        int r = rand();
        char s[16];
        snprintf(s, sizeof(s), "%i", r);
        m->insertField(-1, "antispoof", s);

        m->insertField(-1, "alias", ls->getURL().getHostAndPort());

        set<string>& v = ls->getHostAliases();
        for(set<string>::iterator i=v.begin(); i != v.end(); i++) {
            if((*i).substr(0, 9) != "localhost"
               && (*i) != ls->getURL().getHostAndPort())
            {
                m->insertField(-1, "alias", *i);
            }
        }
        rs->setAntiSpoofID(s);
    }

    rs->sendMessage(&m);

    if(waitforhello) {
        while(rs->isConnected() && !rs->getGreeted()) {
            defaultPeer->flushIncomingBuffers();
        }
    }
}

const map<string,Site*>& Site::getAllSites()
{
    return siteTable;
}

const string Site::getType()
{
    return string("site");
}

void Site::addMessageBlock(MessageBlock* m)
{
    m->acquire(); // released by: addMessageBlock(), removeMessageBlock() or this destructor
    map<string, MessageBlock*>::iterator i = messageBlockTable.find(m->getName());
    if(i != messageBlockTable.end()) (*i).second->release(); // acquired by: addMessageBlock()
    messageBlockTable[m->getName()] = m;
}

void Site::removeMessageBlock(MessageBlock* m)
{
    map<string, MessageBlock*>::iterator i = messageBlockTable.find(m->getName());
    if(i != messageBlockTable.end()) {
        (*i).second->release(); // acquired by: addMessageBlock()
        messageBlockTable.erase(i);
    }
}

MessageBlock* Site::getMessageBlock(const string& s) throw (NoSuchMessageBlockError)
{
    map<string, MessageBlock*>::iterator i = messageBlockTable.find(s);
    if(i != messageBlockTable.end()) {
        (*i).second->acquire(); // released by: the caller of this method
        return (*i).second;
    } else throw NoSuchMessageBlockError("message block not found");
}

void Site::sendMessage(Message* m)
{
    if(m->getNonce() != "") {
        map<string, Message*>::iterator i = outgoingTraps.find(m->getNonce());
        if(i != outgoingTraps.end()) {
            pREF(MessageContext*, mc, (*i).second->getMessageContext(),
                 if(mc) mc->addToReplyRecord(m));
            (*i).second->release(); // acquired by: trapOutgoingReply()
            outgoingTraps.erase(i);
        }
    }
}

void Site::sendMessage(MessageBlock* mb)
{
    for(int i = 0; i < mb->numMessages(); i++) {
        vRef<Message> m = mb->getMessage(i);
        if(m->getNonce() != "") {
            map<string, Message*>::iterator i = outgoingTraps.find(m->getNonce());
            if(i != outgoingTraps.end()) {
                vRef<MessageContext> mc = (*i).second->getMessageContext();
                if(&mc) mc->addToReplyRecord(&m);
                (*i).second->release(); // acquired by: trapOutgoingReply()
                outgoingTraps.erase(i);
            }
        }
    }
}

void Site::trapOutgoingReply(Message* m)
{
    pREF(MessageContext*, mc, m->getMessageContext(), if(!mc) return; );
    string dep = m->getDependency();
    unsigned int i=0;
    unsigned int n=0;
    LOG("site", 3, "Will trap outgoing nonce " << dep);
    while(i < dep.size()) {
        while(i < dep.size() && dep[i] != ',') i++;
        if(outgoingTraps[dep.substr(n, i-n)] == 0) {
            m->acquire(); // released by: sendMessage() or destructor
            outgoingTraps[dep.substr(n, i-n)] = m;
        }
        n=++i;
    }
}

Site& Site::getSite()
{
    acquire(); // released by: the caller of this method
    return *this;
}

bool Site::getGreeted()
{
    return greeted;
}

void Site::setGreeted(bool g)
{
    greeted = g;
}


string Site::generateUniqueName()
{
    char name[256];
    bool inuse;

    do {
        try {
            snprintf(name, sizeof(name), "%i", rand());
            rREF(Vobject&, foo, findObject(name), );
            inuse = true;
        } catch(Vobject::NoSuchObjectError) {
            inuse = false;
        }
    } while(inuse);
    return string(name);
}

string Site::uniqueName(const char* desiredName) {
    string name;
    // need to generate unique name?
    if(desiredName == 0 || strlen(desiredName) == 0) {
        return generateUniqueName();
    } else {
        name = desiredName;
        try {
            while(true) {
                vRef<Vobject> o = findObject(name);
                char tmp[256];
                snprintf(tmp, sizeof(tmp), "%s_%i", desiredName, rand());
                name = tmp;
            }
        } catch(NoSuchObjectError) {
            // yay, name is unique
        }
    }
    return name;
}

void Site::insertMessageFilter(int pos, MessageFilter* mf)
{
    if(pos < 0) pos += (int)messagefilters.size() + 1;
    messagefilters.insert(messagefilters.begin() + pos, mf);
}

void Site::removeMessageFilter(int pos)
{
    if(pos < 0) pos += (int)messagefilters.size() + 1;
    messagefilters.erase(messagefilters.begin() + pos);
}

void Site::removeMessageFilter(MessageFilter* m)
{
    for(unsigned int i = 0; i < messagefilters.size(); i++) {
        if(messagefilters[i] == m) {
            messagefilters.erase(messagefilters.begin() + i);
            i--;
        }
    }
}

const deque<MessageFilter*>& Site::getMessageFilters()
{
    return messagefilters;
}

/* Message filter stuff */

MessageFilter::~MessageFilter() { }

bool MessageFilter::checkMessage(MessageBlock* m)
{
    for(int n = 0; n < m->numMessages(); n++) {
        vRef<Message> msg = m->getMessage(n);
        if(! checkMessage(&msg)) {
            return false;
        }
    }
    return true;
}

/** Notify Event stuff */

VobjectNotifyEvent::VobjectNotifyEvent(ChildChangeListener* cl, VobjectEvent& e)
    : childlistener(cl), parentlistener(0), typelistener(0), event(e)
{
    // the following are released by the destructor
    if(RefCounted* rc = dynamic_cast<RefCounted*>(cl)) rc->acquire();
    e.acquire();
}

VobjectNotifyEvent::VobjectNotifyEvent(ParentChangeListener* pl, VobjectEvent& e)
    : childlistener(0), parentlistener(pl), typelistener(0), event(e)
{
    // the following are released by the destructor
    if(RefCounted* rc = dynamic_cast<RefCounted*>(pl)) rc->acquire();
    e.acquire();
}

VobjectNotifyEvent::VobjectNotifyEvent(TypeChangeListener* tl, VobjectEvent& e)
    : childlistener(0), parentlistener(0), typelistener(tl), event(e)
{
    // the following are released by the destructor
    if(RefCounted* rc = dynamic_cast<RefCounted*>(tl)) rc->acquire();
    e.acquire();
}

VobjectNotifyEvent::~VobjectNotifyEvent()
{
    if(childlistener) if(RefCounted* rc = dynamic_cast<RefCounted*>(childlistener)) rc->release();
    if(parentlistener) if(RefCounted* rc = dynamic_cast<RefCounted*>(parentlistener)) rc->release();
    if(typelistener) if(RefCounted* rc = dynamic_cast<RefCounted*>(typelistener)) rc->release();
    event.release();
}

void VobjectNotifyEvent::notify()
{
    switch(event.getEvent()) {
    case VobjectEvent::TypeInsert:
        typelistener->notifyTypeInserted(event);
        break;
    case VobjectEvent::TypeRemove:
        typelistener->notifyTypeRemoved(event);
        break;
    case VobjectEvent::ParentInsert:
        parentlistener->notifyParentInserted(event);
        break;
    case VobjectEvent::ParentRemove:
        parentlistener->notifyParentRemoved(event);
        break;
    case VobjectEvent::ChildInsert:
        childlistener->notifyChildInserted(event);
        break;
    case VobjectEvent::ChildReplace:
        childlistener->notifyChildReplaced(event);
        break;
    case VobjectEvent::ChildRemove:
        childlistener->notifyChildRemoved(event);
        break;
    default:
        LOG("vobject", 0, "Supplied VobjectNotifyEvent::notify with an unsupported event type!");
    }
}

---