wooo - AsyncStream zrobiony, w efekcie OutputStreamThread usunięty

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

@@ -9,7 +9,13 @@ package net.ranides.assira.io;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import net.ranides.assira.annotations.Meta;
 import net.ranides.assira.events.EventListener;
+import net.ranides.assira.events.EventRouter;
+import net.ranides.assira.events.EventRouterThread;
+import net.ranides.assira.events.ReflectEventListener;
+import net.ranides.assira.trace.AdvLogger;
+import net.ranides.assira.trace.LoggerUtils;
 
 /**
  *
@@ -17,123 +23,88 @@ 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
     }
-
-
-    /**
-     * 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;
-        }
+    
+    public static OutputStream create(OutputStream ostream, String logger) {
+        return create(ostream, LoggerUtils.getLogger(logger));
+    }
+    
+    public static OutputStream create(OutputStream ostream, final AdvLogger logger) {
+        return create(ostream, new OutputListener(ostream) {
+            @Override
+            public void handleEvent(IOEvent event) {
+                if(event instanceof IOEvent.Failure) {
+                    logger.error( ((IOEvent.Failure)event).cause() );
+                }
+            }
+        });
+    }
+    
+    public static OutputStream create(OutputStream ostream, final EventListener<? super IOEvent> listener) {
+        return create(new OutputListener(ostream) {
+            @Override
+            public void handleEvent(IOEvent event) {
+                listener.handleEvent(event);
+            }
+        });
+    }
+    
+    public static OutputStream create(OutputListener listener) {
+        
+        final EventRouter router = EventRouterThread.newInstance("OutputStreamer", Integer.MAX_VALUE, Integer.MAX_VALUE);
+        
+        router.addEventListener(IORequest.class, new OutputDevice(listener));
+        
+        return new OutputDriver() {
+            @Override
+            public void handleEvent(IORequest event) {
+                router.handleEvent(event);
+            }
+        };
+    }
+    
+    public static InputStream create(final InputStream ostream, String logger) {
+        return create(ostream, LoggerUtils.getLogger(logger));
+    }
+    
+    public static InputStream create(final InputStream ostream, final AdvLogger logger) {
+        return create(ostream, new InputListener(ostream) {
+            @Override
+            public void handleEvent(IOEvent event) {
+                if(event instanceof IOEvent.Failure) {
+                    logger.error( ((IOEvent.Failure)event).cause() );
+                }
+            }
+        });
     }
 
-
+    public static InputStream create(final InputStream ostream, final EventListener<? super IOEvent> listener) {
+        return create(new InputListener(ostream) {
+            @Override
+            public void handleEvent(IOEvent event) {
+                listener.handleEvent(event);
+            }
+        });
+    }
+    
+    public static InputStream create(InputListener listener) {
+        
+        final EventRouter router = EventRouterThread.newInstance("OutputStreamer", Integer.MAX_VALUE, Integer.MAX_VALUE);
+        
+        router.addEventListener(IORequest.class, new InputDevice(listener));
+        
+        return new InputDriver() {
+            @Override
+            public void handleEvent(IORequest event) {
+                router.handleEvent(event);
+            }
+        };
+    }
+    
     /**
-     * read -> stream.read -> handle(IOEvent)
+     * Transformacja: {@code read -> stream.read -> handle(IOEvent)}
      */
     public static abstract class InputListener extends InputStream implements EventListener<IOEvent> {
 
@@ -237,17 +208,164 @@ public final class AsyncStream {
             return istream.markSupported();
         }
     }
