io/source/stm/opump.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
  21 #include <sal/log.hxx>
  22
  23 #include <com/sun/star/io/IOException.hpp>
  24 #include <com/sun/star/io/NotConnectedException.hpp>
  25 #include <com/sun/star/io/XActiveDataSource.hpp>
  26 #include <com/sun/star/io/XActiveDataSink.hpp>
  27 #include <com/sun/star/io/XActiveDataControl.hpp>
  28 #include <com/sun/star/io/XConnectable.hpp>
  29 #include <com/sun/star/lang/XServiceInfo.hpp>
  30 #include <com/sun/star/uno/XComponentContext.hpp>
  31
  32 #include <cppuhelper/implbase.hxx>
  33 #include <comphelper/interfacecontainer4.hxx>
  34 #include <cppuhelper/supportsservice.hxx>
  35 #include <osl/thread.h>
  36 #include <mutex>
  37
  38 using namespace cppu;
  39 using namespace com::sun::star::uno;
  40 using namespace com::sun::star::lang;
  41 using namespace com::sun::star::io;
  42
  43 namespace io_stm {
  44
  45     namespace {
  46
  47     class Pump : public WeakImplHelper<
  48           XActiveDataSource, XActiveDataSink, XActiveDataControl, XConnectable, XServiceInfo >
  49     {
  50         std::mutex                              m_aMutex;
  51         oslThread                               m_aThread;
  52
  53         Reference< XConnectable >               m_xPred;
  54         Reference< XConnectable >               m_xSucc;
  55         Reference< XInputStream >               m_xInput;
  56         Reference< XOutputStream >              m_xOutput;
  57         comphelper::OInterfaceContainerHelper4<XStreamListener>  m_cnt;
  58         bool                                m_closeFired;
  59
  60         void run();
  61         static void static_run( void* pObject );
  62
  63         void close();
  64         void fireClose();
  65         void fireStarted();
  66         void fireTerminated();
  67         void fireError( const Any &a );
  68
  69     public:
  70         Pump();
  71         virtual ~Pump() override;
  72
  73         // XActiveDataSource
  74         virtual void SAL_CALL setOutputStream( const Reference< css::io::XOutputStream >& xOutput ) override;
  75         virtual Reference< css::io::XOutputStream > SAL_CALL getOutputStream() override;
  76
  77         // XActiveDataSink
  78         virtual void SAL_CALL setInputStream( const Reference< css::io::XInputStream >& xStream ) override;
  79         virtual Reference< css::io::XInputStream > SAL_CALL getInputStream() override;
  80
  81         // XActiveDataControl
  82         virtual void SAL_CALL addListener( const Reference< css::io::XStreamListener >& xListener ) override;
  83         virtual void SAL_CALL removeListener( const Reference< css::io::XStreamListener >& xListener ) override;
  84         virtual void SAL_CALL start() override;
  85         virtual void SAL_CALL terminate() override;
  86
  87         // XConnectable
  88         virtual void SAL_CALL setPredecessor( const Reference< css::io::XConnectable >& xPred ) override;
  89         virtual Reference< css::io::XConnectable > SAL_CALL getPredecessor() override;
  90         virtual void SAL_CALL setSuccessor( const Reference< css::io::XConnectable >& xSucc ) override;
  91         virtual Reference< css::io::XConnectable > SAL_CALL getSuccessor() override;
  92
  93     public: // XServiceInfo
  94         virtual OUString    SAL_CALL getImplementationName() override;
  95         virtual Sequence< OUString > SAL_CALL getSupportedServiceNames() override;
  96         virtual sal_Bool     SAL_CALL supportsService(const OUString& ServiceName) override;
  97     };
  98
  99     }
 100
 101 Pump::Pump() : m_aThread( nullptr ),
 102                m_closeFired( false )
 103 {
 104 }
 105
 106 Pump::~Pump()
 107 {
 108     // exit gracefully
 109     if( m_aThread )
 110     {
 111         osl_joinWithThread( m_aThread );
 112         osl_destroyThread( m_aThread );
 113     }
 114 }
 115
 116 void Pump::fireError( const  Any & exception )
 117 {
 118     std::unique_lock guard( m_aMutex );
 119     comphelper::OInterfaceIteratorHelper4<XStreamListener> iter( guard, m_cnt );
 120     guard.unlock();
 121     while( iter.hasMoreElements() )
 122     {
 123         try
 124         {
 125             iter.next()->error( exception );
 126         }
 127         catch ( const RuntimeException &e )
 128         {
 129             SAL_WARN("io.streams","com.sun.star.comp.stoc.Pump: unexpected exception during calling listeners" << e);
 130         }
 131     }
 132 }
 133
 134 void Pump::fireClose()
 135 {
 136     bool bFire = false;
 137     {
 138         std::unique_lock guard( m_aMutex );
 139         if( ! m_closeFired  )
 140         {
 141             m_closeFired = true;
 142             bFire = true;
 143         }
 144     }
 145
 146     if( !bFire )
 147         return;
 148
 149     std::unique_lock guard( m_aMutex );
 150     comphelper::OInterfaceIteratorHelper4<XStreamListener> iter( guard, m_cnt );
 151     guard.unlock();
 152     while( iter.hasMoreElements() )
 153     {
 154         try
 155         {
 156             iter.next()->closed( );
 157         }
 158         catch ( const RuntimeException &e )
 159         {
 160             SAL_WARN("io.streams","com.sun.star.comp.stoc.Pump: unexpected exception during calling listeners" << e);
 161         }
 162     }
 163 }
 164
 165 void Pump::fireStarted()
 166 {
 167     std::unique_lock guard( m_aMutex );
 168     comphelper::OInterfaceIteratorHelper4<XStreamListener> iter( guard, m_cnt );
 169     guard.unlock();
 170     while( iter.hasMoreElements() )
 171     {
 172         try
 173         {
 174             iter.next()->started( );
 175         }
 176         catch ( const RuntimeException &e )
 177         {
 178             SAL_WARN("io.streams","com.sun.star.comp.stoc.Pump: unexpected exception during calling listeners" << e);
 179         }
 180     }
 181 }
 182
 183 void Pump::fireTerminated()
 184 {
 185     std::unique_lock guard( m_aMutex );
 186     comphelper::OInterfaceIteratorHelper4<XStreamListener> iter( guard, m_cnt );
 187     guard.unlock();
 188     while( iter.hasMoreElements() )
 189     {
 190         try
 191         {
 192             iter.next()->terminated();
 193         }
 194         catch ( const RuntimeException &e )
 195         {
 196             SAL_WARN("io.streams","com.sun.star.comp.stoc.Pump: unexpected exception during calling listeners" << e);
 197         }
 198     }
 199 }
 200
 201
 202 void Pump::close()
 203 {
 204     // close streams and release references
 205     Reference< XInputStream > rInput;
 206     Reference< XOutputStream > rOutput;
 207     {
 208         std::unique_lock guard( m_aMutex );
 209         rInput = m_xInput;
 210         m_xInput.clear();
 211
 212         rOutput = m_xOutput;
 213         m_xOutput.clear();
 214         m_xSucc.clear();
 215         m_xPred.clear();
 216     }
 217     if( rInput.is() )
 218     {
 219         try
 220         {
 221             rInput->closeInput();
 222         }
 223         catch( Exception & )
 224         {
 225             // go down calm
 226         }
 227     }
 228     if( rOutput.is() )
 229     {
 230         try
 231         {
 232             rOutput->closeOutput();
 233         }
 234         catch( Exception & )
 235         {
 236             // go down calm
 237         }
 238     }
 239 }
 240
 241 void Pump::static_run( void* pObject )
 242 {
 243     osl_setThreadName("io_stm::Pump::run()");
 244     static_cast<Pump*>(pObject)->run();
 245     static_cast<Pump*>(pObject)->release();
 246 }
 247
 248 void Pump::run()
 249 {
 250     try
 251     {
 252         fireStarted();
 253         try
 254         {
 255             Reference< XInputStream > rInput;
 256             Reference< XOutputStream > rOutput;
 257             {
 258                 std::unique_lock aGuard( m_aMutex );
 259                 rInput = m_xInput;
 260                 rOutput = m_xOutput;
 261             }
 262
 263             if( ! rInput.is() )
 264             {
 265                 throw NotConnectedException( u"no input stream set"_ustr, getXWeak() );
 266             }
 267             Sequence< sal_Int8 > aData;
 268             while( rInput->readSomeBytes( aData, 65536 ) )
 269             {
 270                 if( ! rOutput.is() )
 271                 {
 272                     throw NotConnectedException( u"no output stream set"_ustr, getXWeak() );
 273                 }
 274                 rOutput->writeBytes( aData );
 275                 osl_yieldThread();
 276             }
 277         }
 278         catch ( const IOException & e )
 279         {
 280             fireError( Any( e ) );
 281         }
 282         catch ( const RuntimeException & e )
 283         {
 284             fireError( Any( e ) );
 285         }
 286         catch ( const Exception & e )
 287         {
 288             fireError( Any( e ) );
 289         }
 290
 291         close();
 292         fireClose();
 293     }
 294     catch ( const css::uno::Exception &e )
 295     {
 296         // we are the last on the stack.
 297         // this is to avoid crashing the program, when e.g. a bridge crashes
 298         SAL_WARN("io.streams","com.sun.star.comp.stoc.Pump: unexpected exception during calling listeners" << e);
 299     }
 300 }
 301
 302
 303 /*
 304  * XConnectable
 305  */
 306
 307 void Pump::setPredecessor( const Reference< XConnectable >& xPred )
 308 {
 309     std::unique_lock aGuard( m_aMutex );
 310     m_xPred = xPred;
 311 }
 312
 313
 314 Reference< XConnectable > Pump::getPredecessor()
 315 {
 316     std::unique_lock aGuard( m_aMutex );
 317     return m_xPred;
 318 }
 319
 320
 321 void Pump::setSuccessor( const Reference< XConnectable >& xSucc )
 322 {
 323     std::unique_lock aGuard( m_aMutex );
 324     m_xSucc = xSucc;
 325 }
 326
 327
 328 Reference< XConnectable > Pump::getSuccessor()
 329 {
 330     std::unique_lock aGuard( m_aMutex );
 331     return m_xSucc;
 332 }
 333
 334
 335 /*
 336  * XActiveDataControl
 337  */
 338
 339 void Pump::addListener( const Reference< XStreamListener >& xListener )
 340 {
 341     std::unique_lock aGuard( m_aMutex );
 342     m_cnt.addInterface( aGuard, xListener );
 343 }
 344
 345
 346 void Pump::removeListener( const Reference< XStreamListener >& xListener )
 347 {
 348     std::unique_lock aGuard( m_aMutex );
 349     m_cnt.removeInterface( aGuard, xListener );
 350 }
 351
 352
 353 void Pump::start()
 354 {
 355     std::unique_lock aGuard( m_aMutex );
 356     m_aThread = osl_createSuspendedThread(Pump::static_run,this);
 357     if( !m_aThread )
 358     {
 359         throw RuntimeException(
 360             u"Pump::start Couldn't create worker thread"_ustr,
 361             *this);
 362     }
 363
 364     // will be released by OPump::static_run
 365     acquire();
 366     osl_resumeThread( m_aThread );
 367
 368 }
 369
 370
 371 void Pump::terminate()
 372 {
 373     close();
 374
 375     // wait for the worker to die
 376     if( m_aThread )
 377         osl_joinWithThread( m_aThread );
 378
 379     fireTerminated();
 380     fireClose();
 381 }
 382
 383
 384 /*
 385  * XActiveDataSink
 386  */
 387
 388 void Pump::setInputStream( const Reference< XInputStream >& xStream )
 389 {
 390     std::unique_lock aGuard( m_aMutex );
 391     m_xInput = xStream;
 392     Reference< XConnectable > xConnect( xStream, UNO_QUERY );
 393     if( xConnect.is() )
 394         xConnect->setSuccessor( this );
 395     // data transfer starts in XActiveDataControl::start
 396 }
 397
 398
 399 Reference< XInputStream > Pump::getInputStream()
 400 {
 401     std::unique_lock aGuard( m_aMutex );
 402     return m_xInput;
 403 }
 404
 405
 406 /*
 407  * XActiveDataSource
 408  */
 409
 410 void Pump::setOutputStream( const Reference< XOutputStream >& xOut )
 411 {
 412     std::unique_lock aGuard( m_aMutex );
 413     m_xOutput = xOut;
 414     Reference< XConnectable > xConnect( xOut, UNO_QUERY );
 415     if( xConnect.is() )
 416         xConnect->setPredecessor( this );
 417     // data transfer starts in XActiveDataControl::start
 418 }
 419
 420 Reference< XOutputStream > Pump::getOutputStream()
 421 {
 422     std::unique_lock aGuard( m_aMutex );
 423     return m_xOutput;
 424 }
 425
 426 // XServiceInfo
 427 OUString Pump::getImplementationName()
 428 {
 429     return u"com.sun.star.comp.io.Pump"_ustr;
 430 }
 431
 432 // XServiceInfo
 433 sal_Bool Pump::supportsService(const OUString& ServiceName)
 434 {
 435     return cppu::supportsService(this, ServiceName);
 436 }
 437
 438 // XServiceInfo
 439 Sequence< OUString > Pump::getSupportedServiceNames()
 440 {
 441     return { u"com.sun.star.io.Pump"_ustr };
 442 }
 443
 444 }
 445
 446 extern "C" SAL_DLLPUBLIC_EXPORT css::uno::XInterface*
 447 io_Pump_get_implementation(
 448     css::uno::XComponentContext* , css::uno::Sequence<css::uno::Any> const&)
 449 {
 450     return cppu::acquire(new io_stm::Pump());
 451 }
 452
 453
 454 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */