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