-
Notifications
You must be signed in to change notification settings - Fork 21
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[MSHARED-1072] fix blocking in StreamFeeder (#113)
If input stream has no more available data StreamFeeder was block forever
- Loading branch information
1 parent
63c92f8
commit b60e8a2
Showing
4 changed files
with
148 additions
and
117 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,72 +21,62 @@ | |
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.io.OutputStream; | ||
| import java.util.concurrent.atomic.AtomicReference; | ||
| import java.util.Objects; | ||
|
|
||
| /** | ||
| * Read from an InputStream and write the output to an OutputStream. | ||
| * | ||
| * @author <a href="mailto:[email protected]">Trygve Laugstøl</a> | ||
| */ | ||
| class StreamFeeder extends AbstractStreamHandler { | ||
| class StreamFeeder extends Thread { | ||
|
|
||
| private final AtomicReference<InputStream> input; | ||
| private final InputStream input; | ||
|
|
||
| private final AtomicReference<OutputStream> output; | ||
| private final OutputStream output; | ||
|
|
||
| private volatile Throwable exception; | ||
| private Throwable exception; | ||
| private boolean done; | ||
|
|
||
| private final Object lock = new Object(); | ||
|
|
||
| /** | ||
| * Create a new StreamFeeder | ||
| * | ||
| * @param input Stream to read from | ||
| * @param input Stream to read from | ||
| * @param output Stream to write to | ||
| */ | ||
| StreamFeeder(InputStream input, OutputStream output) { | ||
| super(); | ||
| this.input = new AtomicReference<InputStream>(input); | ||
| this.output = new AtomicReference<OutputStream>(output); | ||
| this.input = Objects.requireNonNull(input); | ||
| this.output = Objects.requireNonNull(output); | ||
| this.done = false; | ||
| } | ||
|
|
||
| @Override | ||
| @SuppressWarnings("checkstyle:innerassignment") | ||
| public void run() { | ||
| try { | ||
| feed(); | ||
| } catch (Throwable e) { | ||
| // Catch everything so the streams will be closed and flagged as done. | ||
| if (this.exception != null) { | ||
| this.exception = e; | ||
| for (int data; !isInterrupted() && (data = input.read()) != -1; ) { | ||
| output.write(data); | ||
| } | ||
| output.flush(); | ||
| } catch (IOException e) { | ||
| exception = e; | ||
| } finally { | ||
| close(); | ||
|
|
||
| synchronized (this) { | ||
| notifyAll(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public void close() { | ||
| setDone(); | ||
| final InputStream is = input.getAndSet(null); | ||
| if (is != null) { | ||
| try { | ||
| is.close(); | ||
| } catch (IOException ex) { | ||
| if (this.exception != null) { | ||
| this.exception = ex; | ||
| } | ||
| } | ||
| synchronized (lock) { | ||
| done = true; | ||
| lock.notifyAll(); | ||
| } | ||
| } | ||
|
|
||
| final OutputStream os = output.getAndSet(null); | ||
| if (os != null) { | ||
| try { | ||
| os.close(); | ||
| } catch (IOException ex) { | ||
| if (this.exception != null) { | ||
| this.exception = ex; | ||
| } | ||
| private void close() { | ||
| try { | ||
| output.close(); | ||
| } catch (IOException e) { | ||
| if (exception == null) { | ||
| exception = e; | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -98,23 +88,16 @@ public Throwable getException() { | |
| return this.exception; | ||
| } | ||
|
|
||
| @SuppressWarnings("checkstyle:innerassignment") | ||
| private void feed() throws IOException { | ||
| InputStream is = input.get(); | ||
| OutputStream os = output.get(); | ||
| boolean flush = false; | ||
|
|
||
| if (is != null && os != null) { | ||
| for (int data; !isDone() && (data = is.read()) != -1; ) { | ||
| if (!isDisabled()) { | ||
| os.write(data); | ||
| flush = true; | ||
| public void waitUntilDone() { | ||
| this.interrupt(); | ||
| synchronized (lock) { | ||
| while (!done) { | ||
| try { | ||
| lock.wait(); | ||
| } catch (InterruptedException e) { | ||
| Thread.currentThread().interrupt(); | ||
| } | ||
| } | ||
|
|
||
| if (flush) { | ||
| os.flush(); | ||
| } | ||
| } | ||
| } | ||
| } | ||
91 changes: 91 additions & 0 deletions
91
src/test/java/org/apache/maven/shared/utils/cli/StreamFeederTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.maven.shared.utils.cli; | ||
|
|
||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| import java.io.ByteArrayInputStream; | ||
| import java.io.ByteArrayOutputStream; | ||
|
|
||
| import org.junit.Test; | ||
|
|
||
| import static org.junit.Assert.assertEquals; | ||
|
|
||
| public class StreamFeederTest { | ||
| static class BlockingInputStream extends ByteArrayInputStream { | ||
| public BlockingInputStream(byte[] buf) { | ||
| super(buf); | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized int read() { | ||
| int data = super.read(); | ||
| if (data >= 0) { | ||
| return data; | ||
| } | ||
|
|
||
| // end test data ... block | ||
| try { | ||
| wait(); | ||
| } catch (InterruptedException e) { | ||
| Thread.currentThread().interrupt(); | ||
| } | ||
| return -1; | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| public void waitUntilFeederDone() throws InterruptedException { | ||
|
|
||
| String TEST_DATA = "TestData"; | ||
|
|
||
| BlockingInputStream inputStream = new BlockingInputStream(TEST_DATA.getBytes()); | ||
| ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); | ||
|
|
||
| StreamFeeder streamFeeder = new StreamFeeder(inputStream, outputStream); | ||
|
|
||
| streamFeeder.start(); | ||
|
|
||
| // wait until all data from steam will be read | ||
| while (outputStream.size() < TEST_DATA.length()) { | ||
| Thread.sleep(10); | ||
| } | ||
|
|
||
| streamFeeder.waitUntilDone(); // wait until process finish | ||
|
|
||
| assertEquals(TEST_DATA, outputStream.toString()); | ||
| } | ||
| } |