Skip to content

Commit f4cb0ed

Browse files
JaidenAshmoreJaidenAshmore
authored
Merge pull request #110 from JaidenAshmore/issue/109_reduce_number_of_argument_resolver_resolutions
refs #109: Now only determines the ArgumentResolver needed the first time instead of at every message
2 parents 37d6bb2 + aa2636f commit f4cb0ed

File tree

8 files changed

+195
-192
lines changed

8 files changed

+195
-192
lines changed

‎examples/sqs-listener-library-comparison/src/main/java/com/jashmore/sqs/examples/MessageListeners.java‎

Lines changed: 27 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -8,13 +8,18 @@
88
import static com.jashmore.sqs.examples.Queues.QUEUE_LISTENER_10_QUEUE_NAME;
99
import static com.jashmore.sqs.examples.Queues.QUEUE_LISTENER_30_QUEUE_NAME;
1010
import static com.jashmore.sqs.examples.Queues.SPRING_CLOUD_QUEUE_NAME;
11+
import static software.amazon.awssdk.services.sqs.model.MessageSystemAttributeName.APPROXIMATE_RECEIVE_COUNT;
1112

13+
import com.jashmore.sqs.argument.attribute.MessageSystemAttribute;
14+
import com.jashmore.sqs.argument.messageid.MessageId;
1215
import com.jashmore.sqs.argument.payload.Payload;
1316
import com.jashmore.sqs.spring.container.basic.QueueListener;
1417
import com.jashmore.sqs.spring.container.prefetch.PrefetchingQueueListener;
1518
import lombok.extern.slf4j.Slf4j;
1619
import org.springframework.cloud.aws.messaging.listener.annotation.SqsListener;
1720
import org.springframework.jms.annotation.JmsListener;
21+
import org.springframework.jms.support.JmsHeaders;
22+
import org.springframework.messaging.handler.annotation.Header;
1823
import org.springframework.stereotype.Component;
1924

2025
import java.util.concurrent.atomic.AtomicInteger;
@@ -33,47 +38,60 @@ public class MessageListeners {
3338
private final AtomicLong firstTime = new AtomicLong(0);
3439

3540
@PrefetchingQueueListener(value = PREFETCHING_10_QUEUE_NAME, concurrencyLevel = 10, desiredMinPrefetchedMessages = 50, maxPrefetchedMessages = 60)
36-
public void prefetchingConcurrency10(@Payload final String payload) throws Exception {
41+
public void prefetchingConcurrency10(@Payload final String payload,
42+
@MessageId final String messageId,
43+
@MessageSystemAttribute(APPROXIMATE_RECEIVE_COUNT) int receiveCount) throws Exception {
3744
handleMethod();
3845
}
3946

4047
@PrefetchingQueueListener(value = PREFETCHING_30_QUEUE_NAME, concurrencyLevel = 30, desiredMinPrefetchedMessages = 50, maxPrefetchedMessages = 60)
41-
public void prefetchingConcurrency30(@Payload final String payload) throws Exception {
48+
public void prefetchingConcurrency30(@Payload final String payload,
49+
@MessageId final String messageId,
50+
@MessageSystemAttribute(APPROXIMATE_RECEIVE_COUNT) int receiveCount) throws Exception {
4251
handleMethod();
4352
}
4453

4554
@QueueListener(value = QUEUE_LISTENER_10_QUEUE_NAME, concurrencyLevel = 10)
46-
public void queueListenerMethodConcurrency10(@Payload final String payload) throws Exception {
55+
public void queueListenerMethodConcurrency10(@Payload final String payload,
56+
@MessageId final String messageId,
57+
@MessageSystemAttribute(APPROXIMATE_RECEIVE_COUNT) int receiveCount) throws Exception {
4758
handleMethod();
4859
}
4960

5061
@QueueListener(value = QUEUE_LISTENER_30_QUEUE_NAME, concurrencyLevel = 30)
51-
public void queueListenerMethodConcurrency30(@Payload final String payload) throws Exception {
62+
public void queueListenerMethodConcurrency30(@Payload final String payload,
63+
@MessageId final String messageId,
64+
@MessageSystemAttribute(APPROXIMATE_RECEIVE_COUNT) int receiveCount) throws Exception {
5265
handleMethod();
5366
}
5467

5568
@SqsListener(SPRING_CLOUD_QUEUE_NAME)
56-
public void springCloudConcurrency10(final String payload) throws Exception {
69+
public void springCloudConcurrency10(final String payload,
70+
@Header("ApproximateReceiveCount") String receiveCount,
71+
@Header("MessageId") String messageId) throws Exception {
5772
handleMethod();
5873
}
5974

6075
@JmsListener(destination = JMS_10_QUEUE_NAME, concurrency = "10")
61-
public void jmsConcurrency10(String message) throws Exception {
76+
public void jmsConcurrency10(String message,
77+
@Header("JMSXDeliveryCount") String receiveCount,
78+
@Header(JmsHeaders.MESSAGE_ID) String messageId) throws Exception {
6279
handleMethod();
6380
}
6481

6582
@JmsListener(destination = JMS_30_QUEUE_NAME, concurrency = "30")
66-
public void jmsConcurrency30(String message) throws Exception {
83+
public void jmsConcurrency30(String message,
84+
@Header("JMSXDeliveryCount") String receiveCount,
85+
@Header(JmsHeaders.MESSAGE_ID) String messageId) throws Exception {
6786
handleMethod();
6887
}
6988

7089
private void handleMethod() throws Exception {
7190
firstTime.compareAndSet(0, System.currentTimeMillis());
91+
Thread.sleep(MESSAGE_IO_TIME_IN_MS);
7292
final int currentCount = count.incrementAndGet();
7393
if (currentCount % 100 == 0) {
7494
log.info("Time for processing {} messages is {}ms", currentCount, System.currentTimeMillis() - firstTime.get());
7595
}
76-
77-
Thread.sleep(MESSAGE_IO_TIME_IN_MS);
7896
}
7997
}

‎java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/argument/ArgumentResolverService.java‎

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,7 @@
11
package com.jashmore.sqs.argument;
22

3-
import com.jashmore.sqs.QueueProperties;
43
import com.jashmore.sqs.processor.MessageProcessor;
54
import com.jashmore.sqs.processor.argument.Acknowledge;
6-
import software.amazon.awssdk.services.sqs.model.Message;
75

86
import javax.annotation.concurrent.ThreadSafe;
97

@@ -23,15 +21,13 @@
2321
@ThreadSafe
2422
public interface ArgumentResolverService {
2523
/**
26-
* Resolve the argument value for the given parameter of the method.
24+
* Determine the {@link ArgumentResolver} that should be used for processing an argument of the method.
2725
*
2826
* <p>Note that this does not need to be able to resolve the {@link Acknowledge} argument as that is provided by the {@link MessageProcessor}.
2927
*
30-
* @param queueProperties details about the queue that the message came from
31-
* @param methodParameter the parameter to get the argument value for
32-
* @param message the message being processed by this queue
33-
* @return the value of the argument
34-
* @throws ArgumentResolutionException when there was an error determine the parameter argument value
28+
* @param methodParameter details about the method parameter
29+
* @return the resolver that should be used to resolve this parameter
30+
* @throws UnsupportedArgumentResolutionException if there is no available {@link ArgumentResolver}
3531
*/
36-
Object resolveArgument(QueueProperties queueProperties, MethodParameter methodParameter, Message message) throws ArgumentResolutionException;
32+
ArgumentResolver<?> getArgumentResolver(MethodParameter methodParameter) throws UnsupportedArgumentResolutionException;
3733
}

‎java-dynamic-sqs-listener-api/src/main/java/com/jashmore/sqs/argument/UnsupportedArgumentResolutionException.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
package com.jashmore.sqs.argument;
2+
3+
/**
4+
* Exception thrown if there is no available {@link ArgumentResolver} that is able to resolve the provided parameter for the method.
5+
*/
6+
public class UnsupportedArgumentResolutionException extends RuntimeException {
7+
public UnsupportedArgumentResolutionException() {
8+
super("Unable to resolve message argument");
9+
}
10+
11+
public UnsupportedArgumentResolutionException(final MethodParameter methodParameter) {
12+
super(String.format("No known for parameter[%d] for method: %s",
13+
methodParameter.getParameterIndex(),
14+
methodParameter.getMethod().getDeclaringClass().getName() + "#" + methodParameter.getMethod().getName())
15+
);
16+
}
17+
}

‎java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/DelegatingArgumentResolverService.java‎

Lines changed: 5 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,6 @@
11
package com.jashmore.sqs.argument;
22

3-
import com.jashmore.sqs.QueueProperties;
43
import lombok.AllArgsConstructor;
5-
import software.amazon.awssdk.services.sqs.model.Message;
64

75
import java.util.Set;
86

@@ -17,22 +15,10 @@ public class DelegatingArgumentResolverService implements ArgumentResolverServic
1715
private final Set<ArgumentResolver<?>> argumentResolvers;
1816

1917
@Override
20-
public Object resolveArgument(final QueueProperties queueProperties, final MethodParameter methodParameter, final Message message) {
21-
for (final ArgumentResolver<?> resolver: argumentResolvers) {
22-
if (resolver.canResolveParameter(methodParameter)) {
23-
try {
24-
return resolver.resolveArgumentForParameter(queueProperties, methodParameter, message);
25-
} catch (final RuntimeException runtimeException) {
26-
// Make sure to wrap any unintended exceptions with the expected exception for errors
27-
if (!ArgumentResolutionException.class.isAssignableFrom(runtimeException.getClass())) {
28-
throw new ArgumentResolutionException("Error obtaining an argument value for parameter", runtimeException);
29-
}
30-
31-
throw runtimeException;
32-
}
33-
}
34-
}
35-
36-
throw new ArgumentResolutionException("No ArgumentResolver found that can process this parameter");
18+
public ArgumentResolver<?> getArgumentResolver(final MethodParameter methodParameter) throws UnsupportedArgumentResolutionException {
19+
return argumentResolvers.stream()
20+
.filter(resolver -> resolver.canResolveParameter(methodParameter))
21+
.findFirst()
22+
.orElseThrow(() -> new UnsupportedArgumentResolutionException(methodParameter));
3723
}
3824
}

‎java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/processor/DefaultMessageProcessor.java‎

Lines changed: 61 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,9 @@
11
package com.jashmore.sqs.processor;
22

3+
import static java.util.stream.Collectors.toList;
4+
35
import com.jashmore.sqs.QueueProperties;
4-
import com.jashmore.sqs.argument.ArgumentResolutionException;
6+
import com.jashmore.sqs.argument.ArgumentResolver;
57
import com.jashmore.sqs.argument.ArgumentResolverService;
68
import com.jashmore.sqs.argument.DefaultMethodParameter;
79
import com.jashmore.sqs.argument.MethodParameter;
@@ -14,6 +16,7 @@
1416
import java.lang.reflect.Method;
1517
import java.lang.reflect.Parameter;
1618
import java.util.Arrays;
19+
import java.util.List;
1720
import java.util.concurrent.CompletableFuture;
1821
import java.util.concurrent.ExecutionException;
1922
import java.util.stream.IntStream;
@@ -26,16 +29,31 @@
2629
@ThreadSafe
2730
@AllArgsConstructor
2831
public class DefaultMessageProcessor implements MessageProcessor {
29-
private final ArgumentResolverService argumentResolverService;
3032
private final QueueProperties queueProperties;
3133
private final MessageResolver messageResolver;
3234
private final Method messageConsumerMethod;
3335
private final Object messageConsumerBean;
36+
private final Class<?> returnType;
37+
private final List<InternalArgumentResolver> methodArgumentResolvers;
38+
private final boolean hasAcknowledgeParameter;
39+
40+
public DefaultMessageProcessor(final ArgumentResolverService argumentResolverService,
41+
final QueueProperties queueProperties,
42+
final MessageResolver messageResolver,
43+
final Method messageConsumerMethod,
44+
final Object messageConsumerBean) {
45+
this.queueProperties = queueProperties;
46+
this.messageResolver = messageResolver;
47+
this.messageConsumerMethod = messageConsumerMethod;
48+
this.messageConsumerBean = messageConsumerBean;
49+
this.hasAcknowledgeParameter = hasAcknowledgeParameter();
50+
this.returnType = messageConsumerMethod.getReturnType();
51+
this.methodArgumentResolvers = getArgumentResolvers(argumentResolverService);
52+
}
3453

3554
@Override
3655
public void processMessage(final Message message) throws MessageProcessingException {
37-
final Acknowledge acknowledge = () -> messageResolver.resolveMessage(message);
38-
final Object[] arguments = getArguments(acknowledge, message);
56+
final Object[] arguments = getArguments(message);
3957

4058
final Object result;
4159
try {
@@ -44,12 +62,11 @@ public void processMessage(final Message message) throws MessageProcessingExcept
4462
throw new MessageProcessingException("Error processing message", exception);
4563
}
4664

47-
if (hasAcknowledgeParameter()) {
65+
if (hasAcknowledgeParameter) {
4866
// If the method has the Acknowledge parameter it is up to them to resolve the message
4967
return;
5068
}
5169

52-
final Class<?> returnType = messageConsumerMethod.getReturnType();
5370
if (CompletableFuture.class.isAssignableFrom(returnType)) {
5471
final CompletableFuture<?> resultCompletableFuture = (CompletableFuture) result;
5572

@@ -59,7 +76,7 @@ public void processMessage(final Message message) throws MessageProcessingExcept
5976

6077
try {
6178
resultCompletableFuture
62-
.thenAccept((ignored) -> acknowledge.acknowledgeSuccessful())
79+
.thenAccept((ignored) -> messageResolver.resolveMessage(message))
6380
.get();
6481
} catch (final InterruptedException interruptedException) {
6582
Thread.currentThread().interrupt();
@@ -68,49 +85,64 @@ public void processMessage(final Message message) throws MessageProcessingExcept
6885
throw new MessageProcessingException("Error processing message", executionException.getCause());
6986
}
7087
} else {
71-
acknowledge.acknowledgeSuccessful();
88+
messageResolver.resolveMessage(message);
7289
}
7390
}
7491

75-
private boolean hasAcknowledgeParameter() {
76-
return Arrays.stream(messageConsumerMethod.getParameters())
77-
.anyMatch(DefaultMessageProcessor::isAcknowledgeParameter);
78-
}
79-
80-
private static boolean isAcknowledgeParameter(final Parameter parameter) {
81-
return Acknowledge.class.isAssignableFrom(parameter.getType());
82-
}
83-
8492
/**
8593
* Get the arguments for the method for the message that is being processed.
8694
*
87-
* @param acknowledge the acknowledge object that should be used if a parameter is an {@link Acknowledge}
8895
* @param message the message to populate the arguments from
8996
* @return the array of arguments to call the method with
9097
*/
91-
private Object[] getArguments(final Acknowledge acknowledge, final Message message) {
98+
private Object[] getArguments(final Message message) {
99+
return methodArgumentResolvers.stream()
100+
.map(resolver -> resolver.resolveArgument(message))
101+
.toArray(Object[]::new);
102+
}
103+
104+
private List<InternalArgumentResolver> getArgumentResolvers(final ArgumentResolverService argumentResolverService) {
92105
final Parameter[] parameters = messageConsumerMethod.getParameters();
93106
return IntStream.range(0, parameters.length)
94-
.mapToObj(parameterIndex -> {
107+
.<InternalArgumentResolver>mapToObj(parameterIndex -> {
95108
final Parameter parameter = parameters[parameterIndex];
96109

97-
if (isAcknowledgeParameter(parameter)) {
98-
return acknowledge;
99-
}
100-
101110
final MethodParameter methodParameter = DefaultMethodParameter.builder()
102111
.method(messageConsumerMethod)
103112
.parameter(parameter)
104113
.parameterIndex(parameterIndex)
105114
.build();
106115

107-
try {
108-
return argumentResolverService.resolveArgument(queueProperties, methodParameter, message);
109-
} catch (final ArgumentResolutionException argumentResolutionException) {
110-
throw new MessageProcessingException("Error resolving arguments for message", argumentResolutionException);
116+
if (isAcknowledgeParameter(parameter)) {
117+
return message -> (Acknowledge) () -> messageResolver.resolveMessage(message);
111118
}
119+
120+
final ArgumentResolver<?> argumentResolver = argumentResolverService.getArgumentResolver(methodParameter);
121+
return message -> argumentResolver.resolveArgumentForParameter(queueProperties, methodParameter, message);
112122
})
113-
.toArray(Object[]::new);
123+
.collect(toList());
124+
}
125+
126+
private boolean hasAcknowledgeParameter() {
127+
return Arrays.stream(messageConsumerMethod.getParameters())
128+
.anyMatch(DefaultMessageProcessor::isAcknowledgeParameter);
129+
}
130+
131+
private static boolean isAcknowledgeParameter(final Parameter parameter) {
132+
return Acknowledge.class.isAssignableFrom(parameter.getType());
133+
}
114134

135+
/**
136+
* Internal resolver for resolving the argument given the message.
137+
*/
138+
@FunctionalInterface
139+
interface InternalArgumentResolver {
140+
/**
141+
* Resolve the argument of the method.
142+
*
143+
* @param message the message that is being processed
144+
* @return the argument that should be used for the corresponding parameter
145+
*/
146+
Object resolveArgument(final Message message);
115147
}
116148
}

0 commit comments

Comments
 (0)