tdf#97597: attempt to add test for multithreaded input stream buffering.

But it always passes, even when UseBufferedStream is set to false...
Needs improvement.

Change-Id: I98f65dcd7bec3b47a437fdc6cc42c6e8e3775522
Reviewed-on: https://gerrit.libreoffice.org/33190
Tested-by: Jenkins <ci@libreoffice.org>
Reviewed-by: Kohei Yoshida <libreoffice@kohei.us>

3 files changed, 120 insertions, 1 deletions

diff --git a/package/qa/cppunit/data/a2z.zip b/package/qa/cppunit/data/a2z.zip
new file mode 100644
index 000000000000..4a04508b8f2e
--- /dev/null
+++ b/package/qa/cppunit/data/a2z.zip
diff --git a/package/qa/cppunit/test_package.cxx b/package/qa/cppunit/test_package.cxx
index 07b3abb757d0..80f02d8e4842 100644
--- a/package/qa/cppunit/test_package.cxx
+++ b/package/qa/cppunit/test_package.cxx
@@ -10,8 +10,11 @@
 #include <comphelper/processfactory.hxx>
 #include <unotest/filters-test.hxx>
 #include <unotest/bootstrapfixturebase.hxx>
+#include <comphelper/threadpool.hxx>
 #include "com/sun/star/packages/zip/ZipFileAccess.hpp"
 
+#include <iterator>
+
 using namespace ::com::sun::star;
 
 namespace
@@ -28,9 +31,11 @@ namespace
             SfxFilterFlags, SotClipboardFormatId, unsigned int) override;
 
         void test();
+        void testThreadedStreams();
 
         CPPUNIT_TEST_SUITE(PackageTest);
         CPPUNIT_TEST(test);
+        CPPUNIT_TEST(testThreadedStreams);
         CPPUNIT_TEST_SUITE_END();
     };
 
@@ -56,6 +61,90 @@ namespace
             m_directories.getURLFromSrc("/package/qa/cppunit/data/"));
     }
 
+    // TODO : This test currently doesn't fail even when you set
+    // UseBufferedStream to false. Look into this and replace it with a better
+    // test that actually fails when the aforementioned flag is set to false.
+    void PackageTest::testThreadedStreams()
+    {
+        class Worker : public comphelper::ThreadTask
+        {
+            uno::Reference<io::XInputStream> mxStrm;
+            std::vector<char>& mrBuf;
+
+        public:
+            Worker(
+                const std::shared_ptr<comphelper::ThreadTaskTag>& pTag,
+                const uno::Reference<io::XInputStream>& xStrm,
+                std::vector<char>& rBuf ) :
+                comphelper::ThreadTask(pTag), mxStrm(xStrm), mrBuf(rBuf) {}
+
+            virtual void doWork() override
+            {
+                sal_Int32 nSize = mxStrm->available();
+
+                uno::Sequence<sal_Int8> aBytes;
+                while (nSize > 0)
+                {
+                    sal_Int32 nBytesRead = mxStrm->readBytes(aBytes, 4096);
+                    const sal_Int8* p = aBytes.getArray();
+                    const sal_Int8* pEnd = p + nBytesRead;
+                    std::copy(p, pEnd, std::back_inserter(mrBuf));
+                    nSize -= nBytesRead;
+                }
+            }
+        };
+
+        OUString aURL = m_directories.getURLFromSrc("/package/qa/cppunit/data/a2z.zip");
+
+        uno::Sequence<beans::NamedValue> aArgs(2);
+        aArgs[0].Name = "URL";
+        aArgs[0].Value <<= aURL;
+        aArgs[1].Name = "UseBufferedStream";
+        aArgs[1].Value <<= true;
+
+        uno::Reference<packages::zip::XZipFileAccess2> xZip(
+            packages::zip::ZipFileAccess::createWithArguments(comphelper::getProcessComponentContext(), aArgs));
+
+        CPPUNIT_ASSERT(xZip.is());
+
+        uno::Reference<container::XNameAccess> xNA(xZip, uno::UNO_QUERY);
+        CPPUNIT_ASSERT(xNA.is());
+
+        {
+            comphelper::ThreadPool aPool(4);
+            std::shared_ptr<comphelper::ThreadTaskTag> pTag = comphelper::ThreadPool::createThreadTaskTag();
+
+            std::vector<std::vector<char>> aTestBuffers(26);
+            auto itBuf = aTestBuffers.begin();
+
+            for (char c = 'a'; c <= 'z'; ++c, ++itBuf)
+            {
+                OUString aName(c);
+                aName += ".txt";
+
+                uno::Reference<io::XInputStream> xStrm;
+                xNA->getByName(aName) >>= xStrm;
+
+                CPPUNIT_ASSERT(xStrm.is());
+                aPool.pushTask(new Worker(pTag, xStrm, *itBuf));
+            }
+
+            aPool.waitUntilDone(pTag);
+
+            // Verify the streams.
+            CPPUNIT_ASSERT_EQUAL(size_t(26), aTestBuffers.size());
+            itBuf = aTestBuffers.begin();
+
+            for (char c = 'a'; c <= 'z'; ++c, ++itBuf)
+            {
+                const std::vector<char>& rBuf = *itBuf;
+                CPPUNIT_ASSERT_EQUAL(size_t(1048576), rBuf.size()); // 1 MB each.
+                for (char check : rBuf)
+                    CPPUNIT_ASSERT_EQUAL(c, check);
+            }
+        }
+    }
+
     CPPUNIT_TEST_SUITE_REGISTRATION(PackageTest);
 }
 
