You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by zh...@apache.org on 2019/07/31 03:50:41 UTC
[servicecomb-pack] branch master updated: support fsm redis channel
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
The following commit(s) were added to refs/heads/master by this push:
new 8343590 support fsm redis channel
new 397237d Merge pull request #520 from cmonkey/feature/scb-1374
8343590 is described below
commit 8343590307436422f69abc157c74dba71348a525
Author: CMonkey <42...@gmail.com>
AuthorDate: Mon Jul 29 16:17:23 2019 +0800
support fsm redis channel
---
alpha/alpha-fsm/pom.xml | 14 +++
.../pack/alpha/fsm/FsmAutoConfiguration.java | 77 ++++++++++++-
.../fsm/channel/AbstractActorEventChannel.java | 4 +
.../alpha/fsm/channel/RedisActorEventChannel.java | 12 ++-
.../alpha/fsm/channel/redis/MessagePublisher.java | 23 ++++
.../alpha/fsm/channel/redis/MessageSerializer.java | 86 +++++++++++++++
.../fsm/channel/redis/RedisMessagePublisher.java | 45 ++++++++
.../fsm/channel/redis/RedisMessageSubscriber.java | 70 ++++++++++++
.../pack/alpha/fsm/RedisChannelTest.java | 119 +++++++++++++++++++++
.../src/main/resources/application.yaml | 5 +-
10 files changed, 450 insertions(+), 5 deletions(-)
diff --git a/alpha/alpha-fsm/pom.xml b/alpha/alpha-fsm/pom.xml
index b6c09d3..9f33266 100644
--- a/alpha/alpha-fsm/pom.xml
+++ b/alpha/alpha-fsm/pom.xml
@@ -148,6 +148,20 @@
<groupId>org.scalatest</groupId>
<artifactId>scalatest_2.12</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/src/main/java/org/apache/servicecomb/pack/alpha/fsm/FsmAutoConfiguration.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/FsmAutoConfiguration.java
index ef8248d..9101686 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/FsmAutoConfiguration.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/FsmAutoConfiguration.java
@@ -24,7 +24,12 @@ import akka.actor.ActorSystem;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import java.util.Map;
+
+import org.apache.servicecomb.pack.alpha.core.NodeStatus;
import org.apache.servicecomb.pack.alpha.fsm.channel.ActiveMQActorEventChannel;
+import org.apache.servicecomb.pack.alpha.fsm.channel.redis.MessagePublisher;
+import org.apache.servicecomb.pack.alpha.fsm.channel.redis.RedisMessagePublisher;
+import org.apache.servicecomb.pack.alpha.fsm.channel.redis.RedisMessageSubscriber;
import org.apache.servicecomb.pack.alpha.fsm.metrics.MetricsService;
import org.apache.servicecomb.pack.alpha.fsm.sink.ActorEventSink;
import org.apache.servicecomb.pack.alpha.fsm.channel.ActorEventChannel;
@@ -34,13 +39,26 @@ import org.apache.servicecomb.pack.alpha.fsm.channel.RedisActorEventChannel;
import org.apache.servicecomb.pack.alpha.fsm.sink.SagaActorEventSender;
import org.apache.servicecomb.pack.alpha.fsm.spring.integration.akka.AkkaConfigPropertyAdapter;
import org.apache.servicecomb.pack.alpha.fsm.spring.integration.eventbus.EventSubscribeBeanPostProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
import org.springframework.core.env.ConfigurableEnvironment;
+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
@ConditionalOnProperty(value = {"alpha.feature.akka.enabled"})
@@ -99,8 +117,63 @@ public class FsmAutoConfiguration {
@Bean
@ConditionalOnProperty(value = "alpha.feature.akka.channel.type", havingValue = "redis")
- public ActorEventChannel redisEventChannel(ActorEventSink actorEventSink, MetricsService metricsService){
- return new RedisActorEventChannel(actorEventSink, metricsService);
+ public ActorEventChannel redisEventChannel(ActorEventSink actorEventSink, MetricsService metricsService, @Lazy RedisMessagePublisher redisMessagePublisher){
+ return new RedisActorEventChannel(actorEventSink, metricsService, redisMessagePublisher);
+ }
+
+ @Configuration
+ @ConditionalOnClass(RedisConnection.class)
+ @ConditionalOnProperty(value = "alpha.feature.akka.channel.type", havingValue = "redis")
+ public static class RedisConfig{
+ private static final Logger logger = LoggerFactory.getLogger(RedisConfig.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(ActorEventSink actorEventSink, NodeStatus nodeStatus){
+ return new RedisMessageSubscriber(actorEventSink, nodeStatus);
+ }
+
+ @Bean
+ public MessageListenerAdapter messageListenerAdapter(ActorEventSink actorEventSink, NodeStatus nodeStatus){
+ return new MessageListenerAdapter(redisMessageSubscriber(actorEventSink, nodeStatus));
+ }
+
+ @Bean
+ public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory, ActorEventSink actorEventSink, 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/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/AbstractActorEventChannel.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/AbstractActorEventChannel.java
index 278d7d6..3df4506 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/AbstractActorEventChannel.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/AbstractActorEventChannel.java
@@ -20,8 +20,11 @@ package org.apache.servicecomb.pack.alpha.fsm.channel;
import org.apache.servicecomb.pack.alpha.fsm.event.base.BaseEvent;
import org.apache.servicecomb.pack.alpha.fsm.metrics.MetricsService;
import org.apache.servicecomb.pack.alpha.fsm.sink.ActorEventSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public abstract class AbstractActorEventChannel implements ActorEventChannel {
+ private static final Logger logger = LoggerFactory.getLogger(AbstractActorEventChannel.class);
protected final MetricsService metricsService;
protected final ActorEventSink actorEventSink;
@@ -42,6 +45,7 @@ public abstract class AbstractActorEventChannel implements ActorEventChannel {
this.sendTo(event);
metricsService.metrics().doEventAccepted();
} catch (Exception ex) {
+ logger.error("send Exception = [{}]", ex.getMessage(), ex);
metricsService.metrics().doEventRejected();
}
long end = System.currentTimeMillis();
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/RedisActorEventChannel.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/RedisActorEventChannel.java
index 71319dd..15104e4 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/RedisActorEventChannel.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/RedisActorEventChannel.java
@@ -18,6 +18,8 @@
package org.apache.servicecomb.pack.alpha.fsm.channel;
import java.lang.invoke.MethodHandles;
+
+import org.apache.servicecomb.pack.alpha.fsm.channel.redis.RedisMessagePublisher;
import org.apache.servicecomb.pack.alpha.fsm.event.base.BaseEvent;
import org.apache.servicecomb.pack.alpha.fsm.metrics.MetricsService;
import org.apache.servicecomb.pack.alpha.fsm.sink.ActorEventSink;
@@ -31,13 +33,19 @@ import org.slf4j.LoggerFactory;
public class RedisActorEventChannel extends AbstractActorEventChannel {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private RedisMessagePublisher redisMessagePublisher;
+
public RedisActorEventChannel(
- ActorEventSink actorEventSink, MetricsService metricsService) {
+ ActorEventSink actorEventSink, MetricsService metricsService, RedisMessagePublisher redisMessagePublisher) {
super(actorEventSink, metricsService);
+ this.redisMessagePublisher = redisMessagePublisher;
}
@Override
public void sendTo(BaseEvent event){
- throw new UnsupportedOperationException("Doesn't implement yet!");
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("sendTo message = [{}]", event);
+ }
+ redisMessagePublisher.publish(event);
}
}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessagePublisher.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessagePublisher.java
new file mode 100644
index 0000000..292877f
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessagePublisher.java
@@ -0,0 +1,23 @@
+/*
+ * 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;
+
+public interface MessagePublisher {
+
+ void publish(Object data);
+
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessageSerializer.java b/alpha/alpha-fsm/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/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/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessagePublisher.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessagePublisher.java
new file mode 100644
index 0000000..f1e0363
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessagePublisher.java
@@ -0,0 +1,45 @@
+/*
+ * 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.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/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessageSubscriber.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessageSubscriber.java
new file mode 100644
index 0000000..10d5514
--- /dev/null
+++ b/alpha/alpha-fsm/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.fsm.event.base.BaseEvent;
+import org.apache.servicecomb.pack.alpha.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/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisChannelTest.java b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisChannelTest.java
new file mode 100644
index 0000000..729b859
--- /dev/null
+++ b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisChannelTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.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.apache.servicecomb.pack.alpha.fsm.event.base.BaseEvent;
+import org.apache.servicecomb.pack.alpha.fsm.metrics.MetricsService;
+import org.apache.servicecomb.pack.alpha.fsm.sink.ActorEventSink;
+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.UUID;
+
+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;
+
+ @Mock
+ MetricsService metricsServicee;
+
+ @Spy
+ private NodeStatus nodeStatus = new NodeStatus(NodeStatus.TypeEnum.MASTER);
+
+ @Spy
+ private ActorEventSink 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();
+ SagaEventSender.successfulEvents(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());
+ });
+ }
+}
+
+class RedisEventSink implements ActorEventSink{
+
+ @Override
+ public void send(BaseEvent event) throws Exception {
+
+ }
+}
diff --git a/alpha/alpha-server/src/main/resources/application.yaml b/alpha/alpha-server/src/main/resources/application.yaml
index fa1b35a..6b5180f 100644
--- a/alpha/alpha-server/src/main/resources/application.yaml
+++ b/alpha/alpha-server/src/main/resources/application.yaml
@@ -62,6 +62,9 @@ management:
web:
exposure:
include: "*"
+ health:
+ redis:
+ enabled: false
---
spring:
@@ -115,4 +118,4 @@ akkaConfig:
host: localhost
port: 6379
database: 0
- #password:
\ No newline at end of file
+ #password: