/**
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * 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.
 * </p>
 */

package com.dangdang.ddframe.rdb.sharding.executor;

import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;

import com.dangdang.ddframe.rdb.sharding.exception.ShardingJdbcException;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * 多线程执行框架.
 * 
 * @author gaohongtao
 */
@NoArgsConstructor(access = AccessLevel.PRIVATE)
@Slf4j
public final class ExecutorEngine {
    
    /**
     * 多线程执行任务.
     * 
     * @param inputs 输入参数
     * @param executeUnit 执行单元
     * @param <I> 入参类型
     * @param <O> 出参类型
     * @return 执行结果
     */
    public static <I, O> List<O> execute(final Collection<I> inputs, final ExecuteUnit<I, O> executeUnit) {
        ListenableFuture<List<O>> futures = submitFutures(inputs, executeUnit);
        addCallback(futures);
        return getFutureResults(futures);
    }
    
    /**
     * 多线程执行任务并归并结果.
     * 
     * @param inputs 执行入参
     * @param executeUnit 执行单元
     * @param mergeUnit 合并结果单元
     * @param <I> 入参类型
     * @param <M> 中间结果类型
     * @param <O> 最终结果类型
     * @return 执行结果
     */
    public static <I, M, O> O execute(final Collection<I> inputs, final ExecuteUnit<I, M> executeUnit, final MergeUnit<M, O> mergeUnit) {
        return mergeUnit.merge(execute(inputs, executeUnit));
    }
    
    private static <I, O> ListenableFuture<List<O>> submitFutures(final Collection<I> inputs, final ExecuteUnit<I, O> executeUnit) {
        Set<ListenableFuture<O>> result = new HashSet<>(inputs.size());
        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(inputs.size()));
        for (final I each : inputs) {
            result.add(service.submit(new Callable<O>() {
                
                @Override
                public O call() throws Exception {
                    return executeUnit.execute(each);
                }
            }));
        }
        service.shutdown();
        return Futures.allAsList(result);
    }
    
    private static <T> void addCallback(final ListenableFuture<T> allFutures) {
        Futures.addCallback(allFutures, new FutureCallback<T>() {// 使用 google common 的多线程执行框架
            
            @Override
            public void onSuccess(final T result) {
                log.trace("Concurrent execute result success {}", result);
            }
            
            @Override
            public void onFailure(final Throwable thrown) {
                log.error("Concurrent execute result error {}", thrown);
            }
        });
    }
    
    private static <O> O getFutureResults(final ListenableFuture<O> futures) {
        try {
            return futures.get();
        } catch (final InterruptedException | ExecutionException ex) {
            throw new ShardingJdbcException(ex);
        }
    }
}

@@ -879,27 +879,281 @@
 current.
-Executors;%0A
+LinkedBlockingQueue;%0Aimport java.util.concurrent.ThreadPoolExecutor;%0Aimport java.util.concurrent.TimeUnit;%0A%0Aimport com.dangdang.ddframe.rdb.sharding.api.config.ShardingConfiguration;%0Aimport com.dangdang.ddframe.rdb.sharding.api.config.ShardingConfigurationConstant;
 %0Aimport 
@@ -1511,69 +1511,8 @@
 ors;
-%0A%0Aimport lombok.AccessLevel;%0Aimport lombok.NoArgsConstructor;
 %0Aimp
@@ -1594,98 +1594,882 @@
 */%0A@
-NoArgsConstructor(access = AccessLevel.PRIVATE)%0A@Slf4j%0Apublic final class ExecutorEngine %7B
+Slf4j%0Apublic final class ExecutorEngine %7B%0A    %0A    private final ListeningExecutorService executorService;%0A    %0A    public ExecutorEngine(final ShardingConfiguration configuration) %7B%0A        executorService = MoreExecutors.listeningDecorator(MoreExecutors.getExitingExecutorService(%0A                new ThreadPoolExecutor(configuration.getConfig(ShardingConfigurationConstant.PARALLEL_EXECUTOR_WORKER_MIN_IDLE_SIZE, int.class),%0A                configuration.getConfig(ShardingConfigurationConstant.PARALLEL_EXECUTOR_WORKER_MAX_SIZE, int.class),%0A                configuration.getConfig(ShardingConfigurationConstant.PARALLEL_EXECUTOR_WORKER_MAX_IDLE_TIMEOUT, long.class),%0A                TimeUnit.valueOf(configuration.getConfig(ShardingConfigurationConstant.PARALLEL_EXECUTOR_WORKER_MAX_IDLE_TIMEOUT_TIME_UNIT)),%0A                new LinkedBlockingQueue%3CRunnable%3E())));%0A    %7D
 %0A   
@@ -2640,31 +2640,24 @@
 %0A    public 
-static 
 %3CI, O%3E List%3C
@@ -3134,23 +3134,16 @@
  public 
-static 
 %3CI, M, O
@@ -3339,23 +3339,16 @@
 private 
-static 
 %3CI, O%3E L
@@ -3529,130 +3529,8 @@
 ));%0A
-        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(inputs.size()));%0A
     
@@ -3586,17 +3586,25 @@
 ult.add(
-s
+executorS
 ervice.s
@@ -3826,36 +3826,8 @@
   %7D%0A
-        service.shutdown();%0A
     
@@ -3887,23 +3887,16 @@
 private 
-static 
 %3CT%3E void
@@ -4017,49 +4017,8 @@
 () %7B
-// %E4%BD%BF%E7%94%A8 google common %E7%9A%84%E5%A4%9A%E7%BA%BF%E7%A8%8B%E6%89%A7%E8%A1%8C%E6%A1%86%E6%9E%B6%0A            
 %0A   
@@ -4398,15 +4398,8 @@
 ate 
-static 
 %3CO%3E 

Update ExecutorEngine.java