package/source/zipapi/XBufferedThreadedStream.hxx

   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
  10 #ifndef INCLUDED_PACKAGE_SOURCE_ZIPAPI_XBUFFEREDTHREADEDSTREAM_HXX
  11 #define INCLUDED_PACKAGE_SOURCE_ZIPAPI_XBUFFEREDTHREADEDSTREAM_HXX
  12
  13 #include <com/sun/star/io/XInputStream.hpp>
  14
  15 #include <comphelper/bytereader.hxx>
  16 #include <cppuhelper/implbase.hxx>
  17 #include <rtl/ref.hxx>
  18 #include <salhelper/thread.hxx>
  19
  20 #include <queue>
  21 #include <mutex>
  22 #include <condition_variable>
  23 #include <exception>
  24
  25 typedef css::uno::Sequence< sal_Int8 > Buffer;
  26
  27 class XBufferedThreadedStream : public cppu::WeakImplHelper< css::io::XInputStream >,
  28         public comphelper::ByteReader
  29 {
  30 private:
  31     const css::uno::Reference<XInputStream> mxSrcStream;
  32     sal_Int64 mnPos;                                           /// position in stream
  33     sal_Int64 mnStreamSize;                                    /// available size of stream
  34
  35     Buffer maInUseBuffer;                                   /// Buffer block in use
  36     int mnOffset;                                           /// position in maInUseBuffer
  37     std::queue < Buffer > maPendingBuffers;                 /// Buffers that are available for use
  38     std::queue < Buffer > maUsedBuffers;
  39
  40     rtl::Reference< salhelper::Thread > mxUnzippingThread;
  41     std::mutex maBufferProtector;                           /// mutex protecting Buffer queues.
  42     std::condition_variable maBufferConsumeResume;
  43     std::condition_variable maBufferProduceResume;
  44     bool mbTerminateThread;                                 /// indicates the failure of one of the threads
  45
  46     std::exception_ptr maSavedException;                    /// exception caught during unzipping is saved to be thrown during reading
  47
  48     static const size_t nBufferLowWater = 2;
  49     static const size_t nBufferHighWater = 4;
  50     static const size_t nBufferSize = 32 * 1024;
  51
  52     const Buffer& getNextBlock();
  53     sal_Int64 remainingSize() const { return mnStreamSize - mnPos; }
  54     bool hasBytes() const { return mnPos < mnStreamSize; }
  55
  56     bool canProduce() const
  57     {
  58         return( mbTerminateThread || maPendingBuffers.size() < nBufferHighWater );
  59     }
  60
  61     bool canConsume() const
  62     {
  63         return( mbTerminateThread || !maPendingBuffers.empty() );
  64     }
  65
  66 public:
  67     XBufferedThreadedStream(
  68                   const css::uno::Reference<XInputStream>& xSrcStream,
  69                   sal_Int64 nStreamSize /* cf. sal_Int32 available(); */ );
  70
  71     virtual ~XBufferedThreadedStream() override;
  72
  73     void produce();
  74     void setTerminateThread();
  75     void saveException(const std::exception_ptr& exception) { maSavedException = exception; }
  76
  77     // XInputStream
  78     virtual sal_Int32 SAL_CALL readBytes( css::uno::Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead ) override;
  79     virtual sal_Int32 SAL_CALL readSomeBytes( css::uno::Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead ) override;
  80     virtual void SAL_CALL skipBytes( sal_Int32 nBytesToSkip ) override;
  81     virtual sal_Int32 SAL_CALL available(  ) override;
  82     virtual void SAL_CALL closeInput(  ) override;
  83
  84     // comphelper::ByteReader
  85     virtual sal_Int32 readSomeBytes(sal_Int8* aData, sal_Int32 nBytesToRead) override;
  86 };
  87 #endif
  88
  89 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */