vos/corelibs/vos/remotestreamsite.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-2003 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 "remotestreamsite.hh"
#include "messageblock.hh"
#include "message.hh"

using namespace VOS;

#include <stdio.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

RemoteStreamSite::RemoteStreamSite(FILE* readfd, FILE* writefd)
    : VobjectImplementation("fd", 0, false), RemoteMetaObject("fd", 0),
      RemoteVobject("fd", 0), MetaObject(0),
      readingFILE(readfd), writingFILE(writefd), pretendIsLocal(false), isScript(false),
      partialMessage(new MessageBlock())
{
}

RemoteStreamSite::~RemoteStreamSite()
{
    LOG("remotestreamsite", 5, "deleting " << getURL().getString());
    if(partialMessage) partialMessage->release();
}

int RemoteStreamSite::readStream(char* data, unsigned int datasize)
{
    return (int)fread(data, 1, datasize, readingFILE);
}

int RemoteStreamSite::writeStream(const char* data, unsigned int datasize)
{
    return (int)fwrite(data, 1, datasize, writingFILE);
}

void RemoteStreamSite::sendMessage(Message* m)
{
/*    if(outgoingLock) {
        if(enqueueOutgoing) {
            m->acquire();  // released by: enableOutgoing()
            outgoingQueue.push(pair<Message*, MessageBlock*>(m, 0));
        }
        } else {*/
        Site::sendMessage(m);
        LOG("RemoteStreamSite::sendMessage", 4, ">>>(outgoing)>>> sending to " << getURL().getString() << " [" << m->refcount_debug
            << "/" << m->getCount() <<  "]\n" << m->getFormattedString());
        LOG("refcount", 5, "sendMessage: count on this remotesite is " << getCount());

        if(writingFILE > 0) {
            const string& s = m->getFormattedString();
            if(writeStream(s.c_str(), (unsigned int)s.size()) < 1) {
                throw SiteConnectionError("disconnected");
            }
        }
//    }
}

void RemoteStreamSite::sendMessage(MessageBlock* m)
{
/*    if(outgoingLock) {
        if(enqueueOutgoing) {
            m->acquire();  // released by: enableOutgoing()
            outgoingQueue.push(pair<Message*, MessageBlock*>(0, m));
        }
        } else {*/
        Site::sendMessage(m);
        LOG("RemoteStreamSite::sendMessage", 4, ">>>(outgoing)>>> sending to " << getURL().getString() << "\n" << m->getString());
        LOG("refcount", 5, "sendMessage: count on this remotesite is " << getCount());

        if(writingFILE > 0) {
            const string& s = m->getString();
            if(writeStream(s.c_str(), (unsigned int)s.size()) < 1) {
                throw SiteConnectionError("disconnected");
            }
        }
//    }
}

void RemoteStreamSite::handleDisconnection()
{
    LOG("remotesite", 3, "remote site disconnected: " << strerror(errno));
    LOG("refcount", 5, "handleDisconnection: remote site count is " << getCount());

    //LOG("remotesite", 4, "closing connection rfd " << readingFD << "  wrf " << writingFD);

    if(readingFILE) fclose(readingFILE);
    if(writingFILE) fclose(writingFILE);
    //allOpenSockets.erase(readingFD);
    readingFILE = writingFILE = 0;

    if(! isScript) excise();
}

void RemoteStreamSite::flushIncomingBuffers()
{
    LOG("remotestreamsite", 5, "check baby, remote stream site " << this);
    if(readingFILE == 0) return;
    do {
        char data[1024];

        LOG("remotestreamsite", 5, "about to go into readStream");
        int r=readStream(data, sizeof(data));
        LOG("remotestreamsite", 4, "read stream got " << r);
        if(r == -1 || r == 0) {
            handleDisconnection();
            return;
        }
        string s((const char*)data, r);
        LOG("remotestreamsite", 5, "received " << s);
        int p=partialMessage->parseUpdate(s);
        LOG("remotestreamsite", 4, "parseUpdate returned " << p);
        while(p > 0) {
            vRef<Message> last = partialMessage->lastMessage();
            static int incoming_count = 1;
            last->incoming_debug = incoming_count++;
            localpeer->scheduleMessageBlock(partialMessage, 0, this);
            partialMessage->release(); // acquired by: new MessageBlock() above or below
            s.erase(s.begin(), s.begin() + p);
            partialMessage = new MessageBlock(); // released by: completed message (above) or destructor
            if(s.size() > 0) {
                p = partialMessage->parseUpdate(s);
            } else break;
        }
    } while(1);
}

bool RemoteStreamSite::runScript(const string& filename, LocalSite& ls, Message* lastmessage)
{
    FILE* fd = 0;

#ifdef HAVE_POPEN
    FILE* gzip = 0;
    if(filename.substr(filename.size()-3, 3) == ".gz") {
        char run[512];
        snprintf(run, sizeof(run), "gunzip -c '%s'", filename.c_str());
        if(!(gzip = popen(run, "r"))) return false;
        fd = fileno(gzip);
    } else {
#endif
        fd = fopen(filename.c_str(), "r");
#ifdef HAVE_POPEN
    }
#endif

    if(fd > 0) {
        pREF(RemoteStreamSite*, rss, new RemoteStreamSite(fd, 0),
             rss->pretendIsLocal = true;
             rss->isScript = true;
             string f = filename;
             for(unsigned int i = 0; i < f.size(); i++) if(f[i] == '/') f[i]='_';
             rss->setURL(URL("vop", f, "0", ""));
             rss->addHostAlias(f+":0");
             doSitePeering(&ls, rss, true, false);
             addSite(rss);
             //allOpenSockets.insert(fd);
             double oldtimeout = ls.getTimeoutOnSelect();
             ls.setTimeoutOnSelect(0);
             ls.flushIncomingBuffers();
             //LOG("remotestreamsite", 4, "0 rss readinfd is " << rss->readingFD);
             if(lastmessage) {
                 lastmessage->setSourceSite(rss);
                 ls.sendMessage(lastmessage);
             }
             //LOG("remotestreamsite", 4, "1 rss readinfd is " << rss->readingFD);
             while(rss->readingFILE > 0 || ls.checkScheduleHoldsSite(rss)) {
                 LOG("remotestreamsite", 4, "waiting for script...");
                 ls.flushIncomingBuffers();
             }
             ls.setTimeoutOnSelect(oldtimeout);
             LOG("remotestreamsite", 4, "about to rss excise with count " << rss->getCount());
             rss->excise();
             //allOpenSockets.erase(fd);
            );
#ifdef HAVE_POPEN
        if(gzip) pclose(gzip);
#endif
        return true;
    } else {
        LOG("remotestreamsite", 2, "Cannot open script: " << strerror(errno) << " " << filename);
        return false;
    }
}

bool RemoteStreamSite::isConnected()
{
    return (isScript || writingFILE != 0);
}

---