io/source/stm/opipe.cxx

   1 /*************************************************************************
   2  *
   3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   4  *
   5  * Copyright 2008 by Sun Microsystems, Inc.
   6  *
   7  * OpenOffice.org - a multi-platform office productivity suite
   8  *
   9  * $RCSfile: opipe.cxx,v $
  10  * $Revision: 1.9 $
  11  *
  12  * This file is part of OpenOffice.org.
  13  *
  14  * OpenOffice.org is free software: you can redistribute it and/or modify
  15  * it under the terms of the GNU Lesser General Public License version 3
  16  * only, as published by the Free Software Foundation.
  17  *
  18  * OpenOffice.org is distributed in the hope that it will be useful,
  19  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  20  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  21  * GNU Lesser General Public License version 3 for more details
  22  * (a copy is included in the LICENSE file that accompanied this code).
  23  *
  24  * You should have received a copy of the GNU Lesser General Public License
  25  * version 3 along with OpenOffice.org.  If not, see
  26  * <http://www.openoffice.org/license.html>
  27  * for a copy of the LGPLv3 License.
  28  *
  29  ************************************************************************/
  30
  31 // MARKER(update_precomp.py): autogen include statement, do not remove
  32 #include "precompiled_io.hxx"
  33
  34 // streams
  35 #include <com/sun/star/io/XInputStream.hpp>
  36 #include <com/sun/star/io/XOutputStream.hpp>
  37 #include <com/sun/star/io/XConnectable.hpp>
  38
  39 #include <com/sun/star/lang/XServiceInfo.hpp>
  40
  41 #include <cppuhelper/factory.hxx>
  42
  43 #include <cppuhelper/implbase4.hxx>      // OWeakObject
  44
  45 #include <osl/conditn.hxx>
  46 #include <osl/mutex.hxx>
  47
  48 #include <limits>
  49 #include <string.h>
  50
  51 using namespace ::rtl;
  52 using namespace ::osl;
  53 using namespace ::cppu;
  54 using namespace ::com::sun::star::uno;
  55 using namespace ::com::sun::star::io;
  56 using namespace ::com::sun::star::lang;
  57
  58 #include "factreg.hxx"
  59 #include "streamhelper.hxx"
  60
  61 // Implementation and service names
  62 #define IMPLEMENTATION_NAME     "com.sun.star.comp.io.stm.Pipe"
  63 #define SERVICE_NAME "com.sun.star.io.Pipe"
  64
  65 namespace io_stm{
  66
  67 class OPipeImpl :
  68     public WeakImplHelper4< XInputStream , XOutputStream , XConnectable , XServiceInfo >
  69 {
  70 public:
  71     OPipeImpl( );
  72     ~OPipeImpl();
  73
  74 public: // XInputStream
  75     virtual sal_Int32 SAL_CALL readBytes(Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead)
  76         throw(  NotConnectedException,
  77                 BufferSizeExceededException,
  78                 RuntimeException );
  79     virtual sal_Int32 SAL_CALL readSomeBytes(Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead)
  80         throw( NotConnectedException,
  81                BufferSizeExceededException,
  82                RuntimeException );
  83     virtual void SAL_CALL skipBytes(sal_Int32 nBytesToSkip)
  84         throw( NotConnectedException,
  85                BufferSizeExceededException,
  86                RuntimeException );
  87     virtual sal_Int32 SAL_CALL available(void)
  88         throw( NotConnectedException,
  89                RuntimeException );
  90     virtual void SAL_CALL closeInput(void)
  91         throw( NotConnectedException,
  92                RuntimeException );
  93
  94 public: // XOutputStream
  95
  96     virtual void SAL_CALL writeBytes(const Sequence< sal_Int8 >& aData)
  97         throw( NotConnectedException,
  98                BufferSizeExceededException,
  99                RuntimeException );
 100     virtual void SAL_CALL flush(void)
 101         throw( NotConnectedException,
 102                BufferSizeExceededException,
 103                RuntimeException );
 104     virtual void SAL_CALL closeOutput(void)
 105         throw( NotConnectedException,
 106                BufferSizeExceededException,
 107                RuntimeException );
 108
 109 public: // XConnectable
 110     virtual void SAL_CALL setPredecessor(const Reference< XConnectable >& aPredecessor)
 111         throw( RuntimeException );
 112     virtual Reference< XConnectable > SAL_CALL getPredecessor(void) throw( RuntimeException );
 113     virtual void SAL_CALL setSuccessor(const Reference < XConnectable > & aSuccessor)
 114         throw( RuntimeException );
 115     virtual Reference < XConnectable > SAL_CALL getSuccessor(void) throw( RuntimeException ) ;
 116
 117
 118 public: // XServiceInfo
 119     OUString                    SAL_CALL getImplementationName() throw(  );
 120     Sequence< OUString >         SAL_CALL getSupportedServiceNames(void) throw(  );
 121     sal_Bool                        SAL_CALL supportsService(const OUString& ServiceName) throw(  );
 122
 123 private:
 124
 125     // DEBUG
 126     inline void checkInvariant();
 127
 128     Reference < XConnectable >  m_succ;
 129     Reference < XConnectable >  m_pred;
 130
 131     sal_Int32 m_nBytesToSkip;
 132
 133     sal_Int8  *m_p;
 134
 135     sal_Bool m_bOutputStreamClosed;
 136     sal_Bool m_bInputStreamClosed;
 137
 138     oslCondition m_conditionBytesAvail;
 139     Mutex     m_mutexAccess;
 140     IFIFO               *m_pFIFO;
 141 };
 142
 143
 144
 145 OPipeImpl::OPipeImpl()
 146 {
 147     g_moduleCount.modCnt.acquire( &g_moduleCount.modCnt );
 148     m_nBytesToSkip  = 0;
 149
 150     m_bOutputStreamClosed       = sal_False;
 151     m_bInputStreamClosed        = sal_False;
 152
 153     m_pFIFO = new MemFIFO;
 154     m_conditionBytesAvail = osl_createCondition();
 155 }
 156
 157 OPipeImpl::~OPipeImpl()
 158 {
 159     osl_destroyCondition( m_conditionBytesAvail );
 160     delete m_pFIFO;
 161     g_moduleCount.modCnt.release( &g_moduleCount.modCnt );
 162 }
 163
 164
 165 // These invariants must hold when entering a guarded method or leaving a guarded method.
 166 void OPipeImpl::checkInvariant()
 167 {
 168
 169 }
 170
 171 sal_Int32 OPipeImpl::readBytes(Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead)
 172     throw( NotConnectedException, BufferSizeExceededException,RuntimeException )
 173 {
 174     while( sal_True )
 175     {
 176         { // start guarded section
 177             MutexGuard guard( m_mutexAccess );
 178             if( m_bInputStreamClosed )
 179             {
 180                 throw NotConnectedException(
 181                     OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::readBytes NotConnectedException" )),
 182                     *this );
 183             }
 184             sal_Int32 nOccupiedBufferLen = m_pFIFO->getSize();
 185
 186             if( m_bOutputStreamClosed && nBytesToRead > nOccupiedBufferLen )
 187             {
 188                 nBytesToRead = nOccupiedBufferLen;
 189             }
 190
 191             if( nOccupiedBufferLen < nBytesToRead )
 192             {
 193                 // wait outside guarded section
 194                 osl_resetCondition( m_conditionBytesAvail );
 195             }
 196             else {
 197                 // necessary bytes are available
 198                 m_pFIFO->read( aData , nBytesToRead );
 199                 return nBytesToRead;
 200             }
 201         } // end guarded section
 202
 203         // wait for new data outside guarded section!
 204         osl_waitCondition( m_conditionBytesAvail , 0 );
 205     }
 206 }
 207
 208
 209 sal_Int32 OPipeImpl::readSomeBytes(Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead)
 210     throw( NotConnectedException,
 211            BufferSizeExceededException,
 212            RuntimeException )
 213 {
 214     while( sal_True ) {
 215         {
 216             MutexGuard guard( m_mutexAccess );
 217             if( m_bInputStreamClosed )
 218             {
 219                 throw NotConnectedException(
 220                     OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::readSomeBytes NotConnectedException" )),
 221                     *this );
 222             }
 223             if( m_pFIFO->getSize() )
 224             {
 225                 sal_Int32 nSize = Min( nMaxBytesToRead , m_pFIFO->getSize() );
 226                 aData.realloc( nSize );
 227                 m_pFIFO->read( aData , nSize );
 228                 return nSize;
 229             }
 230
 231             if( m_bOutputStreamClosed )
 232             {
 233                 // no bytes in buffer anymore
 234                 return 0;
 235             }
 236         }
 237
 238         osl_waitCondition( m_conditionBytesAvail , 0 );
 239     }
 240 }
 241
 242
 243 void OPipeImpl::skipBytes(sal_Int32 nBytesToSkip)
 244     throw( NotConnectedException,
 245            BufferSizeExceededException,
 246            RuntimeException )
 247 {
 248     MutexGuard guard( m_mutexAccess );
 249     if( m_bInputStreamClosed )
 250     {
 251         throw NotConnectedException(
 252             OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::skipBytes NotConnectedException" ) ),
 253             *this );
 254     }
 255
 256     if( nBytesToSkip < 0
 257         || (nBytesToSkip
 258             > std::numeric_limits< sal_Int32 >::max() - m_nBytesToSkip) )
 259     {
 260         throw BufferSizeExceededException(
 261             OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::skipBytes BufferSizeExceededException" )),
 262             *this );
 263     }
 264     m_nBytesToSkip += nBytesToSkip;
 265
 266     nBytesToSkip = Min( m_pFIFO->getSize() , m_nBytesToSkip );
 267     m_pFIFO->skip( nBytesToSkip );
 268     m_nBytesToSkip -= nBytesToSkip;
 269 }
 270
 271
 272 sal_Int32 OPipeImpl::available(void)
 273     throw( NotConnectedException,
 274            RuntimeException )
 275  {
 276     MutexGuard guard( m_mutexAccess );
 277     if( m_bInputStreamClosed )
 278     {
 279         throw NotConnectedException(
 280             OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::available NotConnectedException" ) ),
 281             *this );
 282     }
 283     checkInvariant();
 284     return m_pFIFO->getSize();
 285 }
 286
 287 void OPipeImpl::closeInput(void)
 288     throw( NotConnectedException,
 289            RuntimeException)
 290 {
 291     MutexGuard guard( m_mutexAccess );
 292
 293     m_bInputStreamClosed = sal_True;
 294
 295     delete m_pFIFO;
 296     m_pFIFO = 0;
 297
 298     // readBytes may throw an exception
 299     osl_setCondition( m_conditionBytesAvail );
 300
 301     setSuccessor( Reference< XConnectable > () );
 302     return;
 303 }
 304
 305
 306 void OPipeImpl::writeBytes(const Sequence< sal_Int8 >& aData)
 307     throw( NotConnectedException,
 308            BufferSizeExceededException,
 309            RuntimeException)
 310 {
 311     MutexGuard guard( m_mutexAccess );
 312     checkInvariant();
 313
 314     if( m_bOutputStreamClosed )
 315     {
 316         throw NotConnectedException(
 317             OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::writeBytes NotConnectedException (outputstream)" )),
 318             *this );
 319     }
 320
 321     if( m_bInputStreamClosed )
 322     {
 323         throw NotConnectedException(
 324             OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::writeBytes NotConnectedException (inputstream)" )),
 325             *this );
 326     }
 327
 328     // check skipping
 329     sal_Int32 nLen = aData.getLength();
 330     if( m_nBytesToSkip  && m_nBytesToSkip >= nLen  ) {
 331         // all must be skipped - forget whole call
 332         m_nBytesToSkip -= nLen;
 333         return;
 334     }
 335
 336     // adjust buffersize if necessary
 337
 338     try
 339     {
 340         if( m_nBytesToSkip )
 341         {
 342             Sequence< sal_Int8 > seqCopy( nLen - m_nBytesToSkip );
 343             memcpy( seqCopy.getArray() , &( aData.getConstArray()[m_nBytesToSkip] ) , nLen-m_nBytesToSkip );
 344             m_pFIFO->write( seqCopy );
 345         }
 346         else
 347         {
 348             m_pFIFO->write( aData );
 349         }
 350         m_nBytesToSkip = 0;
 351     }
 352     catch ( IFIFO_OutOfBoundsException & )
 353     {
 354         throw BufferSizeExceededException(
 355             OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::writeBytes BufferSizeExceededException" )),
 356             *this );
 357     }
 358     catch ( IFIFO_OutOfMemoryException & )
 359     {
 360         throw BufferSizeExceededException(
 361             OUString( RTL_CONSTASCII_USTRINGPARAM( "Pipe::writeBytes BufferSizeExceededException" )),
 362             *this );
 363     }
 364
 365     // readBytes may check again if enough bytes are available
 366     osl_setCondition( m_conditionBytesAvail );
 367
 368     checkInvariant();
 369 }
 370
 371
 372 void OPipeImpl::flush(void)
 373     throw( NotConnectedException,
 374            BufferSizeExceededException,
 375            RuntimeException)
 376 {
 377     // nothing to do for a pipe
 378     return;
 379 }
 380
 381 void OPipeImpl::closeOutput(void)
 382     throw( NotConnectedException,
 383            BufferSizeExceededException,
 384            RuntimeException)
 385 {
 386     MutexGuard guard( m_mutexAccess );
 387
 388     m_bOutputStreamClosed = sal_True;
 389     osl_setCondition( m_conditionBytesAvail );
 390     setPredecessor( Reference < XConnectable > () );
 391     return;
 392 }
 393
 394
 395 void OPipeImpl::setSuccessor( const Reference < XConnectable >  &r )
 396     throw( RuntimeException )
 397 {
 398      /// if the references match, nothing needs to be done
 399      if( m_succ != r ) {
 400          /// store the reference for later use
 401          m_succ = r;
 402
 403          if( m_succ.is() )
 404          {
 405               m_succ->setPredecessor(
 406                   Reference< XConnectable > ( SAL_STATIC_CAST( XConnectable * , this ) ) );
 407          }
 408      }
 409 }
 410
 411 Reference < XConnectable > OPipeImpl::getSuccessor()    throw( RuntimeException )
 412 {
 413     return m_succ;
 414 }
 415
 416
 417 // XDataSource
 418 void OPipeImpl::setPredecessor( const Reference < XConnectable > &r )
 419     throw( RuntimeException )
 420 {
 421     if( r != m_pred ) {
 422         m_pred = r;
 423         if( m_pred.is() ) {
 424             m_pred->setSuccessor(
 425                 Reference < XConnectable > ( SAL_STATIC_CAST( XConnectable * , this ) ) );
 426         }
 427     }
 428 }
 429
 430 Reference < XConnectable > OPipeImpl::getPredecessor() throw( RuntimeException )
 431 {
 432     return m_pred;
 433 }
 434
 435
 436
 437
 438 // XServiceInfo
 439 OUString OPipeImpl::getImplementationName() throw(  )
 440 {
 441     return OPipeImpl_getImplementationName();
 442 }
 443
 444 // XServiceInfo
 445 sal_Bool OPipeImpl::supportsService(const OUString& ServiceName) throw(  )
 446 {
 447     Sequence< OUString > aSNL = getSupportedServiceNames();
 448     const OUString * pArray = aSNL.getConstArray();
 449
 450     for( sal_Int32 i = 0; i < aSNL.getLength(); i++ )
 451         if( pArray[i] == ServiceName )
 452             return sal_True;
 453
 454     return sal_False;
 455 }
 456
 457 // XServiceInfo
 458 Sequence< OUString > OPipeImpl::getSupportedServiceNames(void) throw(  )
 459 {
 460     return OPipeImpl_getSupportedServiceNames();
 461 }
 462
 463
 464
 465
 466
 467 /* implementation functions
 468 *
 469 *
 470 */
 471
 472
 473 Reference < XInterface > SAL_CALL OPipeImpl_CreateInstance(
 474     const Reference < XComponentContext > & ) throw(Exception)
 475 {
 476     OPipeImpl *p = new OPipeImpl;
 477
 478     return Reference < XInterface > ( SAL_STATIC_CAST( OWeakObject * , p ) );
 479 }
 480
 481
 482 OUString        OPipeImpl_getImplementationName()
 483 {
 484     return OUString( RTL_CONSTASCII_USTRINGPARAM ( IMPLEMENTATION_NAME ) );
 485 }
 486
 487 Sequence<OUString> OPipeImpl_getSupportedServiceNames(void)
 488 {
 489     Sequence<OUString> aRet(1);
 490     aRet.getArray()[0] = OUString( RTL_CONSTASCII_USTRINGPARAM( SERVICE_NAME ));
 491     return aRet;
 492 }
 493 }
 494
 495