+    
+    /**
+     * Transformacja: {@code read -> handle(IORequest)}
+     * <p>
+     * Uwaga! Klasy {@code InputStream} nie zaprojektowano w taki sposób, aby
+     * działała w tle. W efekcie poniższa klasa z semantycznego punktu widzenia
+     * całkowicie nie spełnia kontraktu. W zasadzie Bóg jeden wie po licho 
+     * w ogóle implementujemy {@link InputStream}, bo i tak strach tę klasę do 
+     * normalnych metod przekazać.
+     * </p>
+     * <ul>
+     *      <li>metody zwracają bezużyteczne wyniki</li>
+     *      <li>metody nie sygnalizują błędów wyjątkami</li>
+     *      <li>błędy są sygnalizowane jako {@link IOEvent.Failure IOEvent.Failure}</li>
+     * </ul>
+     */
+    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() {
+            handleEvent(IORequest.read());
+            return 0;
+        }
+        
+        public int read(int size) {
+            handleEvent(IORequest.read(size));
+            return 0;
+        }
+
+        @Override
+        public int read(byte array[]) {
+            handleEvent(IORequest.read(array.length));
+            return 0;
+        }
+
+        @Override
+        public int read(byte array[], int offset, int length) {
+            handleEvent(IORequest.read(length));
+            return 0;
+        }
+
+        @Override
+        public long skip(long delta) {
+            handleEvent(IORequest.skip(delta));
+            return 0;
+        }
+
+        @Override
+        public int available() {
+            return 0;
+        }
+
+        @Override
+        public void close() {
+            handleEvent(IORequest.close());
+        }
+
+        @Override
+        public synchronized void mark(int readlimit) {
+            handleEvent(IORequest.mark(readlimit));
+        }
+
+        @Override
+        public synchronized void reset() {
+            handleEvent(IORequest.reset());
+        }
+
+        @Override
+        public boolean markSupported() {
+            return true;
+        }
+    }
 
     /**
-     * write -> stream.write -> handle(IOEvent)
+     * Transformacja: {@code handle(IORequest) -> stream.read}
+     */
+    public static class InputDevice extends ReflectEventListener<IORequest> {
+        
+        private final InputListener device;
+
+        public InputDevice(InputListener device) {
+            this.device = device;
+        }
+        
+        @Meta.EventHandler
+        void onEvent(IORequest.ReadByte event) {
+            try {
+                device.read();
+            } catch(IOException cause) {
+                device.handleEvent(IOEvent.failure(cause));
+            }
+        }
+        
+        @Meta.EventHandler
+        void onEvent(IORequest.ReadByteArray event) {
+            try {
+                byte[] target = new byte[event.count()];
+                device.read(target);
+            } catch(IOException cause) {
+                device.handleEvent(IOEvent.failure(cause));
+            }
+        }
+        
+        @Meta.EventHandler
+        void onEvent(IORequest.Skip event) {
+            try {
+                device.skip(event.delta());
+            } catch(IOException cause) {
+                device.handleEvent(IOEvent.failure(cause));
+            }
+        }
+        
+        @Meta.EventHandler
+        void onEvent(IORequest.Close event) {
+            try {
+                device.close();
+            } catch(IOException cause) {
+                device.handleEvent(IOEvent.failure(cause));
+            }
+        }
+        
+        @Meta.EventHandler
+        void onEvent(IORequest.Mark event) {
+            device.mark(event.limit());
+        }
+        
+        @Meta.EventHandler
+        void onEvent(IORequest.Reset event) {
+            try {
+                device.reset();
+            } catch(IOException cause) {
+                device.handleEvent(IOEvent.failure(cause));
+            }
+        }
+        
+        @Meta.EventHandler
+        void onEvent(IOEvent.Failure event) {
+            device.handleEvent(event);
+        }
+        
+    }
+
+    /**
+     * Transformacja: {@code 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;
+        public OutputListener(OutputStream ostream) {
+            this.ostream = ostream;
         }
 
         @Override
@@ -308,5 +426,113 @@ public final class AsyncStream {
             }
         }
     }
+    
+    /**
+     * Transformacja: {@code write -> handle(IORequest)}
+     * <ul>
+     *      <li>metody nie sygnalizują błędów wyjątkami</li>
+     *      <li>błędy są sygnalizowane jako {@link IOEvent.Failure IOEvent.Failure}</li>
+     * </ul>
+     */
+    public static abstract class OutputDriver extends OutputStream implements EventListener<IORequest> {
+
+        public OutputDriver() {
+            // do nothing
+        }
+
+        @Override
+        public abstract void handleEvent(IORequest event);
+
+        @Override
+        public void write(int value) {
+            handleEvent(IORequest.write(value));
+        }
+
+        @Override
+        public void write(byte array[]) {
+            handleEvent(IORequest.write(array));
+        }
+
+        @Override
+        public void write(byte array[], int offset, int length) {
+            handleEvent(IORequest.write(array, offset, length));
+        }
+
+        @Override
+        public void flush() {
+            handleEvent(IORequest.flush());
+        }
+
+        @Override
+        public void close() {
+            handleEvent(IORequest.close());
+        }
+    }
+    
+    /**
+     * Transformacja: {@code handle(IORequest) -> stream.write}
+     */   
+    public static class OutputDevice extends ReflectEventListener<IORequest> {
+
+        private final OutputListener device;
+
+        public OutputDevice(OutputListener ostream) {
+            this.device = ostream;
+        }
+
+        @Meta.EventHandler
+        void onEvent(IORequest.WriteByte event) {
+            try {
+                device.write(event.data());
+            } catch(IOException cause) {
+                device.handleEvent(IOEvent.failure(cause));
+            }
+        }
+
+        @Meta.EventHandler
+        void onEvent(IORequest.WriteByteArray event) {
+            try {
+                device.write(event.data());
+            } catch(IOException cause) {
+                device.handleEvent(IOEvent.failure(cause));
+            }
+        }
+
+        @Meta.EventHandler
+        void onEvent(IORequest.Flush event) {
+            try {
+                device.flush();
+            } catch(IOException cause) {
+                device.handleEvent(IOEvent.failure(cause));
+            }
+        }
+
+        @Meta.EventHandler
+        void onEvent(IORequest.Close event) {
+            try {
+                device.close();
+            } catch(IOException cause) {
+                device.handleEvent(IOEvent.failure(cause));
+            }
+        }
+        
+        @Meta.EventHandler
+        void onEvent(IOEvent.Failure event) {
+            device.handleEvent(event);
+        }
+        
+    }
+    
+    public static abstract class ErrorListener implements EventListener<IOEvent> {
+        
+        public abstract void onError(IOException cause);
+
+        @Override
+        public final void handleEvent(IOEvent event) {
+            if (event instanceof IOEvent.Failure) {
+                onError( ((IOEvent.Failure)event).cause() );
+            }
+        }
+    }
 
 }

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

