new: IOEvent.Move

change: IOStreams#pipe
new: IOUtils#close(closeable, EventListener)

assira/src/main/java/net/ranides/assira/io/IOEvent.java

@@ -1,191 +1,215 @@
-/*
- * @author Ranides Atterwim <ranides@gmail.com>
- * @copyright Ranides Atterwim
- * @license WTFPL
- * @url http://ranides.net/projects/assira
- */
-
-package net.ranides.assira.io;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.nio.ByteBuffer;
-import net.ranides.assira.events.Event;
-import net.ranides.assira.events.Events;
-
-/**
- *
- * @author Ranides Atterwim <ranides@gmail.com>
- */
-public interface IOEvent extends Event {
-    
-    static Flush flush(OutputStream source) {
-        return new Flush(source);
-    }
-    
-    static Reset reset(InputStream source) {
-        return new Reset(source);
-    }
-    
-    static Read read(InputStream source, int data) {
-        if(data == -1) {
-            return new Read(source);
-        } else {
-            return new Read(source, new byte[]{(byte)data}, 0, 1);
-        }
-    }
-    
-    static Read read(InputStream source, byte[] data, int offset, int length) {
-        return new Read(source, data, offset, length);
-    }
-    
-    static Read read(InputStream source, byte[] data) {
-        return new Read(source, data, 0, data.length);
-    }
-    
-    static Write write(OutputStream source, int data) {
-        return new Write(source, new byte[]{(byte)data}, 0, 1);
-    }
-    
-    static Write write(OutputStream source, byte[] data, int offset, int length) {
-        return new Write(source, data, offset, length);
-    }
-    
-    static Write write(OutputStream source, byte[] data) {
-        return new Write(source, data, 0, data.length);
-    }
-    
-    static Close close(Closeable source) {
-        return new Close(source);
-    }
-    
-    static Skip skip(InputStream source, long count) {
-        return new Skip(source, count);
-    }
-    
-    static Failure failure(IOException cause) {
-        return new Failure(cause);
-    }
-    
-    class Reset implements IOEvent {
-        
-        private final InputStream source;
-
-        protected Reset(InputStream source) {
-            this.source = source;
-        }
-
-        public InputStream source() {
-            return source;
-        }
-    }
-    
-    class Flush implements IOEvent {
-        
-        private final OutputStream source;
-
-        protected Flush(OutputStream source) {
-            this.source = source;
-        }
-
-        public OutputStream source() {
-            return source;
-        }
-    }
-    
-    class Read implements IOEvent {
-        
-        private static final ByteBuffer NONE = ByteBuffer.wrap(new byte[0]);
-        
-        private final InputStream source;
-        private final ByteBuffer data;
-        
-        protected Read(InputStream source) {
-            this.source = source;
-            this.data = NONE;
-        }
-        
-        protected Read(InputStream source, byte[] data, int offset, int length) {
-            this.source = source;
-            this.data = ByteBuffer.wrap(data, offset, length);
-        }
-        
-        public InputStream source() {
-            return source;
-        }
-        
-        public ByteBuffer data() {
-            return data;
-        }
-        
-    }
-    
-    class Write implements IOEvent {
-        
-        private final OutputStream source;
-        private final ByteBuffer data;
-        
-        protected Write(OutputStream source, byte[] data, int offset, int length) {
-            this.source = source;
-            this.data = ByteBuffer.wrap(data, offset, length);
-        }
-        
-        public OutputStream source() {
-            return source;
-        }
-        
-        public ByteBuffer data() {
-            return data;
-        }
-        
-    }
-
-    class Close implements IOEvent {
-        
-        private final Closeable source;
-
-        protected Close(Closeable source) {
-            this.source = source;
-        }
-
-        public Closeable source() {
-            return source;
-        }
-        
-    }
-    
-    class Skip implements IOEvent {
-        
-        private final InputStream source;
-        private final long count;
-
-        protected Skip(InputStream source, long count) {
-            this.source = source;
-            this.count = count;
-        }
-
-        public InputStream source() {
-            return source;
-        }
-
-        public long count() {
-            return count;
-        }
-        
-    }
-    
-    class Failure extends Events.Failure implements IOEvent {
-		
-        protected Failure(Throwable cause) {
-            super(cause);
-        }
-        
-        @Override
-        public IOException cause() {
-            return (IOException)super.cause();
-        }
-        
-    }
-}
+/*
+ * @author Ranides Atterwim <ranides@gmail.com>
+ * @copyright Ranides Atterwim
+ * @license WTFPL
+ * @url http://ranides.net/projects/assira
+ */
+
+package net.ranides.assira.io;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import net.ranides.assira.events.Event;
+import net.ranides.assira.events.Events;
+
+/**
+ *
+ * @author Ranides Atterwim <ranides@gmail.com>
+ */
+public interface IOEvent extends Event {
+    
+    static Flush flush(OutputStream source) {
+        return new Flush(source);
+    }
+    
+    static Reset reset(InputStream source) {
+        return new Reset(source);
+    }
+    
+    static Read read(InputStream source, int data) {
+        if(data == -1) {
+            return new Read(source);
+        } else {
+            return new Read(source, new byte[]{(byte)data}, 0, 1);
+        }
+    }
+    
+    static Read read(InputStream source, byte[] data, int offset, int length) {
+        return new Read(source, data, offset, length);
+    }
+    
+    static Read read(InputStream source, byte[] data) {
+        return new Read(source, data, 0, data.length);
+    }
+    
+    static Write write(OutputStream source, int data) {
+        return new Write(source, new byte[]{(byte)data}, 0, 1);
+    }
+    
+    static Write write(OutputStream source, byte[] data, int offset, int length) {
+        return new Write(source, data, offset, length);
+    }
+    
+    static Write write(OutputStream source, byte[] data) {
+        return new Write(source, data, 0, data.length);
+    }
+    
+    static Close close(Closeable source) {
+        return new Close(source);
+    }
+    
+    static Skip skip(InputStream source, long count) {
+        return new Skip(source, count);
+    }
+    
+    static Move move(InputStream source, long position) {
+        return new Move(source, position);
+    }
+    
+    static Failure failure(IOException cause) {
+        return new Failure(cause);
+    }
+    
+    class Reset implements IOEvent {
+        
+        private final InputStream source;
+
+        protected Reset(InputStream source) {
+            this.source = source;
+        }
+
+        public InputStream source() {
+            return source;
+        }
+    }
+    
+    class Flush implements IOEvent {
+        
+        private final OutputStream source;
+
+        protected Flush(OutputStream source) {
+            this.source = source;
+        }
+
+        public OutputStream source() {
+            return source;
+        }
+    }
+    
+    class Read implements IOEvent {
+        
+        private static final ByteBuffer NONE = ByteBuffer.wrap(new byte[0]);
+        
+        private final InputStream source;
+        private final ByteBuffer data;
+        
+        protected Read(InputStream source) {
+            this.source = source;
+            this.data = NONE;
+        }
+        
+        protected Read(InputStream source, byte[] data, int offset, int length) {
+            this.source = source;
+            this.data = ByteBuffer.wrap(data, offset, length);
+        }
+        
+        public InputStream source() {
+            return source;
+        }
+        
+        public ByteBuffer data() {
+            return data;
+        }
+        
+    }
+    
+    class Write implements IOEvent {
+        
+        private final OutputStream source;
+        private final ByteBuffer data;
+        
+        protected Write(OutputStream source, byte[] data, int offset, int length) {
+            this.source = source;
+            this.data = ByteBuffer.wrap(data, offset, length);
+        }
+        
+        public OutputStream source() {
+            return source;
+        }
+        
+        public ByteBuffer data() {
+            return data;
+        }
+        
+    }
+
+    class Close implements IOEvent {
+        
+        private final Closeable source;
+
+        protected Close(Closeable source) {
+            this.source = source;
+        }
+
+        public Closeable source() {
+            return source;
+        }
+        
+    }
+    
+    class Skip implements IOEvent {
+        
+        private final InputStream source;
+        private final long count;
+
+        protected Skip(InputStream source, long count) {
+            this.source = source;
+            this.count = count;
+        }
+
+        public InputStream source() {
+            return source;
+        }
+
+        public long count() {
+            return count;
+        }
+        
+    }
+    
+    class Move implements IOEvent {
+        
+        private final InputStream source;
+        private final long position;
+
+        protected Move(InputStream source, long position) {
+            this.source = source;
+            this.position = position;
+        }
+
+        public InputStream source() {
+            return source;
+        }
+
+        public long position() {
+            return position;
+        }
+        
+    }
+    
+    class Failure extends Events.Failure implements IOEvent {
+		
+        protected Failure(Throwable cause) {
+            super(cause);
+        }
+        
+        @Override
+        public IOException cause() {
+            return (IOException)super.cause();
+        }
+        
+    }
+}

