- /**
- * The connection feeds the input from the currently connected source to the
- * input stream that {@link #get(int)} will get its data from.
- *
- * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
- */
- private class Connection implements Runnable {
-
- /** The currently connected source. */
- /* synchronized by syncObject. */
- private Source source;
-
- /** The input stream that {@link #get(int)} will read from. */
- /* synchronized by syncObject. */
- private PipedInputStream pipedInputStream;
-
- /** The output stream that the source will be fed into. */
- /* synchronized by syncObject. */
- private PipedOutputStream pipedOutputStream;
-
- /**
- * Changes the source of the connection.
- *
- * @param source
- * The new source of the connection
- * @return This connection
- * @throws IOException
- * if an I/O error occurs
- */
- public Connection source(Source source) throws IOException {
- synchronized (syncObject) {
- Source oldSource = this.source;
- this.source = source;
- if (oldSource != null) {
- eventBus.post(new SourceFinishedEvent(oldSource));
- }
- pipedInputStream = new PipedInputStream();
- pipedOutputStream = new PipedOutputStream(pipedInputStream);
- syncObject.notifyAll();
- }
- return this;
- }
-
- @Override
- public void run() {
- while (true) {
- /* wait for source to be set. */
- OutputStream outputStream;
- Source source;
- logger.finest("Entering synchronized block...");
- synchronized (syncObject) {
- logger.finest("Entered synchronized block.");
- source = this.source;
- while (source == null) {
- try {
- logger.finest("Waiting for source to connect...");
- syncObject.wait();
- } catch (InterruptedException ie1) {
- /* ignore, keep waiting. */
- }
- source = this.source;
- }
- outputStream = pipedOutputStream;
- }
- logger.finest("Exited synchronized block.");
-
- byte[] buffer = null;
- boolean readSuccessful = false;
- while (!readSuccessful) {
- try {
- buffer = source.get(4096);
- logger.finest(String.format("Read %d Bytes.", buffer.length));
- if (buffer.length > 0) {
- readSuccessful = true;
- }
- } catch (IOException e) {
- /* TODO - notify & wait */
- }
- }
-
- try {
- outputStream.write(buffer);
- logger.finest(String.format("Wrote %d Bytes.", buffer.length));
- } catch (IOException ioe1) {
- /* okay, the sink has died, exit. */
- logger.log(Level.WARNING, "Could not write to pipe!", ioe1);
- break;
- }
- }
-
- logger.info("Exiting.");
- }
-
- }
-