273
|
1 /*
|
|
2 * Copyright (c) 2006-2011 Christian Plattner. All rights reserved.
|
|
3 * Please refer to the LICENSE.txt for licensing details.
|
|
4 */
|
|
5 package ch.ethz.ssh2;
|
|
6
|
|
7 import java.io.IOException;
|
|
8 import java.io.InputStream;
|
|
9 import java.io.InterruptedIOException;
|
|
10
|
|
11 /**
|
|
12 * A <code>StreamGobbler</code> is an InputStream that uses an internal worker
|
|
13 * thread to constantly consume input from another InputStream. It uses a buffer
|
|
14 * to store the consumed data. The buffer size is automatically adjusted, if needed.
|
|
15 * <p/>
|
|
16 * This class is sometimes very convenient - if you wrap a session's STDOUT and STDERR
|
|
17 * InputStreams with instances of this class, then you don't have to bother about
|
|
18 * the shared window of STDOUT and STDERR in the low level SSH-2 protocol,
|
|
19 * since all arriving data will be immediatelly consumed by the worker threads.
|
|
20 * Also, as a side effect, the streams will be buffered (e.g., single byte
|
|
21 * read() operations are faster).
|
|
22 * <p/>
|
|
23 * Other SSH for Java libraries include this functionality by default in
|
|
24 * their STDOUT and STDERR InputStream implementations, however, please be aware
|
|
25 * that this approach has also a downside:
|
|
26 * <p/>
|
|
27 * If you do not call the StreamGobbler's <code>read()</code> method often enough
|
|
28 * and the peer is constantly sending huge amounts of data, then you will sooner or later
|
|
29 * encounter a low memory situation due to the aggregated data (well, it also depends on the Java heap size).
|
|
30 * Joe Average will like this class anyway - a paranoid programmer would never use such an approach.
|
|
31 * <p/>
|
|
32 * The term "StreamGobbler" was taken from an article called "When Runtime.exec() won't",
|
|
33 * see http://www.javaworld.com/javaworld/jw-12-2000/jw-1229-traps.html.
|
|
34 *
|
|
35 * @author Christian Plattner
|
|
36 * @version 2.50, 03/15/10
|
|
37 */
|
|
38
|
307
|
39 public class StreamGobbler extends InputStream {
|
|
40 class GobblerThread extends Thread {
|
|
41 @Override
|
|
42 public void run() {
|
|
43 byte[] buff = new byte[8192];
|
273
|
44
|
307
|
45 while (true) {
|
|
46 try {
|
|
47 int avail = is.read(buff);
|
273
|
48
|
307
|
49 synchronized (synchronizer) {
|
|
50 if (avail <= 0) {
|
|
51 isEOF = true;
|
|
52 synchronizer.notifyAll();
|
|
53 break;
|
|
54 }
|
273
|
55
|
307
|
56 int space_available = buffer.length - write_pos;
|
273
|
57
|
307
|
58 if (space_available < avail) {
|
|
59 /* compact/resize buffer */
|
|
60 int unread_size = write_pos - read_pos;
|
|
61 int need_space = unread_size + avail;
|
|
62 byte[] new_buffer = buffer;
|
273
|
63
|
307
|
64 if (need_space > buffer.length) {
|
|
65 int inc = need_space / 3;
|
|
66 inc = (inc < 256) ? 256 : inc;
|
|
67 inc = (inc > 8192) ? 8192 : inc;
|
|
68 new_buffer = new byte[need_space + inc];
|
|
69 }
|
273
|
70
|
307
|
71 if (unread_size > 0)
|
|
72 System.arraycopy(buffer, read_pos, new_buffer, 0, unread_size);
|
273
|
73
|
307
|
74 buffer = new_buffer;
|
|
75 read_pos = 0;
|
|
76 write_pos = unread_size;
|
|
77 }
|
273
|
78
|
307
|
79 System.arraycopy(buff, 0, buffer, write_pos, avail);
|
|
80 write_pos += avail;
|
|
81 synchronizer.notifyAll();
|
|
82 }
|
|
83 }
|
|
84 catch (IOException e) {
|
|
85 synchronized (synchronizer) {
|
|
86 exception = e;
|
|
87 synchronizer.notifyAll();
|
|
88 break;
|
|
89 }
|
|
90 }
|
|
91 }
|
|
92 }
|
|
93 }
|
273
|
94
|
307
|
95 private InputStream is;
|
273
|
96
|
307
|
97 private final Object synchronizer = new Object();
|
273
|
98
|
307
|
99 private boolean isEOF = false;
|
|
100 private boolean isClosed = false;
|
|
101 private IOException exception = null;
|
273
|
102
|
307
|
103 private byte[] buffer = new byte[2048];
|
|
104 private int read_pos = 0;
|
|
105 private int write_pos = 0;
|
273
|
106
|
307
|
107 public StreamGobbler(InputStream is) {
|
|
108 this.is = is;
|
|
109 GobblerThread t = new GobblerThread();
|
|
110 t.setDaemon(true);
|
|
111 t.start();
|
|
112 }
|
|
113
|
|
114 @Override
|
|
115 public int read() throws IOException {
|
|
116 synchronized (synchronizer) {
|
273
|
117 if (isClosed)
|
|
118 throw new IOException("This StreamGobbler is closed.");
|
|
119
|
307
|
120 while (read_pos == write_pos) {
|
273
|
121 if (exception != null)
|
|
122 throw exception;
|
|
123
|
|
124 if (isEOF)
|
|
125 return -1;
|
|
126
|
307
|
127 try {
|
273
|
128 synchronizer.wait();
|
|
129 }
|
307
|
130 catch (InterruptedException e) {
|
273
|
131 throw new InterruptedIOException();
|
|
132 }
|
|
133 }
|
307
|
134
|
273
|
135 return buffer[read_pos++] & 0xff;
|
|
136 }
|
|
137 }
|
|
138
|
|
139 @Override
|
307
|
140 public int available() throws IOException {
|
|
141 synchronized (synchronizer) {
|
|
142 if (isClosed)
|
|
143 throw new IOException("This StreamGobbler is closed.");
|
273
|
144
|
307
|
145 return write_pos - read_pos;
|
|
146 }
|
|
147 }
|
273
|
148
|
307
|
149 @Override
|
|
150 public int read(byte[] b) throws IOException {
|
|
151 return read(b, 0, b.length);
|
|
152 }
|
273
|
153
|
307
|
154 @Override
|
|
155 public void close() throws IOException {
|
|
156 synchronized (synchronizer) {
|
|
157 if (isClosed)
|
|
158 return;
|
273
|
159
|
307
|
160 isClosed = true;
|
|
161 isEOF = true;
|
|
162 synchronizer.notifyAll();
|
|
163 is.close();
|
|
164 }
|
|
165 }
|
273
|
166
|
307
|
167 @Override
|
|
168 public int read(byte[] b, int off, int len) throws IOException {
|
|
169 if (b == null)
|
|
170 throw new NullPointerException();
|
|
171
|
|
172 if ((off < 0) || (len < 0) || ((off + len) > b.length) || ((off + len) < 0) || (off > b.length))
|
|
173 throw new IndexOutOfBoundsException();
|
273
|
174
|
|
175 if (len == 0)
|
|
176 return 0;
|
307
|
177
|
|
178 synchronized (synchronizer) {
|
273
|
179 if (isClosed)
|
|
180 throw new IOException("This StreamGobbler is closed.");
|
|
181
|
307
|
182 while (read_pos == write_pos) {
|
273
|
183 if (exception != null)
|
|
184 throw exception;
|
|
185
|
|
186 if (isEOF)
|
|
187 return -1;
|
|
188
|
307
|
189 try {
|
273
|
190 synchronizer.wait();
|
|
191 }
|
307
|
192 catch (InterruptedException e) {
|
273
|
193 throw new InterruptedIOException();
|
|
194 }
|
|
195 }
|
|
196
|
|
197 int avail = write_pos - read_pos;
|
|
198 avail = (avail > len) ? len : avail;
|
|
199 System.arraycopy(buffer, read_pos, b, off, avail);
|
|
200 read_pos += avail;
|
|
201 return avail;
|
|
202 }
|
307
|
203 }
|
273
|
204 }
|