assira/src/main/java/net/ranides/assira/io/IOStreams.java

@@ -23,11 +23,6 @@ import java.nio.charset.CharsetDecoder;
 import java.nio.charset.CharsetEncoder;
 import java.nio.charset.CoderResult;
 import java.util.Arrays;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import net.ranides.assira.events.EventListener;
 import net.ranides.assira.generic.CompareUtils;
@@ -139,12 +134,44 @@ public final class IOStreams {
         }
     }
     
-    public static Future<Long> pipe(InputStream input, OutputStream output) {
-        return pipe(input, output, e -> LOGGER.error("Exception ignored in #pipe", e));
+    public static Thread pipe(InputStream input, OutputStream output) {
+        return pipe(input, output, (Exception e) -> LOGGER.error("Exception ignored in #pipe", e));
     }
     
-    public static Future<Long> pipe(InputStream input, OutputStream output, Consumer<? super Exception> handler) {
-        return new PipeThread(input, output, handler).start();
+    public static Thread pipe(InputStream input, OutputStream output, Consumer<? super Exception> handler) {
+        Thread thread = new Thread(() -> {
+            try {
+                copy(input, output);
+            } catch(Exception cause) {
+                handler.accept(cause);
+            } finally {
+                IOUtils.close(input, handler);
+                IOUtils.close(output, handler);
+            }
+        });
+        thread.start();
+        return thread;
+    }
+    
+    public static Thread pipe(InputStream input, OutputStream output, EventListener<? super IOEvent> listener) {
+        if(null == listener) {
+            return pipe(input, output);
+        }
+        Thread thread = new Thread(() -> {
+            try {
+                copy(input, output, p -> listener.handleEvent(IOEvent.move(input, p)));
+                listener.handleEvent(IOEvent.flush(output));
+            } catch(IOException cause) {
+                listener.handleEvent(IOEvent.failure(cause));
+            } finally {
+                IOUtils.close(input, listener);
+                IOUtils.close(output, listener);
+                listener.handleEvent(IOEvent.close(input));
+                listener.handleEvent(IOEvent.close(output));
+            }
+        });
+        thread.start();
+        return thread;
     }
     
     private static final class InputObserver extends FilterInputStream {
@@ -606,65 +633,4 @@ public final class IOStreams {
 
     }
 
-    private static final class PipeThread implements Future<Long> {
-        
-        private final AtomicLong counter = new AtomicLong(0);
-
-        private final Thread thread;
-
-        public PipeThread(InputStream input, OutputStream output, Consumer<? super Exception> handler) {
-            this.thread = new Thread(() -> {
-                try {
-                    byte[] buffer = new byte[32];
-                    int n;
-                    while(-1 != (n = input.read(buffer))) {
-                        output.write(buffer, 0, n);
-                        counter.addAndGet(n);
-                    }
-                } catch (IOException cause) {
-                    handler.accept(cause);
-                } finally {
-                    IOUtils.close(output, handler);
-                    IOUtils.close(input, handler);
-                }
-            });
-        }
-        
-        public PipeThread start() {
-            thread.start();
-            return this;
-        }
-
-        @Override
-        public boolean cancel(boolean mayInterruptIfRunning) {
-            if(!mayInterruptIfRunning || !thread.isAlive() || !thread.isInterrupted()) {
-                return false;
-            }
-            thread.interrupt();
-            return true;
-        }
-
-        @Override
-        public boolean isCancelled() {
-            return thread.isInterrupted();
-        }
-
-        @Override
-        public boolean isDone() {
-            return !thread.isAlive();
-        }
-
-        @Override
-        public Long get() throws InterruptedException, ExecutionException {
-            thread.join();
-            return counter.get();
-        }
-
-        @Override
-        public Long get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
-            thread.join(unit.toMillis(timeout));
-            return counter.get();
-        }
-        
-    }
 }

assira/src/main/java/net/ranides/assira/io/IOUtils.java

@@ -9,6 +9,7 @@ package net.ranides.assira.io;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.function.Consumer;
+import net.ranides.assira.events.EventListener;
 import net.ranides.assira.trace.ExceptionUtils;
 import net.ranides.assira.trace.LoggerUtils;
 
@@ -40,6 +41,19 @@ public final class IOUtils {
         return false;
     }
     
+    public static boolean close(Closeable object, EventListener<? super IOEvent.Failure> listener) {
+        try {
+            return close(object);
+        } catch(IOException cause) {
+            if(null != listener) {
+                listener.handleEvent(IOEvent.failure(cause));
+            } else {
+                LOGGER.warn("Exception ignored on #close", cause);
+            }
+            return false;
+        }
+    }
+    
     public static boolean close(Object object) throws Exception {
         if(object instanceof Closeable) {
             ((Closeable)object).close();

assira/src/test/java/net/ranides/assira/io/IOUtilsTest.java

@@ -87,7 +87,7 @@ public class IOUtilsTest {
         IOUtils.close(new SFile(), listener.consumer(Events::failure));
         
         // consumer
-        IOUtils.close(new CFile("NC"), errors::add);
+        IOUtils.close(new CFile("NC"), (Throwable e) -> errors.add(e));
         
         // real NOP - ignore exception silently
         IOUtils.close(new SFile(), e -> {});