private Object executePiped()

in gshell-parser/src/main/java/org/apache/geronimo/gshell/parser/visitor/ExecutingVisitor.java [161:250]
70 lines of code
12 McCabe index (conditional complexity)

    private Object executePiped(final Object[][] commands) throws CommandLineExecutionFailed, InterruptedException, IOException {
        assert commands != null;

        // Prepare IOs
        final IO[] ios = new IO[commands.length];
        PipedOutputStream pos = null;

        IO io = this.context.getIo();

        for (int i = 0; i < ios.length; i++) {
            InputStream is = (i == 0) ? io.inputStream : new PipedInputStream(pos);
            OutputStream os;

            if (i == ios.length - 1) {
                os = io.outputStream;
            }
            else {
                os = pos = new PipedOutputStream();
            }

            ios[i] = new IO(is, new PrintStream(os), io.errorStream);
        }

        final List<Throwable> errors = new CopyOnWriteArrayList<Throwable>();
        final AtomicReference<Object> ref = new AtomicReference<Object>();
        final CountDownLatch latch = new CountDownLatch(commands.length);

        for (int i = 0; i < commands.length; i++) {
            final int idx = i;

            Runnable r = new Runnable() {
                public void run() {
                    try {
                        ShellContext pipedContext = new ShellContext() {
                            public Shell getShell() {
                                return ExecutingVisitor.this.context.getShell();
                            }

                            public IO getIo() {
                                return ios[idx];
                            }

                            public Variables getVariables() {
                                return ExecutingVisitor.this.context.getVariables();
                            }
                        };

                        Object obj = executor.execute(pipedContext, String.valueOf(commands[idx][0]), Arguments.shift(commands[idx]));

                        if (idx == commands.length - 1) {
                            ref.set(obj);
                        }
                    }
                    catch (Throwable t) {
                        errors.add(t);
                    }
                    finally {
                        if (idx > 0) {
                            Closer.close(ios[idx].inputStream);
                        }
                        if (idx < commands.length - 1) {
                            Closer.close(ios[idx].outputStream);
                        }
                        latch.countDown();
                    }
                }
            };
            if (idx != commands.length - 1) {
                createThread(r).start();
            } else {
                r.run();
            }
        }

        latch.await();

        if (!errors.isEmpty()) {
            Throwable t = errors.get(0);

            // Always preserve the type of notication throwables, reguardless of the trace
            if (t instanceof Notification) {
                throw (Notification)t;
            }

            // Otherwise wrap to preserve the trace
            throw new CommandLineExecutionFailed(t);
        }

        return ref.get();
    }