@@ -174,13 +174,13 @@ public abstract class IOEvent implements Event, Serializable {
 
         private static final long serialVersionUID = 1L;
         
-        private final long limit;
+        private final int limit;
 
-        protected Mark(long limit) {
+        protected Mark(int limit) {
             this.limit = limit;
         }
 
-        public long limit() {
+        public int limit() {
             return limit;
         }
         

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

@@ -7,8 +7,6 @@
 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;
@@ -145,13 +143,13 @@ public abstract class IORequest implements Event, Serializable {
 
         private static final long serialVersionUID = 1L;
 
-        private final long limit;
+        private final int limit;
 
-        protected Mark(long limit) {
+        protected Mark(int limit) {
             this.limit = limit;
         }
 
-        public long limit() {
+        public int limit() {
             return limit;
         }
 

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

@@ -1,121 +0,0 @@
-/*
- *  @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.OutputStream;
-import net.ranides.assira.annotations.Meta;
-import net.ranides.assira.events.*;
-
-/**
- * Strumień delegujący wszystkie bezpośrednie operacje I/O do osobnego wątku.
- * @author ranides
- */
-public final class OutputStreamThread extends OutputStream {
-    
-    private final OutputStream ostream;
-    private final EventRouterThread router;
-    private final EventRouter listeners;
-
-    /**
-     * Tworzy nowy strumień, który wszystkie operacje wykonuje asynchronicznie,
-     * w osobnym wątku, na rzecz podanego argumentu {@code stream}.
-     * @param stream strumień realizujący bezpośrednio operacje I/O
-     */
-    public OutputStreamThread(OutputStream stream) {
-        this.router = EventRouterThread.newInstance("OutputStreamer", Integer.MAX_VALUE, Integer.MAX_VALUE);
-        this.ostream = stream;
-        this.listeners = new EventRouterDispatcher();
-        this.router.addEventListener(IOEvent.class, new ReflectEventListener<IOEvent>(this));
-    }
-    
-    @Override
-    public void write(int data) {
-        router.signalEvent(IOEvent.write(data));
-    }
-
-    @Override
-    public void write(byte[] data) {
-        router.signalEvent(IOEvent.write(data));
-    }
-
-    @Override
-    public void write(byte[] data, int offset, int length) {
-        router.signalEvent(IOEvent.write(data, offset, length));
-    }
-
-    @Override
-    public void close() {
-        router.signalEvent(IOEvent.close());
-        router.dispose();
-    }
-
-    @Override
-    public void flush() {
-        router.signalEvent(IOEvent.flush());
-    }
-    
-    /**
-     * <p>
-     * Rejestruje następnego obserwatora, który jest powiadamiany o wyjątkach
-     * {@link IOException}, które wystąpiły w czasie operacji I/O. Ponieważ 
-     * żadna z metod klasy nie rzuca wyjątków bezpośrednio, wszystkie informacje
-     * o błędach należy rejestrować za pomocą przekazanego obserwatora.
-     * </p><p>
-     * Można zarejestrować dowolną ilość obserwatorów, kolejne wywołania metody
-     * nie usuwają poprzednich obserwatorów.
-     * </p>
-     * @param handler
-     */
-    public void onError(EventListener<IOEvent.Failure> handler) {
-        listeners.addEventListener(IOEvent.Failure.class, handler);
-    }
-    
-/* ************************************************************************** */
-    
-    @Meta.EventHandler
-    void handleEvent(IOEvent.Close event) {
-        try {
-            ostream.close();
-        } catch (IOException ex) {
-            handleError(ex);
-        }
-    }
-    
-    @Meta.EventHandler
-    void handleEvent(IOEvent.Flush event) {
-        try {
-            ostream.flush();
-        } catch (IOException ex) {
-            handleError(ex);
-        }
-    }
-    
-    @Meta.EventHandler
-    void handleEvent(IOEvent.WriteByteArray event) {
-        try {
-            ostream.write(event.array());
-        } catch (IOException ex) {
-            handleError(ex);
-        }
-    }
-    
-    @Meta.EventHandler
-    void handleEvent(IOEvent.WriteByte event) {
-        try {
-            ostream.write(event.head());
-        } catch (IOException ex) {
-            handleError(ex);
-        }
-    }
-    
-    void handleError(IOException cause) {
-        listeners.signalEvent(IOEvent.failure(cause));
-    }
-
-}