forked from eventuate-tram/eventuate-tram-core
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
eventuate-tram#1: Support ActiveMQ. Basic implementation.
- Loading branch information
Showing
18 changed files
with
334 additions
and
43 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,8 @@ | ||
| apply plugin: PublicModulePlugin | ||
|
|
||
| dependencies { | ||
| compile project(":eventuate-tram-consumer-common") | ||
| compile "org.apache.activemq:activemq-core:5.7.0" | ||
| } | ||
|
|
||
|
|
145 changes: 145 additions & 0 deletions
145
...tivemq/src/main/java/io/eventuate/tram/consumer/activemq/MessageConsumerActiveMQImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,145 @@ | ||
| package io.eventuate.tram.consumer.activemq; | ||
|
|
||
| import io.eventuate.javaclient.commonimpl.JSonMapper; | ||
| import io.eventuate.tram.consumer.common.DuplicateMessageDetector; | ||
| import io.eventuate.tram.messaging.common.Message; | ||
| import io.eventuate.tram.messaging.common.MessageImpl; | ||
| import io.eventuate.tram.messaging.consumer.MessageConsumer; | ||
| import io.eventuate.tram.messaging.consumer.MessageHandler; | ||
| import org.apache.activemq.ActiveMQConnectionFactory; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.transaction.support.TransactionTemplate; | ||
|
|
||
| import javax.jms.*; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.ExecutionException; | ||
| import java.util.concurrent.Future; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
|
|
||
| public class MessageConsumerActiveMQImpl implements MessageConsumer { | ||
|
|
||
| private Logger logger = LoggerFactory.getLogger(getClass()); | ||
|
|
||
| @Autowired | ||
| private TransactionTemplate transactionTemplate; | ||
|
|
||
| @Autowired | ||
| private DuplicateMessageDetector duplicateMessageDetector; | ||
|
|
||
| private ActiveMQConnectionFactory connectionFactory; | ||
|
|
||
| private Connection connection; | ||
| private Session session; | ||
| private List<javax.jms.MessageConsumer> consumers = new ArrayList<>(); | ||
| private List<Future<Void>> processingFutures = new ArrayList<>(); | ||
|
|
||
| private AtomicBoolean runFlag = new AtomicBoolean(true); | ||
|
|
||
| public MessageConsumerActiveMQImpl(String url) { | ||
| connectionFactory = new ActiveMQConnectionFactory(url); | ||
| try { | ||
| connection = connectionFactory.createConnection(); | ||
| connection.setExceptionListener(e -> logger.error(e.getMessage(), e)); | ||
| connection.start(); | ||
| session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); | ||
| } catch (JMSException e) { | ||
| logger.error(e.getMessage(), e); | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void subscribe(String subscriberId, Set<String> channels, MessageHandler handler) { | ||
| try { | ||
| for (String channel : channels) { | ||
| Destination destination = session.createQueue(channel); | ||
| javax.jms.MessageConsumer consumer = session.createConsumer(destination); | ||
| consumers.add(consumer); | ||
|
|
||
| processingFutures.add(CompletableFuture.supplyAsync(() -> process(subscriberId, consumer, handler))); | ||
| } | ||
| } catch (JMSException e) { | ||
| logger.error(e.getMessage(), e); | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
|
|
||
| private Void process(String subscriberId, javax.jms.MessageConsumer consumer, MessageHandler handler) { | ||
| while (runFlag.get()) { | ||
| try { | ||
| javax.jms.Message message = consumer.receive(100); | ||
|
|
||
| if (message == null) { | ||
| continue; | ||
| } | ||
|
|
||
| TextMessage textMessage = (TextMessage) message; | ||
| Message tramMessage = JSonMapper.fromJson(textMessage.getText(), MessageImpl.class); | ||
|
|
||
| transactionTemplate.execute(ts -> { | ||
| if (duplicateMessageDetector.isDuplicate(subscriberId, tramMessage.getId())) { | ||
| logger.trace("Duplicate message {} {}", subscriberId, tramMessage.getId()); | ||
| acknowledge(textMessage); | ||
| return null; | ||
| } | ||
|
|
||
| try { | ||
| logger.trace("Invoking handler {} {}", subscriberId, tramMessage.getId()); | ||
| handler.accept(tramMessage); | ||
| logger.trace("handled message {} {}", subscriberId, tramMessage.getId()); | ||
| } catch (Throwable t) { | ||
| logger.trace("Got exception {} {}", subscriberId, tramMessage.getId()); | ||
| logger.trace("Got exception ", t); | ||
| } finally { | ||
| acknowledge(textMessage); | ||
| } | ||
|
|
||
| return null; | ||
| }); | ||
|
|
||
| } catch (JMSException e) { | ||
| logger.error(e.getMessage(), e); | ||
| } | ||
| } | ||
|
|
||
| try { | ||
| consumer.close(); | ||
| } catch (JMSException e) { | ||
| logger.error(e.getMessage(), e); | ||
| } | ||
|
|
||
| return null; | ||
| } | ||
|
|
||
| private void acknowledge(TextMessage textMessage) { | ||
| try { | ||
| textMessage.acknowledge(); | ||
| } catch (JMSException e) { | ||
| logger.error(e.getMessage(), e); | ||
| } | ||
| } | ||
|
|
||
| public void close() { | ||
| runFlag.set(false); | ||
|
|
||
| processingFutures.forEach(f -> { | ||
| try { | ||
| f.get(); | ||
| } catch (InterruptedException | ExecutionException e) { | ||
| logger.error(e.getMessage(), e); | ||
| } | ||
| }); | ||
|
|
||
| try { | ||
| session.close(); | ||
| connection.close(); | ||
| } catch (JMSException e) { | ||
| logger.error(e.getMessage(), e); | ||
| } | ||
| } | ||
| } |
17 changes: 17 additions & 0 deletions
17
.../src/main/java/io/eventuate/tram/consumer/activemq/TramConsumerActiveMQConfiguration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| package io.eventuate.tram.consumer.activemq; | ||
|
|
||
| import io.eventuate.tram.consumer.common.TramConsumerCommonConfiguration; | ||
| import io.eventuate.tram.messaging.consumer.MessageConsumer; | ||
| import org.springframework.beans.factory.annotation.Value; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
| import org.springframework.context.annotation.Import; | ||
|
|
||
| @Configuration | ||
| @Import(TramConsumerCommonConfiguration.class) | ||
| public class TramConsumerActiveMQConfiguration { | ||
| @Bean | ||
| public MessageConsumer messageConsumer(@Value("${activemq.url}") String activeMQURL) { | ||
| return new MessageConsumerActiveMQImpl(activeMQURL); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| apply plugin: PublicModulePlugin | ||
|
|
||
| dependencies { | ||
| compile project(":eventuate-tram-messaging") | ||
|
|
||
| compile "io.eventuate.local.java:eventuate-local-java-common:$eventuateLocalVersion" | ||
| compile "org.springframework.boot:spring-boot-starter-jdbc:$springBootVersion" | ||
| compile 'mysql:mysql-connector-java:5.1.36' | ||
| compile ('org.postgresql:postgresql:9.4-1200-jdbc41') { | ||
| exclude group: "org.slf4j", module: "slf4j-simple" | ||
| } | ||
| compile "io.eventuate.client.java:eventuate-client-java-common-impl:$eventuateClientVersion" | ||
| compile "io.eventuate.client.java:eventuate-client-java-jdbc-common:$eventuateClientVersion" | ||
| } | ||
|
|
||
|
|
2 changes: 1 addition & 1 deletion
2
...sumer/kafka/DuplicateMessageDetector.java → ...umer/common/DuplicateMessageDetector.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...qlTableBasedDuplicateMessageDetector.java → ...qlTableBasedDuplicateMessageDetector.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
22 changes: 22 additions & 0 deletions
22
...mmon/src/main/java/io/eventuate/tram/consumer/common/TramConsumerCommonConfiguration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| package io.eventuate.tram.consumer.common; | ||
|
|
||
| import io.eventuate.javaclient.spring.jdbc.EventuateSchema; | ||
| import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; | ||
| import org.springframework.beans.factory.annotation.Value; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
|
|
||
| @Configuration | ||
| public class TramConsumerCommonConfiguration { | ||
|
|
||
| @Bean | ||
| public EventuateSchema eventuateSchema(@Value("${eventuate.database.schema:#{null}}") String eventuateDatabaseSchema) { | ||
| return new EventuateSchema(eventuateDatabaseSchema); | ||
| } | ||
|
|
||
| @Bean | ||
| @ConditionalOnMissingBean(DuplicateMessageDetector.class) | ||
| public DuplicateMessageDetector duplicateMessageDetector(EventuateSchema eventuateSchema) { | ||
| return new SqlTableBasedDuplicateMessageDetector(eventuateSchema); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.