/*
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
*/
package org.apache.edgent.test.connectors.file;

import static org.apache.edgent.test.connectors.common.FileUtil.createTempFile;

//import static org.junit.Assume.assumeFalse;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;

import org.apache.edgent.connectors.file.CompressedFileWriterPolicy;
import org.apache.edgent.connectors.file.FileStreams;
import org.apache.edgent.connectors.file.FileWriterCycleConfig;
import org.apache.edgent.connectors.file.FileWriterFlushConfig;
import org.apache.edgent.connectors.file.FileWriterPolicy;
import org.apache.edgent.connectors.file.FileWriterRetentionConfig;
import org.apache.edgent.connectors.file.runtime.IFileWriterPolicy;
import org.apache.edgent.function.Predicate;
import org.apache.edgent.test.providers.direct.DirectTopologyTestBase;
import org.apache.edgent.topology.TSink;
import org.apache.edgent.topology.TStream;
import org.apache.edgent.topology.Topology;
import org.apache.edgent.topology.plumbing.PlumbingStreams;
import org.apache.edgent.topology.tester.Condition;
import org.junit.Test;

public class FileStreamsTextFileWriterTest extends DirectTopologyTestBase {
    
    String str = "123456789";
    String[] stdLines = new String[] {
            "1-"+str,
            "2-"+str,
            "3-"+str,
            "4-"+str
    };
    
    private int TMO_SEC = 2;

    public String getStr() {
        return str;
    }

    public String[] getLines() {
        return stdLines;
    }

    @Test
    public void testFlushConfig() throws Exception {
        FileWriterFlushConfig<String> cfg;

        String trueTuple = "true";
        String falseTuple = "false";
        Predicate<String> p = tuple -> tuple.equals("true");

        cfg = FileWriterFlushConfig.newImplicitConfig();
        checkFileWriterConfig(cfg, 0, 0, null, trueTuple, falseTuple);
        
        cfg = FileWriterFlushConfig.newCountBasedConfig(3);
        checkFileWriterConfig(cfg, 3, 0, null, trueTuple, falseTuple);
        expectIAE(() -> FileWriterFlushConfig.newCountBasedConfig(0));
        
        cfg = FileWriterFlushConfig.newTimeBasedConfig(10);
        checkFileWriterConfig(cfg, 0, 10, null, trueTuple, falseTuple);
        expectIAE(() -> FileWriterFlushConfig.newTimeBasedConfig(0));
        
        cfg = FileWriterFlushConfig.newPredicateBasedConfig(p);
        checkFileWriterConfig(cfg, 0, 0, p, trueTuple, falseTuple);
        expectIAE(() -> FileWriterFlushConfig.newPredicateBasedConfig(null));
        
        cfg = FileWriterFlushConfig.newConfig(1, 2, p);
        checkFileWriterConfig(cfg, 1, 2, p, trueTuple, falseTuple);
        cfg = FileWriterFlushConfig.newConfig(0, 0, null);
        checkFileWriterConfig(cfg, 0, 0, null, trueTuple, falseTuple);
        expectIAE(() -> FileWriterFlushConfig.newConfig(-1, 0, null));
        expectIAE(() -> FileWriterFlushConfig.newConfig(0, -1, null));
    }
    
    private static <T> void checkFileWriterConfig(FileWriterFlushConfig<T> cfg,
            int cntTuples, long periodMsec, Predicate<T> tuplePredicate,
            T trueTuple, T falseTuple) {
        assertEquals(cntTuples, cfg.getCntTuples());
        assertEquals(periodMsec, cfg.getPeriodMsec());
        assertEquals(tuplePredicate, cfg.getTuplePredicate());
        cfg.toString();
       
        int falseNTuples = cntTuples==1 ? 0 : cntTuples+1;
        int trueNTuples = 3*cntTuples;
        
        assertFalse("cntTuples:"+cntTuples+" pred:"+tuplePredicate,
                    cfg.evaluate(falseNTuples, falseTuple));
        if (cntTuples!=0)
            assertTrue("cntTuples:"+cntTuples+" pred:"+tuplePredicate,
                    cfg.evaluate(trueNTuples, falseTuple));
        if (tuplePredicate!=null)
            assertTrue("cntTuples:"+cntTuples+" pred:"+tuplePredicate,
                    cfg.evaluate(falseNTuples, trueTuple));
    }

