View Javadoc

1   package com.atlassian.sal.core.executor;
2   
3   import com.google.common.base.Function;
4   import com.google.common.collect.Collections2;
5   
6   import java.util.Collection;
7   import java.util.List;
8   import java.util.concurrent.Callable;
9   import java.util.concurrent.ExecutionException;
10  import java.util.concurrent.ExecutorService;
11  import java.util.concurrent.Future;
12  import java.util.concurrent.TimeUnit;
13  import java.util.concurrent.TimeoutException;
14  
15  /**
16   * Executor service that wraps executing callables and runnables in a wrapper that transfers the thread local state of
17   * the caller to the thread of the executing task.
18   *
19   * @since 2.0
20   */
21  public class ThreadLocalDelegateExecutorService implements ExecutorService
22  {
23      protected final ThreadLocalContextManager manager;
24      private final ExecutorService delegate;
25  
26      public ThreadLocalDelegateExecutorService(ThreadLocalContextManager manager, ExecutorService delegate)
27      {
28          this.manager = manager;
29          this.delegate = delegate;
30      }
31  
32      public void shutdown()
33      {
34          delegate.shutdown();
35      }
36  
37      public List<Runnable> shutdownNow()
38      {
39          return delegate.shutdownNow();
40      }
41  
42      public boolean isShutdown()
43      {
44          return delegate.isShutdown();
45      }
46  
47      public boolean isTerminated()
48      {
49          return delegate.isTerminated();
50      }
51  
52      public boolean awaitTermination(long timeout, TimeUnit unit)
53          throws InterruptedException
54      {
55          return delegate.awaitTermination(timeout, unit);
56      }
57  
58      public <T> Future<T> submit(Callable<T> callable)
59      {
60          return delegate.submit(threadLocalDelegateCallable(callable));
61      }
62  
63      public <T> Future<T> submit(Runnable runnable, T t)
64      {
65          return delegate.submit(threadLocalDelegateRunnable(runnable), t);
66      }
67  
68      public Future<?> submit(Runnable runnable)
69      {
70          return delegate.submit(threadLocalDelegateRunnable(runnable));
71      }
72  
73      public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> callables)
74          throws InterruptedException
75      {
76          return delegate.invokeAll(threadLocalDelegateCallableCollection(callables));
77      }
78  
79      public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> callables, long timeout, TimeUnit unit)
80          throws InterruptedException
81      {
82          return delegate.invokeAll(threadLocalDelegateCallableCollection(callables), timeout, unit);
83      }
84  
85      public <T> T invokeAny(Collection<? extends Callable<T>> callables)
86          throws InterruptedException, ExecutionException
87      {
88          return delegate.invokeAny(threadLocalDelegateCallableCollection(callables));
89      }
90  
91      public <T> T invokeAny(Collection<? extends Callable<T>> callables, long timeout, TimeUnit unit)
92          throws InterruptedException, ExecutionException, TimeoutException
93      {
94          return delegate.invokeAny(threadLocalDelegateCallableCollection(callables), timeout, unit);
95      }
96  
97      public void execute(Runnable runnable)
98      {
99          delegate.execute(threadLocalDelegateRunnable(runnable));
100     }
101 
102     private ThreadLocalDelegateRunnable threadLocalDelegateRunnable(Runnable runnable)
103     {
104         return new ThreadLocalDelegateRunnable(manager, runnable);
105     }
106 
107     private <T> ThreadLocalDelegateCallable<T> threadLocalDelegateCallable(Callable<T> callable)
108     {
109         return new ThreadLocalDelegateCallable<T>(manager, callable);
110     }
111 
112     private <T> Collection<ThreadLocalDelegateCallable<T>> threadLocalDelegateCallableCollection(Collection<? extends Callable<T>> callables)
113     {
114         return Collections2.transform(callables, new Function<Callable<T>, ThreadLocalDelegateCallable<T>>()
115         {
116             public ThreadLocalDelegateCallable<T> apply(Callable<T> callable)
117             {
118                 return threadLocalDelegateCallable(callable);
119             }
120         });
121     }
122 }