// Path: src/main/java/com/github/phantomthief/util/TickerBatchInvoker.java
// public class TickerBatchInvoker<K, V> implements Function<K, CompletableFuture<V>> {
// 
//     private final ThrowableFunction<Collection<K>, Map<K, V>, ? extends Throwable> batchInvoker;
//     private final Executor executor;
//     private final BufferTrigger<TwoTuple<K, CompletableFuture<V>>> bufferTrigger;
// 
//     private TickerBatchInvoker(long ticker,
//             ThrowableFunction<Collection<K>, Map<K, V>, ? extends Throwable> batchInvoker,
//             Executor executor) {
//         this.batchInvoker = batchInvoker;
//         this.executor = executor;
//         this.bufferTrigger = SimpleBufferTrigger.newBuilder()
//                 .setContainer(ConcurrentHashMap::new, this::enqueue)
//                 .on(ticker, MILLISECONDS, 1)
//                 .consumer(this::batchInvoke)
//                 .build();
//     }
// 
//     public static Builder newBuilder() {
//         return new Builder();
//     }
// 
//     private boolean enqueue(Map<K, List<CompletableFuture<V>>> map,
//             TwoTuple<K, CompletableFuture<V>> e) {
//         map.compute(e.getFirst(), (k, list) -> {
//             if (list == null) {
//                 list = new ArrayList<>();
//             }
//             synchronized (list) {
//                 list.add(e.getSecond());
//             }
//             return list;
//         });
//         return true;
//     }
// 
//     private void batchInvoke(Map<K, List<CompletableFuture<V>>> map) {
//         executor.execute(() -> {
//             try {
//                 Map<K, V> result = batchInvoker.apply(map.keySet());
//                 map.forEach((key, futures) -> {
//                     V v = result.get(key);
//                     // 虽然框架会尽力保证enqueue/consume是互斥的,但是这里还是重复保证下
//                     synchronized (futures) {
//                         for (CompletableFuture<V> future : futures) {
//                             future.complete(v);
//                         }
//                     }
//                 });
//             } catch (Throwable e) {
//                 for (List<CompletableFuture<V>> futures : map.values()) {
//                     synchronized (futures) {
//                         futures.stream()
//                                 .filter(future -> !future.isDone())
//                                 .forEach(future -> future.completeExceptionally(e));
//                     }
//                 }
//             }
//         });
//     }
// 
//     @Override
//     public CompletableFuture<V> apply(K key) {
//         CompletableFuture<V> future = new CompletableFuture<>();
//         bufferTrigger.enqueue(tuple(key, future));
//         return future;
//     }
// 
//     public static class Builder {
// 
//         private long ticker;
//         private Executor executor;
// 
//         private Builder() {
//         }
// 
//         public Builder ticker(long time, TimeUnit unit) {
//             this.ticker = unit.toMillis(time);
//             return this;
//         }
// 
//         public Builder executor(Executor executor) {
//             this.executor = executor;
//             return this;
//         }
// 
//         public Builder threads(int nThreads) {
//             this.executor = Executors.newFixedThreadPool(nThreads);
//             return this;
//         }
// 
//         public <K, V> TickerBatchInvoker<K, V> build(
//                 ThrowableFunction<Collection<K>, Map<K, V>, ? extends Throwable> batchInvoker) {
//             checkNotNull(batchInvoker);
//             ensure();
//             return new TickerBatchInvoker<>(ticker, batchInvoker, executor);
//         }
// 
//         private void ensure() {
//             if (ticker <= 0) {
//                 ticker = SECONDS.toMillis(1);
//             }
//             if (executor == null) {
//                 executor = Executors.newCachedThreadPool();
//             }
//         }
//     }
// }


import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.util.Collections.emptyMap;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.function.Function.identity;
import static java.util.stream.Collectors.toMap;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.fail;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
import org.junit.jupiter.api.Test;
import com.github.phantomthief.util.TickerBatchInvoker;
package com.github.phantomthief.test;





/**
 * @author w.vela
 * Created on 16/5/21.
 */
class TickerBatchInvokerTest {

    private Map<Integer, String> load(Collection<Integer> keys) {
        System.out.println("invoke:" + keys);
        sleepUninterruptibly(2, SECONDS);
        return keys.stream().collect(toMap(identity(), Object::toString));
    }

    private Map<Integer, String> loadEmpty(Collection<Integer> keys) {
        System.out.println("invoke empty:" + keys);
        return emptyMap();
    }

    private Map<Integer, String> loadException(Collection<Integer> keys)
            throws IllegalStateException {
        System.out.println("invoke empty:" + keys);
        throw new IllegalStateException("test");
    }

