WebSVN - OpenConcerto - Diff - Rev 177 and 180 - /trunk/OpenConcerto/src/org/openconcerto/utils/ProcessStreams.java



import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Supplier;
 
/**
 * Redirect streams of a process to System.out and System.err.
 *
 * @author Sylvain

    }
 
    // Added to Java 9
    public static final Redirect DISCARD = Redirect.to(StreamUtils.NULL_FILE);
 
    static public final ProcessBuilder redirect(final ProcessBuilder pb) {
        return pb.redirectErrorStream(true).redirectOutput(Redirect.INHERIT);
    }
 
    static public final Process handle(final Process p, final Action action) throws IOException {
        if (action == Action.CLOSE) {
            p.getInputStream().close();
            p.getErrorStream().close();
        } else if (action == Action.REDIRECT) {
            new ProcessStreams(p, System.out, System.err);
        } else if (action == Action.CONSUME) {
            new ProcessStreams(p, StreamUtils.NULL_OS, StreamUtils.NULL_OS);
        }
        return p;
    }

    private final ExecutorService exec = Executors.newFixedThreadPool(2);
    private final CountDownLatch latch;
    private final Future<?> out;
    private final Future<?> err;
 
 
 
 
 
    /**
     * Create a new instance and start reading from the passed process. If a passed
     * {@link OutputStream} is <code>null</code>, then the corresponding {@link InputStream} is not
     * used at all, so the caller should handle it. If the output must be discarded, use
     * {@link StreamUtils#NULL_OS}.

     * @param out where to write the {@link Process#getInputStream() standard output}.
     * @param err where to write the {@link Process#getErrorStream() standard error}.
     */
    public ProcessStreams(final Process p, final OutputStream out, final OutputStream err) {
        this.latch = new CountDownLatch(2);
        this.out = writeToAsync(p::getInputStream, out);
        this.err = writeToAsync(p::getErrorStream, err);
        this.exec.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    ProcessStreams.this.latch.await();
                } catch (final InterruptedException e) {
                    // ne rien faire
                    e.printStackTrace();
                } finally {
                    ProcessStreams.this.exec.shutdown();
                }

 
    protected final void stopErr() {
        this.stop(this.err);
    }
 
    private final void stop(final Future<?> f) {
        if (f == null)
            return;
        // TODO
        // ATTN don't interrupt, hangs in readLine()
        f.cancel(false);
    }
 
    private final Future<?> writeToAsync(final Supplier<InputStream> insSupplier, final Object outs) {
        if (outs == null) {
            this.latch.countDown();
            return null;
        }
        return this.exec.submit(new Callable<Object>() {
            @Override
            public Void call() throws InterruptedException, IOException {
                try (final InputStream ins = insSupplier.get()) {
                    // PrintStream is also an OutputStream
                    if (outs instanceof PrintStream)
                        writeTo(ins, (PrintStream) outs);
                    else
                        StreamUtils.copy(ins, (OutputStream) outs);
 
                    return null;
                } finally {
                    ProcessStreams.this.latch.countDown();
                }
            }

     * @param ins the source.
     * @param outs the destination.
     * @throws InterruptedException if current thread is interrupted.
     * @throws IOException if I/O error.
     */
    public static final void writeTo(final InputStream ins, final PrintStream outs) throws InterruptedException, IOException {
        final BufferedReader r = new BufferedReader(new InputStreamReader(ins));
        String encodedName;
        while ((encodedName = r.readLine()) != null) {
            if (Thread.currentThread().isInterrupted()) {
                throw new InterruptedException();

Rev 177	Rev 180
Line 23...	Line 23...
23	`import java.util.concurrent.Callable;`	23	`import java.util.concurrent.Callable;`
24	`import java.util.concurrent.CountDownLatch;`	24	`import java.util.concurrent.CountDownLatch;`
25	`import java.util.concurrent.ExecutorService;`	25	`import java.util.concurrent.ExecutorService;`
26	`import java.util.concurrent.Executors;`	26	`import java.util.concurrent.Executors;`
27	`import java.util.concurrent.Future;`	27	`import java.util.concurrent.Future;`
-		28	`import java.util.function.Supplier;`
28		29
29	`/**`	30	`/**`
30	`* Redirect streams of a process to System.out and System.err.`	31	`* Redirect streams of a process to System.out and System.err.`
31	`*`	32	`*`
32	`* @author Sylvain`	33	`* @author Sylvain`
Line 58...	Line 59...
58	`}`	59	`}`
59		60
60	`// Added to Java 9`	61	`// Added to Java 9`
61	`public static final Redirect DISCARD = Redirect.to(StreamUtils.NULL_FILE);`	62	`public static final Redirect DISCARD = Redirect.to(StreamUtils.NULL_FILE);`
62		63
63	`static public final ProcessBuilder redirect(final ProcessBuilder pb) throws IOException {`	64	`static public final ProcessBuilder redirect(final ProcessBuilder pb) {`
64	`return pb.redirectErrorStream(true).redirectOutput(Redirect.INHERIT);`	65	`return pb.redirectErrorStream(true).redirectOutput(Redirect.INHERIT);`
65	`}`	66	`}`
66		67
67	`static public final Process handle(final Process p, final Action action) throws IOException {`	68	`static public final Process handle(final Process p, final Action action) throws IOException {`
68	`if (action == Action.CLOSE) {`	69	`if (action == Action.CLOSE) {`
69	`p.getInputStream().close();`	70	`p.getInputStream().close();`
70	`p.getErrorStream().close();`	71	`p.getErrorStream().close();`
71	`} else if (action == Action.REDIRECT) {`	72	`} else if (action == Action.REDIRECT) {`
72	`new ProcessStreams(p);`	73	`new ProcessStreams(p, System.out, System.err);`
73	`} else if (action == Action.CONSUME) {`	74	`} else if (action == Action.CONSUME) {`
74	`new ProcessStreams(p, StreamUtils.NULL_OS, StreamUtils.NULL_OS);`	75	`new ProcessStreams(p, StreamUtils.NULL_OS, StreamUtils.NULL_OS);`
75	`}`	76	`}`
76	`return p;`	77	`return p;`
77	`}`	78	`}`
Line 79...	Line 80...
79	`private final ExecutorService exec = Executors.newFixedThreadPool(2);`	80	`private final ExecutorService exec = Executors.newFixedThreadPool(2);`
80	`private final CountDownLatch latch;`	81	`private final CountDownLatch latch;`
81	`private final Future<?> out;`	82	`private final Future<?> out;`
82	`private final Future<?> err;`	83	`private final Future<?> err;`
83		84
84	`public ProcessStreams(final Process p) {`	-
85	`this(p, System.out, System.err);`	-
86	`}`	-
87		-
88	`/**`	85	`/**`
89	`* Create a new instance and start reading from the passed process. If a passed`	86	`* Create a new instance and start reading from the passed process. If a passed`
90	`* {@link OutputStream} is <code>null</code>, then the corresponding {@link InputStream} is not`	87	`* {@link OutputStream} is <code>null</code>, then the corresponding {@link InputStream} is not`
91	`* used at all, so the caller should handle it. If the output must be discarded, use`	88	`* used at all, so the caller should handle it. If the output must be discarded, use`
92	`* {@link StreamUtils#NULL_OS}.`	89	`* {@link StreamUtils#NULL_OS}.`
Line 95...	Line 92...
95	`* @param out where to write the {@link Process#getInputStream() standard output}.`	92	`* @param out where to write the {@link Process#getInputStream() standard output}.`
96	`* @param err where to write the {@link Process#getErrorStream() standard error}.`	93	`* @param err where to write the {@link Process#getErrorStream() standard error}.`
97	`*/`	94	`*/`
98	`public ProcessStreams(final Process p, final OutputStream out, final OutputStream err) {`	95	`public ProcessStreams(final Process p, final OutputStream out, final OutputStream err) {`
99	`this.latch = new CountDownLatch(2);`	96	`this.latch = new CountDownLatch(2);`
100	`this.out = writeToAsync(p.getInputStream(), out);`	97	`this.out = writeToAsync(p::getInputStream, out);`
101	`this.err = writeToAsync(p.getErrorStream(), err);`	98	`this.err = writeToAsync(p::getErrorStream, err);`
102	`this.exec.submit(new Runnable() {`	99	`this.exec.submit(new Runnable() {`
-		100	`@Override`
103	`public void run() {`	101	`public void run() {`
104	`try {`	102	`try {`
105	`ProcessStreams.this.latch.await();`	103	`ProcessStreams.this.latch.await();`
106	`} catch (InterruptedException e) {`	104	`} catch (final InterruptedException e) {`
107	`// ne rien faire`	105	`// ne rien faire`
108	`e.printStackTrace();`	106	`e.printStackTrace();`
109	`} finally {`	107	`} finally {`
110	`ProcessStreams.this.exec.shutdown();`	108	`ProcessStreams.this.exec.shutdown();`
111	`}`	109	`}`
Line 119...	Line 117...
119		117
120	`protected final void stopErr() {`	118	`protected final void stopErr() {`
121	`this.stop(this.err);`	119	`this.stop(this.err);`
122	`}`	120	`}`
123		121
124	`private final void stop(Future<?> f) {`	122	`private final void stop(final Future<?> f) {`
125	`if (f == null)`	123	`if (f == null)`
126	`return;`	124	`return;`
127	`// TODO`	125	`// TODO`
128	`// ATTN don't interrupt, hangs in readLine()`	126	`// ATTN don't interrupt, hangs in readLine()`
129	`f.cancel(false);`	127	`f.cancel(false);`
130	`}`	128	`}`
131		129
132	`private final Future<?> writeToAsync(final InputStream ins, final Object outs) {`	130	`private final Future<?> writeToAsync(final Supplier<InputStream> insSupplier, final Object outs) {`
133	`if (outs == null) {`	131	`if (outs == null) {`
134	`this.latch.countDown();`	132	`this.latch.countDown();`
135	`return null;`	133	`return null;`
136	`}`	134	`}`
137	`return this.exec.submit(new Callable<Object>() {`	135	`return this.exec.submit(new Callable<Object>() {`
-		136	`@Override`
138	`public Object call() throws InterruptedException, IOException {`	137	`public Void call() throws InterruptedException, IOException {`
139	`try {`	138	`try (final InputStream ins = insSupplier.get()) {`
140	`// PrintStream is also an OutputStream`	139	`// PrintStream is also an OutputStream`
141	`if (outs instanceof PrintStream)`	140	`if (outs instanceof PrintStream)`
142	`writeTo(ins, (PrintStream) outs);`	141	`writeTo(ins, (PrintStream) outs);`
143	`else`	142	`else`
144	`StreamUtils.copy(ins, (OutputStream) outs);`	143	`StreamUtils.copy(ins, (OutputStream) outs);`
145	`ins.close();`	-
146	`return null;`	144	`return null;`
147	`} finally {`	145	`} finally {`
148	`ProcessStreams.this.latch.countDown();`	146	`ProcessStreams.this.latch.countDown();`
149	`}`	147	`}`
150	`}`	148	`}`
Line 157...	Line 155...
157	`* @param ins the source.`	155	`* @param ins the source.`
158	`* @param outs the destination.`	156	`* @param outs the destination.`
159	`* @throws InterruptedException if current thread is interrupted.`	157	`* @throws InterruptedException if current thread is interrupted.`
160	`* @throws IOException if I/O error.`	158	`* @throws IOException if I/O error.`
161	`*/`	159	`*/`
162	`public static final void writeTo(InputStream ins, PrintStream outs) throws InterruptedException, IOException {`	160	`public static final void writeTo(final InputStream ins, final PrintStream outs) throws InterruptedException, IOException {`
163	`final BufferedReader r = new BufferedReader(new InputStreamReader(ins));`	161	`final BufferedReader r = new BufferedReader(new InputStreamReader(ins));`
164	`String encodedName;`	162	`String encodedName;`
165	`while ((encodedName = r.readLine()) != null) {`	163	`while ((encodedName = r.readLine()) != null) {`
166	`if (Thread.currentThread().isInterrupted()) {`	164	`if (Thread.currentThread().isInterrupted()) {`
167	`throw new InterruptedException();`	165	`throw new InterruptedException();`

Dépôt officiel du code source de l'ERP OpenConcerto

svn://code.openconcerto.org/openconcerto

(root)/trunk/OpenConcerto/src/org/openconcerto/utils/ProcessStreams.java - Rev 177 → 180