OSG::GroupSockConnection Class Reference

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

00085 {
00086     // close and remove sockets
00087     while(_sockets.size())
00088     {
00089         try
00090         {
00091             _sockets.begin()->close();
00092             _sockets.erase(_sockets.begin());
00093         }
00094         catch(...)
00095         {
00096         }
00097     }
00098     _acceptSocket.close();
00099 }

00104 {
00105     return &_type;
00106 }

00117 {
00118     Channel channel = -1;
00119     StreamSocket socket;
00120     SocketAddress destination;
00121     if(connectSocket(socket,address,destination,timeout))
00122     {
00123         channel = newChannelIndex(_sockets.size());
00124         _sockets.push_back(socket);
00125         _remoteAddresses.push_back(destination);
00126         _readIndex = 0;
00127     }
00128     return channel;
00129 }

00134 {
00135     ChannelIndex index = channelToIndex(channel);
00136     try
00137     {
00138         _sockets[index].close();
00139     }
00140     catch(...)
00141     {
00142     }
00143     _sockets.erase(_sockets.begin() + index);
00144     delChannelIndex(index);
00145     _readIndex = 0;
00146 }

00152 {
00153     StreamSocket from;
00154     SocketAddress destination;
00155     if(GroupSockConnection::acceptSocket(_acceptSocket,
00156                                          from,
00157                                          destination,
00158                                          timeout))
00159     {
00160         Channel channel = newChannelIndex(_sockets.size());
00161         _sockets.push_back(from);
00162         _remoteAddresses.push_back(destination);
00163         _readIndex = 0;
00164         return channel;
00165     }
00166     else
00167     {
00168         return -1;
00169     }
00170 }

00179 {
00180     int         port=0;
00181     char        localhost[256];
00182     char        portStr[256];
00183     std::string interf;
00184     std::string boundedAddress;
00185 
00186     // get local host name
00187     osgGetHostname(localhost,255);
00188     if(!getInterface().empty())
00189         interf = getInterface();
00190     else
00191         interf = localhost;
00192     // parse address
00193     if(!address.empty())
00194         if(sscanf(address.c_str(),"%*[^:]:%d",&port) != 1)
00195             if(sscanf(address.c_str(),":%d",&port) != 1)
00196                 port = 0;
00197     // bind port
00198     _acceptSocket.setReusePort(true);
00199     _acceptSocket.bind(SocketAddress(interf.c_str(),port));
00200     SINFO << "Connection bound to "
00201           << _acceptSocket.getAddress().getHost() << ":"
00202           << _acceptSocket.getAddress().getPort() << std::endl;
00203     _acceptSocket.listen();
00204     // create address
00205     sprintf(portStr,"%d",_acceptSocket.getAddress().getPort());
00206     return interf + ":" + portStr;
00207 }

00212 {
00213     if(params.empty())
00214         return;
00215 
00216     std::string option = "bufferSize=";
00217     std::string::size_type i = 0;
00218     if((i=params.find(option)) != std::string::npos)
00219     {
00220         std::string str = params.substr(i + option.size());
00221 
00222         std::stringstream ss;
00223         std::string::size_type j = 0;
00224         while(j < str.length() && str[j] != ',' && isdigit(str[j]))
00225         {
00226             ss << str[j++];
00227         }
00228         UInt32 bufferSize;
00229         ss >> bufferSize;
00230 
00231         // clear old buffer.
00232         readBufClear();
00233         writeBufClear();
00234 
00235         _socketReadBuffer.resize(bufferSize);
00236         _socketWriteBuffer.resize(_socketReadBuffer.size());
00237         
00238         // reserve first bytes for buffer size
00239         readBufAdd (&_socketReadBuffer [sizeof(SocketBufferHeader)],
00240                     _socketReadBuffer.size() -sizeof(SocketBufferHeader));
00241         writeBufAdd(&_socketWriteBuffer[sizeof(SocketBufferHeader)],
00242                     _socketWriteBuffer.size()-sizeof(SocketBufferHeader));
00243 
00244         FINFO(("GroupSockConnection::setParams : setting buffer size to %u.\n", bufferSize));
00245     }
00246 }

