OSG::PointSockConnection Class Reference

00065                                         :
00066     Inherited(0)
00067 {
00068     _acceptSocket.open();
00069     _acceptSocket.setReusePort(true);
00070 
00071     _socketReadBuffer.resize(131071);
00072     _socketWriteBuffer.resize( _socketReadBuffer.size() );
00073     // reserve first bytes for buffer size
00074     readBufAdd (&_socketReadBuffer [sizeof(SocketBufferHeader)],
00075                 _socketReadBuffer.size() -sizeof(SocketBufferHeader));
00076     writeBufAdd(&_socketWriteBuffer[sizeof(SocketBufferHeader)],
00077                 _socketWriteBuffer.size()-sizeof(SocketBufferHeader));
00078 }

00083 {
00084     _acceptSocket.close();
00085 }

00090 {
00091     return &_type;
00092 }

00103 {
00104     StreamSocket socket;
00105     if(GroupSockConnection::connectSocket( socket,
00106                                            address,
00107                                           _remoteAddress,
00108                                            timeout))
00109     {
00110         _socket = socket;
00111         _pointToPoint = true;
00112         return true;
00113     }
00114     else
00115     {
00116         return false;
00117     }
00118 }

00126 {
00127     StreamSocket socket;
00128     if(GroupSockConnection::connectSocket( socket,
00129                                            address,
00130                                           _remoteAddress,
00131                                            timeout))
00132     {
00133         _socket = socket;
00134         _pointToPoint = false;
00135         return true;
00136     }
00137     else
00138     {
00139         return false;
00140     }
00141 }

00146 {
00147     _socket.close();
00148 }

00173 {
00174     if(GroupSockConnection::acceptSocket(_acceptSocket,
00175                                          _socket,
00176                                          _remoteAddress,
00177                                           timeout))
00178     {
00179         _pointToPoint = false;
00180         return 0;
00181     }
00182     else
00183     {
00184         return -1;
00185     }
00186 }

00154 {
00155     if(GroupSockConnection::acceptSocket(_acceptSocket,
00156                                          _socket,
00157                                          _remoteAddress,
00158                                           timeout))
00159     {
00160         _pointToPoint = true;
00161         return 0;
00162     }
00163     else
00164     {
00165         return -1;
00166     }
00167 }

00195 {
00196     int         port=0;
00197     char        localhost[256];
00198     char        host[256];
00199     char        portStr[256];
00200     std::string interf;
00201 
00202     // get local host name
00203     osgGetHostname(localhost,255);
00204     if(!getInterface().empty())
00205         interf = getInterface();
00206     else
00207         interf = localhost;
00208 
00209     // parse address
00210     if(!address.empty())
00211         if(sscanf(address.c_str(),"%*[^:]:%d",&port) != 1)
00212             if(sscanf(address.c_str(),":%d",&port) != 1)
00213                 port = 0;
00214     // bind port
00215     _acceptSocket.setReusePort(true);
00216     if(!getInterface().empty())
00217         _acceptSocket.bind(SocketAddress(getInterface().c_str(),port));
00218     else
00219         _acceptSocket.bind(SocketAddress(SocketAddress::ANY,port));
00220 
00221     SINFO << "Connection bound to "
00222           << _acceptSocket.getAddress().getHost() << ":"
00223           << _acceptSocket.getAddress().getPort() << std::endl;
00224     _acceptSocket.listen();
00225     // create address
00226     sprintf(portStr,"%d",_acceptSocket.getAddress().getPort());
00227     return interf + ":" + portStr;
00228 }

00257 {
00258     UInt32 tag;
00259     try
00260     {
00261         if(!_socket.waitReadable(timeout))
00262             return false;
00263         if(!_socket.recv(&tag,sizeof(tag)))
00264             throw ReadError("Channel closed");
00265 
00266         tag = osgNetToHost<UInt32>(tag);
00267         if(tag != 314156)
00268         {
00269             FFATAL(("Stream out of sync in SockConnection\n"));
00270             throw ReadError("Stream out of sync");
00271         }
00272     }
00273     catch(SocketError &e)
00274     {
00275         throw ReadError(e.what());
00276     }
00277     return true;
00278 }

