package io.github.tramchamploo.bufferslayer;

import com.google.common.base.Preconditions;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Record total messages in queue
 */
abstract class MemoryLimiter {

  private static final int MAX_TOTAL_MESSGAES = 1_000_000;

  static MemoryLimiter maxOf(int maxMessages, ReporterMetrics metrics) {
    Preconditions.checkArgument(maxMessages > 0, "maxMessages should be greater than 0.");
    int result = Math.min(MAX_TOTAL_MESSGAES, maxMessages);
    return new DefaultMemoryLimiter(result, metrics);
  }

  abstract boolean isMaximum();

  abstract void waitWhenMaximum();

  abstract void signalAll();

  private static final class DefaultMemoryLimiter extends MemoryLimiter {
    private static final Logger logger = LoggerFactory.getLogger(AsyncReporter.class);

    private final long maxMessages;
    private final ReporterMetrics metrics;

    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();

    private DefaultMemoryLimiter(long maxMessages, ReporterMetrics metrics) {
      this.maxMessages = maxMessages;
      this.metrics = metrics;
    }

    @Override
    boolean isMaximum() {
      return metrics.queuedMessages() >= maxMessages;
    }

    @Override
    void waitWhenMaximum() {
      lock.lock();
      try {
        while (isMaximum()) {
          notFull.await();
        }
      } catch (InterruptedException e) {
        logger.error("Interrupted waiting when full.");
      } finally {
        lock.unlock();
      }
    }

    @Override
    void signalAll() {
      lock.lock();
      try {
        notFull.signalAll();
      } finally {
        lock.unlock();
      }
    }
  }
}


@@ -1438,24 +1438,51 @@
 Maximum() %7B%0A
+      if (isMaximum()) %7B%0A  
       lock.l
@@ -1480,32 +1480,34 @@
    lock.lock();%0A
+  
       try %7B%0A    
@@ -1510,16 +1510,18 @@
         
+  
 while (i
@@ -1536,32 +1536,34 @@
 ()) %7B%0A          
+ 
+ 
 notFull.await();
@@ -1567,26 +1567,30 @@
 ();%0A        
-%7D%0A
+  %7D%0A  
       %7D catc
@@ -1626,16 +1626,18 @@
         
+ 
+ 
 logger.e
@@ -1672,24 +1672,72 @@
 en full.%22);%0A
+          Thread.currentThread().interrupt();%0A  
       %7D fina
@@ -1742,32 +1742,34 @@
 nally %7B%0A        
+  
 lock.unlock();%0A 
@@ -1759,32 +1759,42 @@
  lock.unlock();%0A
+        %7D%0A
       %7D%0A    %7D%0A%0A 

Check memory limit before aquiring lock