diff --git a/package/source/zippackage/zipfileaccess.cxx b/package/source/zippackage/zipfileaccess.cxx
index 88ac19c7dc01..b6875eee9e23 100644
--- a/package/source/zippackage/zipfileaccess.cxx
+++ b/package/source/zippackage/zipfileaccess.cxx
@@ -23,6 +23,7 @@
 #include <com/sun/star/io/XActiveDataSink.hpp>
 #include <com/sun/star/io/XStream.hpp>
 #include <com/sun/star/io/XSeekable.hpp>
+#include <com/sun/star/beans/NamedValue.hpp>
 #include <comphelper/processfactory.hxx>
 #include <cppuhelper/supportsservice.hxx>
 #include <zipfileaccess.hxx>
@@ -182,8 +183,11 @@ void SAL_CALL OZipFileAccess::initialize( const uno::Sequence< uno::Any >& aArgu
     OUString aParamURL;
     uno::Reference< io::XStream > xStream;
     uno::Reference< io::XSeekable > xSeekable;
+    uno::Sequence<beans::NamedValue> aArgs;
 
-    if ( ( aArguments[0] >>= aParamURL ) )
+    bool bUseBufferedStream = false;
+
+    auto openInputStream = [&]()
     {
         ::ucbhelper::Content aContent(
             aParamURL,
@@ -196,6 +200,11 @@ void SAL_CALL OZipFileAccess::initialize( const uno::Sequence< uno::Any >& aArgu
             m_bOwnContent = true;
             xSeekable.set( m_xContentStream, uno::UNO_QUERY );
         }
+    };
+
+    if ( ( aArguments[0] >>= aParamURL ) )
+    {
+        openInputStream();
     }
     else if ( (aArguments[0] >>= xStream ) )
     {
@@ -207,6 +216,25 @@ void SAL_CALL OZipFileAccess::initialize( const uno::Sequence< uno::Any >& aArgu
     {
         xSeekable.set( m_xContentStream, uno::UNO_QUERY );
     }
+    else if (aArguments[0] >>= aArgs)
+    {
+        for (sal_Int32 i = 0; i < aArgs.getLength(); ++i)
+        {
+            const beans::NamedValue& rArg = aArgs[i];
+
+            if (rArg.Name == "URL")
+                rArg.Value >>= aParamURL;
+            else if (rArg.Name == "UseBufferedStream")
+                rArg.Value >>= bUseBufferedStream;
+        }
+
+        if (aParamURL.isEmpty())
+            throw lang::IllegalArgumentException(
+                THROW_WHERE"required argument 'URL' is not given or invalid.",
+                uno::Reference<uno::XInterface>(), 1);
+
+        openInputStream();
+    }
     else
         throw lang::IllegalArgumentException(THROW_WHERE, uno::Reference< uno::XInterface >(), 1 );
 
@@ -224,6 +252,8 @@ void SAL_CALL OZipFileAccess::initialize( const uno::Sequence< uno::Any >& aArgu
                 m_xContentStream,
                 m_xContext,
                 true );
+
+    m_pZipFile->setUseBufferedStream(bUseBufferedStream);
 }
 
 // XNameAccess