class ConcurrentCompositeLineConsumer implements LineConsumer {

  @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();
      }
    }
  }

  public  ConcurrentCompositeLineConsumer(LineConsumer... lineConsumers);

}