blob: beb1cd33c7004e64a8d92271963bacfe53821a59 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
|
/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
/*
* This file is part of the LibreOffice project.
*
* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/.
*/
#ifndef INCLUDED_PACKAGE_SOURCE_ZIPAPI_XBUFFEREDTHREADEDSTREAM_HXX
#define INCLUDED_PACKAGE_SOURCE_ZIPAPI_XBUFFEREDTHREADEDSTREAM_HXX
#include <com/sun/star/io/XInputStream.hpp>
#include <cppuhelper/implbase.hxx>
#include <rtl/ref.hxx>
#include <salhelper/thread.hxx>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <exception>
typedef css::uno::Sequence< sal_Int8 > Buffer;
class XBufferedThreadedStream : public cppu::WeakImplHelper< css::io::XInputStream >
{
private:
const css::uno::Reference<XInputStream> mxSrcStream;
sal_Int64 mnPos; /// position in stream
sal_Int64 mnStreamSize; /// available size of stream
Buffer maInUseBuffer; /// Buffer block in use
int mnOffset; /// position in maInUseBuffer
std::queue < Buffer > maPendingBuffers; /// Buffers that are available for use
std::queue < Buffer > maUsedBuffers;
rtl::Reference< salhelper::Thread > mxUnzippingThread;
std::mutex maBufferProtector; /// mutex protecting Buffer queues.
std::condition_variable maBufferConsumeResume;
std::condition_variable maBufferProduceResume;
bool mbTerminateThread; /// indicates the failure of one of the threads
std::exception_ptr maSavedException; /// exception caught during unzipping is saved to be thrown during reading
static const size_t nBufferLowWater = 2;
static const size_t nBufferHighWater = 4;
static const size_t nBufferSize = 32 * 1024;
const Buffer& getNextBlock();
sal_Int64 remainingSize() const { return mnStreamSize - mnPos; }
bool hasBytes() const { return mnPos < mnStreamSize; }
bool canProduce() const
{
return( mbTerminateThread || maPendingBuffers.size() < nBufferHighWater );
}
bool canConsume() const
{
return( mbTerminateThread || !maPendingBuffers.empty() );
}
public:
XBufferedThreadedStream(
const css::uno::Reference<XInputStream>& xSrcStream,
sal_Int64 nStreamSize /* cf. sal_Int32 available(); */ );
virtual ~XBufferedThreadedStream() override;
void produce();
void setTerminateThread();
void saveException(const std::exception_ptr& exception) { maSavedException = exception; }
// XInputStream
virtual sal_Int32 SAL_CALL readBytes( css::uno::Sequence< sal_Int8 >& aData, sal_Int32 nBytesToRead ) override;
virtual sal_Int32 SAL_CALL readSomeBytes( css::uno::Sequence< sal_Int8 >& aData, sal_Int32 nMaxBytesToRead ) override;
virtual void SAL_CALL skipBytes( sal_Int32 nBytesToSkip ) override;
virtual sal_Int32 SAL_CALL available( ) override;
virtual void SAL_CALL closeInput( ) override;
};
#endif
/* vim:set shiftwidth=4 softtabstop=4 expandtab: */
|