    @Test
    public void testCycleConfig() throws Exception {
        FileWriterCycleConfig<String> cfg;

        String trueTuple = "true";
        String falseTuple = "false";
        Predicate<String> p = tuple -> tuple.equals("true");
        
        cfg = FileWriterCycleConfig.newFileSizeBasedConfig(2);
        checkFileWriterConfig(cfg, 2, 0, 0, null, trueTuple, falseTuple);
        expectIAE(() -> FileWriterCycleConfig.newFileSizeBasedConfig(0));
        
        cfg = FileWriterCycleConfig.newCountBasedConfig(3);
        checkFileWriterConfig(cfg, 0, 3, 0, null, trueTuple, falseTuple);
        expectIAE(() -> FileWriterCycleConfig.newCountBasedConfig(0));
        
        cfg = FileWriterCycleConfig.newTimeBasedConfig(10);
        checkFileWriterConfig(cfg, 0, 0, 10, null, trueTuple, falseTuple);
        expectIAE(() -> FileWriterCycleConfig.newTimeBasedConfig(0));
        
        cfg = FileWriterCycleConfig.newPredicateBasedConfig(p);
        checkFileWriterConfig(cfg, 0, 0, 0, p, trueTuple, falseTuple);
        expectIAE(() -> FileWriterCycleConfig.newPredicateBasedConfig(null));
        
        cfg = FileWriterCycleConfig.newConfig(1, 2, 3, p);
        checkFileWriterConfig(cfg, 1, 2, 3, p, trueTuple, falseTuple);
        expectIAE(() -> FileWriterCycleConfig.newConfig(0, 0, 0, null));
        expectIAE(() -> FileWriterCycleConfig.newConfig(-1, 0, 0, null));
        expectIAE(() -> FileWriterCycleConfig.newConfig(0, -1, 0, null));
        expectIAE(() -> FileWriterCycleConfig.newConfig(0, 0, -1, null));
    }
    
    private static <T> void checkFileWriterConfig(FileWriterCycleConfig<T> cfg,
            long fileSize, int cntTuples, long periodMsec, Predicate<T> tuplePredicate,
            T trueTuple, T falseTuple) {
        assertEquals(fileSize, cfg.getFileSize());
        assertEquals(cntTuples, cfg.getCntTuples());
        assertEquals(periodMsec, cfg.getPeriodMsec());
        assertEquals(tuplePredicate, cfg.getTuplePredicate());
        cfg.toString();
        
        long falseFileSize = fileSize-1;
        long trueFileSize = fileSize+1;
        int falseNTuples = cntTuples==1 ? 0 : cntTuples+1;
        int trueNTuples = 3*cntTuples;
        
        assertFalse("fileSize:"+fileSize+" cntTuples:"+cntTuples+" pred:"+tuplePredicate,
                    cfg.evaluate(falseFileSize, falseNTuples, falseTuple));
        if (fileSize!=0)
            assertTrue("fileSize:"+fileSize+" cntTuples:"+cntTuples+" pred:"+tuplePredicate,
                    cfg.evaluate(trueFileSize, trueNTuples, falseTuple));
        if (cntTuples!=0)
            assertTrue("fileSize:"+fileSize+" cntTuples:"+cntTuples+" pred:"+tuplePredicate,
                    cfg.evaluate(falseFileSize, trueNTuples, falseTuple));
        if (tuplePredicate!=null)
            assertTrue("fileSize:"+fileSize+" cntTuples:"+cntTuples+" pred:"+tuplePredicate,
                    cfg.evaluate(falseFileSize, falseNTuples, trueTuple));
    }

