io/source/acceptor/acc_socket.cxx

   1 /*************************************************************************
   2  *
   3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   4  *
   5  * Copyright 2008 by Sun Microsystems, Inc.
   6  *
   7  * OpenOffice.org - a multi-platform office productivity suite
   8  *
   9  * $RCSfile: acc_socket.cxx,v $
  10  * $Revision: 1.15 $
  11  *
  12  * This file is part of OpenOffice.org.
  13  *
  14  * OpenOffice.org is free software: you can redistribute it and/or modify
  15  * it under the terms of the GNU Lesser General Public License version 3
  16  * only, as published by the Free Software Foundation.
  17  *
  18  * OpenOffice.org is distributed in the hope that it will be useful,
  19  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  20  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  21  * GNU Lesser General Public License version 3 for more details
  22  * (a copy is included in the LICENSE file that accompanied this code).
  23  *
  24  * You should have received a copy of the GNU Lesser General Public License
  25  * version 3 along with OpenOffice.org.  If not, see
  26  * <http://www.openoffice.org/license.html>
  27  * for a copy of the LGPLv3 License.
  28  *
  29  ************************************************************************/
  30
  31 // MARKER(update_precomp.py): autogen include statement, do not remove
  32 #include "precompiled_io.hxx"
  33 #include "acceptor.hxx"
  34
  35 #include <hash_set>
  36 #include <algorithm>
  37
  38 #include <rtl/ustrbuf.hxx>
  39 #include <com/sun/star/connection/XConnectionBroadcaster.hpp>
  40 #include <com/sun/star/connection/ConnectionSetupException.hpp>
  41
  42 #include <cppuhelper/implbase2.hxx>
  43
  44 using namespace ::osl;
  45 using namespace ::rtl;
  46 using namespace ::cppu;
  47 using namespace ::com::sun::star::uno;
  48 using namespace ::com::sun::star::io;
  49 using namespace ::com::sun::star::connection;
  50
  51
  52 namespace io_acceptor {
  53     template<class T>
  54     struct ReferenceHash
  55     {
  56         size_t operator () (const ::com::sun::star::uno::Reference<T> & ref) const
  57         {
  58             return (size_t)ref.get();
  59         }
  60     };
  61
  62     template<class T>
  63     struct ReferenceEqual
  64     {
  65         sal_Bool operator () (const ::com::sun::star::uno::Reference<T> & op1,
  66                               const ::com::sun::star::uno::Reference<T> & op2) const
  67         {
  68             return op1.get() == op2.get();
  69         }
  70     };
  71
  72
  73     typedef ::std::hash_set< ::com::sun::star::uno::Reference< ::com::sun::star::io::XStreamListener>,
  74                              ReferenceHash< ::com::sun::star::io::XStreamListener>,
  75                              ReferenceEqual< ::com::sun::star::io::XStreamListener> >
  76             XStreamListener_hash_set;
  77
  78
  79     class SocketConnection : public ::cppu::WeakImplHelper2<
  80         ::com::sun::star::connection::XConnection,
  81         ::com::sun::star::connection::XConnectionBroadcaster>
  82
  83     {
  84     public:
  85         SocketConnection( const OUString & sConnectionDescription );
  86         ~SocketConnection();
  87
  88         virtual sal_Int32 SAL_CALL read( ::com::sun::star::uno::Sequence< sal_Int8 >& aReadBytes,
  89                                          sal_Int32 nBytesToRead )
  90             throw(::com::sun::star::io::IOException,
  91                   ::com::sun::star::uno::RuntimeException);
  92         virtual void SAL_CALL write( const ::com::sun::star::uno::Sequence< sal_Int8 >& aData )
  93             throw(::com::sun::star::io::IOException,
  94                   ::com::sun::star::uno::RuntimeException);
  95         virtual void SAL_CALL flush(  ) throw(
  96             ::com::sun::star::io::IOException,
  97             ::com::sun::star::uno::RuntimeException);
  98         virtual void SAL_CALL close(  )
  99             throw(::com::sun::star::io::IOException,
 100                   ::com::sun::star::uno::RuntimeException);
 101         virtual ::rtl::OUString SAL_CALL getDescription(  )
 102             throw(::com::sun::star::uno::RuntimeException);
 103
 104         // XConnectionBroadcaster
 105         virtual void SAL_CALL addStreamListener(const ::com::sun::star::uno::Reference< ::com::sun::star::io::XStreamListener>& aListener)
 106             throw(::com::sun::star::uno::RuntimeException);
 107         virtual void SAL_CALL removeStreamListener(const ::com::sun::star::uno::Reference< ::com::sun::star::io::XStreamListener>& aListener)
 108             throw(::com::sun::star::uno::RuntimeException);
 109
 110     public:
 111         void completeConnectionString();
 112
 113         ::osl::StreamSocket m_socket;
 114         ::osl::SocketAddr m_addr;
 115         oslInterlockedCount m_nStatus;
 116         ::rtl::OUString m_sDescription;
 117
 118         ::osl::Mutex _mutex;
 119         sal_Bool     _started;
 120         sal_Bool     _closed;
 121         sal_Bool     _error;
 122         XStreamListener_hash_set _listeners;
 123     };
 124
 125     template<class T>
 126     void notifyListeners(SocketConnection * pCon, sal_Bool * notified, T t)
 127     {
 128           XStreamListener_hash_set listeners;
 129
 130         {
 131             ::osl::MutexGuard guard(pCon->_mutex);
 132             if(!*notified)
 133             {
 134                 *notified = sal_True;
 135                 listeners = pCon->_listeners;
 136             }
 137         }
 138
 139         ::std::for_each(listeners.begin(), listeners.end(), t);
 140     }
 141
 142     static void callStarted(Reference<XStreamListener> xStreamListener)
 143     {
 144         xStreamListener->started();
 145     }
 146
 147     struct callError {
 148         const Any & any;
 149
 150         callError(const Any & any);
 151
 152         void operator () (Reference<XStreamListener> xStreamListener);
 153     };
 154
 155     callError::callError(const Any & aAny)
 156         : any(aAny)
 157     {
 158     }
 159
 160     void callError::operator () (Reference<XStreamListener> xStreamListener)
 161     {
 162         xStreamListener->error(any);
 163     }
 164
 165     static void callClosed(Reference<XStreamListener> xStreamListener)
 166     {
 167         xStreamListener->closed();
 168     }
 169
 170
 171     SocketConnection::SocketConnection( const OUString &sConnectionDescription) :
 172         m_nStatus( 0 ),
 173         m_sDescription( sConnectionDescription ),
 174         _started(sal_False),
 175         _closed(sal_False),
 176         _error(sal_False)
 177     {
 178         g_moduleCount.modCnt.acquire( &g_moduleCount.modCnt );
 179         // make it unique
 180         m_sDescription += OUString( RTL_CONSTASCII_USTRINGPARAM( ",uniqueValue=" ) );
 181         m_sDescription += OUString::valueOf(
 182             sal::static_int_cast< sal_Int64 >(
 183                 reinterpret_cast< sal_IntPtr >(&m_socket)),
 184             10 );
 185     }
 186
 187     SocketConnection::~SocketConnection()
 188     {
 189         g_moduleCount.modCnt.release( &g_moduleCount.modCnt );
 190     }
 191
 192     void SocketConnection::completeConnectionString()
 193     {
 194         OUStringBuffer buf( 256 );
 195         buf.appendAscii( ",peerPort=" );
 196         buf.append( (sal_Int32) m_socket.getPeerPort() );
 197         buf.appendAscii( ",peerHost=" );
 198         buf.append( m_socket.getPeerHost( ) );
 199
 200         buf.appendAscii( ",localPort=" );
 201         buf.append( (sal_Int32) m_socket.getLocalPort() );
 202         buf.appendAscii( ",localHost=" );
 203         buf.append( m_socket.getLocalHost() );
 204
 205         m_sDescription += buf.makeStringAndClear();
 206     }
 207
 208     sal_Int32 SocketConnection::read( Sequence < sal_Int8 > & aReadBytes , sal_Int32 nBytesToRead )
 209             throw(::com::sun::star::io::IOException,
 210                   ::com::sun::star::uno::RuntimeException)
 211     {
 212         if( ! m_nStatus )
 213         {
 214             notifyListeners(this, &_started, callStarted);
 215
 216             if( aReadBytes.getLength() != nBytesToRead )
 217             {
 218                 aReadBytes.realloc( nBytesToRead );
 219             }
 220
 221             sal_Int32 i = 0;
 222             i = m_socket.read( aReadBytes.getArray()  , aReadBytes.getLength() );
 223
 224             if(i != nBytesToRead)
 225             {
 226                 OUString message(RTL_CONSTASCII_USTRINGPARAM("acc_socket.cxx:SocketConnection::read: error - "));
 227                 message +=      m_socket.getErrorAsString();
 228
 229                 IOException ioException(message, Reference<XInterface>(static_cast<XConnection *>(this)));
 230
 231                 Any any;
 232                 any <<= ioException;
 233
 234                 notifyListeners(this, &_error, callError(any));
 235
 236                 throw ioException;
 237             }
 238
 239             return i;
 240         }
 241         else
 242         {
 243             OUString message(RTL_CONSTASCII_USTRINGPARAM("acc_socket.cxx:SocketConnection::read: error - connection already closed"));
 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
 256     void SocketConnection::write( const Sequence < sal_Int8 > &seq )
 257             throw(::com::sun::star::io::IOException,
 258                   ::com::sun::star::uno::RuntimeException)
 259     {
 260         if( ! m_nStatus )
 261         {
 262             if( m_socket.write( seq.getConstArray() , seq.getLength() ) != seq.getLength() )
 263             {
 264                 OUString message(RTL_CONSTASCII_USTRINGPARAM("acc_socket.cxx:SocketConnection::write: error - "));
 265                 message += m_socket.getErrorAsString();
 266
 267                 IOException ioException(message, Reference<XInterface>(static_cast<XConnection *>(this)));
 268
 269                 Any any;
 270                 any <<= ioException;
 271
 272                 notifyListeners(this, &_error, callError(any));
 273
 274                 throw ioException;
 275             }
 276         }
 277         else
 278         {
 279             OUString message(RTL_CONSTASCII_USTRINGPARAM("acc_socket.cxx:SocketConnection::write: error - connection already closed"));
 280
 281             IOException ioException(message, Reference<XInterface>(static_cast<XConnection *>(this)));
 282
 283             Any any;
 284             any <<= ioException;
 285
 286             notifyListeners(this, &_error, callError(any));
 287
 288             throw ioException;
 289         }
 290     }
 291
 292     void SocketConnection::flush( )
 293             throw(::com::sun::star::io::IOException,
 294                   ::com::sun::star::uno::RuntimeException)
 295     {
 296
 297     }
 298
 299     void SocketConnection::close()
 300             throw(::com::sun::star::io::IOException,
 301                   ::com::sun::star::uno::RuntimeException)
 302     {
 303         // enshure close is called only once
 304         if(  1 == osl_incrementInterlockedCount( (&m_nStatus) ) )
 305         {
 306             m_socket.shutdown();
 307             notifyListeners(this, &_closed, callClosed);
 308         }
 309     }
 310
 311     OUString SocketConnection::getDescription()
 312             throw( ::com::sun::star::uno::RuntimeException)
 313     {
 314         return m_sDescription;
 315     }
 316
 317
 318     // XConnectionBroadcaster
 319     void SAL_CALL SocketConnection::addStreamListener(const Reference<XStreamListener> & aListener) throw(RuntimeException)
 320     {
 321         MutexGuard guard(_mutex);
 322
 323         _listeners.insert(aListener);
 324     }
 325
 326     void SAL_CALL SocketConnection::removeStreamListener(const Reference<XStreamListener> & aListener) throw(RuntimeException)
 327     {
 328         MutexGuard guard(_mutex);
 329
 330         _listeners.erase(aListener);
 331     }
 332
 333     SocketAcceptor::SocketAcceptor( const OUString &sSocketName,
 334                                     sal_uInt16 nPort,
 335                                     sal_Bool bTcpNoDelay,
 336                                     const OUString &sConnectionDescription) :
 337         m_sSocketName( sSocketName ),
 338         m_sConnectionDescription( sConnectionDescription ),
 339         m_nPort( nPort ),
 340         m_bTcpNoDelay( bTcpNoDelay ),
 341         m_bClosed( sal_False )
 342     {
 343     }
 344
 345
 346     void SocketAcceptor::init()
 347     {
 348         if( ! m_addr.setPort( m_nPort ) )
 349         {
 350             OUStringBuffer message( 128 );
 351             message.appendAscii( "acc_socket.cxx:SocketAcceptor::init - error - invalid tcp/ip port " );
 352             message.append( (sal_Int32) m_nPort );
 353             throw ConnectionSetupException(
 354                 message.makeStringAndClear() , Reference< XInterface> () );
 355         }
 356         if( ! m_addr.setHostname( m_sSocketName.pData ) )
 357         {
 358             OUStringBuffer message( 128 );
 359             message.appendAscii( "acc_socket.cxx:SocketAcceptor::init - error - invalid host " );
 360             message.append( m_sSocketName );
 361             throw ConnectionSetupException(
 362                 message.makeStringAndClear(), Reference< XInterface > () );
 363         }
 364         m_socket.setOption( osl_Socket_OptionReuseAddr, 1);
 365
 366         if(! m_socket.bind(m_addr) )
 367         {
 368             OUStringBuffer message( 128 );
 369             message.appendAscii( "acc_socket.cxx:SocketAcceptor::init - error - couldn't bind on " );
 370             message.append( m_sSocketName ).appendAscii( ":" ).append((sal_Int32)m_nPort);
 371             throw ConnectionSetupException(
 372                 message.makeStringAndClear(),
 373                 Reference<XInterface>());
 374         }
 375
 376         if(! m_socket.listen() )
 377         {
 378             OUStringBuffer message( 128 );
 379             message.appendAscii( "acc_socket.cxx:SocketAcceptor::init - error - can't listen on " );
 380             message.append( m_sSocketName ).appendAscii( ":" ).append( (sal_Int32) m_nPort);
 381             throw ConnectionSetupException(     message.makeStringAndClear(),Reference<XInterface>() );
 382         }
 383     }
 384
 385     Reference< XConnection > SocketAcceptor::accept( )
 386     {
 387         SocketConnection *pConn = new SocketConnection( m_sConnectionDescription );
 388
 389         if( m_socket.acceptConnection( pConn->m_socket )!= osl_Socket_Ok )
 390         {
 391             // stopAccepting was called
 392             delete pConn;
 393             return Reference < XConnection > ();
 394         }
 395         if( m_bClosed )
 396         {
 397             delete pConn;
 398             return Reference < XConnection > ();
 399         }
 400
 401         pConn->completeConnectionString();
 402         if( m_bTcpNoDelay )
 403         {
 404             sal_Int32 nTcpNoDelay = sal_True;
 405             pConn->m_socket.setOption( osl_Socket_OptionTcpNoDelay , &nTcpNoDelay,
 406                                        sizeof( nTcpNoDelay ) , osl_Socket_LevelTcp );
 407         }
 408
 409         return Reference < XConnection > ( (XConnection * ) pConn );
 410     }
 411
 412     void SocketAcceptor::stopAccepting()
 413     {
 414         m_bClosed = sal_True;
 415         m_socket.close();
 416     }
 417 }
 418
 419