1 /*
2 * Copyright (c) 2013, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation.
8 *
9 * This code is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
12 * version 2 for more details (a copy is included in the LICENSE file that
13 * accompanied this code).
14 *
15 * You should have received a copy of the GNU General Public License version
16 * 2 along with this work; if not, write to the Free Software Foundation,
17 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18 *
19 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
20 * or visit www.oracle.com if you need additional information or have any
21 * questions.
22 */
23
24 package jdk.testlibrary;
25
26 import java.io.BufferedInputStream;
27 import java.io.ByteArrayOutputStream;
28 import java.io.OutputStream;
29 import java.io.InputStream;
30 import java.io.IOException;
31 import java.util.HashSet;
32 import java.util.Set;
33 import java.util.concurrent.Future;
34 import java.util.concurrent.FutureTask;
35 import java.util.concurrent.atomic.AtomicBoolean;
36
37 public final class StreamPumper implements Runnable {
38
39 private static final int BUF_SIZE = 256;
40
41 /**
42 * Pump will be called by the StreamPumper to process the incoming data
43 */
44 abstract public static class Pump {
45 abstract void register(StreamPumper d);
46 }
47
48 /**
49 * OutputStream -> Pump adapter
50 */
51 final public static class StreamPump extends Pump {
52 private final OutputStream out;
53 public StreamPump(OutputStream out) {
54 this.out = out;
55 }
56
57 @Override
58 void register(StreamPumper sp) {
59 sp.addOutputStream(out);
60 }
61 }
62
63 /**
64 * Used to process the incoming data line-by-line
65 */
66 abstract public static class LinePump extends Pump {
67 @Override
68 final void register(StreamPumper sp) {
69 sp.addLineProcessor(this);
70 }
71
72 abstract protected void processLine(String line);
73 }
74
75 private final InputStream in;
76 private final Set<OutputStream> outStreams = new HashSet<OutputStream>();
77 private final Set<LinePump> linePumps = new HashSet<LinePump>();
78
79 private final AtomicBoolean processing = new AtomicBoolean(false);
80 private final FutureTask<Void> processingTask = new FutureTask<Void>(this, null);
81
82 public StreamPumper(InputStream in) {
83 this.in = in;
84 }
85
86 /**
87 * Create a StreamPumper that reads from in and writes to out.
88 *
89 * @param in
90 * The stream to read from.
91 * @param out
92 * The stream to write to.
93 */
94 public StreamPumper(InputStream in, OutputStream out) {
95 this(in);
96 this.addOutputStream(out);
97 }
98
99 /**
100 * Implements Thread.run(). Continuously read from {@code in} and write to
101 * {@code out} until {@code in} has reached end of stream. Abort on
102 * interruption. Abort on IOExceptions.
103 */
104 @Override
105 public void run() {
106 BufferedInputStream is = null;
107 try {
108 is = new BufferedInputStream(in);
109 ByteArrayOutputStream lineBos = new ByteArrayOutputStream();
110 byte[] buf = new byte[BUF_SIZE];
111 int len = 0;
112 int linelen = 0;
113
114 while ((len = is.read(buf)) > 0 && !Thread.interrupted()) {
115 for(OutputStream out : outStreams) {
116 out.write(buf, 0, len);
117 }
118 if (!linePumps.isEmpty()) {
119 int i = 0;
120 int lastcrlf = -1;
121 while (i < len) {
122 if (buf[i] == '\n' || buf[i] == '\r') {
123 int bufLinelen = i - lastcrlf - 1;
124 if (bufLinelen > 0) {
125 lineBos.write(buf, lastcrlf + 1, bufLinelen);
126 }
127 linelen += bufLinelen;
128
129 if (linelen > 0) {
130 lineBos.flush();
131 final String line = lineBos.toString();
132 for (LinePump lp : linePumps) {
133 lp.processLine(line);
134 };
135 lineBos.reset();
136 linelen = 0;
137 }
138 lastcrlf = i;
139 }
140
141 i++;
142 }
143 if (lastcrlf == -1) {
144 lineBos.write(buf, 0, len);
145 linelen += len;
146 } else if (lastcrlf < len - 1) {
147 lineBos.write(buf, lastcrlf + 1, len - lastcrlf - 1);
148 linelen += len - lastcrlf - 1;
149 }
150 }
151 }
152
153 } catch (IOException e) {
154 e.printStackTrace();
155 } finally {
156 for(OutputStream out : outStreams) {
157 try {
158 out.flush();
159 } catch (IOException e) {}
160 }
161 if (is != null) {
162 try {
163 is.close();
164 } catch (IOException e) {}
165 }
166 try {
167 in.close();
168 } catch (IOException e) {}
169 }
170 }
171
172 final void addOutputStream(OutputStream out) {
173 outStreams.add(out);
174 }
175
176 final void addLineProcessor(LinePump lp) {
177 linePumps.add(lp);
178 }
179
180 final public StreamPumper addPump(Pump ... pump) {
181 if (processing.get()) {
182 throw new IllegalStateException("Can not modify pumper while " +
183 "processing is in progress");
184 }
185 for(Pump p : pump) {
186 p.register(this);
187 }
188 return this;
189 }
190
191 final public Future<Void> process() {
192 if (!processing.compareAndSet(false, true)) {
193 throw new IllegalStateException("Can not re-run the processing");
194 }
195 Thread t = new Thread(new Runnable() {
196 @Override
197 public void run() {
198 processingTask.run();
199 }
200 });
201 t.setDaemon(true);
202 t.start();
203
204 return processingTask;
205 }
206 }