    @Test
    public void testRetentionConfig() throws Exception {
        FileWriterRetentionConfig cfg;

        cfg = FileWriterRetentionConfig.newFileCountBasedConfig(2);
        checkFileWriterConfig(cfg, 2, 0, 0, 0);
        expectIAE(() -> FileWriterRetentionConfig.newFileCountBasedConfig(0));
        
        cfg = FileWriterRetentionConfig.newAggregateFileSizeBasedConfig(3);
        checkFileWriterConfig(cfg, 0, 3, 0, 0);
        expectIAE(() -> FileWriterRetentionConfig.newAggregateFileSizeBasedConfig(0));
        
        cfg = FileWriterRetentionConfig.newAgeBasedConfig(10,11);
        checkFileWriterConfig(cfg, 0, 0, 10, 11);
        expectIAE(() -> FileWriterRetentionConfig.newAgeBasedConfig(0,1));
        expectIAE(() -> FileWriterRetentionConfig.newAgeBasedConfig(1,0));
        
        cfg = FileWriterRetentionConfig.newConfig(1, 2, 3, 0);
        checkFileWriterConfig(cfg, 1, 2, 3, 0);
        expectIAE(() -> FileWriterRetentionConfig.newConfig(0, 0, 0, 0));
        expectIAE(() -> FileWriterRetentionConfig.newConfig(0, 0, 1, 0));
        expectIAE(() -> FileWriterRetentionConfig.newConfig(0, 0, 0, 1));
        expectIAE(() -> FileWriterRetentionConfig.newConfig(-1, 0, 0, 0));
        expectIAE(() -> FileWriterRetentionConfig.newConfig(0, -1, 0, 0));
        expectIAE(() -> FileWriterRetentionConfig.newConfig(0, 0, -1, 0));
        expectIAE(() -> FileWriterRetentionConfig.newConfig(0, 0, 0, -1));
    }
    
    private void expectIAE(Runnable fn) {
        try { 
            fn.run();
            fail("expected IAE");
        } catch (IllegalArgumentException e) { /* expected */ }
    }
    
    private static <T> void checkFileWriterConfig(FileWriterRetentionConfig cfg,
            int fileCnt, long aggSize, long ageSec, long periodMsec) {
        assertEquals(fileCnt, cfg.getFileCount());
        assertEquals(aggSize, cfg.getAggregateFileSize());
        assertEquals(ageSec, cfg.getAgeSec());
        assertEquals(periodMsec, cfg.getPeriodMsec());
        cfg.toString();

        int falseFileCnt = fileCnt-1;
        int trueFileCnt = fileCnt+1;
        long falseAggSize = aggSize-1;
        long trueAggSize = aggSize+1;
        
        assertFalse("fileCnt:"+fileCnt+" aggSize:"+aggSize,
                    cfg.evaluate(falseFileCnt, falseAggSize));
        if (fileCnt!=0)
            assertTrue("fileCnt:"+fileCnt+" aggSize:"+aggSize,
                    cfg.evaluate(trueFileCnt, falseAggSize));
        if (aggSize!=0)
            assertTrue("fileCnt:"+fileCnt+" aggSize:"+aggSize,
                    cfg.evaluate(falseFileCnt, trueAggSize));
    }

    @Test
    public void testDefaultConfig() throws Exception {
        FileWriterPolicy<String> policy = new FileWriterPolicy<>();
        checkFileWriterConfig(policy.getFlushConfig(), 0, TimeUnit.SECONDS.toMillis(10), null, null, null);
        checkFileWriterConfig(policy.getCycleConfig(), 1*1024*1024, 0, 0, null, null, null);
        checkFileWriterConfig(policy.getRetentionConfig(), 10, 0, 0, 0);
        policy.toString();
        policy.close();
    }

