AsyncStream - draft

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

@@ -0,0 +1,151 @@
+/*
+ * @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.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.TreeMap;
+import net.ranides.assira.events.EventListener;
+
+/**
+ *
+ * @author ranides
+ */
+public final class AsyncStream {
+
+    private AsyncStream() {
+        // utility class
+    }
+
+    private static abstract class InputAdapter extends InputStream implements EventListener<IOEvent> {
+        
+        private final InputStream istream;
+
+        public InputAdapter(final InputStream istream) {
+            this.istream = istream;
+        }
+
+        @Override
+        public abstract void handleEvent(IOEvent event);
+
+        public int read() throws IOException {
+            try {
+                int data = istream.read();
+                handleEvent(IOEvent.read(data));
+                return data;
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
+        }
+
+        public int read(byte array[]) throws IOException {
+            try {
+                int delta = istream.read(array);
+                handleEvent(IOEvent.read(array, 0, delta));
+                return delta;
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
+        }
+        
+        public int read(byte array[], int offset, int length) throws IOException {
+            try {
+                int delta = istream.read(array, offset, length);
+                handleEvent(IOEvent.read(array, offset, delta));
+                return delta;
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
+        }
+
+        public long skip(long n) throws IOException {
+            try {
+                long delta = istream.skip(n);
+                handleEvent(IOEvent.skip(delta));
+                return delta;
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
+        }
+
+        public int available() throws IOException {
+            try {
+                return istream.available();
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
+        }
+
+        public void close() throws IOException {
+            try {
+                istream.close();
+                handleEvent(IOEvent.close());
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
+        }
+
+        public synchronized void mark(int readlimit) {
+            istream.mark(readlimit);
+            handleEvent(IOEvent.mark(readlimit));
+        }
+
+        public synchronized void reset() throws IOException {
+            try {
+                istream.reset();
+                handleEvent(IOEvent.reset());
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
+        }
+
+        public boolean markSupported() {
+            return istream.markSupported();
+        }
+    }
+    
+    public static abstract class OutputListener extends OutputStream implements EventListener<IOEvent> {
+
+        @Override
+        public abstract void handleEvent(IOEvent event);
+
+        @Override
+        public void write(int value) throws IOException {
+            handleEvent(IOEvent.write(value));
+        }
+
+        @Override
+        public void write(byte array[]) throws IOException {
+            handleEvent(IOEvent.write(array));
+        }
+
+        @Override
+        public void write(byte array[], int offset, int length) throws IOException {
+            handleEvent(IOEvent.write(array, offset, length));
+        }
+
+        @Override
+        public void flush() throws IOException {
+            handleEvent(IOEvent.flush());
+        }
+
+        @Override
+        public void close() throws IOException {
+            handleEvent(IOEvent.close());
+        }
+    }
+}

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

@@ -29,6 +29,7 @@ public abstract class IOEvent implements Event, Serializable {
         static final Close CLOSE = new Close();
         static final Flush FLUSH = new Flush();
         static final Open OPEN = new Open();
+        static final Reset RESET = new Reset();
     }
 
     public static Close close() {
@@ -42,6 +43,18 @@ public abstract class IOEvent implements Event, Serializable {
     public static Open open() {
         return Li.OPEN;
     }
+    
+    public static Skip skip(long delta) {
+        return new Skip(delta);
+    }
+    
+    public static Reset reset() {
+        return Li.RESET;
+    }
+    
+    public static Mark mark(int limit) {
+        return new Mark(limit);
+    }
 
     public static Failure failure(IOException cause) {
         return new Failure(cause);
@@ -122,6 +135,60 @@ public abstract class IOEvent implements Event, Serializable {
             return "IOEvent.Flush";
         }
     }
+    
+    public static class Skip extends IOEvent {
+
+        private static final long serialVersionUID = 1L;
+        
+        private final long delta;
+
+        protected Skip(long delta) {
+            this.delta = delta;
+        }
+
+        public long delta() {
+            return delta;
+        }
+        
+        @Override
+        public String toString() {
+            return "IOEvent.Skip";
+        }
+    }
+    
+    public static class Reset extends IOEvent {
+
+        private static final long serialVersionUID = 1L;
+        
+        protected Reset() {
+            // do nothing
+        }
+
+        @Override
+        public String toString() {
+            return "IOEvent.Reset";
+        }
+    }
+    
+    public static class Mark extends IOEvent {
+
+        private static final long serialVersionUID = 1L;
+        
+        private final long limit;
+
+        protected Mark(long limit) {
+            this.limit = limit;
+        }
+
+        public long limit() {
+            return limit;
+        }
+        
+        @Override
+        public String toString() {
+            return "IOEvent.Mark";
+        }
+    }
 
     public static class Open extends IOEvent {