00283 {
00284     UInt32 tag = osgHostToNet<UInt32>(314156);
00285     try
00286     {
00287         _socket.send(&tag,sizeof(tag));
00288     }
00289     catch(SocketError &e)
00290     {
00291         throw ReadError(e.what());
00292     }
00293 }

00238 {
00239     try
00240     {
00241         if(_socket.waitReadable(timeout))
00242             return 0;
00243     }
00244     catch(SocketError &e)
00245     {
00246         throw ReadError(e.what());
00247     }
00248     return -1;
00249 }

00301 {
00302     return new PointSockConnection();
00303 }

00316 {
00317     int len;
00318 
00319     // read data
00320     len=_socket.recv(mem,size);
00321     if(len==0)
00322     {
00323         throw ReadError("read got 0 bytes!");
00324     }
00325 }

00335 {
00336     int size;
00337     int len;
00338 
00339     // read buffer header
00340     len=_socket.recv(&_socketReadBuffer[0],sizeof(SocketBufferHeader));
00341     if(len==0)
00342         throw ReadError("peek got 0 bytes!");
00343     // read remaining data
00344     size=osgNetToHost<UInt32>(((SocketBufferHeader*)&_socketReadBuffer[0])->size);
00345     len=_socket.recv(&_socketReadBuffer[sizeof(SocketBufferHeader)],
00346                      size);
00347     if(len==0)
00348         throw ReadError("read got 0 bytes!");
00349     readBufBegin()->setDataSize(size);
00350 }    

00360 {
00361     _socket.send(mem,size);
00362 }

00370 {
00371     Int32 index;
00372     UInt32 size = writeBufBegin()->getDataSize();
00373 
00374     // write size to header
00375     ((SocketBufferHeader*)&_socketWriteBuffer[0])->size = 
00376         osgHostToNet<UInt32>(size);
00377 
00378     if(size)
00379     {
00380         // write whole block
00381         _socket.send(&_socketWriteBuffer[0],
00382                      size+sizeof(SocketBufferHeader));
00383     }
00384 }

00102 {
00103     return _interface;
00104 }

00110 {
00111     _interface = interf;
00112 }

00086 {
00087     UInt8 const *data = static_cast<UInt8 const *>(src);
00088 
00089     if(_zeroCopyThreshold && size >= _zeroCopyThreshold)
00090     {
00091         if(_zeroCopyThreshold == 1)
00092         {
00093             write(const_cast<MemoryHandle>(data), size);
00094         }
00095         else
00096         {
00097             UInt8 tag = 1;
00098 
00099             // we have to write a tag, to indicate the membership
00100             // of this zero copy block to the current data block
00101             put(&tag, sizeof(tag));
00102 
00103             _zeroCopyBuffers.push_back(
00104                 MemoryBlock(const_cast<MemoryHandle>(data), size, size));
00105         }
00106     }
00107     else
00108     {
00109         UInt32 copySize;
00110 
00111         while(size != 0)
00112         {
00113             if(_currentWriteBuffer == writeBufEnd())
00114             {
00115                 pushBuffer();
00116             }
00117 
00118             copySize = osgMin((_currentWriteBuffer->getSize() -
00119                                _currentWriteBufferPos),
00120                               size);
00121 
00122             memcpy(_currentWriteBuffer->getMem() + _currentWriteBufferPos,
00123                     data,
00124                     copySize);
00125 
00126              size                  -= copySize;
00127             _currentWriteBufferPos += copySize;
00128              data                  += copySize;
00129 
00130             // skip to next buffer if current buffer is full
00131             if(_currentWriteBufferPos == _currentWriteBuffer->getSize())
00132             {
00133                 _currentWriteBuffer->setDataSize(_currentWriteBufferPos);
00134                 _currentWriteBuffer++;
00135                 _currentWriteBufferPos = 0;
00136             }
00137         }
00138     }
00139 }