package io.trane.ndbc.pool;

import java.time.Duration;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.function.Function;
import java.util.function.Supplier;

import io.trane.future.Future;
import io.trane.future.Promise;

public class Pool<T extends Pool.Item> {

  private static final Future<Object> POOL_EXHAUSTED = Future.exception(new RuntimeException("Pool exhausted"));

  public interface Item {
    Future<Void> release();

    Future<Boolean> validate();
  }

  public static <T extends Pool.Item> Pool<T> apply(Supplier<Future<T>> supplier, int maxSize, int maxWaiters,
      Duration validationInterval) {
    return new Pool<>(supplier, maxSize, maxWaiters, validationInterval);
  }

  private final Supplier<Future<T>> supplier;
  private final Semaphore sizeSemaphore;
  private final Semaphore waitersSemaphore;
  private final Queue<T> items;
  private final Queue<Waiter<T, ?>> waiters;

  private Pool(Supplier<Future<T>> supplier, int maxSize, int maxWaiters, Duration validationInterval) {
    this.supplier = supplier;
    this.sizeSemaphore = semaphore(maxSize);
    this.waitersSemaphore = semaphore(maxWaiters);
    this.items = new ConcurrentLinkedQueue<>();
    this.waiters = new ConcurrentLinkedQueue<>();
    if (validationInterval.toMillis() != Long.MAX_VALUE)
      scheduleValidation(validationInterval, new ScheduledThreadPoolExecutor(1));
  }

  public <R> Future<R> apply(Function<T, Future<R>> f) {
    final T item = items.poll();
    if (item != null)
      return f.apply(item).ensure(() -> release(item));
    else if (sizeSemaphore.tryAcquire()) {
      return supplier.get().flatMap(i -> f.apply(i).ensure(() -> release(i)));
    } else if (waitersSemaphore.tryAcquire()) {
      Waiter<T, R> p = new Waiter<>(f);
      waiters.offer(p);
      return p;
    } else
      return POOL_EXHAUSTED.unsafeCast();
  }

  private final void release(T item) {
    Waiter<T, ?> waiter = waiters.poll();
    if (waiter != null) {
      waitersSemaphore.release();
      waiter.apply(item).ensure(() -> release(item));
    } else
      items.offer(item);
  };

  private Future<Void> validateN(int n) {
    if (n >= 0) {
      final T item = items.poll();
      if (item == null) {
        return Future.VOID;
      } else
        // TODO logging
        return item.validate().rescue(e -> Future.FALSE).flatMap(valid -> {
          if (!valid) {
            sizeSemaphore.release();
            return item.release().rescue(e -> Future.VOID);
          } else {
            items.offer(item);
            return Future.VOID;
          }
        }).flatMap(v -> validateN(n - 1));
    } else
      return Future.VOID;
  }

  private Future<Void> scheduleValidation(Duration validationInterval, ScheduledExecutorService scheduler) {
    return Future.VOID.delayed(validationInterval, scheduler).flatMap(v1 -> {
      long start = System.currentTimeMillis();
      return validateN(items.size()).flatMap(v2 -> {
        long next = validationInterval.toMillis() - System.currentTimeMillis() - start;
        if (next <= 0) {
          // TODO logging
          return scheduleValidation(validationInterval, scheduler);
        } else
          return scheduleValidation(Duration.ofMillis(next), scheduler);
      });
    });
  }

  private static class Waiter<T, R> extends Promise<R> {

    private final Function<T, Future<R>> f;

    public Waiter(Function<T, Future<R>> f) {
      super();
      this.f = f;
    }

    public Waiter<T, R> apply(T value) {
      become(f.apply(value));
      return this;
    }
  }

  private Semaphore semaphore(int permits) {
    if (permits == Integer.MAX_VALUE)
      return new Semaphore(permits) {
        private static final long serialVersionUID = 1L;

        @Override
        public void release() {
        }

        @Override
        public boolean tryAcquire() {
          return true;
        }
      };
    else
      return new Semaphore(permits);
  }
}

@@ -430,34 +430,16 @@
 s Pool%3CT
- extends Pool.Item
 %3E %7B%0A%0A  p
@@ -562,28 +562,74 @@
 lic 
-interface Item %7B%0A   
+static %3CT%3E Pool%3CT%3E apply(Supplier%3CFuture%3CT%3E%3E supplier, Function%3CT,
  Fut
@@ -637,24 +637,25 @@
 re%3CVoid%3E
+%3E
  release
 ();%0A%0A   
@@ -650,16 +650,27 @@
 ease
-();%0A%0A
+,%0A  
    
+ Function%3CT,
  Fut
@@ -681,16 +681,17 @@
 Boolean%3E
+%3E
  validat
@@ -695,97 +695,8 @@
 date
-();%0A  %7D%0A%0A  public static %3CT extends Pool.Item%3E Pool%3CT%3E apply(Supplier%3CFuture%3CT%3E%3E supplier
 , in
@@ -721,22 +721,16 @@
 Waiters,
-%0A     
  Duratio
@@ -784,16 +784,35 @@
 upplier,
+ release, validate,
  maxSize
@@ -897,16 +897,122 @@
 pplier;%0A
+  private final Function%3CT, Future%3CVoid%3E%3E release;%0A  private final Function%3CT, Future%3CBoolean%3E%3E validate;%0A
   privat
@@ -1210,16 +1210,96 @@
 upplier,
+ Function%3CT, Future%3CVoid%3E%3E release, Function%3CT, Future%3CBoolean%3E%3E validate,%0A     
  int max
@@ -1381,16 +1381,74 @@
 pplier;%0A
+    this.release = release;%0A    this.validate = validate;%0A
     this
@@ -1535,16 +1535,61 @@
 iters);%0A
+    // TODO is this the best data structure?%0A
     this
@@ -2026,26 +2026,24 @@
 ryAcquire())
- %7B
 %0A      retur
@@ -2109,26 +2109,24 @@
 se(i)));%0A   
- %7D
  else if (wa
@@ -2533,32 +2533,38 @@
   %7D;%0A%0A  private 
+final 
 Future%3CVoid%3E val
@@ -2742,22 +2742,27 @@
 urn 
-item.
 valida
+te.apply(i
 te
-(
+m
 ).re
@@ -2829,47 +2829,8 @@
 lid)
- %7B%0A            sizeSemaphore.release();
 %0A   
@@ -2849,21 +2849,26 @@
 urn 
-item.
 release
-(
+.apply(item
 ).re
@@ -2889,16 +2889,54 @@
 re.VOID)
+.ensure(() -%3E sizeSemaphore.release())
 ;%0A      
@@ -2934,26 +2934,24 @@
 );%0A         
- %7D
  else %7B%0A    

remove Pool.Item interface