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