CODENVY-587: Handle closed consumers in composite line comsumer

Signed-off-by: Mykola Morhun <[email protected]>

core/che-core-api-core/src/main/java/org/eclipse/che/api/core/util/CompositeLineConsumer.java

@@ -14,38 +14,58 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.nio.channels.ClosedByInterruptException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 
 /**
  * @author andrew00x
+ * @author Mykola Morhun
  */
 public class CompositeLineConsumer implements LineConsumer {
     private static final Logger LOG = LoggerFactory.getLogger(CompositeLineConsumer.class);
 
-    private final LineConsumer[] lineConsumers;
+    private final List<LineConsumer> lineConsumers;
+    private       boolean            isClosed;
 
     public CompositeLineConsumer(LineConsumer... lineConsumers) {
-        this.lineConsumers = lineConsumers;
+        this.lineConsumers = new ArrayList<>(lineConsumers.length);
+        Arrays.stream(lineConsumers).forEach(this.lineConsumers::add);
+
+        this.isClosed = false;
     }
 
     @Override
     public void close() throws IOException {
-        for (LineConsumer lineConsumer : lineConsumers) {
-            try {
-                lineConsumer.close();
-            } catch (IOException e) {
-                LOG.error(String.format("An error occurred while closing the line consumer %s", lineConsumer), e);
+        if (!isClosed) {
+            isClosed = true;
+            for (LineConsumer lineConsumer : lineConsumers) {
+                try {
+                    lineConsumer.close();
+                } catch (IOException e) {
+                    LOG.error(String.format("An error occurred while closing the line consumer %s", lineConsumer), e);
+                }
             }
         }
     }
 
     @Override
     public void writeLine(String line) throws IOException {
-        for (LineConsumer lineConsumer : lineConsumers) {
-            try {
-                lineConsumer.writeLine(line);
-            } catch (IOException e) {
-                LOG.error(String.format("An error occurred while writing line to the line consumer %s", lineConsumer), e);
+        if (!isClosed) {
+            for (LineConsumer lineConsumer : lineConsumers) {
+                try {
+                    lineConsumer.writeLine(line);
+                } catch (ConsumerAlreadyClosedException | ClosedByInterruptException e) {
+                    lineConsumers.remove(lineConsumer); // consumer is already closed, so we cannot write into it any more
+                    if (lineConsumers.size() == 0) { // if all consumers are closed then we can close this one
+                        isClosed = true;
+                    }
+                } catch (IOException e) {
+                    LOG.error(String.format("An error occurred while writing line to the line consumer %s", lineConsumer), e);
+                }
             }
         }
     }
+
 }

core/che-core-api-core/src/main/java/org/eclipse/che/api/core/util/ConsumerAlreadyClosedException.java

@@ -0,0 +1,22 @@
+/*******************************************************************************
+ * Copyright (c) 2012-2016 Codenvy, S.A.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *   Codenvy, S.A. - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.che.api.core.util;
+
+import java.io.IOException;
+
+/**
+ * @author Mykola Morhun
+ */
+public class ConsumerAlreadyClosedException extends IOException {
+    public ConsumerAlreadyClosedException(String message) {
+        super(message);
+    }
+}

core/che-core-api-core/src/main/java/org/eclipse/che/api/core/util/FileLineConsumer.java

@@ -34,11 +34,18 @@ public File getFile() {
 
     @Override
     public void writeLine(String line) throws IOException {
-        if (line != null) {
-            writer.write(line);
+        try {
+            if (line != null) {
+                writer.write(line);
+            }
+            writer.write('\n');
+            writer.flush();
+        } catch (IOException e) {
+            if("Stream closed".equals(e.getMessage())) {
+                throw new ConsumerAlreadyClosedException(e.getMessage());
+            }
+            throw e;
         }
-        writer.write('\n');
-        writer.flush();
     }
 
     @Override

plugins/plugin-docker/che-plugin-docker-client/src/main/java/org/eclipse/che/plugin/docker/client/ProgressMonitor.java

@@ -10,6 +10,7 @@
  *******************************************************************************/
 package org.eclipse.che.plugin.docker.client;
 
+import org.eclipse.che.api.core.util.ConsumerAlreadyClosedException;
 import org.eclipse.che.plugin.docker.client.json.ProgressStatus;
 
 /**
@@ -19,7 +20,7 @@
  * @author Alexander Garagatyi
  */
 public interface ProgressMonitor {
-    void updateProgress(ProgressStatus currentProgressStatus);
+    void updateProgress(ProgressStatus currentProgressStatus) throws ConsumerAlreadyClosedException;
 
     ProgressMonitor DEV_NULL = new ProgressMonitor() {
         @Override