001: /*
002: * Licensed to the Apache Software Foundation (ASF) under one or more
003: * contributor license agreements. See the NOTICE file distributed with
004: * this work for additional information regarding copyright ownership.
005: * The ASF licenses this file to You under the Apache License, Version 2.0
006: * (the "License"); you may not use this file except in compliance with
007: * the License. You may obtain a copy of the License at
008: *
009: * http://www.apache.org/licenses/LICENSE-2.0
010: *
011: * Unless required by applicable law or agreed to in writing, software
012: * distributed under the License is distributed on an "AS IS" BASIS,
013: * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014: * See the License for the specific language governing permissions and
015: * limitations under the License.
016: *
017: */
018:
019: package org.apache.tools.ant.taskdefs;
020:
021: import java.io.IOException;
022: import java.io.InputStream;
023: import java.io.OutputStream;
024:
025: /**
026: * Copies all data from an input stream to an output stream.
027: *
028: * @since Ant 1.2
029: */
030: public class StreamPumper implements Runnable {
031:
032: private InputStream is;
033: private OutputStream os;
034: private volatile boolean finish;
035: private volatile boolean finished;
036: private boolean closeWhenExhausted;
037: private boolean autoflush = false;
038: private Exception exception = null;
039: private int bufferSize = 128;
040: private boolean started = false;
041:
042: /**
043: * Create a new stream pumper.
044: *
045: * @param is input stream to read data from
046: * @param os output stream to write data to.
047: * @param closeWhenExhausted if true, the output stream will be closed when
048: * the input is exhausted.
049: */
050: public StreamPumper(InputStream is, OutputStream os,
051: boolean closeWhenExhausted) {
052: this .is = is;
053: this .os = os;
054: this .closeWhenExhausted = closeWhenExhausted;
055: }
056:
057: /**
058: * Create a new stream pumper.
059: *
060: * @param is input stream to read data from
061: * @param os output stream to write data to.
062: */
063: public StreamPumper(InputStream is, OutputStream os) {
064: this (is, os, false);
065: }
066:
067: /**
068: * Set whether data should be flushed through to the output stream.
069: * @param autoflush if true, push through data; if false, let it be buffered
070: * @since Ant 1.6.3
071: */
072: /*package*/void setAutoflush(boolean autoflush) {
073: this .autoflush = autoflush;
074: }
075:
076: /**
077: * Copies data from the input stream to the output stream.
078: *
079: * Terminates as soon as the input stream is closed or an error occurs.
080: */
081: public void run() {
082: synchronized (this ) {
083: started = true;
084: }
085: finished = false;
086: finish = false;
087:
088: final byte[] buf = new byte[bufferSize];
089:
090: int length;
091: try {
092: while ((length = is.read(buf)) > 0 && !finish) {
093: os.write(buf, 0, length);
094: if (autoflush) {
095: os.flush();
096: }
097: }
098: os.flush();
099: } catch (Exception e) {
100: synchronized (this ) {
101: exception = e;
102: }
103: } finally {
104: if (closeWhenExhausted) {
105: try {
106: os.close();
107: } catch (IOException e) {
108: // ignore
109: }
110: }
111: finished = true;
112: synchronized (this ) {
113: notifyAll();
114: }
115: }
116: }
117:
118: /**
119: * Tells whether the end of the stream has been reached.
120: * @return true is the stream has been exhausted.
121: */
122: public boolean isFinished() {
123: return finished;
124: }
125:
126: /**
127: * This method blocks until the stream pumper finishes.
128: * @throws InterruptedException if interrupted.
129: * @see #isFinished()
130: */
131: public synchronized void waitFor() throws InterruptedException {
132: while (!isFinished()) {
133: wait();
134: }
135: }
136:
137: /**
138: * Set the size in bytes of the read buffer.
139: * @param bufferSize the buffer size to use.
140: * @throws IllegalStateException if the StreamPumper is already running.
141: */
142: public synchronized void setBufferSize(int bufferSize) {
143: if (started) {
144: throw new IllegalStateException(
145: "Cannot set buffer size on a running StreamPumper");
146: }
147: this .bufferSize = bufferSize;
148: }
149:
150: /**
151: * Get the size in bytes of the read buffer.
152: * @return the int size of the read buffer.
153: */
154: public synchronized int getBufferSize() {
155: return bufferSize;
156: }
157:
158: /**
159: * Get the exception encountered, if any.
160: * @return the Exception encountered.
161: */
162: public synchronized Exception getException() {
163: return exception;
164: }
165:
166: /**
167: * Stop the pumper as soon as possible.
168: * Note that it may continue to block on the input stream
169: * but it will really stop the thread as soon as it gets EOF
170: * or any byte, and it will be marked as finished.
171: * @since Ant 1.6.3
172: */
173: /*package*/synchronized void stop() {
174: finish = true;
175: notifyAll();
176: }
177: }
|