|
|
@@ -26,11 +26,11 @@ public final class AsyncStream {
|
|
|
private AsyncStream() {
|
|
|
// utility class
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
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
|
|
|
@@ -41,7 +41,7 @@ public final class AsyncStream {
|
|
|
}
|
|
|
});
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
public static OutputStream create(OutputStream ostream, final EventListener<? super IOEvent> listener) {
|
|
|
return create(new OutputListener(ostream) {
|
|
|
@Override
|
|
|
@@ -50,13 +50,13 @@ public final class AsyncStream {
|
|
|
}
|
|
|
});
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
public static OutputStream create(OutputListener listener) {
|
|
|
-
|
|
|
+
|
|
|
final EventRouter router = EventReactor.newInstance("OutputStreamer", Integer.MAX_VALUE, Integer.MAX_VALUE);
|
|
|
-
|
|
|
+
|
|
|
router.addEventListener(IORequest.class, new OutputDevice(listener));
|
|
|
-
|
|
|
+
|
|
|
return new OutputDriver() {
|
|
|
@Override
|
|
|
public void handleEvent(IORequest event) {
|
|
|
@@ -64,11 +64,11 @@ public final class AsyncStream {
|
|
|
}
|
|
|
};
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
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
|
|
|
@@ -81,28 +81,14 @@ public final class AsyncStream {
|
|
|
}
|
|
|
|
|
|
public static InputStream create(final InputStream ostream, final EventListener<? super IOEvent> listener) {
|
|
|
- return create(new InputListener(ostream) {
|
|
|
+ return new InputListener(ostream) {
|
|
|
@Override
|
|
|
public void handleEvent(IOEvent event) {
|
|
|
listener.handleEvent(event);
|
|
|
}
|
|
|
- });
|
|
|
- }
|
|
|
-
|
|
|
- public static InputStream create(InputListener listener) {
|
|
|
-
|
|
|
- final EventRouter router = EventReactor.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);
|
|
|
- }
|
|
|
};
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
/**
|
|
|
* Transformacja: {@code read -> stream.read -> handle(IOEvent)}
|
|
|
*/
|
|
|
@@ -208,97 +194,18 @@ 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;
|
|
|
- }
|
|
|
- }
|
|
|
|
|
|
/**
|
|
|
* 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 {
|
|
|
@@ -307,7 +214,7 @@ public final class AsyncStream {
|
|
|
device.handleEvent(IOEvent.failure(cause));
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
@Meta.EventHandler
|
|
|
void onEvent(IORequest.ReadByteArray event) {
|
|
|
try {
|
|
|
@@ -317,7 +224,7 @@ public final class AsyncStream {
|
|
|
device.handleEvent(IOEvent.failure(cause));
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
@Meta.EventHandler
|
|
|
void onEvent(IORequest.Skip event) {
|
|
|
try {
|
|
|
@@ -326,7 +233,7 @@ public final class AsyncStream {
|
|
|
device.handleEvent(IOEvent.failure(cause));
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
@Meta.EventHandler
|
|
|
void onEvent(IORequest.Close event) {
|
|
|
try {
|
|
|
@@ -335,12 +242,12 @@ public final class AsyncStream {
|
|
|
device.handleEvent(IOEvent.failure(cause));
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
@Meta.EventHandler
|
|
|
void onEvent(IORequest.Mark event) {
|
|
|
device.mark(event.limit());
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
@Meta.EventHandler
|
|
|
void onEvent(IORequest.Reset event) {
|
|
|
try {
|
|
|
@@ -349,12 +256,12 @@ public final class AsyncStream {
|
|
|
device.handleEvent(IOEvent.failure(cause));
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
@Meta.EventHandler
|
|
|
void onEvent(IOEvent.Failure event) {
|
|
|
device.handleEvent(event);
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -426,7 +333,7 @@ public final class AsyncStream {
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
/**
|
|
|
* Transformacja: {@code write -> handle(IORequest)}
|
|
|
* <ul>
|
|
|
@@ -468,10 +375,10 @@ public final class AsyncStream {
|
|
|
handleEvent(IORequest.close());
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
/**
|
|
|
* Transformacja: {@code handle(IORequest) -> stream.write}
|
|
|
- */
|
|
|
+ */
|
|
|
public static class OutputDevice extends ReflectEventListener<IORequest> {
|
|
|
|
|
|
private final OutputListener device;
|
|
|
@@ -515,24 +422,12 @@ public final class AsyncStream {
|
|
|
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() );
|
|
|
- }
|
|
|
- }
|
|
|
}
|
|
|
|
|
|
}
|