@Override
@Nullable
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
	try {
		if (this.enterpriseConcurrentScheduler) {
			return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger);
		}
		else {
			ErrorHandler errorHandler =
					(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
			return new ReschedulingRunnable(task, trigger, this.scheduledExecutor, errorHandler).schedule();
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public <T> Future<T> submit(Callable<T> task) {
	try {
		if (this.taskDecorator == null && this.concurrentExecutor instanceof ExecutorService) {
			return ((ExecutorService) this.concurrentExecutor).submit(task);
		}
		else {
			FutureTask<T> future = new FutureTask<T>(task);
			doExecute(this.concurrentExecutor, this.taskDecorator, future);
			return future;
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException(
				"Executor [" + this.concurrentExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public Future<?> submit(Runnable task) {
	try {
		if (this.taskDecorator == null && this.concurrentExecutor instanceof ExecutorService) {
			return ((ExecutorService) this.concurrentExecutor).submit(task);
		}
		else {
			FutureTask<Object> future = new FutureTask<>(task, null);
			doExecute(this.concurrentExecutor, this.taskDecorator, future);
			return future;
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException(
				"Executor [" + this.concurrentExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public <T> Future<T> submit(Callable<T> task) {
	try {
		if (this.taskDecorator == null && this.concurrentExecutor instanceof ExecutorService) {
			return ((ExecutorService) this.concurrentExecutor).submit(task);
		}
		else {
			FutureTask<T> future = new FutureTask<>(task);
			doExecute(this.concurrentExecutor, this.taskDecorator, future);
			return future;
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException(
				"Executor [" + this.concurrentExecutor + "] did not accept task: " + task, ex);
	}
}
 

public void onMessage(ChannelHandlerContext ctx, ByteBuf message) {
    /*
     use thread pool first onMessage may block
    */
    Node node = (Node) (ctx.channel().attr(AttributeKey.valueOf("node")).get());

    if (threadPool == null) {
        callBack.onMessage(ctx, node, message);
    } else {
        try {
            threadPool.execute(
                    new Runnable() {
                        @Override
                        public void run() {
                            callBack.onMessage(ctx, node, message);
                        }
                    });
        } catch (TaskRejectedException e) {
            logger.warn(" TaskRejectedException : {}, message: {}", e, message);
        }
    }
}
 

private void runBlockHeaderCallbackTasks() {
    Queue<Runnable> tasks = getBlockHeaderCallbackTasks();

    ConcurrentLinkedQueue<Runnable> concurrentLinkedQueue = new ConcurrentLinkedQueue<>();
    setBlockHeaderCallbackTasks(concurrentLinkedQueue);

    boolean writeBack = false;
    for (Runnable task : tasks) {
        if (writeBack) {
            concurrentLinkedQueue.add(task);
            if (logger.isDebugEnabled()) {
                logger.debug(" write task back to queue, task: {}", task);
            }
        } else {
            try {
                threadPool.execute(task);
            } catch (TaskRejectedException e) {
                logger.warn(" TaskRejectedException: {}", e);
                concurrentLinkedQueue.add(task);
                writeBack = true;
            }
        }
    }
}
 

@Override
public Future<?> submit(Runnable task) {
	try {
		if (this.taskDecorator == null && this.concurrentExecutor instanceof ExecutorService) {
			return ((ExecutorService) this.concurrentExecutor).submit(task);
		}
		else {
			FutureTask<Object> future = new FutureTask<Object>(task, null);
			doExecute(this.concurrentExecutor, this.taskDecorator, future);
			return future;
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException(
				"Executor [" + this.concurrentExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public Future<?> submit(Runnable task) {
	try {
		if (this.taskDecorator == null && this.concurrentExecutor instanceof ExecutorService) {
			return ((ExecutorService) this.concurrentExecutor).submit(task);
		}
		else {
			FutureTask<Object> future = new FutureTask<>(task, null);
			doExecute(this.concurrentExecutor, this.taskDecorator, future);
			return future;
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException(
				"Executor [" + this.concurrentExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public <T> Future<T> submit(Callable<T> task) {
	try {
		if (this.taskDecorator == null && this.concurrentExecutor instanceof ExecutorService) {
			return ((ExecutorService) this.concurrentExecutor).submit(task);
		}
		else {
			FutureTask<T> future = new FutureTask<>(task);
			doExecute(this.concurrentExecutor, this.taskDecorator, future);
			return future;
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException(
				"Executor [" + this.concurrentExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
	try {
		if (this.enterpriseConcurrentScheduler) {
			return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger);
		}
		else {
			ErrorHandler errorHandler =
					(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
			return new ReschedulingRunnable(task, trigger, this.scheduledExecutor, errorHandler).schedule();
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public void execute(Runnable task) {
	Executor executor = getThreadPoolExecutor();
	try {
		executor.execute(task);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public Future<?> submit(Runnable task) {
	ExecutorService executor = getThreadPoolExecutor();
	try {
		return executor.submit(task);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public <T> Future<T> submit(Callable<T> task) {
	ExecutorService executor = getThreadPoolExecutor();
	try {
		return executor.submit(task);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ListenableFuture<?> submitListenable(Runnable task) {
	ExecutorService executor = getThreadPoolExecutor();
	try {
		ListenableFutureTask<Object> future = new ListenableFutureTask<>(task, null);
		executor.execute(future);
		return future;
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

/**
 * Delegates to the specified JDK concurrent executor.
 * @see java.util.concurrent.Executor#execute(Runnable)
 */
@Override
public void execute(Runnable task) {
	try {
		doExecute(this.concurrentExecutor, this.taskDecorator, task);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException(
				"Executor [" + this.concurrentExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> schedule(Runnable task, Date startTime) {
	long initialDelay = startTime.getTime() - System.currentTimeMillis();
	try {
		return this.scheduledExecutor.schedule(decorateTask(task, false), initialDelay, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Date startTime, long period) {
	long initialDelay = startTime.getTime() - System.currentTimeMillis();
	try {
		return this.scheduledExecutor.scheduleAtFixedRate(decorateTask(task, true), initialDelay, period, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long period) {
	try {
		return this.scheduledExecutor.scheduleAtFixedRate(decorateTask(task, true), 0, period, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Date startTime, long delay) {
	long initialDelay = startTime.getTime() - System.currentTimeMillis();
	try {
		return this.scheduledExecutor.scheduleWithFixedDelay(decorateTask(task, true), initialDelay, delay, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long delay) {
	try {
		return this.scheduledExecutor.scheduleWithFixedDelay(decorateTask(task, true), 0, delay, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public void execute(Runnable task) {
	Executor executor = getScheduledExecutor();
	try {
		executor.execute(errorHandlingTask(task, false));
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public Future<?> submit(Runnable task) {
	ExecutorService executor = getScheduledExecutor();
	try {
		return executor.submit(errorHandlingTask(task, false));
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ListenableFuture<?> submitListenable(Runnable task) {
	ExecutorService executor = getScheduledExecutor();
	try {
		ListenableFutureTask<Object> listenableFuture = new ListenableFutureTask<>(task, null);
		executeAndTrack(executor, listenableFuture);
		return listenableFuture;
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
	ExecutorService executor = getScheduledExecutor();
	try {
		ListenableFutureTask<T> listenableFuture = new ListenableFutureTask<>(task);
		executeAndTrack(executor, listenableFuture);
		return listenableFuture;
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
@Nullable
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
	ScheduledExecutorService executor = getScheduledExecutor();
	try {
		ErrorHandler errorHandler = this.errorHandler;
		if (errorHandler == null) {
			errorHandler = TaskUtils.getDefaultErrorHandler(true);
		}
		return new ReschedulingRunnable(task, trigger, executor, errorHandler).schedule();
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> schedule(Runnable task, Date startTime) {
	ScheduledExecutorService executor = getScheduledExecutor();
	long initialDelay = startTime.getTime() - System.currentTimeMillis();
	try {
		return executor.schedule(errorHandlingTask(task, false), initialDelay, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Date startTime, long period) {
	ScheduledExecutorService executor = getScheduledExecutor();
	long initialDelay = startTime.getTime() - System.currentTimeMillis();
	try {
		return executor.scheduleAtFixedRate(errorHandlingTask(task, true), initialDelay, period, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long period) {
	ScheduledExecutorService executor = getScheduledExecutor();
	try {
		return executor.scheduleAtFixedRate(errorHandlingTask(task, true), 0, period, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Date startTime, long delay) {
	ScheduledExecutorService executor = getScheduledExecutor();
	long initialDelay = startTime.getTime() - System.currentTimeMillis();
	try {
		return executor.scheduleWithFixedDelay(errorHandlingTask(task, true), initialDelay, delay, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long delay) {
	ScheduledExecutorService executor = getScheduledExecutor();
	try {
		return executor.scheduleWithFixedDelay(errorHandlingTask(task, true), 0, delay, TimeUnit.MILLISECONDS);
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}