package org.cobbzilla.util.io;

import lombok.extern.slf4j.Slf4j;
import org.cobbzilla.util.system.Bytes;

import java.io.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static org.cobbzilla.util.daemon.ZillaRuntime.*;
import static org.cobbzilla.util.reflect.ReflectionUtil.closeQuietly;
import static org.cobbzilla.util.reflect.ReflectionUtil.instantiate;

@Slf4j
public class FilterInputStreamViaOutputStream extends PipedInputStream implements Runnable {

    private static final int DEFAULT_PIPE_BUFFER_SIZE = (int) (64 * Bytes.KB);
    private static final long THREAD_TERMINATE_TIMEOUT = TimeUnit.SECONDS.toMillis(10);

    private InputStream in;
    private PipedOutputStream pipeOut;
    private OutputStream out;
    private Thread thread;

    public FilterInputStreamViaOutputStream(InputStream in, Class<? extends OutputStream> outStreamClass) {
        super(DEFAULT_PIPE_BUFFER_SIZE);
        this.in = in;
        try {
            this.pipeOut = new PipedOutputStream(this);
        } catch (Exception e) {
            die("FilterInputStreamViaOutputStream: error creating pipeOut: "+shortError(e));
        }
        this.out = instantiate(outStreamClass, this.pipeOut);
        start();
    }

    public FilterInputStreamViaOutputStream(InputStream in, Function<OutputStream, OutputStream> outFactory) {
        super(DEFAULT_PIPE_BUFFER_SIZE);
        this.in = in;
        try {
            this.pipeOut = new PipedOutputStream(this);
        } catch (Exception e) {
            die("FilterInputStreamViaOutputStream: error creating pipeOut: "+shortError(e));
        }
        try {
            this.out = outFactory.apply(this.pipeOut);
        } catch (Exception e) {
            die("FilterInputStreamViaOutputStream: error creating out: "+shortError(e));
        }
        start();
    }

    private void start() {
        this.thread = new Thread(this);
        this.thread.setName(getClass().getSimpleName()+"/"+System.identityHashCode(this));
        this.thread.setDaemon(true);
        this.thread.start();
    }

    @Override public void run() {
        try {
            int c;
            int counter = 0;
            while((c = in.read()) >= 0) {
                out.write(c);
                counter++;
            }
            out.flush();

        } catch (IOException e) {
            log.error("run: error copying bytes: "+shortError(e));
            throw new RuntimeException(e);

        } finally {
            closeQuietly(out);
        }
    }

    @Override public void close() {
        log.info("close called from "+stacktrace());
        try {
            super.close();
        } catch (Exception e) {
            log.warn("close: error closing pipeIn: "+shortError(e));
        }
        closeQuietly(in);
        closeQuietly(out);
        closeQuietly(pipeOut);
        if (this.thread.isAlive()) {
            background(() -> terminate(this.thread, THREAD_TERMINATE_TIMEOUT));
        }
    }
}

@@ -569,16 +569,94 @@
 es.KB);%0A
+    private static final int DEFAULT_COPY_BUFFER_SIZE = (int) (8 * Bytes.KB);%0A
     priv
@@ -2217,21 +2217,69 @@
         
+f
 in
-t c
+al byte%5B%5D buf = new byte%5BDEFAULT_COPY_BUFFER_SIZE%5D
 ;%0A      
@@ -2292,19 +2292,17 @@
 int 
-counter = 0
+bytesRead
 ;%0A  
@@ -2320,11 +2320,20 @@
 hile
+ 
 ((
-c
+bytesRead
  = i
@@ -2339,16 +2339,19 @@
 in.read(
+buf
 )) %3E= 0)
@@ -2383,37 +2383,26 @@
 ite(
-c);%0A                counter++
+buf, 0, bytesRead)
 ;%0A  
@@ -2489,18 +2489,27 @@
     
-log.error(
+final String msg = 
 %22run
@@ -2532,17 +2532,19 @@
 bytes: %22
-+
+ + 
 shortErr
@@ -2540,32 +2540,59 @@
  + shortError(e)
+;%0A            log.error(msg
 );%0A            t
@@ -2617,16 +2617,21 @@
 ception(
+msg, 
 e);%0A%0A   

use buffer to copy