From 410a04a58ee59de3055483a2f971c2e42015267e Mon Sep 17 00:00:00 2001
From: David Steele <david@pgmasters.net>
Date: Wed, 28 Nov 2018 14:20:12 -0500
Subject: [PATCH] Allow arbitary InOut filters to be chained in IoFilterGroup.

If InOut filters were placed next to each other then the second filter would never get a NULL input signaling it to flush.  This arrangement only worked if the second filter had some other indication that it should flush, such as a decompression filter where the flush is indicated in the input stream.

This is not a live issue because currently no InOut filters are chained together.
---
 doc/xml/release.xml             |  4 ++++
 src/common/io/filter/group.c    | 16 +++++++++++++++-
 test/src/module/common/ioTest.c |  7 +++++--
 3 files changed, 24 insertions(+), 3 deletions(-)

diff --git a/doc/xml/release.xml b/doc/xml/release.xml
index 6053cd92d..ddfb0693c 100644
--- a/doc/xml/release.xml
+++ b/doc/xml/release.xml
@@ -39,6 +39,10 @@
                         <p>Add <code>IoFilter</code> interface to <code>CipherBlock</code> object.</p>
                     </release-item>
 
+                    <release-item>
+                        <p>Allow arbitary <code>InOut</code> filters to be chained in <code>IoFilterGroup</code>.</p>
+                    </release-item>
+
                     <release-item>
                         <release-item-contributor-list>
                             <release-item-contributor id="cynthia.shang"/>
diff --git a/src/common/io/filter/group.c b/src/common/io/filter/group.c
index 62b53d7c7..707c2164a 100644
--- a/src/common/io/filter/group.c
+++ b/src/common/io/filter/group.c
@@ -254,6 +254,20 @@ ioFilterGroupProcess(IoFilterGroup *this, const Buffer *input, Buffer *output)
                 // Keep processing while the filter is not done or there is input
                 if (!ioFilterDone(filterData->filter) || filterData->input != NULL)
                 {
+                    // If we are flushing and the prior filter is done and is not producing any more output then this filter should
+                    // be flushing as well.  Set filterData->input = NULL so it knows there is no more input coming.
+                    //
+                    // If the filter is already done then there is no need to set input to NULL because it has already flushed and
+                    // the filter shouldn't need to hand NULL input if it doesn't need it to know when to flush.
+                    //
+                    // Checking filterIdx - 1 is safe because the first filter's filterData->input is always set to NULL when input
+                    // is NULL.
+                    if (input == NULL && filterData->input != NULL && !ioFilterDone(filterData->filter) &&
+                        ioFilterDone(ioFilterGroupGet(this, filterIdx - 1)->filter) && bufUsed(filterData->input) == 0)
+                    {
+                        filterData->input = NULL;
+                    }
+
                     ioFilterProcessInOut(filterData->filter, filterData->input, filterData->output);
 
                     // If inputSame is set then the output buffer for this filter is full and it will need to be pre-processed with
@@ -262,7 +276,7 @@ ioFilterGroupProcess(IoFilterGroup *this, const Buffer *input, Buffer *output)
                         this->inputSame = true;
 
                     // Else clear the buffer if it was locally allocated.  If this is an input buffer that was passed in then the
-                    // caller is reponsible for clearing it.
+                    // caller is responsible for clearing it.
                     else if (filterData->inputLocal != NULL)
                         bufUsedZero(filterData->inputLocal);
                 }
diff --git a/test/src/module/common/ioTest.c b/test/src/module/common/ioTest.c
index 71b4c4f76..04da6a545 100644
--- a/test/src/module/common/ioTest.c
+++ b/test/src/module/common/ioTest.c
@@ -416,6 +416,9 @@ testRun(void)
         TEST_RESULT_VOID(ioFilterGroupAdd(filterGroup, ioSizeFilter(sizeFilter)), "    add filter to filter group");
         TEST_RESULT_VOID(
             ioFilterGroupAdd(filterGroup, ioTestFilterMultiplyNew("double", 2, 3, 'X')->filter), "    add filter to filter group");
+        TEST_RESULT_VOID(
+            ioFilterGroupAdd(filterGroup, ioTestFilterMultiplyNew("single", 1, 1, 'Y')->filter),
+            "    add filter to filter group");
         TEST_RESULT_VOID(ioFilterGroupAdd(filterGroup, ioTestFilterSizeNew("size2")->filter), "    add filter to filter group");
         TEST_RESULT_VOID(ioWriteFilterGroupSet(ioBufferWriteIo(bufferWrite), filterGroup), "    add filter group to write io");
 
@@ -436,12 +439,12 @@ testRun(void)
         TEST_RESULT_STR(strPtr(strNewBuf(buffer)), "AABB\n\nZZ112233445", "    check write");
 
         TEST_RESULT_VOID(ioWriteClose(ioBufferWriteIo(bufferWrite)), " close buffer write object");
-        TEST_RESULT_STR(strPtr(strNewBuf(buffer)), "AABB\n\nZZ1122334455XXX", "    check write after close");
+        TEST_RESULT_STR(strPtr(strNewBuf(buffer)), "AABB\n\nZZ1122334455XXXY", "    check write after close");
 
         TEST_RESULT_PTR(ioWriteFilterGroup(ioBufferWriteIo(bufferWrite)), filterGroup, "    check filter group");
         TEST_RESULT_UINT(
             varUInt64(ioFilterGroupResult(filterGroup, ioFilterType(ioSizeFilter(sizeFilter)))), 9, "    check filter result");
-        TEST_RESULT_UINT(varUInt64(ioFilterGroupResult(filterGroup, strNew("size2"))), 21, "    check filter result");
+        TEST_RESULT_UINT(varUInt64(ioFilterGroupResult(filterGroup, strNew("size2"))), 22, "    check filter result");
 
         TEST_RESULT_VOID(ioBufferWriteFree(bufferWrite), "    free buffer write object");
         TEST_RESULT_VOID(ioBufferWriteFree(NULL), "    free NULL buffer write object");