1 /************************************************************** 2 * 3 * Licensed to the Apache Software Foundation (ASF) under one 4 * or more contributor license agreements. See the NOTICE file 5 * distributed with this work for additional information 6 * regarding copyright ownership. The ASF licenses this file 7 * to you under the Apache License, Version 2.0 (the 8 * "License"); you may not use this file except in compliance 9 * with the License. You may obtain a copy of the License at 10 * 11 * http://www.apache.org/licenses/LICENSE-2.0 12 * 13 * Unless required by applicable law or agreed to in writing, 14 * software distributed under the License is distributed on an 15 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 16 * KIND, either express or implied. See the License for the 17 * specific language governing permissions and limitations 18 * under the License. 19 * 20 *************************************************************/ 21 22 23 24 // MARKER(update_precomp.py): autogen include statement, do not remove 25 #include "precompiled_io.hxx" 26 #include "acceptor.hxx" 27 28 #include <hash_set> 29 #include <algorithm> 30 31 #include <rtl/ustrbuf.hxx> 32 #include <com/sun/star/connection/XConnectionBroadcaster.hpp> 33 #include <com/sun/star/connection/ConnectionSetupException.hpp> 34 35 #include <cppuhelper/implbase2.hxx> 36 37 using namespace ::osl; 38 using namespace ::rtl; 39 using namespace ::cppu; 40 using namespace ::com::sun::star::uno; 41 using namespace ::com::sun::star::io; 42 using namespace ::com::sun::star::connection; 43 44 45 namespace io_acceptor { 46 template<class T> 47 struct ReferenceHash 48 { operator ()io_acceptor::ReferenceHash49 size_t operator () (const ::com::sun::star::uno::Reference<T> & ref) const 50 { 51 return (size_t)ref.get(); 52 } 53 }; 54 55 template<class T> 56 struct ReferenceEqual 57 { operator ()io_acceptor::ReferenceEqual58 sal_Bool operator () (const ::com::sun::star::uno::Reference<T> & op1, 59 const ::com::sun::star::uno::Reference<T> & op2) const 60 { 61 return op1.get() == op2.get(); 62 } 63 }; 64 65 66 typedef ::std::hash_set< ::com::sun::star::uno::Reference< ::com::sun::star::io::XStreamListener>, 67 ReferenceHash< ::com::sun::star::io::XStreamListener>, 68 ReferenceEqual< ::com::sun::star::io::XStreamListener> > 69 XStreamListener_hash_set; 70 71 72 class SocketConnection : public ::cppu::WeakImplHelper2< 73 ::com::sun::star::connection::XConnection, 74 ::com::sun::star::connection::XConnectionBroadcaster> 75 76 { 77 public: 78 SocketConnection( const OUString & sConnectionDescription ); 79 ~SocketConnection(); 80 81 virtual sal_Int32 SAL_CALL read( ::com::sun::star::uno::Sequence< sal_Int8 >& aReadBytes, 82 sal_Int32 nBytesToRead ); 83 virtual void SAL_CALL write( const ::com::sun::star::uno::Sequence< sal_Int8 >& aData ); 84 virtual void SAL_CALL flush( ); 85 virtual void SAL_CALL close( ); 86 virtual ::rtl::OUString SAL_CALL getDescription( ); 87 88 // XConnectionBroadcaster 89 virtual void SAL_CALL addStreamListener(const ::com::sun::star::uno::Reference< ::com::sun::star::io::XStreamListener>& aListener); 90 virtual void SAL_CALL removeStreamListener(const ::com::sun::star::uno::Reference< ::com::sun::star::io::XStreamListener>& aListener); 91 92 public: 93 void completeConnectionString(); 94 95 ::osl::StreamSocket m_socket; 96 ::osl::SocketAddr m_addr; 97 oslInterlockedCount m_nStatus; 98 ::rtl::OUString m_sDescription; 99 100 ::osl::Mutex _mutex; 101 sal_Bool _started; 102 sal_Bool _closed; 103 sal_Bool _error; 104 XStreamListener_hash_set _listeners; 105 }; 106 107 template<class T> notifyListeners(SocketConnection * pCon,sal_Bool * notified,T t)108 void notifyListeners(SocketConnection * pCon, sal_Bool * notified, T t) 109 { 110 XStreamListener_hash_set listeners; 111 112 { 113 ::osl::MutexGuard guard(pCon->_mutex); 114 if(!*notified) 115 { 116 *notified = sal_True; 117 listeners = pCon->_listeners; 118 } 119 } 120 121 ::std::for_each(listeners.begin(), listeners.end(), t); 122 } 123 callStarted(Reference<XStreamListener> xStreamListener)124 static void callStarted(Reference<XStreamListener> xStreamListener) 125 { 126 xStreamListener->started(); 127 } 128 129 struct callError { 130 const Any & any; 131 132 callError(const Any & any); 133 134 void operator () (Reference<XStreamListener> xStreamListener); 135 }; 136 callError(const Any & aAny)137 callError::callError(const Any & aAny) 138 : any(aAny) 139 { 140 } 141 operator ()(Reference<XStreamListener> xStreamListener)142 void callError::operator () (Reference<XStreamListener> xStreamListener) 143 { 144 xStreamListener->error(any); 145 } 146 callClosed(Reference<XStreamListener> xStreamListener)147 static void callClosed(Reference<XStreamListener> xStreamListener) 148 { 149 xStreamListener->closed(); 150 } 151 152 SocketConnection(const OUString & sConnectionDescription)153 SocketConnection::SocketConnection( const OUString &sConnectionDescription) : 154 m_nStatus( 0 ), 155 m_sDescription( sConnectionDescription ), 156 _started(sal_False), 157 _closed(sal_False), 158 _error(sal_False) 159 { 160 g_moduleCount.modCnt.acquire( &g_moduleCount.modCnt ); 161 // make it unique 162 m_sDescription += OUString( RTL_CONSTASCII_USTRINGPARAM( ",uniqueValue=" ) ); 163 m_sDescription += OUString::valueOf( 164 sal::static_int_cast< sal_Int64 >( 165 reinterpret_cast< sal_IntPtr >(&m_socket)), 166 10 ); 167 } 168 ~SocketConnection()169 SocketConnection::~SocketConnection() 170 { 171 g_moduleCount.modCnt.release( &g_moduleCount.modCnt ); 172 } 173 completeConnectionString()174 void SocketConnection::completeConnectionString() 175 { 176 OUStringBuffer buf( 256 ); 177 buf.appendAscii( ",peerPort=" ); 178 buf.append( (sal_Int32) m_socket.getPeerPort() ); 179 buf.appendAscii( ",peerHost=" ); 180 buf.append( m_socket.getPeerHost( ) ); 181 182 buf.appendAscii( ",localPort=" ); 183 buf.append( (sal_Int32) m_socket.getLocalPort() ); 184 buf.appendAscii( ",localHost=" ); 185 buf.append( m_socket.getLocalHost() ); 186 187 m_sDescription += buf.makeStringAndClear(); 188 } 189 read(Sequence<sal_Int8> & aReadBytes,sal_Int32 nBytesToRead)190 sal_Int32 SocketConnection::read( Sequence < sal_Int8 > & aReadBytes , sal_Int32 nBytesToRead ) 191 { 192 if( ! m_nStatus ) 193 { 194 notifyListeners(this, &_started, callStarted); 195 196 if( aReadBytes.getLength() != nBytesToRead ) 197 { 198 aReadBytes.realloc( nBytesToRead ); 199 } 200 201 sal_Int32 i = 0; 202 i = m_socket.read( aReadBytes.getArray() , aReadBytes.getLength() ); 203 204 if(i != nBytesToRead) 205 { 206 OUString message(RTL_CONSTASCII_USTRINGPARAM("acc_socket.cxx:SocketConnection::read: error - ")); 207 message += m_socket.getErrorAsString(); 208 209 IOException ioException(message, Reference<XInterface>(static_cast<XConnection *>(this))); 210 211 Any any; 212 any <<= ioException; 213 214 notifyListeners(this, &_error, callError(any)); 215 216 throw ioException; 217 } 218 219 return i; 220 } 221 else 222 { 223 OUString message(RTL_CONSTASCII_USTRINGPARAM("acc_socket.cxx:SocketConnection::read: error - connection already closed")); 224 225 IOException ioException(message, Reference<XInterface>(static_cast<XConnection *>(this))); 226 227 Any any; 228 any <<= ioException; 229 230 notifyListeners(this, &_error, callError(any)); 231 232 throw ioException; 233 } 234 } 235 write(const Sequence<sal_Int8> & seq)236 void SocketConnection::write( const Sequence < sal_Int8 > &seq ) 237 { 238 if( ! m_nStatus ) 239 { 240 if( m_socket.write( seq.getConstArray() , seq.getLength() ) != seq.getLength() ) 241 { 242 OUString message(RTL_CONSTASCII_USTRINGPARAM("acc_socket.cxx:SocketConnection::write: error - ")); 243 message += m_socket.getErrorAsString(); 244 245 IOException ioException(message, Reference<XInterface>(static_cast<XConnection *>(this))); 246 247 Any any; 248 any <<= ioException; 249 250 notifyListeners(this, &_error, callError(any)); 251 252 throw ioException; 253 } 254 } 255 else 256 { 257 OUString message(RTL_CONSTASCII_USTRINGPARAM("acc_socket.cxx:SocketConnection::write: error - connection already closed")); 258 259 IOException ioException(message, Reference<XInterface>(static_cast<XConnection *>(this))); 260 261 Any any; 262 any <<= ioException; 263 264 notifyListeners(this, &_error, callError(any)); 265 266 throw ioException; 267 } 268 } 269 flush()270 void SocketConnection::flush( ) 271 { 272 273 } 274 close()275 void SocketConnection::close() 276 { 277 // ensure close is called only once 278 if( 1 == osl_incrementInterlockedCount( (&m_nStatus) ) ) 279 { 280 m_socket.shutdown(); 281 notifyListeners(this, &_closed, callClosed); 282 } 283 } 284 getDescription()285 OUString SocketConnection::getDescription() 286 { 287 return m_sDescription; 288 } 289 290 291 // XConnectionBroadcaster addStreamListener(const Reference<XStreamListener> & aListener)292 void SAL_CALL SocketConnection::addStreamListener(const Reference<XStreamListener> & aListener) 293 { 294 MutexGuard guard(_mutex); 295 296 _listeners.insert(aListener); 297 } 298 removeStreamListener(const Reference<XStreamListener> & aListener)299 void SAL_CALL SocketConnection::removeStreamListener(const Reference<XStreamListener> & aListener) 300 { 301 MutexGuard guard(_mutex); 302 303 _listeners.erase(aListener); 304 } 305 SocketAcceptor(const OUString & sSocketName,sal_uInt16 nPort,sal_Bool bTcpNoDelay,const OUString & sConnectionDescription)306 SocketAcceptor::SocketAcceptor( const OUString &sSocketName, 307 sal_uInt16 nPort, 308 sal_Bool bTcpNoDelay, 309 const OUString &sConnectionDescription) : 310 m_sSocketName( sSocketName ), 311 m_sConnectionDescription( sConnectionDescription ), 312 m_nPort( nPort ), 313 m_bTcpNoDelay( bTcpNoDelay ), 314 m_bClosed( sal_False ) 315 { 316 } 317 318 init()319 void SocketAcceptor::init() 320 { 321 if( ! m_addr.setPort( m_nPort ) ) 322 { 323 OUStringBuffer message( 128 ); 324 message.appendAscii( "acc_socket.cxx:SocketAcceptor::init - error - invalid tcp/ip port " ); 325 message.append( (sal_Int32) m_nPort ); 326 throw ConnectionSetupException( 327 message.makeStringAndClear() , Reference< XInterface> () ); 328 } 329 if( ! m_addr.setHostname( m_sSocketName.pData ) ) 330 { 331 OUStringBuffer message( 128 ); 332 message.appendAscii( "acc_socket.cxx:SocketAcceptor::init - error - invalid host " ); 333 message.append( m_sSocketName ); 334 throw ConnectionSetupException( 335 message.makeStringAndClear(), Reference< XInterface > () ); 336 } 337 m_socket.setOption( osl_Socket_OptionReuseAddr, 1); 338 339 if(! m_socket.bind(m_addr) ) 340 { 341 OUStringBuffer message( 128 ); 342 message.appendAscii( "acc_socket.cxx:SocketAcceptor::init - error - couldn't bind on " ); 343 message.append( m_sSocketName ).appendAscii( ":" ).append((sal_Int32)m_nPort); 344 throw ConnectionSetupException( 345 message.makeStringAndClear(), 346 Reference<XInterface>()); 347 } 348 349 if(! m_socket.listen() ) 350 { 351 OUStringBuffer message( 128 ); 352 message.appendAscii( "acc_socket.cxx:SocketAcceptor::init - error - can't listen on " ); 353 message.append( m_sSocketName ).appendAscii( ":" ).append( (sal_Int32) m_nPort); 354 throw ConnectionSetupException( message.makeStringAndClear(),Reference<XInterface>() ); 355 } 356 } 357 accept()358 Reference< XConnection > SocketAcceptor::accept( ) 359 { 360 SocketConnection *pConn = new SocketConnection( m_sConnectionDescription ); 361 362 if( m_socket.acceptConnection( pConn->m_socket )!= osl_Socket_Ok ) 363 { 364 // stopAccepting was called 365 delete pConn; 366 return Reference < XConnection > (); 367 } 368 if( m_bClosed ) 369 { 370 delete pConn; 371 return Reference < XConnection > (); 372 } 373 374 pConn->completeConnectionString(); 375 if( m_bTcpNoDelay ) 376 { 377 sal_Int32 nTcpNoDelay = sal_True; 378 pConn->m_socket.setOption( osl_Socket_OptionTcpNoDelay , &nTcpNoDelay, 379 sizeof( nTcpNoDelay ) , osl_Socket_LevelTcp ); 380 } 381 382 return Reference < XConnection > ( (XConnection * ) pConn ); 383 } 384 stopAccepting()385 void SocketAcceptor::stopAccepting() 386 { 387 m_bClosed = sal_True; 388 m_socket.close(); 389 } 390 } 391