    @Test
    public void testNoFilesCreated() throws Exception {
        // complete before any files are generated
        Topology t = newTopology("testNoFilesCreated");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        // build expected results
        List<List<String>> expResults = Collections.emptyList();

        TStream<String> s = t.events(eventSetup -> { /* no tuples generated */ });
        
        FileStreams.textFileWriter(s, () -> basePath.toString());

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testOneFileCreated() throws Exception {
        // all lines into a single (the first) file
        Topology t = newTopology("testOneFileCreated");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();

        // build expected results
        // net all in one, the first, file
        List<List<String>> expResults = buildExpResults(lines, tuple -> false);
        assertEquals(1, expResults.size());
        
        TStream<String> s = t.strings(lines);
        
        // default writer policy
        TSink<String> sink = FileStreams.textFileWriter(s, () -> basePath.toString());

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
        
        assertNotNull(sink);
    }

    @Test
    public void testManyFiles() throws Exception {
        Topology t = newTopology("testManyFiles");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();

        // build expected results
        // net one tuples per file
        List<List<String>> expResults = buildExpResults(lines, tuple -> true);

        // in this config files are create very fast hence they end
        // up exercising the _<n> suffix to basePath_YYYYMMDD_HHMMSS
        
        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),  // no extra flush
                FileWriterCycleConfig.newCountBasedConfig(1), // yield one line per file
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testManyFilesSlow() throws Exception {
        Topology t = newTopology("testManyFilesSlow");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();

        // build expected results
        // net one tuples per file
        List<List<String>> expResults = buildExpResults(lines, tuple -> true);
        
        // add delay so we get different files w/o a _<n> suffix
        
        int throttleSec = 2;
        TStream<String> s = PlumbingStreams.blockingThrottle(
                t.strings(lines), throttleSec, TimeUnit.SECONDS);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),  // no extra flush
                FileWriterCycleConfig.newCountBasedConfig(1), // yield one line per file
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, (lines.length*throttleSec)+TMO_SEC,
                basePath, expResults);
    }

    @Test
    public void testRetainCntBased() throws Exception {
        // more lines than configured retained numFiles; only keep the last numFiles
        Topology t = newTopology("testRetainCntBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();
        
        // build expected results
        // net one tuples per file
        List<List<String>> expResults = buildExpResults(lines, tuple -> true);
        int keepCnt = 2;  // only keep the last n files
        for (int i = 0; i < keepCnt; i++)
            expResults.remove(0);
        assertEquals(keepCnt, expResults.size());
        
        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newCountBasedConfig(1),
                FileWriterRetentionConfig.newFileCountBasedConfig(keepCnt)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);
        
        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testRetainAggSizeBased() throws Exception {
        // more aggsize than configured; only keep aggsize worth
        Topology t = newTopology("testRetainAggSizeBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();
        
        // build expected results
        // net one tuple per file
        List<List<String>> expResults = buildExpResults(lines, tuple -> true);
        // agg size only enough for last two lines
        long aggregateFileSize = 2 * (("1-"+getStr()).getBytes(StandardCharsets.UTF_8).length + 1/*eol*/);
        expResults.remove(0);
        expResults.remove(0);
        assertEquals(2, expResults.size());
        
        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newCountBasedConfig(1),
                FileWriterRetentionConfig.newAggregateFileSizeBasedConfig(aggregateFileSize)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);
        
        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testRetainAgeBased() throws Exception {
        Topology t = newTopology("testRetainAgeBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();
        
        // build expected results
        int keepCnt = 2;  // only keep the last n files with throttling, age,
                          // and TMO_SEC
        int ageSec = 5;
        long periodMsec = TimeUnit.SECONDS.toMillis(1);
        // net one tuple per file
        List<List<String>> expResults = buildExpResults(lines, tuple -> true);
        for (int i = 0; i < keepCnt; i++)
            expResults.remove(0);
        assertEquals(keepCnt, expResults.size());
        
        // add delay so we can age out things
        //
        // After several runs this test seems reliable but
        // I suspect it may be fragile wrt timing hence the results.
        //
        // With 4 tuples, throttleDelay=2sec, and ageSec=5
        // t0=add-f1, t1, t2=add-f2, t3, t4=add-f3, t5-rm-f1, t6=add-f4, t7=rm-f2, t8, t9=rm-f3, ...
        //
        // So we want to check somewhere around t8 (after t7 and definitely before t9)
        // so all 4 files were created and the first 2 have been aged out.
        // with complete delay = #files-1*throttle + TMO_SEC, should be 6+2 == t8.
        
        int throttleSec = 2;
        TStream<String> s = PlumbingStreams.blockingThrottle(
                t.strings(lines), throttleSec, TimeUnit.SECONDS);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newCountBasedConfig(1),
                FileWriterRetentionConfig.newAgeBasedConfig(ageSec, periodMsec)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);
        
        completeAndValidateWriter(t, ((lines.length-1)*throttleSec)+TMO_SEC,
                basePath, expResults);
    }

    @Test
    public void testFlushImplicit() throws Exception {
        Topology t = newTopology("testFlushImplicit");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();

        // build expected results
        // net all in one, the first, file
        List<List<String>> expResults = buildExpResults(lines, tuple -> false);

        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newCountBasedConfig(1000),
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testFlushCntBased() throws Exception {
        Topology t = newTopology("testFlushCntBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();

        // build expected results
        // net all in one, the first, file
        List<List<String>> expResults = buildExpResults(lines, tuple -> false);

        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newCountBasedConfig(1),  // every tuple
                FileWriterCycleConfig.newCountBasedConfig(1000),  // all in 1 file
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testFlushTimeBased() throws Exception {
        Topology t = newTopology("testFlushTimeBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();

        // build expected results
        // net all in one, the first, file
        List<List<String>> expResults = buildExpResults(lines, tuple -> false);
        
        // add delay so time flush happens
        
        int throttleSec = 1;
        TStream<String> s = PlumbingStreams.blockingThrottle(
                t.strings(lines), throttleSec, TimeUnit.SECONDS);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newTimeBasedConfig(TimeUnit.MILLISECONDS.toMillis(250)),
                FileWriterCycleConfig.newCountBasedConfig(1000),  // all in 1 file
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, (lines.length*throttleSec)+TMO_SEC,
                basePath, expResults);
    }

    @Test
    public void testFlushTupleBased() throws Exception {
        Topology t = newTopology("testFlushTupleBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();

        // build expected results
        // net all in one, the first, file
        List<List<String>> expResults = buildExpResults(lines, tuple -> false);

        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newPredicateBasedConfig(
                        tuple -> tuple.startsWith("1-") || tuple.startsWith("3-")),
                FileWriterCycleConfig.newCountBasedConfig(1000),  // all in 1 file
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testCycleCntBased() throws Exception {
        Topology t = newTopology("testCycleCntBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();
        
        // build expected results
        // net two tuples per file
        int cntTuples = 2;
        AtomicInteger cnt = new AtomicInteger();
        Predicate<String> cycleIt = tuple -> cnt.incrementAndGet() % cntTuples == 0;
        List<List<String>> expResults = buildExpResults(lines, cycleIt);
        assertEquals(lines.length / cntTuples, expResults.size());

        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newCountBasedConfig(cntTuples),
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testCycleSizeBased() throws Exception {
        Topology t = newTopology("testCycleSizeBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();
        
        // build expected results
        // net one tuple per file 
        List<List<String>> expResults = buildExpResults(lines, tuple -> true);
        int fileSize = 2;

        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newFileSizeBasedConfig(fileSize),
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testCycleTimeBased() throws Exception {
        Topology t = newTopology("testCycleTimeBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();
        
        // build expected results
        // net one tuple per file with 250msec cycle config and 1 throttle
        List<List<String>> expResults = buildExpResults(lines, tuple -> true);
        
        // add delay so time cycle happens
        // also verifies only cycle if there's something to cycle
        // (i.e., these cycles happen faster than tuples are written)
        
        int throttleSec = 1;
        TStream<String> s = PlumbingStreams.blockingThrottle(
                t.strings(lines), throttleSec, TimeUnit.SECONDS);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newTimeBasedConfig(TimeUnit.MILLISECONDS.toMillis(250)),
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, (lines.length*throttleSec)+TMO_SEC,
                basePath, expResults);
    }

    @Test
    public void testCycleTupleBased() throws Exception {
        Topology t = newTopology("testCycleTupleBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();

        // build expected results
        // a tuple based config / predicate.  in this case should end up with 3 files.
        Predicate<String> cycleIt = tuple -> tuple.startsWith("1-") || tuple.startsWith("3-");
        List<List<String>> expResults = buildExpResults(lines, cycleIt);
        assertEquals(3, expResults.size());

        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newPredicateBasedConfig(cycleIt),
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testAllTimeBased() throws Exception {
        // exercise case with multiple timer based policies
        Topology t = newTopology("testAllTimeBased");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();
        
        // build expected results
        // keep all given age and TMO_SEC
        int ageSec = 10;
        long periodMsec = TimeUnit.SECONDS.toMillis(1);
        // net one tuple per file
        List<List<String>> expResults = buildExpResults(lines, tuple -> true);
        
        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newTimeBasedConfig(TimeUnit.MILLISECONDS.toMillis(250)),
                FileWriterCycleConfig.newConfig(1, 2000, TimeUnit.SECONDS.toMillis(10), null),
                FileWriterRetentionConfig.newAgeBasedConfig(ageSec, periodMsec)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);
        
        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    @Test
    public void testWriterWatcherReader() throws Exception {
        // verify all the pieces work together
        Topology t = newTopology("testWriterWatcherReader");
        
        String testDirPrefix = "testWriterWatcherReader";
        Path dir = Files.createTempDirectory(testDirPrefix);
        Path basePath = dir.resolve("writerCreated");
        
        String[] lines = getLines();

        System.out.println("########## "+t.getName());
        
        // Write the files
        // add delay so watcher starts first and gets to "see" file additions
        int throttleSec = 2;
        TStream<String> contents = PlumbingStreams.blockingOneShotDelay(
                t.strings(lines), 2, TimeUnit.SECONDS);
        contents = PlumbingStreams.blockingThrottle(
                contents, throttleSec, TimeUnit.SECONDS);
        
        IFileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newCountBasedConfig(1),  // one per file
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(contents, () -> basePath.toString(), () -> policy);
        
        // Watch and read contents
        TStream<String> pathnames = FileStreams.directoryWatcher(t,
                () -> dir.toAbsolutePath().toString())
          .peek(tuple -> System.out.println(new Date() + " watcher added "+tuple))
          .peek(tuple -> { if (new File(tuple).getName().startsWith("."))
            throw new RuntimeException("Not filtering active/hidden files "+tuple); });
        TStream<String> readContents = FileStreams.textFileReader(pathnames);

        boolean dump = true;
        try {
            completeAndValidate("", t, readContents,
                    (lines.length*throttleSec)+TMO_SEC+3/*on-the-edge*/, lines);
            dump = false;
        }
        finally {
            deleteDirAndFiles(dir, testDirPrefix, dump);
        }
    }

    @Test
    public void testCompressedFileWriterPolicy() throws Exception {
        Topology t = newTopology("testCompressedFileWriterPolicy");
        
        // establish a base path
        Path basePath = createTempFile("test1", "txt", new String[0]);
        
        String[] lines = getLines();
        
        // build expected results
        // net 2 tuples per file
        int cntTuples = 2;
        AtomicInteger cnt = new AtomicInteger();
        Predicate<String> cycleIt = tuple -> cnt.incrementAndGet() % cntTuples == 0;
        List<List<String>> expResults = buildExpResults(lines, cycleIt);
        assertEquals(lines.length / cntTuples, expResults.size());

        TStream<String> s = t.strings(lines);
        
        IFileWriterPolicy<String> policy = new CompressedFileWriterPolicy<String>(
                FileWriterFlushConfig.newImplicitConfig(),
                FileWriterCycleConfig.newCountBasedConfig(cntTuples),
                FileWriterRetentionConfig.newFileCountBasedConfig(10)
                );
        FileStreams.textFileWriter(s, () -> basePath.toString(), () -> policy);

        completeAndValidateWriter(t, TMO_SEC, basePath, expResults);
    }

    private void deleteDirAndFiles(Path dir, String dirPrefix, boolean dump) {
        // exercise caution before removing all files in dir
        if (!dirPrefix.startsWith("test"))
            throw new IllegalStateException("Yikes. dir:"+dir+" dirPrefix:"+dirPrefix);
        String leaf = dir.getFileName().toString();
        if (!leaf.startsWith(dirPrefix))
            throw new IllegalStateException("Yikes. dir:"+dir+" dirPrefix:"+dirPrefix);
        
        // Ok, delete all the files in the dir and then the dir
        for (File file : dir.toFile().listFiles()) {
            if (dump)
                dumpFile(file);
            file.delete();
        }
        dir.toFile().delete();
    }
    
    private void dumpFile(File f) {
        System.out.println("<<<<< Dumping "+f);
        try {
            Path path = f.toPath();
            try (BufferedReader br = Files.newBufferedReader(path)) {
                br.lines().forEach(line -> System.out.println(line));
            }
        }
        catch (Exception e) {
            System.out.println("##### exception: " + e.getLocalizedMessage());
        }
        System.out.println(">>>>> DONE "+f);
    }
    
    private <T> List<List<T>> buildExpResults(T[] tuples, Predicate<T> cycleIt) {
        List<List<T>> expResults = new ArrayList<>();
        List<T> oneFile = null;
        for (T tuple : tuples) {
            if (oneFile==null) {
                oneFile = new ArrayList<>();
                expResults.add(oneFile);
            }
            oneFile.add(tuple);
            if (cycleIt.test(tuple))
                oneFile = null;
        }
        return expResults;
    }
    
    private <T> void completeAndValidateWriter(Topology t, int tmoSec,
            Path basePath, List<List<T>> expResults) throws Exception {
        
        try {
            // just let it run to the tmo before we check the file contents
            Condition<Object> tc = new Condition<Object>() {
                public boolean valid() { return false; }
                public Object getResult() { return null; }
            };
            
            complete(t, tc, tmoSec, TimeUnit.SECONDS);

            System.out.println("########## "+t.getName());

            // right number of files?
            List<Path> actFiles = getActFiles(basePath);
            System.out.println("actFiles: "+actFiles);
            assertEquals(actFiles.toString(), expResults.size(), actFiles.size());
            
            // do the file(s) have the right contents?
            System.out.println("expResults: "+expResults);
            int i = 0;
            for (List<T> expFile : expResults) {
                Path path = actFiles.get(i++);
                checkContents(path, expFile.toArray(new String[0]));
            }
        }
        finally {
            deleteAll(basePath);
        }
    }
    
    private void deleteAll(Path basePath) {
        Path parent = basePath.getParent();
        String baseLeaf = basePath.getFileName().toString();
        String[] actLeafs = parent.toFile().list(
                (dir,leaf) -> leaf.startsWith(baseLeaf));
        for (String leaf : actLeafs) {
            parent.resolve(leaf).toFile().delete();
        }
    }
    
    private List<Path> getActFiles(Path basePath) {
        List<Path> paths = new ArrayList<>();
        Path parent = basePath.getParent();
        String baseLeaf = basePath.getFileName().toString();
        String[] actLeafs = parent.toFile().list(
                (dir,leaf) -> leaf.startsWith(baseLeaf+"_"));
        Arrays.sort(actLeafs, (o1,o2) -> o1.compareTo(o2));
        for (String leaf : actLeafs) {
            paths.add(parent.resolve(leaf));
        }
        return paths;
    }
    
    private void checkContents(Path path, String[] lines) {
        if (path.getFileName().toString().endsWith(".zip")) {
          checkZipContents(path, lines);
          return;
        }
        System.out.println("checking file "+path);
        int lineCnt = 0;
        try (BufferedReader br = Files.newBufferedReader(path)) {
            for (String line : lines) {
                ++lineCnt;
                String actLine = br.readLine();
                assertEquals("path:"+path+" line "+lineCnt, line, actLine);
            }
            assertNull("path:"+path+" line "+lineCnt+" expected EOF", br.readLine());
        }
        catch (IOException e) {
            assertNull("path:"+path+" line "+lineCnt+" unexpected IOException "+e, e);
        }
    }
    
    private void checkZipContents(Path path, String[] lines) {
        System.out.println("checking file "+path);
        String fileName = path.getFileName().toString();
        String entryName = fileName.substring(0, fileName.length() - ".zip".length());
        int lineCnt = 0;
        try (
            FileInputStream fis = new FileInputStream(path.toFile());
            ZipInputStream zin = new ZipInputStream((new BufferedInputStream(fis)));
            )
        {
          ZipEntry entry = zin.getNextEntry();
          
          assertEquals(entryName, entry.getName());

          BufferedReader br = new BufferedReader(new InputStreamReader(zin, StandardCharsets.UTF_8));
          for (String line : lines) {
            ++lineCnt;
            String actLine = br.readLine();
            assertEquals("path:"+path+" line "+lineCnt, line, actLine);
          }
          assertNull("path:"+path+" line "+lineCnt+" expected EOF", br.readLine());
        }
        catch (IOException e) {
          assertNull("path:"+path+" line "+lineCnt+" unexpected IOException "+e, e);
        }
    }
}