io/source/stm/opipe.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 #include <sal/config.h>
  21
  22 #include <com/sun/star/io/BufferSizeExceededException.hpp>
  23 #include <com/sun/star/io/NotConnectedException.hpp>
  24 #include <com/sun/star/io/XPipe.hpp>
  25 #include <com/sun/star/io/XConnectable.hpp>
  26
  27 #include <com/sun/star/lang/XServiceInfo.hpp>
  28
  29 #include <cppuhelper/implbase.hxx>
  30 #include <cppuhelper/supportsservice.hxx>
  31
  32 #include <osl/conditn.hxx>
  33 #include <osl/mutex.hxx>
  34
  35 #include <limits>
  36 #include <memory>
  37 #include <string.h>
  38
  39 using namespace ::osl;
  40 using namespace ::cppu;
  41 using namespace ::com::sun::star::uno;
  42 using namespace ::com::sun::star::io;
  43 using namespace ::com::sun::star::lang;
  44
  45 #include <services.hxx>
  46 #include "streamhelper.hxx"
  47
  48 namespace com::sun::star::uno { class XComponentContext; }
  49
  50 // Implementation and service names
  51 #define IMPLEMENTATION_NAME "com.sun.star.comp.io.stm.Pipe"
  52
  53 namespace io_stm{
  54
  55 class OPipeImpl :
  56     public WeakImplHelper< XPipe , XConnectable , XServiceInfo >
  57 {
  58 public:
  59     OPipeImpl( );
  60
  61 public: // XInputStream
  62     virtual sal_Int32 SAL_CALL readBytes(Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead) override;
  63     virtual sal_Int32 SAL_CALL readSomeBytes(Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead) override;
  64     virtual void SAL_CALL skipBytes(sal_Int32 nBytesToSkip) override;
  65     virtual sal_Int32 SAL_CALL available() override;
  66     virtual void SAL_CALL closeInput() override;
  67
  68 public: // XOutputStream
  69
  70     virtual void SAL_CALL writeBytes(const Sequence< sal_Int8 >& aData) override;
  71     virtual void SAL_CALL flush() override;
  72     virtual void SAL_CALL closeOutput() override;
  73
  74 public: // XConnectable
  75     virtual void SAL_CALL setPredecessor(const Reference< XConnectable >& aPredecessor) override;
  76     virtual Reference< XConnectable > SAL_CALL getPredecessor() override;
  77     virtual void SAL_CALL setSuccessor(const Reference < XConnectable > & aSuccessor) override;
  78     virtual Reference < XConnectable > SAL_CALL getSuccessor() override ;
  79
  80
  81 public: // XServiceInfo
  82     OUString                    SAL_CALL getImplementationName() override;
  83     Sequence< OUString >         SAL_CALL getSupportedServiceNames() override;
  84     sal_Bool                        SAL_CALL supportsService(const OUString& ServiceName) override;
  85
  86 private:
  87
  88     Reference < XConnectable >  m_succ;
  89     Reference < XConnectable >  m_pred;
  90
  91     sal_Int32 m_nBytesToSkip;
  92
  93     bool m_bOutputStreamClosed;
  94     bool m_bInputStreamClosed;
  95
  96     osl::Condition m_conditionBytesAvail;
  97     Mutex          m_mutexAccess;
  98     std::unique_ptr<MemFIFO> m_pFIFO;
  99 };
 100
 101
 102 OPipeImpl::OPipeImpl()
 103     : m_nBytesToSkip(0 )
 104     , m_bOutputStreamClosed(false )
 105     , m_bInputStreamClosed( false )
 106     , m_pFIFO( new MemFIFO )
 107 {
 108 }
 109
 110
 111
 112 sal_Int32 OPipeImpl::readBytes(Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead)
 113 {
 114     while( true )
 115     {
 116         { // start guarded section
 117             MutexGuard guard( m_mutexAccess );
 118             if( m_bInputStreamClosed )
 119             {
 120                 throw NotConnectedException(
 121                     "Pipe::readBytes NotConnectedException",
 122                     *this );
 123             }
 124             sal_Int32 nOccupiedBufferLen = m_pFIFO->getSize();
 125
 126             if( m_bOutputStreamClosed && nBytesToRead > nOccupiedBufferLen )
 127             {
 128                 nBytesToRead = nOccupiedBufferLen;
 129             }
 130
 131             if( nOccupiedBufferLen < nBytesToRead )
 132             {
 133                 // wait outside guarded section
 134                 m_conditionBytesAvail.reset();
 135             }
 136             else {
 137                 // necessary bytes are available
 138                 m_pFIFO->read( aData , nBytesToRead );
 139                 return nBytesToRead;
 140             }
 141         } // end guarded section
 142
 143         // wait for new data outside guarded section!
 144         m_conditionBytesAvail.wait();
 145     }
 146 }
 147
 148
 149 sal_Int32 OPipeImpl::readSomeBytes(Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead)
 150 {
 151     while( true ) {
 152         {
 153             MutexGuard guard( m_mutexAccess );
 154             if( m_bInputStreamClosed )
 155             {
 156                 throw NotConnectedException(
 157                     "Pipe::readSomeBytes NotConnectedException",
 158                     *this );
 159             }
 160             if( m_pFIFO->getSize() )
 161             {
 162                 sal_Int32 nSize = std::min( nMaxBytesToRead , m_pFIFO->getSize() );
 163                 aData.realloc( nSize );
 164                 m_pFIFO->read( aData , nSize );
 165                 return nSize;
 166             }
 167
 168             if( m_bOutputStreamClosed )
 169             {
 170                 // no bytes in buffer anymore
 171                 return 0;
 172             }
 173         }
 174
 175         m_conditionBytesAvail.wait();
 176     }
 177 }
 178
 179
 180 void OPipeImpl::skipBytes(sal_Int32 nBytesToSkip)
 181 {
 182     MutexGuard guard( m_mutexAccess );
 183     if( m_bInputStreamClosed )
 184     {
 185         throw NotConnectedException(
 186             "Pipe::skipBytes NotConnectedException",
 187             *this );
 188     }
 189
 190     if( nBytesToSkip < 0
 191         || (nBytesToSkip
 192             > std::numeric_limits< sal_Int32 >::max() - m_nBytesToSkip) )
 193     {
 194         throw BufferSizeExceededException(
 195             "Pipe::skipBytes BufferSizeExceededException",
 196             *this );
 197     }
 198     m_nBytesToSkip += nBytesToSkip;
 199
 200     nBytesToSkip = std::min( m_pFIFO->getSize() , m_nBytesToSkip );
 201     m_pFIFO->skip( nBytesToSkip );
 202     m_nBytesToSkip -= nBytesToSkip;
 203 }
 204
 205
 206 sal_Int32 OPipeImpl::available()
 207  {
 208     MutexGuard guard( m_mutexAccess );
 209     if( m_bInputStreamClosed )
 210     {
 211         throw NotConnectedException(
 212             "Pipe::available NotConnectedException",
 213             *this );
 214     }
 215     return m_pFIFO->getSize();
 216 }
 217
 218 void OPipeImpl::closeInput()
 219 {
 220     MutexGuard guard( m_mutexAccess );
 221
 222     m_bInputStreamClosed = true;
 223
 224     m_pFIFO.reset();
 225
 226     // readBytes may throw an exception
 227     m_conditionBytesAvail.set();
 228
 229     setSuccessor( Reference< XConnectable > () );
 230 }
 231
 232
 233 void OPipeImpl::writeBytes(const Sequence< sal_Int8 >& aData)
 234 {
 235     MutexGuard guard( m_mutexAccess );
 236
 237     if( m_bOutputStreamClosed )
 238     {
 239         throw NotConnectedException(
 240             "Pipe::writeBytes NotConnectedException (outputstream)",
 241             *this );
 242     }
 243
 244     if( m_bInputStreamClosed )
 245     {
 246         throw NotConnectedException(
 247             "Pipe::writeBytes NotConnectedException (inputstream)",
 248             *this );
 249     }
 250
 251     // check skipping
 252     sal_Int32 nLen = aData.getLength();
 253     if( m_nBytesToSkip  && m_nBytesToSkip >= nLen  ) {
 254         // all must be skipped - forget whole call
 255         m_nBytesToSkip -= nLen;
 256         return;
 257     }
 258
 259     // adjust buffersize if necessary
 260     if( m_nBytesToSkip )
 261     {
 262         Sequence< sal_Int8 > seqCopy( nLen - m_nBytesToSkip );
 263         memcpy( seqCopy.getArray() , &( aData.getConstArray()[m_nBytesToSkip] ) , nLen-m_nBytesToSkip );
 264         m_pFIFO->write( seqCopy );
 265     }
 266     else
 267     {
 268         m_pFIFO->write( aData );
 269     }
 270     m_nBytesToSkip = 0;
 271
 272     // readBytes may check again if enough bytes are available
 273     m_conditionBytesAvail.set();
 274 }
 275
 276
 277 void OPipeImpl::flush()
 278 {
 279     // nothing to do for a pipe
 280 }
 281
 282 void OPipeImpl::closeOutput()
 283 {
 284     MutexGuard guard( m_mutexAccess );
 285
 286     m_bOutputStreamClosed = true;
 287     m_conditionBytesAvail.set();
 288     setPredecessor( Reference < XConnectable > () );
 289 }
 290
 291
 292 void OPipeImpl::setSuccessor( const Reference < XConnectable >  &r )
 293 {
 294      /// if the references match, nothing needs to be done
 295      if( m_succ != r ) {
 296          /// store the reference for later use
 297          m_succ = r;
 298
 299          if( m_succ.is() )
 300          {
 301               m_succ->setPredecessor(
 302                   Reference< XConnectable > ( static_cast< XConnectable * >(this) ) );
 303          }
 304      }
 305 }
 306
 307 Reference < XConnectable > OPipeImpl::getSuccessor()
 308 {
 309     return m_succ;
 310 }
 311
 312
 313 // XDataSource
 314 void OPipeImpl::setPredecessor( const Reference < XConnectable > &r )
 315 {
 316     if( r != m_pred ) {
 317         m_pred = r;
 318         if( m_pred.is() ) {
 319             m_pred->setSuccessor(
 320                 Reference < XConnectable > ( static_cast< XConnectable * >(this) ) );
 321         }
 322     }
 323 }
 324
 325 Reference < XConnectable > OPipeImpl::getPredecessor()
 326 {
 327     return m_pred;
 328 }
 329
 330
 331 // XServiceInfo
 332 OUString OPipeImpl::getImplementationName()
 333 {
 334     return OPipeImpl_getImplementationName();
 335 }
 336
 337 // XServiceInfo
 338 sal_Bool OPipeImpl::supportsService(const OUString& ServiceName)
 339 {
 340     return cppu::supportsService(this, ServiceName);
 341 }
 342
 343 // XServiceInfo
 344 Sequence< OUString > OPipeImpl::getSupportedServiceNames()
 345 {
 346     return OPipeImpl_getSupportedServiceNames();
 347 }
 348
 349 /* implementation functions
 350 *
 351 *
 352 */
 353
 354
 355 Reference < XInterface > OPipeImpl_CreateInstance(
 356     SAL_UNUSED_PARAMETER const Reference < XComponentContext > & )
 357 {
 358     OPipeImpl *p = new OPipeImpl;
 359
 360     return Reference < XInterface > ( static_cast< OWeakObject * >(p) );
 361 }
 362
 363
 364 OUString    OPipeImpl_getImplementationName()
 365 {
 366     return IMPLEMENTATION_NAME;
 367 }
 368
 369 Sequence<OUString> OPipeImpl_getSupportedServiceNames()
 370 {
 371     Sequence<OUString> aRet { "com.sun.star.io.Pipe" };
 372     return aRet;
 373 }
 374 }
 375
 376
 377 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */