|
1 | 1 | package com.jashmore.sqs.spring; |
2 | 2 |
|
3 | | -import static java.util.stream.Collectors.toList; |
4 | | - |
5 | 3 | import com.google.common.annotations.VisibleForTesting; |
| 4 | +import com.google.common.base.Suppliers; |
6 | 5 | import com.google.common.collect.ImmutableMap; |
7 | 6 | import com.google.common.collect.ImmutableSet; |
8 | 7 |
|
|
19 | 18 | import java.util.Map; |
20 | 19 | import java.util.Optional; |
21 | 20 | import java.util.Set; |
| 21 | +import java.util.concurrent.CompletableFuture; |
22 | 22 | import java.util.concurrent.ExecutionException; |
23 | | -import java.util.concurrent.ExecutorService; |
24 | | -import java.util.concurrent.Executors; |
25 | | -import java.util.concurrent.Future; |
26 | 23 | import java.util.concurrent.atomic.AtomicBoolean; |
27 | 24 | import java.util.function.Consumer; |
| 25 | +import java.util.function.Supplier; |
28 | 26 | import javax.annotation.Nonnull; |
29 | | -import javax.annotation.concurrent.GuardedBy; |
30 | 27 | import javax.annotation.concurrent.ThreadSafe; |
31 | 28 |
|
32 | 29 | /** |
|
38 | 35 | @Slf4j |
39 | 36 | @ThreadSafe |
40 | 37 | public class DefaultQueueContainerService implements QueueContainerService, ApplicationContextAware, SmartLifecycle { |
41 | | - /** |
42 | | - * Used to be able to start and stop containers concurrently. |
43 | | - */ |
44 | | - private final ExecutorService executorService; |
45 | | - |
46 | 38 | /** |
47 | 39 | * These {@link QueueWrapper}s should be injected by the spring application and therefore to add more wrappers into the system a corresponding bean |
48 | 40 | * with this interface must be included in the application. |
49 | 41 | */ |
50 | 42 | private final List<QueueWrapper> queueWrappers; |
51 | 43 |
|
52 | 44 | /** |
53 | | - * This contains all of the containers that have been created from wrapping the Spring Application's bean's methods. |
| 45 | + * This contains a supplier that can obtain all of the {@link MessageListenerContainer}s that have been built for this application. |
54 | 46 | * |
55 | | - * <p>This is only modified via the {@link #setApplicationContext(ApplicationContext)} method, which will only be called during the lifecycle of the spring |
56 | | - * application. This method protects from multiple calls to setting this application context so this will maintain its thread safety. |
| 47 | + * <p>This must be contained within a {@link Supplier} because the {@link QueueContainerService} can be dependency injected into the application and |
| 48 | + * because the construction of these containers needs to look at all beans there can be a cyclic dependency. |
57 | 49 | */ |
58 | | - @GuardedBy("this") |
59 | | - private Map<String, MessageListenerContainer> containers = null; |
| 50 | + private Supplier<Map<String, MessageListenerContainer>> containersLazilyLoaded; |
60 | 51 |
|
61 | 52 | /** |
62 | 53 | * Determines whether this container service is currently running in the Spring lifecycle. |
63 | 54 | */ |
64 | 55 | private AtomicBoolean isRunning = new AtomicBoolean(false); |
65 | 56 |
|
66 | 57 | public DefaultQueueContainerService(final List<QueueWrapper> queueWrappers) { |
67 | | - this.executorService = Executors.newCachedThreadPool(); |
68 | 58 | this.queueWrappers = queueWrappers; |
69 | 59 | } |
70 | 60 |
|
71 | | - /** |
72 | | - * Initialise all of the containers for this application by finding all bean methods that need to be wrapped. |
73 | | - */ |
74 | 61 | @Override |
75 | | - public synchronized void setApplicationContext(@Nonnull final ApplicationContext applicationContext) throws BeansException { |
76 | | - if (containers != null) { |
77 | | - log.warn("Trying to set application context when already set up previously"); |
78 | | - return; |
79 | | - } |
80 | | - |
81 | | - if (queueWrappers.isEmpty()) { |
82 | | - containers = ImmutableMap.of(); |
83 | | - return; |
84 | | - } |
85 | | - |
86 | | - log.debug("Initialising QueueContainerService..."); |
87 | | - final Map<String, MessageListenerContainer> messageContainers = new HashMap<>(); |
88 | | - |
89 | | - for (final String beanName : applicationContext.getBeanDefinitionNames()) { |
90 | | - final Object bean = applicationContext.getBean(beanName); |
91 | | - for (final Method method : bean.getClass().getMethods()) { |
92 | | - for (final QueueWrapper annotationProcessor : queueWrappers) { |
93 | | - if (annotationProcessor.canWrapMethod(method)) { |
94 | | - final IdentifiableMessageListenerContainer identifiableMessageListenerContainer = annotationProcessor.wrapMethod(bean, method); |
95 | | - if (messageContainers.containsKey(identifiableMessageListenerContainer.getIdentifier())) { |
96 | | - throw new IllegalStateException("Created two MessageListenerContainers with the same identifier: " |
97 | | - + identifiableMessageListenerContainer.getIdentifier()); |
98 | | - } |
99 | | - log.debug("Created MessageListenerContainer with id: {}", identifiableMessageListenerContainer.getIdentifier()); |
100 | | - messageContainers.put(identifiableMessageListenerContainer.getIdentifier(), identifiableMessageListenerContainer.getContainer()); |
101 | | - } |
102 | | - } |
103 | | - } |
104 | | - } |
105 | | - |
106 | | - this.containers = ImmutableMap.copyOf(messageContainers); |
| 62 | + public void setApplicationContext(@Nonnull final ApplicationContext applicationContext) throws BeansException { |
| 63 | + containersLazilyLoaded = Suppliers.memoize(() -> calculateMessageListenerContainers(queueWrappers, applicationContext)); |
107 | 64 | } |
108 | 65 |
|
109 | 66 | @Override |
110 | | - public synchronized void startAllContainers() { |
| 67 | + public void startAllContainers() { |
111 | 68 | runForAllContainers(MessageListenerContainer::start); |
112 | 69 | } |
113 | 70 |
|
114 | 71 | @Override |
115 | | - public synchronized void startContainer(final String queueIdentifier) { |
| 72 | + public void startContainer(final String queueIdentifier) { |
116 | 73 | runForQueue(queueIdentifier, MessageListenerContainer::start); |
117 | 74 | } |
118 | 75 |
|
119 | 76 | @Override |
120 | | - public synchronized void stopAllContainers() { |
| 77 | + public void stopAllContainers() { |
121 | 78 | runForAllContainers(MessageListenerContainer::stop); |
122 | 79 | } |
123 | 80 |
|
124 | 81 | @Override |
125 | | - public synchronized void stopContainer(final String queueIdentifier) { |
| 82 | + public void stopContainer(final String queueIdentifier) { |
126 | 83 | runForQueue(queueIdentifier, MessageListenerContainer::stop); |
127 | 84 | } |
128 | 85 |
|
| 86 | + /** |
| 87 | + * For each of the containers run the following {@link Consumer} asynchronously and wait for them all to finish. |
| 88 | + * |
| 89 | + * @param containerConsumer the consumer to call |
| 90 | + */ |
129 | 91 | private void runForAllContainers(final Consumer<MessageListenerContainer> containerConsumer) { |
130 | | - final List<? extends Future<?>> taskFutures = containers.values().stream() |
131 | | - .map(container -> executorService.submit(() -> containerConsumer.accept(container))) |
132 | | - .collect(toList()); |
133 | | - |
134 | | - for (final Future<?> future : taskFutures) { |
135 | | - try { |
136 | | - future.get(); |
137 | | - } catch (InterruptedException interruptedException) { |
138 | | - log.warn("Thread interrupted while running command across all containers"); |
139 | | - return; |
140 | | - } catch (ExecutionException executionException) { |
141 | | - log.error("Error running command on container", executionException); |
142 | | - } |
| 92 | + final CompletableFuture<?>[] allTaskCompletableFutures = containersLazilyLoaded.get().values().stream() |
| 93 | + .map(container -> CompletableFuture.runAsync(() -> containerConsumer.accept(container))) |
| 94 | + .toArray(CompletableFuture[]::new); |
| 95 | + |
| 96 | + try { |
| 97 | + CompletableFuture.allOf(allTaskCompletableFutures).get(); |
| 98 | + } catch (InterruptedException interruptedException) { |
| 99 | + log.warn("Thread interrupted while running command across all containers"); |
| 100 | + } catch (ExecutionException executionException) { |
| 101 | + log.error("Error running command on container", executionException); |
143 | 102 | } |
144 | 103 | } |
145 | 104 |
|
146 | | - private void runForQueue(final String queueIdentifier, Consumer<MessageListenerContainer> runnable) { |
147 | | - final MessageListenerContainer container = Optional.ofNullable(containers.get(queueIdentifier)) |
| 105 | + /** |
| 106 | + * For the given queue with the identifier run the following {@link Consumer} for the container and wait until it is finished. |
| 107 | + * |
| 108 | + * @param queueIdentifier the identifier of the queue |
| 109 | + * @param containerConsumer the container consumer to run |
| 110 | + */ |
| 111 | + private void runForQueue(final String queueIdentifier, final Consumer<MessageListenerContainer> containerConsumer) { |
| 112 | + final MessageListenerContainer container = Optional.ofNullable(containersLazilyLoaded.get().get(queueIdentifier)) |
148 | 113 | .orElseThrow(() -> new IllegalArgumentException("No container with the provided identifier")); |
149 | 114 |
|
150 | | - runnable.accept(container); |
| 115 | + containerConsumer.accept(container); |
151 | 116 | } |
152 | 117 |
|
153 | 118 | @Override |
@@ -191,6 +156,39 @@ public synchronized int getPhase() { |
191 | 156 |
|
192 | 157 | @VisibleForTesting |
193 | 158 | synchronized Set<MessageListenerContainer> getContainers() { |
194 | | - return ImmutableSet.copyOf(containers.values()); |
| 159 | + return ImmutableSet.copyOf(containersLazilyLoaded.get().values()); |
| 160 | + } |
| 161 | + |
| 162 | + /** |
| 163 | + * Initialise all of the containers for this application by finding all bean methods that need to be wrapped. |
| 164 | + */ |
| 165 | + private static Map<String, MessageListenerContainer> calculateMessageListenerContainers( |
| 166 | + @Nonnull final List<QueueWrapper> queueWrappers, |
| 167 | + @Nonnull final ApplicationContext applicationContext) { |
| 168 | + if (queueWrappers.isEmpty()) { |
| 169 | + return ImmutableMap.of(); |
| 170 | + } |
| 171 | + |
| 172 | + log.debug("Initialising QueueContainerService..."); |
| 173 | + final Map<String, MessageListenerContainer> messageContainers = new HashMap<>(); |
| 174 | + |
| 175 | + for (final String beanName : applicationContext.getBeanDefinitionNames()) { |
| 176 | + final Object bean = applicationContext.getBean(beanName); |
| 177 | + for (final Method method : bean.getClass().getMethods()) { |
| 178 | + for (final QueueWrapper annotationProcessor : queueWrappers) { |
| 179 | + if (annotationProcessor.canWrapMethod(method)) { |
| 180 | + final IdentifiableMessageListenerContainer identifiableMessageListenerContainer = annotationProcessor.wrapMethod(bean, method); |
| 181 | + if (messageContainers.containsKey(identifiableMessageListenerContainer.getIdentifier())) { |
| 182 | + throw new IllegalStateException("Created two MessageListenerContainers with the same identifier: " |
| 183 | + + identifiableMessageListenerContainer.getIdentifier()); |
| 184 | + } |
| 185 | + log.debug("Created MessageListenerContainer with id: {}", identifiableMessageListenerContainer.getIdentifier()); |
| 186 | + messageContainers.put(identifiableMessageListenerContainer.getIdentifier(), identifiableMessageListenerContainer.getContainer()); |
| 187 | + } |
| 188 | + } |
| 189 | + } |
| 190 | + } |
| 191 | + |
| 192 | + return ImmutableMap.copyOf(messageContainers); |
195 | 193 | } |
196 | 194 | } |
0 commit comments