001
002 /*
003 * Copyright (C) 2011 Archie L. Cobbs. All rights reserved.
004 *
005 * $Id: AsyncInputStream.java 298 2012-02-23 23:30:22Z archie.cobbs $
006 */
007
008 package org.dellroad.stuff.io;
009
010 import java.io.IOException;
011 import java.io.InputStream;
012
013 import org.slf4j.Logger;
014 import org.slf4j.LoggerFactory;
015
016 /**
017 * Performs asynchonous reads on an {@link InputStream} and notifies of input events.
018 *
019 * <p>
020 * Reads are performed in a dedicated background thread, from which the configured listener is notified.
021 * The background thread runs until this instance is {@linkplain #close closed}, EOF or an exception is detected
022 * on the input, or a listener callback method throws an exception.
023 * </p>
024 *
025 * <p>
026 * Instances of this class are thread-safe. The {@link #close} method may be safely invoked re-entrantly from the
027 * listener callback methods.
028 * </p>
029 */
030 public class AsyncInputStream {
031
032 private static final int BUFFER_SIZE = 4096;
033
034 protected final Logger log = LoggerFactory.getLogger(getClass());
035
036 private final InputStream input;
037 private final String name;
038 private final Listener listener;
039
040 private boolean closed; // this instance has been close()'d
041
042 /**
043 * Constructor.
044 *
045 * <p>
046 * If {@code listener} is null, this instance effectively reads and discards all of the input in a background thread.
047 * </p>
048 *
049 * @param input underlying input stream
050 * @param name name for this instance; used to create the name of the background thread
051 * @param listener callback object for input events, or null for none
052 * @throws IllegalArgumentException if any parameter is null
053 */
054 public AsyncInputStream(InputStream input, String name, Listener listener) {
055 if (input == null)
056 throw new IllegalArgumentException("null input");
057 if (name == null)
058 throw new IllegalArgumentException("name input");
059 this.input = input;
060 this.name = name;
061 this.listener = listener;
062 new Thread(this.name) {
063 @Override
064 public void run() {
065 AsyncInputStream.this.threadMain();
066 }
067 }.start();
068 }
069
070 /**
071 * Close this instance.
072 *
073 * <p>
074 * Does nothing if already closed.
075 */
076 public synchronized void close() {
077 if (this.closed)
078 return;
079 try {
080 this.input.close();
081 } catch (IOException e) {
082 // ignore; we assume main thread will awake in any case
083 }
084 this.closed = true;
085 }
086
087 /**
088 * Writer thread main entry point.
089 */
090 private void threadMain() {
091 try {
092 this.runLoop();
093 } catch (Throwable t) {
094 synchronized (this) {
095 if (this.closed)
096 return;
097 }
098 try {
099 if (this.listener != null)
100 this.listener.handleException(t);
101 } catch (Exception e) {
102 this.log.error(this.name + ": caught unexpected exception", e);
103 }
104 }
105 }
106
107 /**
108 * Async reader thread main loop.
109 */
110 private void runLoop() throws IOException {
111 byte[] buf = new byte[BUFFER_SIZE];
112 while (true) {
113 int r = this.input.read(buf);
114 if (r == -1) {
115 if (this.listener != null)
116 this.listener.handleEOF();
117 break;
118 }
119 if (this.listener != null)
120 this.listener.handleInput(buf, 0, r);
121 }
122 }
123
124 /**
125 * Callback interface required by {@link AsyncInputStream}.
126 */
127 public interface Listener {
128
129 /**
130 * Handle new data read from the underlying input.
131 * This method must not write to buffer bytes outside of the defined region.
132 *
133 * @param buf data buffer
134 * @param off starting offset of data in buffer
135 * @param len number of bytes of data
136 */
137 void handleInput(byte[] buf, int off, int len);
138
139 /**
140 * Handle an exception detected on the underlying input.
141 * No further events will be delivered.
142 *
143 * <p>
144 * Typically the assocaited {@link AsyncInputStream} will be closed in this callback.
145 *
146 * @param e the exception received (usually {@link IOException} but could also be any other {@link RuntimeException})
147 */
148 void handleException(Throwable e);
149
150 /**
151 * Handle end-of-file detected on the underlying input.
152 * No further events will be delivered.
153 *
154 * <p>
155 * Typically the assocaited {@link AsyncInputStream} will be closed in this callback.
156 */
157 void handleEOF();
158 }
159 }
160