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:14 UTC
[servicecomb-pack] branch master updated (70adb95 -> 7ba16cb)
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git.
from 70adb95 SCB-1423 Use the AtomInteger counter instead of ConcurrentLinkedQueue.size()
new 8eeaa12 SCB-1417 add alpha-fsm-channel-redis dependency to pom.xml
new 441a138 SCB-1417 add alpha-fsm-channel-redis module to alpha/pom.xml
new 4c5125c SCB-1417 add MessagePublisher to core
new 249d86e SCB-1417 delete fsm RedisChannelTest
new 572266b SCB-1417 delete FsmAutoConfiguration.RedisConfig
new 703d4b3 SCB-1417 delete fsm channel redis
new 6b675d2 SCB-1417 add alpha-fsm-channel-redis dependency to alpha-fsm
new 18f660e SCB-1417 change import package to FsmAutoConfiguration
new e62f336 SCB-1417 add alpha-fsm-channel-redis module
new 7ba16cb add License
The 10 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../alpha/core/fsm/channel}/MessagePublisher.java | 2 +-
alpha/alpha-fsm-channel-redis/README.md | 17 ++++
.../{alpha-fsm => alpha-fsm-channel-redis}/pom.xml | 73 +---------------
.../alpha/fsm/channel/redis/MessageSerializer.java | 0
.../redis/RedisChannelAutoConfiguration.java | 99 ++++++++++++++++++++++
.../fsm/channel/redis/RedisMessagePublisher.java | 1 +
.../fsm/channel/redis/RedisMessageSubscriber.java | 0
.../src/main/resources/META-INF/spring.factories | 2 +-
.../pack/alpha/fsm/RedisChannelTest.java | 39 ++++++---
.../pack/alpha/fsm/RedisEventSink.java} | 26 ++----
.../src/test/resources/log4j2.xml | 0
alpha/alpha-fsm/pom.xml | 4 +
.../pack/alpha/fsm/FsmAutoConfiguration.java | 70 ---------------
alpha/pom.xml | 1 +
pom.xml | 5 ++
15 files changed, 165 insertions(+), 174 deletions(-)
rename alpha/{alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis => alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/fsm/channel}/MessagePublisher.java (93%)
create mode 100644 alpha/alpha-fsm-channel-redis/README.md
copy alpha/{alpha-fsm => alpha-fsm-channel-redis}/pom.xml (61%)
rename alpha/{alpha-fsm => alpha-fsm-channel-redis}/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessageSerializer.java (100%)
create mode 100644 alpha/alpha-fsm-channel-redis/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisChannelAutoConfiguration.java
rename alpha/{alpha-fsm => alpha-fsm-channel-redis}/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessagePublisher.java (95%)
rename alpha/{alpha-fsm => alpha-fsm-channel-redis}/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessageSubscriber.java (100%)
copy {omega/omega-spring-cloud-consul-starter => alpha/alpha-fsm-channel-redis}/src/main/resources/META-INF/spring.factories (94%)
rename alpha/{alpha-fsm => alpha-fsm-channel-redis}/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisChannelTest.java (66%)
copy alpha/{alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/fsm/event/internal/StopEvent.java => alpha-fsm-channel-redis/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisEventSink.java} (68%)
copy alpha/{alpha-server => alpha-fsm-channel-redis}/src/test/resources/log4j2.xml (100%)
[servicecomb-pack] 09/10: SCB-1417 add alpha-fsm-channel-redis
module
Posted by ni...@apache.org.
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>
[servicecomb-pack] 05/10: SCB-1417 delete
FsmAutoConfiguration.RedisConfig
Posted by ni...@apache.org.
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 572266b2ccd505215278290f34190b8811b0c1b6
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 10:43:32 2019 +0800
SCB-1417 delete FsmAutoConfiguration.RedisConfig
---
.../pack/alpha/fsm/FsmAutoConfiguration.java | 55 ----------------------
1 file changed, 55 deletions(-)
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 a37f337..018d93a 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
@@ -144,61 +144,6 @@ public class FsmAutoConfiguration {
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);
- }
- }
-
@Bean
@ConditionalOnMissingBean(TransactionRepository.class)
public TransactionRepository transactionRepository() {
[servicecomb-pack] 03/10: SCB-1417 add MessagePublisher to core
Posted by ni...@apache.org.
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 4c5125cec0bd7ca0438a65456cddde76ca70c2e3
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 10:41:11 2019 +0800
SCB-1417 add MessagePublisher to core
---
.../alpha/core/fsm/channel/MessagePublisher.java | 23 ++++++++++++++++++++++
1 file changed, 23 insertions(+)
diff --git a/alpha/alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/fsm/channel/MessagePublisher.java b/alpha/alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/fsm/channel/MessagePublisher.java
new file mode 100644
index 0000000..a03cfb3
--- /dev/null
+++ b/alpha/alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/fsm/channel/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.core.fsm.channel;
+
+public interface MessagePublisher {
+
+ void publish(Object data);
+
+}
[servicecomb-pack] 01/10: SCB-1417 add alpha-fsm-channel-redis
dependency to pom.xml
Posted by ni...@apache.org.
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 8eeaa12d203cd2c174676315c74e81029f2b8335
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 10:39:44 2019 +0800
SCB-1417 add alpha-fsm-channel-redis dependency to pom.xml
---
pom.xml | 5 +++++
1 file changed, 5 insertions(+)
diff --git a/pom.xml b/pom.xml
index 62d7ae4..8c43768 100644
--- a/pom.xml
+++ b/pom.xml
@@ -321,6 +321,11 @@
</dependency>
<dependency>
<groupId>org.apache.servicecomb.pack</groupId>
+ <artifactId>alpha-fsm-channel-redis</artifactId>
+ <version>0.5.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.servicecomb.pack</groupId>
<artifactId>alpha-server</artifactId>
<version>0.5.0-SNAPSHOT</version>
</dependency>
[servicecomb-pack] 10/10: add License
Posted by ni...@apache.org.
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 7ba16cbae80ef3c0c5f445a5121a7a9f7e72b1ea
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 15:58:26 2019 +0800
add License
---
.../fsm/channel/redis/RedisChannelAutoConfiguration.java | 16 ++++++++++++++++
1 file changed, 16 insertions(+)
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
index 9bf285b..0a91531 100644
--- 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
@@ -1,3 +1,19 @@
+/*
+ * 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;
[servicecomb-pack] 07/10: SCB-1417 add alpha-fsm-channel-redis
dependency to alpha-fsm
Posted by ni...@apache.org.
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 6b675d246e14bd0949b6fbb8398217cf1d91c36f
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 10:47:09 2019 +0800
SCB-1417 add alpha-fsm-channel-redis dependency to alpha-fsm
---
alpha/alpha-fsm/pom.xml | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/alpha/alpha-fsm/pom.xml b/alpha/alpha-fsm/pom.xml
index 4102be6..4ec848b 100644
--- a/alpha/alpha-fsm/pom.xml
+++ b/alpha/alpha-fsm/pom.xml
@@ -67,6 +67,10 @@
<groupId>org.apache.servicecomb.pack</groupId>
<artifactId>alpha-core</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.servicecomb.pack</groupId>
+ <artifactId>alpha-fsm-channel-redis</artifactId>
+ </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
[servicecomb-pack] 06/10: SCB-1417 delete fsm channel redis
Posted by ni...@apache.org.
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 703d4b3f6fef8f8f804916a523ca36ab212374c0
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 10:45:15 2019 +0800
SCB-1417 delete fsm channel redis
---
.../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 ------------------
4 files changed, 224 deletions(-)
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
deleted file mode 100644
index 292877f..0000000
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessagePublisher.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * 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
deleted file mode 100644
index 711bb07..0000000
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/MessageSerializer.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * 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
deleted file mode 100644
index f1e0363..0000000
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessagePublisher.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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
deleted file mode 100644
index 3f11134..0000000
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/channel/redis/RedisMessageSubscriber.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * 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");
- }
- }
- }
-}
[servicecomb-pack] 02/10: SCB-1417 add alpha-fsm-channel-redis
module to alpha/pom.xml
Posted by ni...@apache.org.
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 441a1383cdff3df2e7e332a4fd9e2da83281cde4
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 10:40:04 2019 +0800
SCB-1417 add alpha-fsm-channel-redis module to alpha/pom.xml
---
alpha/pom.xml | 1 +
1 file changed, 1 insertion(+)
diff --git a/alpha/pom.xml b/alpha/pom.xml
index 01969a7..91ec0ed 100644
--- a/alpha/pom.xml
+++ b/alpha/pom.xml
@@ -33,6 +33,7 @@
<modules>
<module>alpha-core</module>
<module>alpha-fsm</module>
+ <module>alpha-fsm-channel-redis</module>
<module>alpha-benchmark</module>
<module>alpha-spring-cloud-starter-eureka</module>
<module>alpha-spring-cloud-starter-consul</module>
[servicecomb-pack] 04/10: SCB-1417 delete fsm RedisChannelTest
Posted by ni...@apache.org.
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 249d86ebc1be3e06f2c85cb184d57ba9ae59b48c
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 10:42:06 2019 +0800
SCB-1417 delete fsm RedisChannelTest
---
.../pack/alpha/fsm/RedisChannelTest.java | 119 ---------------------
1 file changed, 119 deletions(-)
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
deleted file mode 100644
index 9c9d11d..0000000
--- a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/RedisChannelTest.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * 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.core.fsm.event.base.BaseEvent;
-import org.apache.servicecomb.pack.alpha.fsm.metrics.MetricsService;
-import org.apache.servicecomb.pack.alpha.core.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 {
-
- }
-}
[servicecomb-pack] 08/10: SCB-1417 change import package to
FsmAutoConfiguration
Posted by ni...@apache.org.
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 18f660e2b38277d561d8d082c573a7f5021566b3
Author: CMonkey <42...@gmail.com>
AuthorDate: Wed Aug 7 10:48:06 2019 +0800
SCB-1417 change import package to FsmAutoConfiguration
---
.../servicecomb/pack/alpha/fsm/FsmAutoConfiguration.java | 15 ---------------
1 file changed, 15 deletions(-)
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 018d93a..c26b82c 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
@@ -25,11 +25,8 @@ import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import java.util.Map;
import javax.annotation.PostConstruct;
-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.repository.NoneTransactionRepository;
import org.apache.servicecomb.pack.alpha.fsm.repository.elasticsearch.ElasticsearchTransactionRepository;
@@ -43,10 +40,7 @@ import org.apache.servicecomb.pack.alpha.fsm.channel.MemoryActorEventChannel;
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.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;
@@ -54,15 +48,6 @@ 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;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
@Configuration