io/source/acceptor/acc_socket.cxx

   1 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
   2 /*
   3  * This file is part of the LibreOffice project.
   4  *
   5  * This Source Code Form is subject to the terms of the Mozilla Public
   6  * License, v. 2.0. If a copy of the MPL was not distributed with this
   7  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
   8  *
   9  * This file incorporates work covered by the following license notice:
  10  *
  11  *   Licensed to the Apache Software Foundation (ASF) under one or more
  12  *   contributor license agreements. See the NOTICE file distributed
  13  *   with this work for additional information regarding copyright
  14  *   ownership. The ASF licenses this file to you under the Apache
  15  *   License, Version 2.0 (the "License"); you may not use this file
  16  *   except in compliance with the License. You may obtain a copy of
  17  *   the License at http://www.apache.org/licenses/LICENSE-2.0 .
  18  */
  19
  20 #include "acceptor.hxx"
  21
  22 #include <unordered_set>
  23
  24 #include <osl/mutex.hxx>
  25 #include <rtl/ustrbuf.hxx>
  26 #include <com/sun/star/connection/XConnection.hpp>
  27 #include <com/sun/star/connection/XConnectionBroadcaster.hpp>
  28 #include <com/sun/star/connection/ConnectionSetupException.hpp>
  29 #include <com/sun/star/io/IOException.hpp>
  30 #include <cppuhelper/implbase.hxx>
  31
  32 using namespace ::osl;
  33 using namespace ::cppu;
  34 using namespace ::com::sun::star::uno;
  35 using namespace ::com::sun::star::io;
  36 using namespace ::com::sun::star::connection;
  37
  38
  39 namespace io_acceptor {
  40
  41     typedef std::unordered_set< css::uno::Reference< css::io::XStreamListener> >
  42             XStreamListener_hash_set;
  43
  44
  45     class SocketConnection : public ::cppu::WeakImplHelper<
  46         css::connection::XConnection,
  47         css::connection::XConnectionBroadcaster>
  48
  49     {
  50     public:
  51         explicit SocketConnection( const OUString & sConnectionDescription );
  52
  53         virtual sal_Int32 SAL_CALL read( css::uno::Sequence< sal_Int8 >& aReadBytes,
  54                                          sal_Int32 nBytesToRead ) override;
  55         virtual void SAL_CALL write( const css::uno::Sequence< sal_Int8 >& aData ) override;
  56         virtual void SAL_CALL flush(  ) override;
  57         virtual void SAL_CALL close(  ) override;
  58         virtual OUString SAL_CALL getDescription(  ) override;
  59
  60         // XConnectionBroadcaster
  61         virtual void SAL_CALL addStreamListener(const css::uno::Reference< css::io::XStreamListener>& aListener) override;
  62         virtual void SAL_CALL removeStreamListener(const css::uno::Reference< css::io::XStreamListener>& aListener) override;
  63
  64     public:
  65         void completeConnectionString();
  66
  67         ::osl::StreamSocket m_socket;
  68         oslInterlockedCount m_nStatus;
  69         OUString m_sDescription;
  70
  71         ::osl::Mutex _mutex;
  72         bool     _started;
  73         bool     _closed;
  74         bool     _error;
  75         XStreamListener_hash_set _listeners;
  76     };
  77
  78     template<class T>
  79     static void notifyListeners(SocketConnection * pCon, bool * notified, T t)
  80     {
  81         XStreamListener_hash_set listeners;
  82
  83         {
  84             ::osl::MutexGuard guard(pCon->_mutex);
  85             if(!*notified)
  86             {
  87                 *notified = true;
  88                 listeners = pCon->_listeners;
  89             }
  90         }
  91
  92         for(auto& listener : listeners)
  93             t(listener);
  94     }
  95
  96     static void callStarted(const Reference<XStreamListener>& xStreamListener)
  97     {
  98         xStreamListener->started();
  99     }
 100
 101     struct callError {
 102         const Any & any;
 103
 104         explicit callError(const Any & any);
 105
 106         void operator () (const Reference<XStreamListener>& xStreamListener);
 107     };
 108
 109     callError::callError(const Any & aAny)
 110         : any(aAny)
 111     {
 112     }
 113
 114     void callError::operator () (const Reference<XStreamListener>& xStreamListener)
 115     {
 116         xStreamListener->error(any);
 117     }
 118
 119     static void callClosed(const Reference<XStreamListener>& xStreamListener)
 120     {
 121         xStreamListener->closed();
 122     }
 123
 124
 125     SocketConnection::SocketConnection( const OUString &sConnectionDescription) :
 126         m_nStatus( 0 ),
 127         m_sDescription( sConnectionDescription ),
 128         _started(false),
 129         _closed(false),
 130         _error(false)
 131     {
 132         // make it unique
 133         m_sDescription += ",uniqueValue=" ;
 134         m_sDescription += OUString::number(
 135             sal::static_int_cast< sal_Int64 >(
 136                 reinterpret_cast< sal_IntPtr >(&m_socket)) );
 137     }
 138
 139     void SocketConnection::completeConnectionString()
 140     {
 141         OUStringBuffer buf( 256 );
 142         buf.append( ",peerPort=" );
 143         buf.append( m_socket.getPeerPort() );
 144         buf.append( ",peerHost=" );
 145         buf.append( m_socket.getPeerHost( ) );
 146
 147         buf.append( ",localPort=" );
 148         buf.append( m_socket.getLocalPort() );
 149         buf.append( ",localHost=" );
 150         buf.append( m_socket.getLocalHost() );
 151
 152         m_sDescription += buf;
 153     }
 154
 155     sal_Int32 SocketConnection::read( Sequence < sal_Int8 > & aReadBytes , sal_Int32 nBytesToRead )
 156     {
 157         if( ! m_nStatus )
 158         {
 159             notifyListeners(this, &_started, callStarted);
 160
 161             if( aReadBytes.getLength() != nBytesToRead )
 162             {
 163                 aReadBytes.realloc( nBytesToRead );
 164             }
 165
 166             sal_Int32 i = m_socket.read(
 167                 aReadBytes.getArray(), aReadBytes.getLength());
 168
 169             if(i != nBytesToRead)
 170             {
 171                 OUString message = "acc_socket.cxx:SocketConnection::read: error - " +
 172                     m_socket.getErrorAsString();
 173
 174                 IOException ioException(message, static_cast<XConnection *>(this));
 175
 176                 Any any;
 177                 any <<= ioException;
 178
 179                 notifyListeners(this, &_error, callError(any));
 180
 181                 throw ioException;
 182             }
 183
 184             return i;
 185         }
 186         else
 187         {
 188             IOException ioException("acc_socket.cxx:SocketConnection::read: error - connection already closed", static_cast<XConnection *>(this));
 189
 190             Any any;
 191             any <<= ioException;
 192
 193             notifyListeners(this, &_error, callError(any));
 194
 195             throw ioException;
 196         }
 197     }
 198
 199     void SocketConnection::write( const Sequence < sal_Int8 > &seq )
 200     {
 201         if( ! m_nStatus )
 202         {
 203             if( m_socket.write( seq.getConstArray() , seq.getLength() ) != seq.getLength() )
 204             {
 205                 OUString message = "acc_socket.cxx:SocketConnection::write: error - " +
 206                     m_socket.getErrorAsString();
 207
 208                 IOException ioException(message, static_cast<XConnection *>(this));
 209
 210                 Any any;
 211                 any <<= ioException;
 212
 213                 notifyListeners(this, &_error, callError(any));
 214
 215                 throw ioException;
 216             }
 217         }
 218         else
 219         {
 220             IOException ioException("acc_socket.cxx:SocketConnection::write: error - connection already closed", static_cast<XConnection *>(this));
 221
 222             Any any;
 223             any <<= ioException;
 224
 225             notifyListeners(this, &_error, callError(any));
 226
 227             throw ioException;
 228         }
 229     }
 230
 231     void SocketConnection::flush( )
 232     {
 233
 234     }
 235
 236     void SocketConnection::close()
 237     {
 238         // ensure close is called only once
 239         if(  1 == osl_atomic_increment( (&m_nStatus) ) )
 240         {
 241             m_socket.shutdown();
 242             notifyListeners(this, &_closed, callClosed);
 243         }
 244     }
 245
 246     OUString SocketConnection::getDescription()
 247     {
 248         return m_sDescription;
 249     }
 250
 251
 252     // XConnectionBroadcaster
 253     void SAL_CALL SocketConnection::addStreamListener(const Reference<XStreamListener> & aListener)
 254     {
 255         MutexGuard guard(_mutex);
 256
 257         _listeners.insert(aListener);
 258     }
 259
 260     void SAL_CALL SocketConnection::removeStreamListener(const Reference<XStreamListener> & aListener)
 261     {
 262         MutexGuard guard(_mutex);
 263
 264         _listeners.erase(aListener);
 265     }
 266
 267     SocketAcceptor::SocketAcceptor( const OUString &sSocketName,
 268                                     sal_uInt16 nPort,
 269                                     bool bTcpNoDelay,
 270                                     const OUString &sConnectionDescription) :
 271         m_sSocketName( sSocketName ),
 272         m_sConnectionDescription( sConnectionDescription ),
 273         m_nPort( nPort ),
 274         m_bTcpNoDelay( bTcpNoDelay ),
 275         m_bClosed( false )
 276     {
 277     }
 278
 279
 280     void SocketAcceptor::init()
 281     {
 282         if( ! m_addr.setPort( m_nPort ) )
 283         {
 284             throw ConnectionSetupException(
 285                 "acc_socket.cxx:SocketAcceptor::init - error - invalid tcp/ip port " +
 286                 OUString::number( m_nPort ));
 287         }
 288         if( ! m_addr.setHostname( m_sSocketName.pData ) )
 289         {
 290             throw ConnectionSetupException(
 291                 "acc_socket.cxx:SocketAcceptor::init - error - invalid host " + m_sSocketName );
 292         }
 293         m_socket.setOption( osl_Socket_OptionReuseAddr, 1);
 294
 295         if(! m_socket.bind(m_addr) )
 296         {
 297             throw ConnectionSetupException(
 298                 "acc_socket.cxx:SocketAcceptor::init - error - couldn't bind on " +
 299                 m_sSocketName + ":" + OUString::number(m_nPort));
 300         }
 301
 302         if(! m_socket.listen() )
 303         {
 304             throw ConnectionSetupException(
 305                 "acc_socket.cxx:SocketAcceptor::init - error - can't listen on " +
 306                 m_sSocketName  + ":" + OUString::number(m_nPort) );
 307         }
 308     }
 309
 310     Reference< XConnection > SocketAcceptor::accept( )
 311     {
 312         std::unique_ptr<SocketConnection> pConn(new SocketConnection( m_sConnectionDescription ));
 313
 314         if( m_socket.acceptConnection( pConn->m_socket )!= osl_Socket_Ok )
 315         {
 316             // stopAccepting was called
 317             return Reference < XConnection > ();
 318         }
 319         if( m_bClosed )
 320         {
 321             return Reference < XConnection > ();
 322         }
 323
 324         pConn->completeConnectionString();
 325         ::osl::SocketAddr remoteAddr;
 326         pConn->m_socket.getPeerAddr(remoteAddr);
 327         OUString remoteHostname = remoteAddr.getHostname();
 328         // we enable tcpNoDelay for loopback connections because
 329         // it can make a significant speed difference on linux boxes.
 330         if( m_bTcpNoDelay || remoteHostname == "localhost" ||
 331             remoteHostname.startsWith("127.0.0.") )
 332         {
 333             sal_Int32 nTcpNoDelay = sal_Int32(true);
 334             pConn->m_socket.setOption( osl_Socket_OptionTcpNoDelay , &nTcpNoDelay,
 335                                        sizeof( nTcpNoDelay ) , osl_Socket_LevelTcp );
 336         }
 337
 338         return Reference < XConnection > ( static_cast<XConnection *>(pConn.release()) );
 339     }
 340
 341     void SocketAcceptor::stopAccepting()
 342     {
 343         m_bClosed = true;
 344         m_socket.close();
 345     }
 346 }
 347
 348
 349 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */