RemoteContext.cc

Go to the documentation of this file.

//=============================================================================
//  Computational Process Networks class library
//  Copyright (C) 1997-2006  Gregory E. Allen and The University of Texas
//
//  This library is free software; you can redistribute it and/or modify it
//  under the terms of the GNU Library 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
//  Library General Public License for more details.
//
//  The GNU Public License is available in the file LICENSE, or you
//  can write to the Free Software Foundation, Inc., 59 Temple Place -
//  Suite 330, Boston, MA 02111-1307, USA, or you can find it on the
//  World Wide Web at http://www.fsf.org.
//=============================================================================
#include "common_priv.h"
#include <cpn/remote_context/RemoteContext.h>
#include <cpn/utils/ErrnoException.h>
#include <cpn/threading/PthreadFunctional.h>
#include <vector>
#include <stdio.h>
#include <Variant/Variant.h>

namespace CPN {

    RemoteContext::RemoteContext(const SocketAddress &addr)
        : endwrite(false)
    { 
        thread.reset(CreatePthreadFunctional(this, &RemoteContext::EntryPoint));
        sock.Connect(addr);
        sock.SetNoDelay(true);
        thread->Start();
    }

    RemoteContext::RemoteContext(const SockAddrList &addrs)
        : endwrite(false)
    { 
        thread.reset(CreatePthreadFunctional(this, &RemoteContext::EntryPoint));
        sock.Connect(addrs);
        sock.SetNoDelay(true);
        thread->Start();
    }

    RemoteContext::RemoteContext(int fd)
        : sock(fd)
    {
        thread.reset(CreatePthreadFunctional(this, &RemoteContext::EntryPoint));
        sock.SetNoDelay(true);
        thread->Start();
    }

    RemoteContext::~RemoteContext() {
        EndWrite();
        thread->Join();
    }

    void RemoteContext::SendMessage(const Variant &msg) {
        // We have the lock
        if (!sock.Closed() && !endwrite) {
            std::string message = libvariant::SerializeJSON(msg);
            //printf("<<< %s\n", message.c_str());
            unsigned numwritten = 0;
            while (numwritten < message.size() + 1) {
                numwritten += sock.Write(message.c_str() + numwritten, (message.size() + 1) - numwritten);
            }
        }
    }

    void *RemoteContext::EntryPoint() {
        std::vector<char> buf(4*1024);
        unsigned num = 0;
        try {
            sock.Readable(false);
            sock.Poll(-1);
            while (sock.Good()) {
                if (buf.size() - num <= 0) {
                    buf.resize(buf.size()*2);
                }
                unsigned numread = sock.Recv(&buf[num], buf.size() - num, false);
                if (numread == 0) {
                    if (sock.Eof()) {
                        Terminate();
                        return 0;
                    }
                    sock.Poll(-1);
                } else {
                    num += numread;
                    char *end = (char*)memchr(&buf[0], 0, num);
                    while (end != 0) {
                        Variant v = libvariant::DeserializeJSON(&buf[0]);
                        DispatchMessage(v);
                        end += 1;
                        num -= (end - &buf[0]);
                        memmove(&buf[0], end, num);
                        end = (char*)memchr(&buf[0], 0, num);
                    }
                }
            }
        } catch (const ErrnoException &e) {
            fprintf(stderr, "RemoteContext: Uncaught errno exception (%d): %s\n", e.Error(), e.what());
            Terminate();
        }
        sock.Close();
        return 0;
    }

    void RemoteContext::EndWrite() {
        PthreadMutexProtected alock(lock);
        endwrite = true;
        sock.ShutdownWrite();
    }

    bool RemoteContext::IsEndWrite() {
        PthreadMutexProtected alock(lock);
        return endwrite;
    }

}