00256 {
00257     Int32 maxnread=0,nread;
00258     ChannelIndex index;
00259     SocketSelection selection,result;
00260 
00261     // if there is data in the read buffer, return current channel
00262     if(_zeroCopyThreshold != 1 &&
00263        _currentReadBuffer != readBufEnd())
00264     {
00265         FFATAL(("Channel change ignores data in current buffer"))
00266         return indexToChannel(_readIndex);
00267     }    
00268 
00269     if(_selection[_readIndex] &&
00270        _sockets[_readIndex].getAvailable())
00271     {
00272         return indexToChannel(_readIndex);;
00273     }
00274 
00275     // wait for first socket to deliver data
00276     for(index = 0 ; 
00277     index < static_cast<ChannelIndex>(_sockets.size()); 
00278     ++index)
00279     {
00280         if(_selection[index])
00281             selection.setRead(_sockets[index]);
00282     }
00283     
00284     try 
00285     {
00286         // select ok ?
00287         if(!selection.select(timeout,result))
00288             return -1;
00289 
00290         // use socket with most data
00291         for(index = 0 ; index < _sockets.size() ; ++index)
00292         {
00293             if(result.isSetRead(_sockets[index]))
00294             {
00295                 nread=_sockets[index].getAvailable();
00296                 if(maxnread < nread)
00297                 {
00298                     maxnread = nread;
00299                     _readIndex=index;
00300                 }
00301             }
00302         }
00303     }
00304     catch(SocketException &e)
00305     {
00306         throw ReadError(e.what());
00307     }
00308 
00309     // return channel id
00310     return indexToChannel(_readIndex);
00311 }

00385 {
00386     return new GroupSockConnection();
00387 }

00400 {
00401     int len;
00402 
00403     // read data
00404     len=_sockets[_readIndex].recv(mem,size);
00405     if(len==0)
00406     {
00407 //        throw ChannelClosed(indexToChannel(_readIndex));
00408         throw ReadError("Channel closed");
00409     }
00410 } 

00420 {
00421     int size;
00422     int len;
00423 
00424     // read buffer header
00425     len=_sockets[_readIndex].recv(&_socketReadBuffer[0],sizeof(SocketBufferHeader));
00426     if(len==0)
00427         throw ReadError("Channel closed");
00428     // read remaining data
00429     size=osgNetToHost<UInt32>(((SocketBufferHeader*)&_socketReadBuffer[0])->size);
00430     len=_sockets[_readIndex].recv(&_socketReadBuffer[sizeof(SocketBufferHeader)],
00431                          size);
00432     if(len==0)
00433         throw ReadError("Channel closed");
00434     readBufBegin()->setDataSize(size);
00435 }    

00445 {
00446     Int32 index;
00447 
00448     try
00449     {
00450         // write to all connected sockets
00451         for(index = 0 ; index < _sockets.size() ; ++index)
00452             _sockets[index].send(mem,size);
00453     }
00454     catch(SocketException &e)
00455     {
00456         throw WriteError(e.what());
00457     }
00458 }

00466 {
00467     Int32 index;
00468     UInt32 size = writeBufBegin()->getDataSize();
00469     // write size to header
00470     ((SocketBufferHeader*)&_socketWriteBuffer[0])->size = 
00471         osgHostToNet<UInt32>(size);
00472     if(size)
00473     {
00474         // write data to all sockets
00475         for(index = 0 ; index < _sockets.size() ; ++index)
00476         {
00477             // write whole block
00478             _sockets[index].send(&_socketWriteBuffer[0],
00479                                  size+sizeof(SocketBufferHeader));
00480         }
00481     }
00482 }

00319 {
00320     UInt32 len;
00321     UInt32 index;
00322     UInt32 tag=314156;
00323     UInt32 missing = _sockets.size();
00324     SocketSelection selection,result;
00325 
00326     for(index = 0 ; index < _sockets.size() ; ++index)
00327         selection.setRead(_sockets[index]);
00328 
00329     try
00330     {
00331         while(missing)
00332         {
00333             if(!selection.select(timeout,result))
00334                 return false;
00335             for(index = 0 ; index < _sockets.size() ; ++index)
00336             {
00337                 if(result.isSetRead(_sockets[index]))
00338                 {
00339                     len = _sockets[index].recv(&tag,sizeof(tag));
00340                     tag = osgNetToHost<UInt32>(tag);
00341                     if(len == 0)
00342                         throw ReadError("Channel closed");
00343                     selection.clearRead(_sockets[index]);
00344                     missing--;
00345                     if(tag != 314156)
00346                     {
00347                         FFATAL(("Stream out of sync in SockConnection\n"));
00348                         throw ReadError("Stream out of sync");
00349                     }
00350                 }
00351             }
00352         }
00353     }
00354     catch(SocketException &e)
00355     {
00356         throw ReadError(e.what());
00357     }
00358     return true;
00359 }