Source/Base/Network/Socket/OSGGroupSockPipeline.cpp

   1 /*---------------------------------------------------------------------------*\
   2  *                                OpenSG                                     *
   3  *                                                                           *
   4  *                                                                           *
   5  *             Copyright (C) 2000-2002 by the OpenSG Forum                   *
   6  *                                                                           *
   7  *                            www.opensg.org                                 *
   8  *                                                                           *
   9  *   contact: dirk@opensg.org, gerrit.voss@vossg.org, jbehr@zgdv.de          *
  10  *                                                                           *
  11 \*---------------------------------------------------------------------------*/
  12 /*---------------------------------------------------------------------------*\
  13  *                                License                                    *
  14  *                                                                           *
  15  * This library is free software; you can redistribute it and/or modify it   *
  16  * under the terms of the GNU Library General Public License as published    *
  17  * by the Free Software Foundation, version 2.                               *
  18  *                                                                           *
  19  * This library is distributed in the hope that it will be useful, but       *
  20  * WITHOUT ANY WARRANTY; without even the implied warranty of                *
  21  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU         *
  22  * Library General Public License for more details.                          *
  23  *                                                                           *
  24  * You should have received a copy of the GNU Library General Public         *
  25  * License along with this library; if not, write to the Free Software       *
  26  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.                 *
  27  *                                                                           *
  28 \*---------------------------------------------------------------------------*/
  29 /*---------------------------------------------------------------------------*\
  30  *                                Changes                                    *
  31  *                                                                           *
  32  *                                                                           *
  33  *                                                                           *
  34  *                                                                           *
  35  *                                                                           *
  36  *                                                                           *
  37 \*---------------------------------------------------------------------------*/
  38
  39 //---------------------------------------------------------------------------
  40 //  Includes
  41 //---------------------------------------------------------------------------
  42 #include <cstdlib>
  43 #include <cstdio>
  44 #include <cassert>
  45
  46 #include <set>
  47
  48 #include "OSGConfig.h"
  49 #include "OSGLog.h"
  50 #include "OSGBaseThread.h"
  51 #include "OSGSocketSelection.h"
  52 #include "OSGBinaryMessage.h"
  53 #include "OSGGroupSockPipeline.h"
  54 #include "OSGConnectionType.h"
  55 #include "OSGBinaryMessage.h"
  56
  57 OSG_USING_NAMESPACE
  58
  59 /** \class OSG::GroupSockPipeline
  60  **/
  61
  62 /*-------------------------------------------------------------------------*/
  63 /*                            constructor destructor                       */
  64
  65 /*! Constructor
  66  */
  67
  68 GroupSockPipeline::GroupSockPipeline():
  69      Inherited  (     ),
  70     _next       (     ),
  71     _initialized(false)
  72 {
  73     _next.open();
  74 }
  75
  76 /*! Destructor
  77  */
  78 GroupSockPipeline::~GroupSockPipeline(void)
  79 {
  80     _next.close();
  81 }
  82
  83 /*! get connection type
  84  */
  85 const ConnectionType *GroupSockPipeline::getType()
  86 {
  87     return &_type;
  88 }
  89
  90 /*-------------------------------------------------------------------------*/
  91 /*                            connection                                   */
  92
  93 /*! connect to the given point. If timeout is reached, -1 is
  94     returned
  95  */
  96 GroupConnection::Channel GroupSockPipeline::connectPoint(
  97     const std::string &address,
  98     Time               timeout)
  99 {
 100     Channel channel = Inherited::connectPoint(address,timeout);
 101     return channel;
 102 }
 103
 104 /*! disconnect the given channel
 105  */
 106 void GroupSockPipeline::disconnect(Channel channel)
 107 {
 108     Inherited::disconnect(channel);
 109 }
 110
 111 /*! accept an icomming point connection. If timeout is reached,
 112     -1 is returned. If timeout is -1 then wait without timeout
 113  */
 114 GroupConnection::Channel GroupSockPipeline::acceptPoint(Time timeout)
 115 {
 116     Connection::Channel channel = Inherited::acceptPoint(timeout);
 117     return channel;
 118 }
 119
 120 /*-------------------------- helpers --------------------------------------*/
 121
 122 /** \brief create conneciton
 123  */
 124
 125 GroupConnection *GroupSockPipeline::create(void)
 126 {
 127     return new GroupSockPipeline();
 128 }
 129
 130 /*-------------------------------------------------------------------------*/
 131 /*                              read write                                 */
 132
 133 /** Write data to all destinations
 134  *
 135  * \param mem   Pointer to data buffer
 136  * \param size  Size of bytes to write
 137  *
 138  **/
 139
 140 void GroupSockPipeline::write(MemoryHandle mem,UInt32 size)
 141 {
 142     if(!_initialized)
 143         initialize();
 144
 145     try
 146     {
 147         if(getChannelCount())
 148             _next.send(mem,size);
 149     }
 150     catch(SocketException &e)
 151     {
 152         throw WriteError(e.what());
 153     }
 154 }
 155
 156 /** Write buffer
 157  *
 158  * Write blocksize and data.
 159  *
 160  **/
 161 void GroupSockPipeline::writeBuffer(void)
 162 {
 163     if(!_initialized)
 164         initialize();
 165
 166     UInt32 size = writeBufBegin()->getDataSize();
 167
 168     // write size to header
 169     (reinterpret_cast<SocketBufferHeader*>(&_socketWriteBuffer[0]))->size =
 170         osgHostToNet<UInt32>(size);
 171
 172     if(size)
 173     {
 174         _next.send(&_socketWriteBuffer[0],
 175                           size+sizeof(SocketBufferHeader));
 176     }
 177 }
 178
 179 /*-------------------------------------------------------------------------*/
 180 /*                              private helpers                            */
 181
 182 /*! initialize pipeline
 183  */
 184 void GroupSockPipeline::initialize(void)
 185 {
 186     UInt32         index,len;
 187     UInt32         nextPort;
 188     std::string    nextHost;
 189     BinaryMessage  message;
 190
 191     for(index = 0 ; index<_sockets.size() ; ++index)
 192     {
 193         len = _sockets[index].recv(message);
 194         if(len == 0)
 195             throw ReadError("Channel closed\n");
 196         nextHost = message.getString();
 197         nextPort = message.getUInt32();
 198
 199         message.clear();
 200         if(index == 0)
 201         {
 202             message.putUInt32(true);
 203             _sockets[_sockets.size()-1].send(message);
 204             for(;;)
 205             {
 206                 try
 207                 {
 208                     _next.connect(SocketAddress(nextHost.c_str(),
 209                                                 nextPort));
 210                     break;
 211                 }
 212                 catch(...)
 213                 {
 214                 }
 215             }
 216         }
 217         else
 218         {
 219             message.clear();
 220             message.putUInt32(false);
 221             message.putString(nextHost);
 222             message.putUInt32(nextPort);
 223             _sockets[index-1].send(message);
 224         }
 225     }
 226     _initialized = true;
 227 }
 228
 229 /*-------------------------------------------------------------------------*/
 230 /*                              static type                                */
 231
 232 ConnectionType GroupSockPipeline::_type(
 233     &GroupSockPipeline::create,
 234     "SockPipeline");