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