You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by ni...@apache.org on 2019/08/08 02:15:23 UTC
[servicecomb-pack] 09/10: SCB-1417 add alpha-fsm-channel-redis
module
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit e62f336a3659afc487f80550e7d87f74e64d0bbf
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 10:49:21 2019 +0800
SCB-1417 add alpha-fsm-channel-redis module
---
alpha/alpha-fsm-channel-redis/README.md | 17 +++
alpha/alpha-fsm-channel-redis/pom.xml | 110 +++++++++++++++++
.../alpha/fsm/channel/redis/MessageSerializer.java | 86 ++++++++++++++
.../redis/RedisChannelAutoConfiguration.java | 83 +++++++++++++
.../fsm/channel/redis/RedisMessagePublisher.java | 46 ++++++++
.../fsm/channel/redis/RedisMessageSubscriber.java | 70 +++++++++++
.../src/main/resources/META-INF/spring.factories | 17 +++
.../pack/alpha/fsm/RedisChannelTest.java | 130 +++++++++++++++++++++
.../servicecomb/pack/alpha/fsm/RedisEventSink.java | 32 +++++
.../src/test/resources/log4j2.xml | 30 +++++
10 files changed, 621 insertions(+)
diff --git a/alpha/alpha-fsm-channel-redis/README.md b/alpha/alpha-fsm-channel-redis/README.md
new file mode 100644
index 0000000..98ecb93
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/README.md
@@ -0,0 +1,17 @@
+# FSM Redis channel
+## Enabled Saga State Machine Module
+
+Using `alpha.feature.akka.enabled=true` launch Alpha and Omega Side
+Using `alpha.feature.akka.channel.type=redis` launch Alpha and Omega Side
+
+```properties
+alpha.feature.akka.enabled=true
+alpha.feature.akka.channel.type=redis
+```
+
+setting spring boot redis
+```
+spring.redis.host=your_redis_host
+spring.redis.port=your_redis_port
+spring.redis.password=your_redis_password
+```
diff --git a/alpha/alpha-fsm-channel-redis/pom.xml b/alpha/alpha-fsm-channel-redis/pom.xml
new file mode 100644
index 0000000..a896264
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/pom.xml
@@ -0,0 +1,110 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>alpha</artifactId>
+ <groupId>org.apache.servicecomb.pack</groupId>
+ <version>0.5.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>alpha-fsm-channel-redis</artifactId>
+ <name>Pack::Alpha::Fsm::channel::redis</name>
+
+ <properties>
+ </properties>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-dependencies</artifactId>
+ <version>${spring.boot.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <dependencies>
+ <!-- spring boot -->
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-autoconfigure</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.servicecomb.pack</groupId>
+ <artifactId>pack-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.servicecomb.pack</groupId>
+ <artifactId>alpha-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-log4j2</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <!-- For testing the artifacts scope are test-->
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-test</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-data-redis</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-pool2</artifactId>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git a/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessageSerializer.java b/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessageSerializer.java
new file mode 100644
index 0000000..711bb07
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessageSerializer.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.pack.alpha.fsm.channel.redis;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.serializer.SerializationException;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Optional;
+
+public class MessageSerializer {
+
+ private static final Logger logger = LoggerFactory.getLogger(MessageSerializer.class);
+
+ private static MessageSerializerImpl serializer = null;
+
+ public MessageSerializer() {
+ serializer = new MessageSerializerImpl();
+ }
+
+ public Optional<byte[]> serializer(Object data){
+ return Optional.ofNullable(serializer.serialize(data));
+ }
+
+ public Optional<Object> deserialize(byte[] bytes){
+ return Optional.ofNullable(serializer.deserialize(bytes));
+ }
+
+ private class MessageSerializerImpl implements RedisSerializer<Object>{
+ @Override
+ public byte[] serialize(Object data) throws SerializationException {
+ try {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ ObjectOutputStream outputStream = new ObjectOutputStream(byteArrayOutputStream);
+ outputStream.writeObject(data);
+
+ byte[] bytes = byteArrayOutputStream.toByteArray();
+
+ outputStream.close();
+
+ return bytes;
+ }catch (Exception e){
+ logger.error("serialize Exception = [{}]", e.getMessage(), e);
+ }
+
+ return null;
+ }
+
+ @Override
+ public Object deserialize(byte[] bytes) throws SerializationException {
+ try {
+ ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
+ ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream);
+
+ Object object = objectInputStream.readObject();
+
+ objectInputStream.close();
+
+ return object;
+ }catch (Exception e){
+ logger.error("deserialize Exception = [{}]", e.getMessage(), e);
+ }
+
+ return null;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisChannelAutoConfiguration.java b/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisChannelAutoConfiguration.java
new file mode 100644
index 0000000..9bf285b
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisChannelAutoConfiguration.java
@@ -0,0 +1,83 @@
+package org.apache.servicecomb.pack.alpha.fsm.channel.redis;
+
+import org.apache.servicecomb.pack.alpha.core.NodeStatus;
+import org.apache.servicecomb.pack.alpha.core.fsm.channel.MessagePublisher;
+import org.apache.servicecomb.pack.alpha.core.fsm.sink.ActorEventSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.data.redis.serializer.GenericToStringSerializer;
+import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+@Configuration
+@ConditionalOnClass(RedisConnection.class)
+@ConditionalOnProperty(value = "alpha.feature.akka.channel.type", havingValue = "redis")
+public class RedisChannelAutoConfiguration {
+ private static final Logger logger = LoggerFactory.getLogger(RedisChannelAutoConfiguration.class);
+
+ @Value("${alpha.feature.akka.channel.redis.topic:servicecomb-pack-actor-event}")
+ private String topic;
+
+ @Bean
+ public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
+ RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
+ redisTemplate.setKeySerializer(new StringRedisSerializer());
+ redisTemplate.setHashKeySerializer(new GenericToStringSerializer<>(Object.class));
+ redisTemplate.setHashValueSerializer(new JdkSerializationRedisSerializer());
+ redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
+ redisTemplate.setConnectionFactory(redisConnectionFactory);
+
+ return redisTemplate;
+ }
+
+ @Bean
+ RedisMessageSubscriber redisMessageSubscriber(@Lazy @Qualifier("actorEventSink") ActorEventSink actorEventSink,
+ @Lazy @Qualifier("nodeStatus") NodeStatus nodeStatus) {
+ return new RedisMessageSubscriber(actorEventSink, nodeStatus);
+ }
+
+ @Bean
+ public MessageListenerAdapter messageListenerAdapter(@Lazy @Qualifier("actorEventSink") ActorEventSink actorEventSink,
+ @Lazy @Qualifier("nodeStatus") NodeStatus nodeStatus) {
+ return new MessageListenerAdapter(redisMessageSubscriber(actorEventSink, nodeStatus));
+ }
+
+ @Bean
+ public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory,
+ @Lazy @Qualifier("actorEvetSink") ActorEventSink actorEventSink,
+ @Lazy @Qualifier("nodeStatus") NodeStatus nodeStatus) {
+ RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
+
+ redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
+ redisMessageListenerContainer.addMessageListener(redisMessageSubscriber(actorEventSink, nodeStatus), channelTopic());
+
+ return redisMessageListenerContainer;
+ }
+
+ @Bean
+ MessagePublisher redisMessagePublisher(RedisTemplate<String, Object> redisTemplate) {
+ return new RedisMessagePublisher(redisTemplate, channelTopic());
+ }
+
+ @Bean
+ ChannelTopic channelTopic() {
+ if (logger.isDebugEnabled()) {
+ logger.debug("build channel topic = [{}]", topic);
+ }
+ return new ChannelTopic(topic);
+ }
+
+}
diff --git a/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessagePublisher.java b/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessagePublisher.java
new file mode 100644
index 0000000..8957bb7
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessagePublisher.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.pack.alpha.fsm.channel.redis;
+
+
+import org.apache.servicecomb.pack.alpha.core.fsm.channel.MessagePublisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.ChannelTopic;
+
+public class RedisMessagePublisher implements MessagePublisher {
+
+ private static final Logger logger = LoggerFactory.getLogger(RedisMessagePublisher.class);
+
+ private RedisTemplate<String, Object> redisTemplate;
+ private ChannelTopic channelTopic;
+
+ public RedisMessagePublisher(RedisTemplate<String, Object> redisTemplate, ChannelTopic channelTopic) {
+ this.redisTemplate = redisTemplate;
+ this.channelTopic = channelTopic;
+ }
+
+ @Override
+ public void publish(Object data) {
+ if(logger.isDebugEnabled()) {
+ logger.debug("send message [{}] to [{}]", data, channelTopic.getTopic());
+ }
+ redisTemplate.convertAndSend(channelTopic.getTopic(), data);
+
+ }
+}
diff --git a/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessageSubscriber.java b/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessageSubscriber.java
new file mode 100644
index 0000000..3f11134
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessageSubscriber.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.pack.alpha.fsm.channel.redis;
+
+import org.apache.servicecomb.pack.alpha.core.NodeStatus;
+import org.apache.servicecomb.pack.alpha.core.fsm.event.base.BaseEvent;
+import org.apache.servicecomb.pack.alpha.core.fsm.sink.ActorEventSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+
+import java.nio.charset.StandardCharsets;
+
+public class RedisMessageSubscriber implements MessageListener {
+
+ private static final Logger logger = LoggerFactory.getLogger(RedisMessageSubscriber.class);
+
+ private ActorEventSink actorEventSink;
+ private NodeStatus nodeStatus;
+
+ private MessageSerializer messageSerializer = new MessageSerializer();
+
+ public RedisMessageSubscriber(ActorEventSink actorEventSink, NodeStatus nodeStatus) {
+ this.actorEventSink = actorEventSink;
+ this.nodeStatus = nodeStatus;
+ }
+
+ @Override
+ public void onMessage(Message message, byte[] pattern) {
+ if(nodeStatus.isMaster()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("pattern = [{}]", new String(pattern, StandardCharsets.UTF_8));
+ }
+
+ messageSerializer.deserialize(message.getBody()).ifPresent(data -> {
+
+ BaseEvent event = (BaseEvent) data;
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("event = [{}]", event);
+ }
+
+ try {
+ actorEventSink.send(event);
+ } catch (Exception e) {
+ logger.error("subscriber Exception = [{}]", e.getMessage(), e);
+ }
+ });
+ }else{
+ if(logger.isDebugEnabled()){
+ logger.debug("nodeStatus is not master and cancel this time subscribe");
+ }
+ }
+ }
+}
diff --git a/alpha/alpha-fsm-channel-redis/src/main/resources/META-INF/spring.factories b/alpha/alpha-fsm-channel-redis/src/main/resources/META-INF/spring.factories
new file mode 100644
index 0000000..0810009
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,17 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.apache.servicecomb.pack.alpha.fsm.channel.redis.RedisChannelAutoConfiguration
diff --git a/alpha/alpha-fsm-channel-redis/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisChannelTest.java b/alpha/alpha-fsm-channel-redis/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisChannelTest.java
new file mode 100644
index 0000000..5e1077a
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisChannelTest.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.pack.alpha.fsm;
+
+import org.apache.servicecomb.pack.alpha.core.NodeStatus;
+import org.apache.servicecomb.pack.alpha.core.fsm.event.SagaEndedEvent;
+import org.apache.servicecomb.pack.alpha.core.fsm.event.SagaStartedEvent;
+import org.apache.servicecomb.pack.alpha.core.fsm.event.TxEndedEvent;
+import org.apache.servicecomb.pack.alpha.core.fsm.event.TxStartedEvent;
+import org.apache.servicecomb.pack.alpha.core.fsm.event.base.BaseEvent;
+import org.apache.servicecomb.pack.alpha.fsm.channel.redis.MessageSerializer;
+import org.apache.servicecomb.pack.alpha.fsm.channel.redis.RedisMessagePublisher;
+import org.apache.servicecomb.pack.alpha.fsm.channel.redis.RedisMessageSubscriber;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.redis.connection.DefaultMessage;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.*;
+
+
+@RunWith(MockitoJUnitRunner.class)
+public class RedisChannelTest {
+
+ @Mock
+ private RedisConnection redisConnection;
+
+ @Mock
+ private RedisTemplate<String, Object> redisTemplate;
+
+ @Mock
+ private RedisConnectionFactory redisConnectionFactory;
+
+ @Spy
+ private ChannelTopic channelTopic = new ChannelTopic("redis-channel");
+
+ private RedisMessageListenerContainer redisMessageListenerContainer;
+
+ @Spy
+ private NodeStatus nodeStatus = new NodeStatus(NodeStatus.TypeEnum.MASTER);
+
+ @Spy
+ private RedisEventSink actorEventSink = new RedisEventSink();
+
+ private RedisMessagePublisher redisMessagePublisher;
+
+ private RedisMessageSubscriber redisMessageSubscriber;
+
+ private MessageListenerAdapter messageListenerAdapter;
+
+ @Before
+ public void setup(){
+ when(redisConnectionFactory.getConnection()).thenReturn(redisConnection);
+
+ redisTemplate.afterPropertiesSet();
+
+ redisMessageSubscriber = new RedisMessageSubscriber(actorEventSink, nodeStatus);
+ messageListenerAdapter = new MessageListenerAdapter(redisMessageSubscriber);
+ messageListenerAdapter.afterPropertiesSet();
+
+ redisMessageListenerContainer = new RedisMessageListenerContainer();
+ redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
+ redisMessageListenerContainer.addMessageListener(messageListenerAdapter, channelTopic);
+ redisMessageListenerContainer.afterPropertiesSet();
+ redisMessageListenerContainer.start();
+
+ redisMessagePublisher = new RedisMessagePublisher(redisTemplate, channelTopic);
+
+ }
+
+
+ @Test
+ public void testRedisPubSub(){
+ final String globalTxId = UUID.randomUUID().toString().replaceAll("-", "");
+ final String localTxId1 = UUID.randomUUID().toString().replaceAll("-", "");
+ final String localTxId2 = UUID.randomUUID().toString().replaceAll("-", "");
+ final String localTxId3 = UUID.randomUUID().toString().replaceAll("-", "");
+
+ MessageSerializer messageSerializer = new MessageSerializer();
+ buildData(globalTxId, localTxId1, localTxId2, localTxId3).forEach(baseEvent -> {
+ redisMessagePublisher.publish(baseEvent);
+ redisMessageSubscriber.onMessage(new DefaultMessage(channelTopic.getTopic().getBytes(), messageSerializer.serializer(baseEvent).orElse(new byte[0])), channelTopic.getTopic().getBytes());
+ });
+
+ assertEquals(0, actorEventSink.countDownLatch.getCount());
+ }
+
+ private List<BaseEvent> buildData(String globalTxId, String localTxId_1, String localTxId_2, String localTxId_3){
+ List<BaseEvent> sagaEvents = new ArrayList<>();
+ sagaEvents.add(SagaStartedEvent.builder().serviceName("service_g").instanceId("instance_g").globalTxId(globalTxId).build());
+ sagaEvents.add(TxStartedEvent.builder().serviceName("service_c1").instanceId("instance_c1").globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build());
+ sagaEvents.add(TxEndedEvent.builder().serviceName("service_c1").instanceId("instance_c1").globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build());
+ sagaEvents.add(TxStartedEvent.builder().serviceName("service_c2").instanceId("instance_c2").globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build());
+ sagaEvents.add(TxEndedEvent.builder().serviceName("service_c2").instanceId("instance_c2").globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build());
+ sagaEvents.add(TxStartedEvent.builder().serviceName("service_c3").instanceId("instance_c3").globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build());
+ sagaEvents.add(TxEndedEvent.builder().serviceName("service_c3").instanceId("instance_c3").globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build());
+ sagaEvents.add(SagaEndedEvent.builder().serviceName("service_g").instanceId("instance_g").globalTxId(globalTxId).build());
+ return sagaEvents;
+ }
+}
+
+
diff --git a/alpha/alpha-fsm-channel-redis/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisEventSink.java b/alpha/alpha-fsm-channel-redis/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisEventSink.java
new file mode 100644
index 0000000..f681e2b
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisEventSink.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.pack.alpha.fsm;
+
+import org.apache.servicecomb.pack.alpha.core.fsm.event.base.BaseEvent;
+import org.apache.servicecomb.pack.alpha.core.fsm.sink.ActorEventSink;
+
+import java.util.concurrent.CountDownLatch;
+
+public class RedisEventSink implements ActorEventSink {
+
+ public static final CountDownLatch countDownLatch = new CountDownLatch(8);
+
+ @Override
+ public void send(BaseEvent event) throws Exception {
+ countDownLatch.countDown();
+ }
+}
\ No newline at end of file
diff --git a/alpha/alpha-fsm-channel-redis/src/test/resources/log4j2.xml b/alpha/alpha-fsm-channel-redis/src/test/resources/log4j2.xml
new file mode 100644
index 0000000..58924c6
--- /dev/null
+++ b/alpha/alpha-fsm-channel-redis/src/test/resources/log4j2.xml
@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+
+<Configuration status="WARN">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Root level="info">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</Configuration>