Hi I am building a library for messaging between AWS components and so I want a lightweight solution. One part of the solution requires me to listen in on when an annotated method is invoked, so I thought I'd use a pointcut and implement an advice on what to do when that pointcut was reached.
The gradle.build script looks like this:
buildscript {
ext {
// some company-specific config here
nexus = {
credentials {
username nexusBuildUserToken
password nexusBuildPassToken
}
url nexusRepoURL
}
}
repositories {
mavenCentral()
maven(nexus)
}
dependencies {
classpath("net.researchgate:gradle-release:$gradleReleasePluginVersion")
classpath("gradle.plugin.aspectj:gradle-aspectj:$gradleAspectJPluginVersion")
}
}
apply plugin: 'java'
// IDE
apply plugin: 'idea'
apply plugin: 'eclipse-wtp'
apply plugin: "aspectj.gradle"
jar {
enabled = true
}
// project artifact info
group = groupId
archivesBaseName = artifactId
repositories {
mavenCentral()
maven(nexus)
maven {
url "https://dl.bintray.com/findify/maven"
}
}
dependencies {
compile("org.aspectj:aspectjtools:$aspectjVersion")
compile("org.apache.commons:commons-lang3:${commonsLang3Version}")
compile("com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:${jacksonDataformatYamlVersion}")
compile("org.elasticmq:elasticmq-rest-sqs_2.11:0.14.1")
compile("com.amazonaws:aws-java-sdk-sqs:${awsMessagingVersion}")
compile("com.amazonaws:aws-java-sdk-sns:${awsMessagingVersion}")
compile("au.com.auspost:json-encryption:${jsonEncryptionVersion}")
compile("org.apache.commons:commons-lang3:${commonsLang3Version}")
compile("org.reflections:reflections:${reflectionsVersion}")
compile("redis.clients:jedis:${jedisVersion}")
compile("org.aspectj:aspectjweaver:$aspectjVersion")
compile("org.aspectj:aspectjrt:$aspectjVersion")
testCompile("junit:junit:${jUnitVersion}")
testCompile("org.mockito:mockito-core:${mockitoCoreVersion}")
testCompile("org.assertj:assertj-core:${assertjVersion}")
testCompile("ai.grakn:redis-mock:${embeddedRedisVersion}")
testCompile("org.slf4j:slf4j-simple:1.7.25")
testCompile("ch.qos.logback:logback-core:1.2.3")
testCompile(group: 'io.findify', name: 'sqsmock_2.11', version: '0.3.2')
}
As you can see I included all the aspectj libs in an effort to make sure I'm not missing anything I need (feel free to tell me what I don't need).
The class I expect to get weaved is this:
@Aspect
public class TopicSenderManager {
private ThreadPoolFactory threadPoolFactory;
private CorrelationService correlationService = new CorrelationService.Default();
private Map<String, TopicSenderProcessor> topicSenderProcessors = new HashMap<>();
private ExecutorService executor;
public TopicSenderManager(Map<String, TopicSenderProcessor> topicSenderProcessors) {
this.threadPoolFactory = new ThreadPoolFactory.Default();
this.topicSenderProcessors = topicSenderProcessors;
executor = threadPoolFactory.create(topicSenderProcessors.size(), "TopicSender");
}
@Around("@annotation(topicSender) && execution(* *(..))")
public Object sendMessageToTopic(ProceedingJoinPoint pjp, TopicSender topicSender) throws Throwable {
TopicSenderProcessor topicSenderProcessor = getProcessor(topicSender.topicAlias(), topicSender.eventType());
topicSenderProcessor.setCorrelationId(correlationService.read());
Object[] args = pjp.getArgs();
if (args == null || args.length != 1) {
throw new Exception("naughty, naughty");
} else {
topicSenderProcessor.setEventMessage((EventMessage) args[0]);
executor.execute(topicSenderProcessor);
}
return pjp.proceed();
}
public Map<String, TopicSenderProcessor> getTopicSenderProcessors() {
return topicSenderProcessors;
}
public TopicSenderProcessor getProcessor(String topic, String eventType) {
String senderKey = topic + "," + eventType;
return topicSenderProcessors.get(senderKey);
}
}
What I'm hoping this will do is pick up every execution (invocation) of a method annotated with @TopicSender and execute the associated processor in a thread from the threadpool.
I had a look at the decompiled class of the TopicSenderManager which is shown below:
@Aspect
public class TopicSenderManager {
private ThreadPoolFactory threadPoolFactory = new Default();
private CorrelationService correlationService = new au.com.auspost.messaging.CorrelationService.Default();
private Map<String, TopicSenderProcessor> topicSenderProcessors = new HashMap();
private ExecutorService executor;
public TopicSenderManager(Map<String, TopicSenderProcessor> topicSenderProcessors) {
this.topicSenderProcessors = topicSenderProcessors;
this.executor = this.threadPoolFactory.create(topicSenderProcessors.size(), "TopicSender");
}
@Around("@annotation(topicSender) && execution(* *(..))")
public Object sendMessageToTopic(ProceedingJoinPoint pjp, TopicSender topicSender) throws Throwable {
TopicSenderProcessor topicSenderProcessor = this.getProcessor(topicSender.topicAlias(), topicSender.eventType());
topicSenderProcessor.setCorrelationId(ajc$inlineAccessFieldGet$au_com_auspost_messaging_send_topic_TopicSenderManager$au_com_auspost_messaging_send_topic_TopicSenderManager$correlationService(this).read());
Object[] args = pjp.getArgs();
if (args != null && args.length == 1) {
topicSenderProcessor.setEventMessage((EventMessage)args[0]);
ajc$inlineAccessFieldGet$au_com_auspost_messaging_send_topic_TopicSenderManager$au_com_auspost_messaging_send_topic_TopicSenderManager$executor(this).execute(topicSenderProcessor);
return pjp.proceed();
} else {
throw new Exception("naughty, naughty");
}
}
public Map<String, TopicSenderProcessor> getTopicSenderProcessors() {
return this.topicSenderProcessors;
}
public TopicSenderProcessor getProcessor(String topic, String eventType) {
String senderKey = topic + "," + eventType;
return (TopicSenderProcessor)this.topicSenderProcessors.get(senderKey);
}
public static TopicSenderManager aspectOf() {
if (ajc$perSingletonInstance == null) {
throw new NoAspectBoundException("au.com.auspost.messaging.send.topic.TopicSenderManager", ajc$initFailureCause);
} else {
return ajc$perSingletonInstance;
}
}
public static boolean hasAspect() {
return ajc$perSingletonInstance != null;
}
static {
try {
ajc$postClinit();
} catch (Throwable var1) {
ajc$initFailureCause = var1;
}
}
}
I assume this is the weaving that is supposed to have happened.
The next thing is to set up a (sort of) unit test that demonstrates the behaviour that I want:
Firstly I define a simple example of the sort of method that I want to intercept:
public class SenderT1E1 {
@TopicSender(topicAlias = "t1", eventType = "a.c.a.e1")
public void aSendingMethod(EventMessage<TestMessage> eventMessage) {
// do what you like before sending the message
}
}
Then the test that looks at it:
public class SenderE1T1Test {
static TopicSenderManager manager;
@BeforeClass
public static void setUp() throws Exception {
MockAmazonClient snsClient = new MockAmazonClient();
TopicSenderFactory senderFactory = new TopicSenderFactory();
PublishResult publishResult = new PublishResult().withMessageId("E1T1");
manager = senderFactory.make(snsClient.amazonSNS(publishResult), "aws-send-test.properties");
}
@Test
public void whenSenderIsCalledMessageIsSent() {
SenderT1E1 target = new SenderT1E1();
EventMessage message = new EventMessage<>();
message.setEventType("a.c.a.e1");
message.setPayload(new TestMessage());
target.aSendingMethod(message);
TopicSenderProcessor processor = manager.getProcessor("http://localhost:8001/topic/t1", "a.c.a.e1");
assertThat(manager.getTopicSenderProcessors().entrySet().size(), is(2));
manager.getTopicSenderProcessors().forEach((k,v) -> System.out.println(k + ", " + v));
assertThat(processor, is(notNullValue()));
// now, did it execute...
assertThat(processor.getEventMessage(), is(notNullValue()));
assertThat(processor.getLastPublishRequest(), is(notNullValue()));
assertThat(processor.getLastPublishResult(), is(notNullValue()));
}
}
Basically, everything up to the // now did it execute... line works, but everything is null after that. So it looks like the pointcut is never reached.
So, is the weaving not working correctly? is the pointcut and advice in the @Around spec not correct? is there some kind of runtime trigger, I have forgotten about? or is it something else?
I thought I should make a statement here: the reason I'm not using spring is that I have found that the number of dependencies leads to a really bloated library and I'm trying to keep it as trim as possible. I want the speed benefits of aspectJ and i would prefer compile time weaving because this is a library to be used in applications I will resort to spring if I have to, but it really is a last resort.
The problem was in the something else category. The reason the Pointcut was not being triggered was because I was using compile-time weaving, and the weaved target class was not in the classpath at runtime. The weaved target class was in src/test/java, while the pointcut was in src/main/java and when the application is compiled, src/test/java is no where in sight. What I needed was runtime weaving, so that the test class could be discovered and weaved, and then the pointcut would work. Unfortunately, runtime weaving is an expensive operation, so not an option for my purposes, so I have gone over to using listener pattern. Its not as clean from an implementors perspective, but much better performance-wise