OSGPointSockPipeline.cpp

Go to the documentation of this file.

/*---------------------------------------------------------------------------*\
 *                                OpenSG                                     *
 *                                                                           *
 *                                                                           *
 *             Copyright (C) 2000-2002 by the OpenSG Forum                   *
 *                                                                           *
 *                            www.opensg.org                                 *
 *                                                                           *
 *   contact: dirk@opensg.org, gerrit.voss@vossg.org, jbehr@zgdv.de          *
 *                                                                           *
\*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*\
 *                                License                                    *
 *                                                                           *
 * 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, version 2.                               *
 *                                                                           *
 * 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.                          *
 *                                                                           *
 * You should have received a copy of the GNU Library General Public         *
 * License along with this library; if not, write to the Free Software       *
 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.                 *
 *                                                                           *
\*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*\
 *                                Changes                                    *
 *                                                                           *
 *                                                                           *
 *                                                                           *
 *                                                                           *
 *                                                                           *
 *                                                                           *
\*---------------------------------------------------------------------------*/

//---------------------------------------------------------------------------
//  Includes
//---------------------------------------------------------------------------
#include <stdlib.h>
#include <stdio.h>
#include <assert.h>

#include "OSGConfig.h"
#include "OSGLog.h"
#include "OSGBaseFunctions.h"
#include "OSGBaseThread.h"
#include "OSGSocketSelection.h"
#include "OSGBinaryMessage.h"
#include "OSGPointSockPipeline.h"
#include "OSGGroupSockPipeline.h"
#include "OSGConnectionType.h"
#include "OSGBinaryMessage.h"

OSG_USING_NAMESPACE

/*-------------------------------------------------------------------------*/
/*                            constructor destructor                       */

PointSockPipeline::PointSockPipeline():
    Inherited(),
    _initialized(false)
{
    _prev.open();
    _next.open();
}

PointSockPipeline::~PointSockPipeline(void)
{
    _prev.close();
    _next.close();
}

const ConnectionType *PointSockPipeline::getType()
{
    return &_type;
}

/*-------------------------------------------------------------------------*/
/*                            connection                                   */

Connection::Channel PointSockPipeline::connectGroup(
    const std::string &address,
    Time               timeout)
{
    Channel channel = Inherited::connectGroup(address,timeout);
    return channel;
}

void PointSockPipeline::disconnect(void)
{
    _socket.close();
}

Connection::Channel PointSockPipeline::acceptGroup(Time timeout)
{
    Channel channel = Inherited::acceptGroup(timeout);
    return channel;
}

/*-------------------------------------------------------------------------*/
/*                              channel handling                           */

Connection::Channel PointSockPipeline::selectChannel(Time timeout)
    throw (ReadError)
{
    if(!_initialized)
        initialize();
    try
    {
        if(_prev.waitReadable(timeout))
            return 0;
    }
    catch(SocketError &e)
    {
        throw ReadError(e.what());
    }
    return -1;
}

/*-------------------------- create ---------------------------------------*/

PointConnection *PointSockPipeline::create(void)
{
    return new PointSockPipeline();
}

/*-------------------------------------------------------------------------*/
/*                              read write                                 */

void PointSockPipeline::read(MemoryHandle mem,UInt32 size)
{
    int len;

    if(!_initialized)
        initialize();
    // read data
    len=_prev.recv(mem,size);
    if(len==0)
    {
        throw ReadError("read got 0 bytes!");
    }
    // send to next in chain
    if(!_last)
        _next.send(mem,size);
}

void PointSockPipeline::readBuffer()
{
    int size;
    int len;

    if(!_initialized)
        initialize();

    // read buffer header
    len=_prev.recv(&_socketReadBuffer[0],sizeof(SocketBufferHeader));
    if(len==0)
        throw ReadError("peek got 0 bytes!");
    // read remaining data
    size = osgntohl(
        reinterpret_cast<SocketBufferHeader *>(&_socketReadBuffer[0])->size);

    len=_prev.recv(&_socketReadBuffer[sizeof(SocketBufferHeader)],
                   size);
    if(len==0)
        throw ReadError("read got 0 bytes!");
    readBufBegin()->setDataSize(size);
    // send to next in chain
    if(!_last)
        _next.send(&_socketReadBuffer[0],
                   sizeof(SocketBufferHeader)+size);
}    

/*-------------------------------------------------------------------------*/
/*                              private helpers                            */

void PointSockPipeline::initialize(void)
{
    BinaryMessage message;
    StreamSocket  sock;
    UInt32        nextPort;
    std::string   nextHost;
    UInt32        len;
    char          localhost[256];
    std::string   interf;

    // get local host name
    osgGetHostname(localhost,255);
    if(!getInterface().empty())
        interf = getInterface();
    else
        interf = localhost;

    sock.open();
    sock.bind(SocketAddress(interf.c_str(),0));
    sock.listen();

    // send my own address
    message.putString(interf);
    message.putUInt32(sock.getAddress().getPort());
    _socket.send(message);
    // accept prev
    _prev = sock.accept();
    sock.close();

    len = _socket.recv(message);
    if(len == 0)
        throw ReadError("Channel closed\n");
    _last = message.getUInt32();
    if(!_last)
    {
        nextHost = message.getString();
        nextPort = message.getUInt32();
        for(;;)
        {
            try
            {
                _next.connect(SocketAddress(nextHost.c_str(),
                                            nextPort));
                break;
            }
            catch(...)
            {
            }
        }
    }

    _initialized = true;
}

/*-------------------------------------------------------------------------*/
/*                              static type                                */

ConnectionType PointSockPipeline::_type(
    &PointSockPipeline::create,
    "SockPipeline");

/*-------------------------------------------------------------------------*/
/*                              cvs id's                                   */

#ifdef __sgi
#pragma set woff 1174
#endif

#ifdef OSG_LINUX_ICC
#pragma warning( disable : 177 )
#endif

namespace
{
    static Char8 cvsid_cpp       [] = "@(#)$Id: $";
    static Char8 cvsid_hpp       [] = OSG_GROUPSOCKPIPELINE_HEADER_CVSID;
}

---