CODENVY-587: Handle closed consumers in composite line comsumer

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

@@ -10,42 +10,76 @@
  *******************************************************************************/
 package org.eclipse.che.api.core.util;
 
+import org.eclipse.che.api.core.util.lineconsumer.ConsumerAlreadyClosedException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.nio.channels.ClosedByInterruptException;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
+ * This line consumer consists of several consumers and copies each consumed line into all subconsumers.
+ * Is used when lines should be written into two or more consumers.
+ * <br/>
+ * This class is not thread safe.
+ * Also see multithreaded implementation {@link org.eclipse.che.api.core.util.lineconsumer.ConcurrentCompositeLineConsumer}
+ *
  * @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                  isOpen;
 
     public CompositeLineConsumer(LineConsumer... lineConsumers) {
-        this.lineConsumers = lineConsumers;
+        this.lineConsumers = new CopyOnWriteArrayList<>(lineConsumers);
+        this.isOpen = true;
     }
 
+    /**
+     * Closes all unclosed subconsumers.
+     */
     @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);
+    public void close() {
+        if (isOpen) {
+            isOpen = false;
+            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);
+                }
             }
         }
     }
 
+    /**
+     * Writes given line to each subconsumer.
+     * Do nothing if this consumer is closed or all subconsumers are closed.
+     *
+     * @param line
+     *         line to write
+     */
     @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);
+    public void writeLine(String line) {
+        if (isOpen) {
+            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
+                        isOpen = false;
+                    }
+                } 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/FileLineConsumer.java

@@ -10,22 +10,33 @@
  *******************************************************************************/
 package org.eclipse.che.api.core.util;
 
+import org.eclipse.che.api.core.util.lineconsumer.ConsumerAlreadyClosedException;
+
 import java.io.File;
 import java.io.IOException;
 import java.io.Writer;
 import java.nio.charset.Charset;
 import java.nio.file.Files;
 
 /**
+ * Consumes logs and writes them into file.
+ * <br/>
+ * This class is not thread safe.
+ * Also see multithreaded implementation {@link org.eclipse.che.api.core.util.lineconsumer.ConcurrentFileLineConsumer}
+ *
  * @author andrew00x
+ * @author Mykola Morhun
  */
 public class FileLineConsumer implements LineConsumer {
     private final File   file;
     private final Writer writer;
 
+    private boolean isOpen;
+
     public FileLineConsumer(File file) throws IOException {
         this.file = file;
         writer = Files.newBufferedWriter(file.toPath(), Charset.defaultCharset());
+        isOpen = true;
     }
 
     public File getFile() {
@@ -34,15 +45,27 @@ public File getFile() {
 
     @Override
     public void writeLine(String line) throws IOException {
-        if (line != null) {
-            writer.write(line);
+        if (isOpen) {
+            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
     public void close() throws IOException {
-        writer.close();
+        if (isOpen) {
+            isOpen = false;
+            writer.close();
+        }
     }
 }

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

@@ -0,0 +1,100 @@
+/*******************************************************************************
+ * 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.lineconsumer;
+
+import org.eclipse.che.api.core.util.LineConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.channels.ClosedByInterruptException;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * This line consumer consists of several consumers and copies each consumed line into all subconsumers.
+ * Is used when lines should be written into two or more consumers.
+ * This implementation is thread safe.
+ *
+ * @author andrew00x
+ * @author Mykola Morhun
+ */
+public class ConcurrentCompositeLineConsumer implements LineConsumer {
+    private static final Logger LOG = LoggerFactory.getLogger(ConcurrentCompositeLineConsumer.class);
+
+    private final List<LineConsumer>     lineConsumers;
+    private final ReentrantReadWriteLock lock;
+
+    private volatile boolean isOpen;
+
+    public ConcurrentCompositeLineConsumer(LineConsumer... lineConsumers) {
+        this.lineConsumers = new CopyOnWriteArrayList<>(lineConsumers);
+        this.lock = new ReentrantReadWriteLock();
+        this.isOpen = true;
+    }
+
+    public boolean isOpen() {
+        return isOpen;
+    }
+
+    /**
+     * Closes all unclosed subconsumers.
+     */
+    @Override
+    public void close() {
+        if (isOpen) {
+            lock.writeLock().lock();
+            try {
+                isOpen = false;
+                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);
+                    }
+                }
+            } finally {
+                lock.writeLock().unlock();
+            }
+        }
+    }
+
+    /**
+     * Writes given line to each subconsumer.
+     * Do nothing if this consumer is closed or all subconsumers are closed.
+     *
+     * @param line
+     *         line to write
+     */
+    @Override
+    public void writeLine(String line) {
+        if (isOpen && lock.readLock().tryLock()) {
+            try {
+                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
+                            isOpen = false;
+                        }
+                    } catch (IOException e) {
+                        LOG.error(String.format("An error occurred while writing line to the line consumer %s", lineConsumer), e);
+                    }
+                }
+            } finally {
+                lock.readLock().unlock();
+            }
+        }
+    }
+
+}

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

@@ -0,0 +1,84 @@
+/*******************************************************************************
+ * 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.lineconsumer;
+
+import org.eclipse.che.api.core.util.LineConsumer;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Writer;
+import java.nio.charset.Charset;
+import java.nio.file.Files;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * Consumes logs and writes them into file.
+ * This implementation is thread safe.
+ *
+ * @author andrew00x
+ * @author Mykola Morhun
+ */
+public class ConcurrentFileLineConsumer implements LineConsumer {
+    private final File                   file;
+    private final Writer                 writer;
+    private final ReentrantReadWriteLock lock;
+
+    private volatile boolean isOpen;
+
+    public ConcurrentFileLineConsumer(File file) throws IOException {
+        this.file = file;
+        writer = Files.newBufferedWriter(file.toPath(), Charset.defaultCharset());
+        isOpen = true;
+        lock = new ReentrantReadWriteLock();
+    }
+
+    public File getFile() {
+        return file;
+    }
+
+    public boolean isOpen() {
+        return isOpen;
+    }
+
+    @Override
+    public void writeLine(String line) throws IOException {
+        if (isOpen && lock.readLock().tryLock()) {
+            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;
+            } finally {
+                lock.readLock().unlock();
+            }
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (isOpen) {
+            lock.writeLock().lock();
+            try {
+                isOpen = false;
+                writer.close();
+            } finally {
+                lock.writeLock().unlock();
+            }
+        }
+    }
+
+}

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

@@ -0,0 +1,24 @@
+/*******************************************************************************
+ * 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.lineconsumer;
+
+import java.io.IOException;
+
+/**
+ * Is thrown as result of attempt to write a line into closed line consumer.
+ *
+ * @author Mykola Morhun
+ */
+public class ConsumerAlreadyClosedException extends IOException {
+    public ConsumerAlreadyClosedException(String message) {
+        super(message);
+    }
+}