code: /*
   * Copyright 2018 Mitsunori Komatsu (komamitsu)
   *
   * Licensed 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.komamitsu.fluency.flusher;
  
  import org.komamitsu.fluency.buffer.Buffer;
  import org.komamitsu.fluency.ingester.Ingester;
  import org.komamitsu.fluency.util.ExecutorServiceUtils;
  import org.komamitsu.fluency.validation.Validatable;
  import org.komamitsu.fluency.validation.annotation.Max;
  import org.komamitsu.fluency.validation.annotation.Min;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.Closeable;
  import java.io.Flushable;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  public class Flusher
          implements Flushable, Closeable
  {
      private static final Logger LOG = LoggerFactory.getLogger(Flusher.class);
      private static final int DEFAULT_EVENT_QUEUE_SIZE = 16;
      protected final Buffer buffer;
      protected final Ingester ingester;
      private final AtomicBoolean isTerminated = new AtomicBoolean();
      private final Config config;
      private final BlockingQueue<Boolean> eventQueue = new LinkedBlockingQueue<>(DEFAULT_EVENT_QUEUE_SIZE);
      private final ExecutorService executorService = ExecutorServiceUtils.newSingleThreadDaemonExecutor();
  
      public Flusher(Config config, Buffer buffer, Ingester ingester)
      {
          config.validateValues();
          this.config = config;
          this.buffer = buffer;
          this.ingester = ingester;
          executorService.execute(this::runLoop);
      }
  
      private void runLoop() {
          Boolean wakeup = null;
          do {
              try {
                  wakeup = eventQueue.poll(config.getFlushAttemptIntervalMillis(), TimeUnit.MILLISECONDS);
                  boolean force = wakeup != null;
                  buffer.flush(ingester, force);
              }
              catch (InterruptedException e) {
                  Thread.currentThread().interrupt();
              }
              catch (Throwable e) {
                  LOG.error("Failed to flush", e);
              }
          }
          while (!executorService.isShutdown());
  
          if (wakeup == null) {
              // The above run loop can quit without force buffer flush in the following cases
              // - close() is called right after the repeated non-force buffer flush executed in the run loop
              //
              // In these cases, remaining buffers won't be flushed.
              // So force buffer flush is executed here just in case
              try {
                  buffer.flush(ingester, true);
              }
              catch (Throwable e) {
                  LOG.error("Failed to flush", e);
              }
          }
      }
  
      public Buffer getBuffer()
      {
          return buffer;
      }
  
      @Override
      public void flush()
      {
          eventQueue.offer(true);
      }
  
      private void flushBufferQuietly()
      {
          LOG.trace("Flushing the buffer");
  
          try {
              flush();
          }
          catch (Throwable e) {
              LOG.error("Failed to call flush()", e);
          }
      }
  
      private void finishExecutorQuietly()
      {
          LOG.trace("Finishing the executor");
  
          ExecutorServiceUtils.finishExecutorService(
                  executorService,
                  config.getWaitUntilBufferFlushed());
      }
  
      private void closeBufferQuietly()
      {
          LOG.trace("Closing the buffer");
  
          try {
              buffer.close();
          }
          catch (Throwable e) {
              LOG.warn("Failed to close the buffer", e);
          }
      }
  
      private void closeIngesterQuietly()
      {
          LOG.trace("Closing the ingester");
  
          try {
              // Close the socket at the end to prevent the server from failing to read from the connection
              ingester.close();
          }
          catch (Throwable e) {
              LOG.error("Failed to close the ingester", e);
          }
      }
  
      private void closeResourcesQuietly()
      {
          ExecutorService executorService = Executors.newSingleThreadExecutor();
          try {
              Future<Void> future = executorService.submit(() -> {
                  closeBufferQuietly();
                  closeIngesterQuietly();
                  isTerminated.set(true);
                  return null;
              });
              future.get(getWaitUntilTerminated(), TimeUnit.SECONDS);
          }
          catch (InterruptedException e) {
              LOG.warn("Interrupted", e);
              Thread.currentThread().interrupt();
          }
          catch (ExecutionException e) {
              LOG.warn("closeBuffer() failed", e);
          }
          catch (TimeoutException e) {
              LOG.warn("closeBuffer() timed out", e);
          }
          finally {
              executorService.shutdown();;
          }
      }
  
      @Override
      public void close()
      {
          flushBufferQuietly();
  
          finishExecutorQuietly();
  
          closeResourcesQuietly();
      }
  
      public boolean isTerminated()
      {
          return isTerminated.get();
      }
  
      public Ingester getIngester()
      {
          return ingester;
      }
  
      /**
       * @deprecated  As of release 2.4.0, replaced by {@link #getFlushAttemptIntervalMillis()}
       */
      @Deprecated
      public int getFlushIntervalMillis()
      {
          return config.getFlushAttemptIntervalMillis();
      }
  
      public int getFlushAttemptIntervalMillis()
      {
          return config.getFlushAttemptIntervalMillis();
      }
  
      public int getWaitUntilBufferFlushed()
      {
          return config.getWaitUntilBufferFlushed();
      }
  
      public int getWaitUntilTerminated()
      {
          return config.getWaitUntilTerminated();
      }
  
      @Override
      public String toString()
      {
          return "Flusher{" +
                  "isTerminated=" + isTerminated +
                  ", buffer=" + buffer +
                  ", ingester=" + ingester +
                  ", config=" + config +
                  '}';
      }
  
      public static class Config
          implements Validatable
      {
          @Min(20)
          @Max(2000)
          private int flushAttemptIntervalMillis = 600;
          @Min(1)
          private int waitUntilBufferFlushed = 60;
          @Min(1)
          private int waitUntilTerminated = 60;
  
          /**
           * @deprecated  As of release 2.4.0, replaced by {@link #getFlushAttemptIntervalMillis()}
           */
          @Deprecated
          public int getFlushIntervalMillis()
          {
              return flushAttemptIntervalMillis;
          }
  
          /**
           * @deprecated  As of release 2.4.0, replaced by {@link #setFlushAttemptIntervalMillis(int flushAttemptIntervalMillis)}
           */
          @Deprecated
          public void setFlushIntervalMillis(int flushAttemptIntervalMillis)
          {
              this.flushAttemptIntervalMillis = flushAttemptIntervalMillis;
          }
  
          public int getFlushAttemptIntervalMillis()
          {
              return flushAttemptIntervalMillis;
          }
  
          public void setFlushAttemptIntervalMillis(int flushAttemptIntervalMillis)
          {
              this.flushAttemptIntervalMillis = flushAttemptIntervalMillis;
          }
  
          public int getWaitUntilBufferFlushed()
          {
              return waitUntilBufferFlushed;
          }
  
          public void setWaitUntilBufferFlushed(int waitUntilBufferFlushed)
          {
              this.waitUntilBufferFlushed = waitUntilBufferFlushed;
          }
  
          public int getWaitUntilTerminated()
          {
              return waitUntilTerminated;
          }
  
          public void setWaitUntilTerminated(int waitUntilTerminated)
          {
              this.waitUntilTerminated = waitUntilTerminated;
          }
  
          void validateValues()
          {
              validate();
          }
  
          @Override
          public String toString()
          {
              return "Config{" +
                      "flushAttemptIntervalMillis=" + flushAttemptIntervalMillis +
                      ", waitUntilBufferFlushed=" + waitUntilBufferFlushed +
                      ", waitUntilTerminated=" + waitUntilTerminated +
                      '}';
          }
      }
  }
  