001    
002    /*
003     * Copyright (C) 2011 Archie L. Cobbs. All rights reserved.
004     *
005     * $Id: NullModemOutputStream.java 148 2011-10-21 03:17:02Z archie.cobbs $
006     */
007    
008    package org.dellroad.stuff.io;
009    
010    import java.io.FilterOutputStream;
011    import java.io.IOException;
012    import java.io.InputStream;
013    import java.io.PipedInputStream;
014    import java.io.PipedOutputStream;
015    
016    /**
017     * Presents an {@link java.io.OutputStream OutputStream} interface given a {@link ReadCallback} that can read from an
018     * {@link InputStream}. A separate thread is created to perform the actual reading.
019     *
020     * <p>
021     * If data is written beyond what the reader is willing to consume, an {@link IOException} is thrown.
022     *
023     * @since 1.0.82
024     */
025    public class NullModemOutputStream extends FilterOutputStream {
026    
027        /**
028         * Constructor.
029         *
030         * @param reader    {@link InputStream} reader callback
031         * @param name      name for this instance; used to create the name of the background thread
032         */
033        public NullModemOutputStream(final ReadCallback reader, String name) {
034            super(new PipedOutputStream());
035    
036            // Sanity check
037            if (reader == null)
038                throw new IllegalArgumentException("null reader");
039    
040            // Create other end of pipe
041            PipedInputStream input;
042            try {
043                input = new PipedInputStream(this.getPipedOutputStream());
044            } catch (IOException e) {
045                throw new RuntimeException("unexpected exception", e);
046            }
047    
048            // Launch reader thread
049            Thread thread = new ReaderThread(reader, input, name);
050            thread.setDaemon(true);
051            thread.start();
052        }
053    
054        /**
055         * Get the wrapped stream cast as a {@link PipedOutputStream}.
056         */
057        protected PipedOutputStream getPipedOutputStream() {
058            return (PipedOutputStream)this.out;
059        }
060    
061        /**
062         * Ensure output stream is closed when this instance is no longer referenced.
063         *
064         * <p>
065         * This ensures the reader thread wakes up (and exits, avoiding a memory leak) when an instance of this class
066         * is created but never read from.
067         */
068        @Override
069        protected void finalize() throws Throwable {
070            try {
071                try {
072                    this.getPipedOutputStream().close();
073                } catch (IOException e) {
074                    // ignore
075                }
076            } finally {
077                super.finalize();
078            }
079        }
080    
081        /**
082         * Callback interface used by {@link NullModemOutputStream}.
083         */
084        public interface ReadCallback {
085    
086            /**
087             * Read from the given input stream.
088             *
089             * <p>
090             * This method will be invoked (once) asynchronously in a dedicated reader thread.
091             * </p>
092             *
093             * @param input input providing the data written to the corresponding {@link NullModemOutputStream}
094             * @throws IOException if an I/O error occurs
095             */
096            void readFrom(InputStream input) throws IOException;
097        }
098    
099        /**
100         * Reader thread. This is designed to not hold a reference to the {@link NullModemOutputStream}.
101         */
102        private static class ReaderThread extends Thread {
103    
104            private final ReadCallback reader;
105            private final PipedInputStream input;
106    
107            ReaderThread(ReadCallback reader, PipedInputStream input, String name) {
108                super(name);
109                this.reader = reader;
110                this.input = input;
111            }
112    
113            @Override
114            public void run() {
115                try {
116                    this.reader.readFrom(this.input);
117                } catch (IOException e) {
118                    // ignore - writer will get another IOException because pipe is about to be broken
119                } finally {
120                    try {
121                        this.input.close();
122                    } catch (IOException e) {
123                        // ignore
124                    }
125                }
126            }
127        }
128    }
129