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