    @Test
    void test() {
// Path: src/main/java/com/github/phantomthief/util/TickerBatchInvoker.java
// public class TickerBatchInvoker<K, V> implements Function<K, CompletableFuture<V>> {
// 
//     private final ThrowableFunction<Collection<K>, Map<K, V>, ? extends Throwable> batchInvoker;
//     private final Executor executor;
//     private final BufferTrigger<TwoTuple<K, CompletableFuture<V>>> bufferTrigger;
// 
//     private TickerBatchInvoker(long ticker,
//             ThrowableFunction<Collection<K>, Map<K, V>, ? extends Throwable> batchInvoker,
//             Executor executor) {
//         this.batchInvoker = batchInvoker;
//         this.executor = executor;
//         this.bufferTrigger = SimpleBufferTrigger.newBuilder()
//                 .setContainer(ConcurrentHashMap::new, this::enqueue)
//                 .on(ticker, MILLISECONDS, 1)
//                 .consumer(this::batchInvoke)
//                 .build();
//     }
// 
//     public static Builder newBuilder() {
//         return new Builder();
//     }
// 
//     private boolean enqueue(Map<K, List<CompletableFuture<V>>> map,
//             TwoTuple<K, CompletableFuture<V>> e) {
//         map.compute(e.getFirst(), (k, list) -> {
//             if (list == null) {
//                 list = new ArrayList<>();
//             }
//             synchronized (list) {
//                 list.add(e.getSecond());
//             }
//             return list;
//         });
//         return true;
//     }
// 
//     private void batchInvoke(Map<K, List<CompletableFuture<V>>> map) {
//         executor.execute(() -> {
//             try {
//                 Map<K, V> result = batchInvoker.apply(map.keySet());
//                 map.forEach((key, futures) -> {
//                     V v = result.get(key);
//                     // 虽然框架会尽力保证enqueue/consume是互斥的,但是这里还是重复保证下
//                     synchronized (futures) {
//                         for (CompletableFuture<V> future : futures) {
//                             future.complete(v);
//                         }
//                     }
//                 });
//             } catch (Throwable e) {
//                 for (List<CompletableFuture<V>> futures : map.values()) {
//                     synchronized (futures) {
//                         futures.stream()
//                                 .filter(future -> !future.isDone())
//                                 .forEach(future -> future.completeExceptionally(e));
//                     }
//                 }
//             }
//         });
//     }
// 
//     @Override
//     public CompletableFuture<V> apply(K key) {
//         CompletableFuture<V> future = new CompletableFuture<>();
//         bufferTrigger.enqueue(tuple(key, future));
//         return future;
//     }
// 
//     public static class Builder {
// 
//         private long ticker;
//         private Executor executor;
// 
//         private Builder() {
//         }
// 
//         public Builder ticker(long time, TimeUnit unit) {
//             this.ticker = unit.toMillis(time);
//             return this;
//         }
// 
//         public Builder executor(Executor executor) {
//             this.executor = executor;
//             return this;
//         }
// 
//         public Builder threads(int nThreads) {
//             this.executor = Executors.newFixedThreadPool(nThreads);
//             return this;
//         }
// 
//         public <K, V> TickerBatchInvoker<K, V> build(
//                 ThrowableFunction<Collection<K>, Map<K, V>, ? extends Throwable> batchInvoker) {
//             checkNotNull(batchInvoker);
//             ensure();
//             return new TickerBatchInvoker<>(ticker, batchInvoker, executor);
//         }
// 
//         private void ensure() {
//             if (ticker <= 0) {
//                 ticker = SECONDS.toMillis(1);
//             }
//             if (executor == null) {
//                 executor = Executors.newCachedThreadPool();
//             }
//         }
//     }
// }


// Path: src/test/java/com/github/phantomthief/test/TickerBatchInvokerTest.java
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.util.Collections.emptyMap;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.function.Function.identity;
import static java.util.stream.Collectors.toMap;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.fail;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
import org.junit.jupiter.api.Test;
import com.github.phantomthief.util.TickerBatchInvoker;

package com.github.phantomthief.test;





/**
 * @author w.vela
 * Created on 16/5/21.
 */
class TickerBatchInvokerTest {

    private Map<Integer, String> load(Collection<Integer> keys) {
        System.out.println("invoke:" + keys);
        sleepUninterruptibly(2, SECONDS);
        return keys.stream().collect(toMap(identity(), Object::toString));
    }

    private Map<Integer, String> loadEmpty(Collection<Integer> keys) {
        System.out.println("invoke empty:" + keys);
        return emptyMap();
    }

    private Map<Integer, String> loadException(Collection<Integer> keys)
            throws IllegalStateException {
        System.out.println("invoke empty:" + keys);
        throw new IllegalStateException("test");
    }

    @Test
    void test() {
        TickerBatchInvoker<Integer, String> batchInvoker = TickerBatchInvoker.newBuilder()