AsyncStream - szkic

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

@@ -9,9 +9,6 @@ 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;
 
 /**
@@ -20,21 +17,136 @@ import net.ranides.assira.events.EventListener;
  */
 public final class AsyncStream {
 
+    // InputDevice      handle(IORequest) -> stream.read
+    //
+    // InputDriver      read -> handle(IORequest)
+    //
+    // InputListener    read -> stream.read -> handle(IOEvent)
+
+
+    // OutputDevice     handle(IORequest) -> stream.write
+    //
+    // OutputDriver     write -> handle(IORequest)
+    //
+    // OutputListener   write -> stream.write -> handle(IOEvent)
+
+    // komentarz: to chyba jest bez sensu, bo stream nie ma zaimplementowanego
+    // interfejsu w taki sposób, żeby działał w tle (np metody powinny zwracać
+    // wyniki).
+
+    // W praktyce InputListener oraz OutputListener mają sens w 100%.
+
+    // OutputDriver oraz OutputDevice mają sens w 75% (obsługa błędów leży)
+    // 1. OutputDriver nie rzuca wyjątków
+    // 2. OutputDevice nie wie, komu przesłać zdarzenia o wyjątkach
+
+    // InputDriver oraz InputDevice mają sens w 25% (obsługa wszystkiego leży)
+    // 3. InputDriver nie rzuca wyjątków
+    // 4. InputDriver nie zwraca sensownych wyników
+    // 5. InputDevice nie wie komu przesłać zdarzenia o wyjątkach
+    // 6. InputDevice nie wie komu zwrócić wyniki
+    // 7. kto ma tworzyć bufor na czytane dane ???
+    //
+    // W zasadzie to działa rozsądnie tylko dla układu:
+    //      InputDriver -> InputDevice -> InputListener -> InputStream
+    //
+    // A poniższy układ do niczego się nie nadaje:
+    //      InputDriver -> InputDevice -> InputStream
+
+    // Może więc tworzenie Device będzie możliwie tylko dla Listener'a?
+    // A dopiero Listener by był tworzony ze stream'a?
+    // To by rozwiązało problemy 2, 5, 6.
+
+    // Problem 7 powinno się rozwiązać alokując pamięć ZAWSZE. W końcu ten, co wywołał
+    // metodę, to chuj wie co będzie z tą tablicą za moment robić...
+
     private AsyncStream() {
         // utility class
     }
 
-    private static abstract class InputAdapter extends InputStream implements EventListener<IOEvent> {
-        
+
+    /**
+     * Uwaga! Wszystkie metody zwracają bezużyteczne wyniki. W zasadzie Bóg jeden
+     * wie po licho w ogóle implementujemy {@link InputStream}, bo i tak tego
+     * nie można nigdzie do normalnych metod przekazać.
+     *
+     * read -> handle(IORequest)
+     */
+    public static abstract class InputDriver extends InputStream implements EventListener<IORequest> {
+
+        public InputDriver() {
+            // do nothing
+        }
+
+        @Override
+        public abstract void handleEvent(IORequest event);
+
+        @Override
+        public int read() throws IOException {
+            handleEvent(IORequest.read());
+            return 0;
+        }
+
+        @Override
+        public int read(byte array[]) throws IOException {
+            handleEvent(IORequest.read(array.length));
+            return 0;
+        }
+
+        @Override
+        public int read(byte array[], int offset, int length) throws IOException {
+            handleEvent(IORequest.read(length));
+            return 0;
+        }
+
+        @Override
+        public long skip(long delta) throws IOException {
+            handleEvent(IORequest.skip(delta));
+            return 0;
+        }
+
+        @Override
+        public int available() throws IOException {
+            return 0;
+        }
+
+        @Override
+        public void close() throws IOException {
+            handleEvent(IORequest.close());
+        }
+
+        @Override
+        public synchronized void mark(int readlimit) {
+            handleEvent(IORequest.mark(readlimit));
+        }
+
+        @Override
+        public synchronized void reset() throws IOException {
+            handleEvent(IORequest.reset());
+        }
+
+        @Override
+        public boolean markSupported() {
+            return true;
+        }
+    }
+
+
+    /**
+     * read -> stream.read -> handle(IOEvent)
+     */
+    public static abstract class InputListener extends InputStream implements EventListener<IOEvent> {
+
         private final InputStream istream;
 
-        public InputAdapter(final InputStream istream) {
+        public InputListener(final InputStream istream) {
             this.istream = istream;
         }
 
         @Override
         public abstract void handleEvent(IOEvent event);
 
+        @Override
         public int read() throws IOException {
             try {
                 int data = istream.read();
@@ -46,6 +158,7 @@ public final class AsyncStream {
             }
         }
 
+        @Override
         public int read(byte array[]) throws IOException {
             try {
                 int delta = istream.read(array);
@@ -56,7 +169,8 @@ public final class AsyncStream {
                 throw cause;
             }
         }
-        
+
+        @Override
         public int read(byte array[], int offset, int length) throws IOException {
             try {
                 int delta = istream.read(array, offset, length);
@@ -68,6 +182,7 @@ public final class AsyncStream {
             }
         }
 
+        @Override
         public long skip(long n) throws IOException {
             try {
                 long delta = istream.skip(n);
@@ -79,6 +194,7 @@ public final class AsyncStream {
             }
         }
 
+        @Override
         public int available() throws IOException {
             try {
                 return istream.available();
@@ -88,6 +204,7 @@ public final class AsyncStream {
             }
         }
 
+        @Override
         public void close() throws IOException {
             try {
                 istream.close();
@@ -98,11 +215,13 @@ public final class AsyncStream {
             }
         }
 
+        @Override
         public synchronized void mark(int readlimit) {
             istream.mark(readlimit);
             handleEvent(IOEvent.mark(readlimit));
         }
 
+        @Override
         public synchronized void reset() throws IOException {
             try {
                 istream.reset();
@@ -113,39 +232,81 @@ public final class AsyncStream {
             }
         }
 
+        @Override
         public boolean markSupported() {
             return istream.markSupported();
         }
     }
-    
+
+
+    /**
+     * write -> stream.write -> handle(IOEvent)
+     */
     public static abstract class OutputListener extends OutputStream implements EventListener<IOEvent> {
 
+        private final OutputStream ostream;
+
+        public OutputListener(OutputStream istream) {
+            this.ostream = istream;
+        }
+
         @Override
         public abstract void handleEvent(IOEvent event);
 
         @Override
         public void write(int value) throws IOException {
-            handleEvent(IOEvent.write(value));
+            try {
+                ostream.write(value);
+                handleEvent(IOEvent.write(value));
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
         }
 
         @Override
         public void write(byte array[]) throws IOException {
-            handleEvent(IOEvent.write(array));
+            try {
+                ostream.write(array);
+                handleEvent(IOEvent.write(array));
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
         }
 
         @Override
         public void write(byte array[], int offset, int length) throws IOException {
-            handleEvent(IOEvent.write(array, offset, length));
+            try {
+                ostream.write(array, offset, length);
+                handleEvent(IOEvent.write(array, offset, length));
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
         }
 
         @Override
         public void flush() throws IOException {
-            handleEvent(IOEvent.flush());
+            try {
+                ostream.flush();
+                handleEvent(IOEvent.flush());
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
         }
 
         @Override
         public void close() throws IOException {
-            handleEvent(IOEvent.close());
+            try {
+                ostream.close();
+                handleEvent(IOEvent.close());
+            } catch(IOException cause) {
+                handleEvent(IOEvent.failure(cause));
+                throw cause;
+            }
         }
     }
+
 }

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

@@ -0,0 +1,271 @@
+/*
+ * @author Ranides Atterwim <ranides@gmail.com>
+ * @copyright Ranides Atterwim
+ * @license WTFPL
+ * @url http://ranides.net/projects/assira
+ */
+package net.ranides.assira.io;
+
+import edu.umd.cs.findbugs.annotations.SuppressWarnings;
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Arrays;
+import net.ranides.assira.events.Event;
+
+/**
+ *
+ * @author ranides
+ */
+@SuppressWarnings({
+    "EI_EXPOSE_REP",
+    "PMD"
+})
+public abstract class IORequest implements Event, Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private static final class Li { // NOPMD lazy init idiom
+        static final Close CLOSE = new Close();
+        static final Flush FLUSH = new Flush();
+        static final Reset RESET = new Reset();
+    }
+
+    public static Close close() {
+        return Li.CLOSE;
+    }
+
+    public static Flush flush() {
+        return Li.FLUSH;
+    }
+
+    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 ReadByte read() {
+        return new ReadByte();
+    }
+
+    public static ReadByteArray read(int count) {
+        return new ReadByteArray(count);
+    }
+
+    @Deprecated
+    public static ReadByteBuffer read(byte[] buffer, int offset, int count) {
+        return new ReadByteBuffer(buffer, offset, count);
+    }
+
+    public static WriteByte write(int value) {
+        return new WriteByte(value);
+    }
+
+    public static WriteByteArray write(byte[] array) {
+        return new WriteByteArray(Arrays.copyOf(array, array.length));
+    }
+
+    public static WriteByteArray write(byte[] array, int offset, int length) {
+        return new WriteByteArray(Arrays.copyOfRange(array, offset, offset + length));
+    }
+
+/* ************************************************************************** */
+
+
+    public static class Close extends IORequest {
+
+        private static final long serialVersionUID = 1L;
+
+        protected Close() {
+            // do nothing
+        }
+
+        @Override
+        public String toString() {
+            return "IORequest.Close";
+        }
+    }
+
+    public static class Flush extends IORequest {
+
+        private static final long serialVersionUID = 1L;
+
+        protected Flush() {
+            // do nothing
+        }
+
+        @Override
+        public String toString() {
+            return "IORequest.Flush";
+        }
+    }
+
+    public static class Skip extends IORequest {
+
+        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 "IORequest.Skip";
+        }
+    }
+
+    public static class Reset extends IORequest {
+
+        private static final long serialVersionUID = 1L;
+
+        protected Reset() {
+            // do nothing
+        }
+
+        @Override
+        public String toString() {
+            return "IORequest.Reset";
+        }
+    }
+
+    public static class Mark extends IORequest {
+
+        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 "IORequest.Mark";
+        }
+    }
+
+    public static class ReadByte extends IORequest {
+
+        private static final long serialVersionUID = 1L;
+
+        protected ReadByte() {
+            // do nothing
+        }
+
+        @Override
+        public String toString() {
+            return "IORequest.ReadByte";
+        }
+    }
+
+    public static class ReadByteArray extends IORequest {
+
+        private static final long serialVersionUID = 1L;
+
+        private final int count;
+
+        protected ReadByteArray(int count) {
+            this.count = count;
+        }
+
+        public final int count() {
+            return count;
+        }
+
+        @Override
+        public String toString() {
+            return "IORequest.ReadByteArray: " + count;
+        }
+    }
+
+    @Deprecated
+    public static class ReadByteBuffer extends IORequest {
+
+        private static final long serialVersionUID = 1L;
+
+        private final byte[] buffer;
+        private final int offset;
+        private final int count;
+
+        protected ReadByteBuffer(byte[] buffer, int offset, int count) {
+            this.buffer = buffer;
+            this.offset = offset;
+            this.count = count;
+        }
+
+        public byte[] buffer() {
+            return buffer;
+        }
+
+        public int offset() {
+            return offset;
+        }
+
+        public final int count() {
+            return count;
+        }
+
+        @Override
+        public String toString() {
+            return "IORequest.ReadByteBuffer: " + count;
+        }
+    }
+
+    public static class WriteByte extends IORequest {
+
+        private static final long serialVersionUID = 1L;
+
+        private final int data;
+
+        protected WriteByte(int data) {
+            this.data = data;
+        }
+
+        public final int data() {
+            return data;
+        }
+
+        @Override
+        public String toString() {
+            return "IORequest.WriteByte: " + data;
+        }
+    }
+
+    public static class WriteByteArray extends IORequest {
+
+        private static final long serialVersionUID = 1L;
+
+        private final byte[] data;
+
+        protected WriteByteArray(byte[] data) {
+            this.data = data;
+        }
+
+        public byte[] data() {
+            return data;
+        }
+
+        @Override
+        public String toString() {
+            return "IORequest.WriteByteArray: " + data.length;
+        }
+    }
+
+}