/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you under the Apache License, Version 2.0 (the
    * "License"); you may not use this file except in compliance
    * with the License.  You may obtain a copy of the License at
    *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   */
  package org.apache.maven.plugin.surefire.booterclient.output;
  
  import javax.annotation.Nonnull;
  
  import java.io.Closeable;
  import java.io.IOException;
  import java.util.Map;
  import java.util.concurrent.ConcurrentLinkedDeque;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.locks.AbstractQueuedSynchronizer;
  
  import org.apache.maven.surefire.api.event.Event;
  import org.apache.maven.surefire.extensions.EventHandler;
  import org.slf4j.MDC;
  
  import static java.lang.Thread.currentThread;
  import static org.apache.maven.surefire.api.util.internal.DaemonThreadFactory.newDaemonThread;
  
  /**
   * Knows how to reconstruct *all* the state transmitted over Channel by the forked process.
   * <br>
   * After applying the performance improvements with {@link QueueSynchronizer} the throughput becomes
   * 6.33 mega messages per second
   * (158 nanoseconds per message, 5 million messages within 0.79 seconds - see the test ThreadedStreamConsumerTest)
   * on CPU i5 Dual Core 2.6 GHz and Oracle JDK 11.
   *
   * @author Kristian Rosenvold
   */
  public final class ThreadedStreamConsumer implements EventHandler<Event>, Closeable {
      private static final int QUEUE_MAX_ITEMS = 10_000;
      private static final Event END_ITEM = new FinalEvent();
  
      private final QueueSynchronizer<Event> synchronizer = new QueueSynchronizer<>(QUEUE_MAX_ITEMS, END_ITEM);
      private final AtomicBoolean stop = new AtomicBoolean();
      private final AtomicBoolean isAlive = new AtomicBoolean(true);
      private final Thread consumer;
      private final Pumper pumper;
  
      final class Pumper implements Runnable {
          private final EventHandler<Event> target;
  
          private final MultipleFailureException errors = new MultipleFailureException();
          private final Map<String, String> parentThreadsMdcContextMap;
  
          Pumper(EventHandler<Event> target, Map<String, String> mdcContextMap) {
              this.target = target;
              this.parentThreadsMdcContextMap = mdcContextMap;
          }
  
          /**
           * Calls {@link ForkClient#handleEvent(Event)} which may throw any {@link RuntimeException}.<br>
           * Even if {@link ForkClient} is not fault-tolerant, this method MUST be fault-tolerant and thus the
           * try-catch block must be inside of the loop which prevents from loosing events from {@link EventHandler}.
           * <br>
           * If {@link org.apache.maven.plugin.surefire.report.ConsoleOutputFileReporter#writeTestOutput} throws
           * {@link java.io.IOException} and then {@code target.consumeLine()} throws any RuntimeException, this method
           * MUST NOT skip reading the events from the forked JVM; otherwise we could simply lost events
           * e.g. acquire-next-test which means that {@link ForkClient} could hang on waiting for old test to complete
           * and therefore the plugin could be permanently in progress.
           */
          @Override
          public void run() {
              // copy mdc context map from parent thread so that mvn
              // tools that build in parallel can use it in there logging
              // to make clear what module is producing the logs
              MDC.setContextMap(parentThreadsMdcContextMap);
              while (!stop.get() || !synchronizer.isEmptyQueue()) {
                  try {
                      Event item = synchronizer.awaitNext();
  
                      if (shouldStopQueueing(item)) {
                          break;
                      }
  
                      target.handleEvent(item);
                  } catch (Throwable t) {
                      // ensure the stack trace to be at the instance of the exception
                      t.getStackTrace();
                      errors.addException(t);
                  }
             }
 
             isAlive.set(false);
         }
 
         boolean hasErrors() {
             return errors.hasNestedExceptions();
         }
 
         void throwErrors() throws IOException {
             throw errors;
         }
     }
 
     public ThreadedStreamConsumer(EventHandler<Event> target) {
         pumper = new Pumper(target, MDC.getCopyOfContextMap());
         Thread consumer = newDaemonThread(pumper, "ThreadedStreamConsumer");
         consumer.setUncaughtExceptionHandler((t, e) -> isAlive.set(false));
         consumer.start();
         this.consumer = consumer;
     }
 
     @Override
     public void handleEvent(@Nonnull Event event) {
         // Do NOT call Thread.isAlive() - slow.
         // It makes worse performance from 790 ms to 1250 ms for 5 million messages.
         if (!stop.get() && isAlive.get()) {
             synchronizer.pushNext(event);
         }
     }
 
     @Override
     public void close() throws IOException {
         isAlive.compareAndSet(true, consumer.isAlive());
         if (stop.compareAndSet(false, true) && isAlive.get()) {
             if (currentThread().isInterrupted()) {
                 synchronizer.markStopped();
                 consumer.interrupt();
             } else {
                 synchronizer.markStopped();
 
                 try {
                     consumer.join();
                 } catch (InterruptedException e) {
                     // we should not set interrupted=true in this Thread
                     // if consumer's Thread was interrupted which is indicated by InterruptedException
                 }
 
                 synchronizer.clearQueue();
             }
         }
 
         if (pumper.hasErrors()) {
             pumper.throwErrors();
         }
     }
 
     /**
      * Compared item with {@link #END_ITEM} by identity.
      *
      * @param item    element from <code>items</code>
      * @return {@code true} if tail of the queue
      */
     private static boolean shouldStopQueueing(Event item) {
         return item == END_ITEM;
     }
 
     /**
      *
      */
     private static class FinalEvent extends Event {
         FinalEvent() {
             super(null);
         }
 
         @Override
         public boolean isControlCategory() {
             return false;
         }
 
         @Override
         public boolean isConsoleCategory() {
             return false;
         }
 
         @Override
         public boolean isConsoleErrorCategory() {
             return false;
         }
 
         @Override
         public boolean isStandardStreamCategory() {
             return false;
         }
 
         @Override
         public boolean isSysPropCategory() {
             return false;
         }
 
         @Override
         public boolean isTestCategory() {
             return false;
         }
 
         @Override
         public boolean isJvmExitError() {
             return false;
         }
     }
 
     /**
      * This synchronization helper mostly avoids the locks.
      * If the queue size has reached zero or {@code maxQueueSize} then the threads are locked (parked/unparked).
      * The thread instance T1 is reader (see the class "Pumper") and T2 is the writer (see the method "handleEvent").
      *
      * @param <T> element type in the queue
      */
     static class QueueSynchronizer<T> {
         private final SyncT1 t1 = new SyncT1();
         private final SyncT2 t2 = new SyncT2();
         private final ConcurrentLinkedDeque<T> queue = new ConcurrentLinkedDeque<>();
         private final AtomicInteger queueSize = new AtomicInteger();
         private final int maxQueueSize;
         private final T stopItemMarker;
 
         QueueSynchronizer(int maxQueueSize, T stopItemMarker) {
             this.maxQueueSize = maxQueueSize;
             this.stopItemMarker = stopItemMarker;
         }
 
         private class SyncT1 extends AbstractQueuedSynchronizer {
             private static final long serialVersionUID = 1L;
 
             @Override
             protected int tryAcquireShared(int arg) {
                 return queueSize.get() == 0 ? -1 : 1;
             }
 
             @Override
             protected boolean tryReleaseShared(int arg) {
                 return true;
             }
 
             void waitIfZero() throws InterruptedException {
                 acquireSharedInterruptibly(1);
             }
 
             void release() {
                 releaseShared(0);
             }
         }
 
         private class SyncT2 extends AbstractQueuedSynchronizer {
             private static final long serialVersionUID = 1L;
 
             @Override
             protected int tryAcquireShared(int arg) {
                 return queueSize.get() < maxQueueSize ? 1 : -1;
             }
 
             @Override
             protected boolean tryReleaseShared(int arg) {
                 return true;
             }
 
             void awaitMax() {
                 acquireShared(1);
             }
 
             void tryRelease() {
                 if (queueSize.get() == 0) {
                     releaseShared(0);
                 }
             }
         }
 
         void markStopped() {
             addNext(stopItemMarker);
         }
 
         void pushNext(T t) {
             t2.awaitMax();
             addNext(t);
         }
 
         T awaitNext() throws InterruptedException {
             t2.tryRelease();
             t1.waitIfZero();
             queueSize.decrementAndGet();
             return queue.pollFirst();
         }
 
         boolean isEmptyQueue() {
             return queue.isEmpty();
         }
 
         void clearQueue() {
             queue.clear();
         }
 
         private void addNext(T t) {
             queue.addLast(t);
             if (queueSize.getAndIncrement() == 0) {
                 t1.release();
             }
         }
     }
 }