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 2018/08/31 00:43:32 UTC
[incubator-servicecomb-saga] 06/07: SCB-876 Add omega Tcc workflow
unit test case & refactor.
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/incubator-servicecomb-saga.git
commit 39a69c5b7d426f15c4c64835f1f3f572900e8386
Author: cherrylzhao <zh...@126.com>
AuthorDate: Thu Aug 30 23:53:31 2018 +0800
SCB-876 Add omega Tcc workflow unit test case & refactor.
---
.../saga/omega/context/CallbackContext.java | 2 +-
.../saga/omega/context/OmegaContext.java | 18 ++-
.../spring/CompensableMethodCheckingCallback.java | 3 +-
.../transaction/spring/ExecutorFieldCallback.java | 2 -
.../spring/ParticipateMethodCheckingCallback.java | 3 +-
.../spring/TransactionAspectConfig.java | 10 +-
.../omega/transaction/spring/MessageConfig.java | 143 +++++++++++++++++
.../transaction/spring/TccInterceptorTest.java | 170 +++++++++++++++++++++
.../omega/transaction/spring/TccUserService.java} | 46 ++++--
.../transaction/spring/TccUserServiceMain.java} | 29 ++--
.../spring/TransactionInterceptionTest.java | 142 +++--------------
.../transaction/tcc/CoordinateMessageHandler.java | 2 +-
.../transaction/tcc/TccParticipatorAspect.java | 18 ++-
.../transaction/tcc/events/CoordinatedEvent.java | 11 ++
.../transaction/tcc/events/ParticipatedEvent.java | 12 ++
.../transaction/tcc/events/TccEndedEvent.java | 9 ++
.../transaction/tcc/events/TccStartedEvent.java | 8 +
.../tcc/CoordinateMessageHandlerTest.java | 8 +-
18 files changed, 456 insertions(+), 180 deletions(-)
diff --git a/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/CallbackContext.java b/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/CallbackContext.java
index 77e2aff..0fe2613 100644
--- a/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/CallbackContext.java
+++ b/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/CallbackContext.java
@@ -43,7 +43,7 @@ public class CallbackContext {
public void apply(String globalTxId, String localTxId, String callbackMethod, Object... payloads) {
CallbackContextInternal contextInternal = contexts.get(callbackMethod);
String oldGlobalTxId = omegaContext.globalTxId();
- String oldLocalTxId= omegaContext.localTxId();
+ String oldLocalTxId = omegaContext.localTxId();
try {
omegaContext.setGlobalTxId(globalTxId);
omegaContext.setLocalTxId(localTxId);
diff --git a/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/OmegaContext.java b/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/OmegaContext.java
index 8ad6083..4055a69 100644
--- a/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/OmegaContext.java
+++ b/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/OmegaContext.java
@@ -17,7 +17,8 @@
package org.apache.servicecomb.saga.omega.context;
-import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
/**
* OmegaContext holds the globalTxId and localTxId which are used to build the invocation map
@@ -29,7 +30,12 @@ public class OmegaContext {
private final ThreadLocal<String> globalTxId = new InheritableThreadLocal<>();
private final ThreadLocal<String> localTxId = new InheritableThreadLocal<>();
private final IdGenerator<String> idGenerator;
- private final ThreadLocal<Collection> parameters = new InheritableThreadLocal<>();
+ private final ThreadLocal<Map<String, Object[]>> parameters = new InheritableThreadLocal<Map<String, Object[]>>() {
+ @Override
+ protected Map<String, Object[]> initialValue() {
+ return new HashMap<>();
+ }
+ };
public OmegaContext(IdGenerator<String> idGenerator) {
this.idGenerator = idGenerator;
@@ -63,12 +69,12 @@ public class OmegaContext {
return localTxId.get();
}
- public void setParameters(Collection parameters) {
- this.parameters.set(parameters);
+ public void setParameters(String localTxId, Object ... parameters) {
+ this.parameters.get().put(localTxId, parameters);
}
- public Collection parameters() {
- return parameters.get();
+ public Object[] parameters(String localTxId) {
+ return parameters.get().get(localTxId);
}
public void clear() {
diff --git a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/CompensableMethodCheckingCallback.java b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/CompensableMethodCheckingCallback.java
index 18a6b32..e9bbd42 100644
--- a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/CompensableMethodCheckingCallback.java
+++ b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/CompensableMethodCheckingCallback.java
@@ -20,9 +20,8 @@ package org.apache.servicecomb.saga.omega.transaction.spring;
import java.lang.reflect.Method;
import org.apache.servicecomb.saga.omega.context.CallbackContext;
import org.apache.servicecomb.saga.omega.transaction.annotations.Compensable;
-import org.springframework.util.ReflectionUtils.MethodCallback;
-class CompensableMethodCheckingCallback extends MethodCheckingCallback implements MethodCallback {
+class CompensableMethodCheckingCallback extends MethodCheckingCallback {
public CompensableMethodCheckingCallback(Object bean, CallbackContext callbackContext) {
super(bean, callbackContext, "compensation");
diff --git a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ExecutorFieldCallback.java b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ExecutorFieldCallback.java
index 9ef937e..07a9466 100644
--- a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ExecutorFieldCallback.java
+++ b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ExecutorFieldCallback.java
@@ -27,8 +27,6 @@ import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
-import java.util.stream.Collectors;
-
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.transaction.spring.annotations.OmegaContextAware;
import org.slf4j.Logger;
diff --git a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java
index e8b4b77..b704f89 100644
--- a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java
+++ b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java
@@ -20,9 +20,8 @@ package org.apache.servicecomb.saga.omega.transaction.spring;
import java.lang.reflect.Method;
import org.apache.servicecomb.saga.omega.context.CallbackContext;
import org.apache.servicecomb.saga.omega.transaction.annotations.Participate;
-import org.springframework.util.ReflectionUtils.MethodCallback;
-public class ParticipateMethodCheckingCallback extends MethodCheckingCallback implements MethodCallback {
+public class ParticipateMethodCheckingCallback extends MethodCheckingCallback {
public ParticipateMethodCheckingCallback(Object bean, CallbackContext callbackContext) {
super(bean, callbackContext, "coordinate");
diff --git a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionAspectConfig.java b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionAspectConfig.java
index 26c6512..1516394 100644
--- a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionAspectConfig.java
+++ b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionAspectConfig.java
@@ -64,17 +64,19 @@ public class TransactionAspectConfig {
@Bean
org.apache.servicecomb.saga.omega.transaction.tcc.MessageHandler coordinateMessageHandler(
- TccEventService tccEventService) {
- return new CoordinateMessageHandler(tccEventService);
+ TccEventService tccEventService,
+ @Qualifier("coordinateContext") CallbackContext coordinateContext,
+ OmegaContext omegaContext) {
+ return new CoordinateMessageHandler(tccEventService, coordinateContext, omegaContext);
}
- @Order(2)
+ @Order(0)
@Bean
TccStartAspect tccStartAspect(TccEventService tccEventService, OmegaContext context) {
return new TccStartAspect(tccEventService, context);
}
- @Order(3)
+ @Order(1)
@Bean
TccParticipatorAspect tccParticipatorAspect(TccEventService tccEventService, OmegaContext context) {
return new TccParticipatorAspect(tccEventService, context);
diff --git a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/MessageConfig.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/MessageConfig.java
new file mode 100644
index 0000000..6651248
--- /dev/null
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/MessageConfig.java
@@ -0,0 +1,143 @@
+/*
+ * 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.saga.omega.transaction.spring;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.servicecomb.saga.omega.context.CallbackContext;
+import org.apache.servicecomb.saga.omega.context.IdGenerator;
+import org.apache.servicecomb.saga.omega.context.OmegaContext;
+import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
+import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+import org.apache.servicecomb.saga.omega.transaction.TxEvent;
+import org.apache.servicecomb.saga.omega.transaction.tcc.TccEventService;
+import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent;
+import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
+import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
+import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccStartedEvent;
+import org.mockito.Mockito;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class MessageConfig {
+ private final List<String> messages = new ArrayList<>();
+
+ @Bean
+ IdGenerator<String> idGenerator() {
+ return Mockito.mock(IdGenerator.class);
+ }
+
+ @Bean(name = "compensationContext")
+ CallbackContext recoveryCompensationContext(OmegaContext omegaContext) {
+ return new CallbackContext(omegaContext);
+ }
+
+ @Bean(name = "coordinateContext")
+ CallbackContext coordinateContext(OmegaContext omegaContext) {
+ return new CallbackContext(omegaContext);
+ }
+
+ @Bean
+ OmegaContext omegaContext(IdGenerator<String> idGenerator) {
+ return new OmegaContext(idGenerator);
+ }
+
+ @Bean
+ List<String> messages() {
+ return messages;
+ }
+
+ @Bean
+ MessageSender sender() {
+ return new MessageSender() {
+ @Override
+ public void onConnected() {
+
+ }
+
+ @Override
+ public void onDisconnected() {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOW";
+ }
+
+ @Override
+ public AlphaResponse send(TxEvent event) {
+ messages.add(event.toString());
+ return new AlphaResponse(false);
+ }
+ };
+ }
+
+ @Bean
+ TccEventService tccEventService() {
+ return new TccEventService() {
+ @Override
+ public void onConnected() {
+ }
+
+ @Override
+ public void onDisconnected() {
+
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public String target() {
+ return "UNKNOWN";
+ }
+
+ @Override
+ public AlphaResponse participate(ParticipatedEvent participateEvent) {
+ messages.add(participateEvent.toString());
+ return new AlphaResponse(false);
+ }
+
+ @Override
+ public AlphaResponse tccTransactionStart(TccStartedEvent tccStartEvent) {
+ messages.add(tccStartEvent.toString());
+ return new AlphaResponse(false);
+ }
+
+ @Override
+ public AlphaResponse tccTransactionStop(TccEndedEvent tccEndEvent) {
+ messages.add(tccEndEvent.toString());
+ return new AlphaResponse(false);
+ }
+
+ @Override
+ public AlphaResponse coordinate(CoordinatedEvent coordinatedEvent) {
+ messages.add(coordinatedEvent.toString());
+ return new AlphaResponse(false);
+ }
+ };
+ }
+}
diff --git a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TccInterceptorTest.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TccInterceptorTest.java
new file mode 100644
index 0000000..2a52264
--- /dev/null
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TccInterceptorTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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.saga.omega.transaction.spring;
+
+import static com.seanyinx.github.unit.scaffolding.AssertUtils.expectFailing;
+import static com.seanyinx.github.unit.scaffolding.Randomness.uniquify;
+import static org.apache.servicecomb.saga.omega.transaction.spring.TransactionalUserService.ILLEGAL_USER;
+import static org.assertj.core.util.IterableUtil.toArray;
+import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.UUID;
+import org.apache.servicecomb.saga.common.TransactionStatus;
+import org.apache.servicecomb.saga.omega.context.IdGenerator;
+import org.apache.servicecomb.saga.omega.context.OmegaContext;
+import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent;
+import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
+import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
+import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccStartedEvent;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = {TransactionTestMain.class, MessageConfig.class})
+public class TccInterceptorTest {
+
+ private static final String globalTxId = UUID.randomUUID().toString();
+ private final String newLocalTxId = UUID.randomUUID().toString();
+ private final String anotherLocalTxId = UUID.randomUUID().toString();
+ private final String username = uniquify("username");
+ private final String email = uniquify("email");
+
+ private final User user = new User(username, email);
+ private final User illegalUser = new User(ILLEGAL_USER, email);
+
+ private final String usernameJack = uniquify("Jack");
+ private final User jack = new User(usernameJack, uniquify("jack@gmail.com"));
+
+ @Autowired
+ private IdGenerator<String> idGenerator;
+
+ @Autowired
+ private List<String> messages;
+
+ @Autowired
+ private OmegaContext omegaContext;
+
+ @Autowired
+ private TccUserServiceMain tccUserServiceMain;
+
+ @Autowired
+ private UserRepository userRepository;
+
+ private String confirmMethod;
+
+ private String cancelMethod;
+
+ @Autowired
+ private org.apache.servicecomb.saga.omega.transaction.tcc.MessageHandler coordinateMessageHandler;
+
+ @Before
+ public void setUp() throws Exception {
+ when(idGenerator.nextId()).thenReturn(globalTxId, newLocalTxId, anotherLocalTxId);
+// omegaContext.setGlobalTxId(globalTxId);
+// omegaContext.setLocalTxId(globalTxId);
+ confirmMethod = TccUserService.class.getDeclaredMethod("confirm", User.class).toString();
+ cancelMethod = TccUserService.class.getDeclaredMethod("cancel", User.class).toString();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ messages.clear();
+ userRepository.deleteAll();
+ omegaContext.clear();
+ tccUserServiceMain.resetCount();
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ }
+
+ @Test
+ public void tccWorkflowSucceed() {
+ tccUserServiceMain.add(user, jack);
+
+ // TODO could't get parameters, for OmegaContext has been cleared after sending TccEndedEvent synchronous.
+ // TODO CoordinateCommand will be send from alpha asynchronous.
+ omegaContext.setParameters(newLocalTxId, user);
+ omegaContext.setParameters(anotherLocalTxId, jack);
+ coordinateMessageHandler.onReceive(globalTxId, newLocalTxId, globalTxId, confirmMethod);
+ coordinateMessageHandler.onReceive(globalTxId, newLocalTxId, anotherLocalTxId, confirmMethod);
+
+ assertArrayEquals(
+ new String[] {
+ new TccStartedEvent(globalTxId, globalTxId).toString(),
+ new ParticipatedEvent(globalTxId, newLocalTxId, globalTxId, confirmMethod, cancelMethod, TransactionStatus.Succeed).toString(),
+ new ParticipatedEvent(globalTxId, anotherLocalTxId, globalTxId, confirmMethod, cancelMethod, TransactionStatus.Succeed).toString(),
+ new TccEndedEvent(globalTxId, globalTxId, TransactionStatus.Succeed).toString(),
+ new CoordinatedEvent(globalTxId, newLocalTxId, globalTxId, confirmMethod, TransactionStatus.Succeed).toString(),
+ new CoordinatedEvent(globalTxId, anotherLocalTxId, globalTxId, confirmMethod, TransactionStatus.Succeed).toString()
+ },
+ toArray(messages)
+ );
+
+ User result = userRepository.findByUsername(user.username());
+ assertThat(result.username(), is(user.username()));
+ assertThat(result.email(), is(user.email()));
+
+ result = userRepository.findByUsername(jack.username());
+ assertThat(result.username(), is(jack.username()));
+ assertThat(result.email(), is(jack.email()));
+ }
+
+ @Test
+ public void tccWorkflowFailed() {
+ try {
+ tccUserServiceMain.add(user, illegalUser);
+ expectFailing(IllegalArgumentException.class);
+ } catch (IllegalArgumentException ignored) {
+
+ }
+
+ // TODO could't get parameters, for OmegaContext has been cleared after sending TccEndedEvent synchronous.
+ // TODO CoordinateCommand will be send from alpha asynchronous.
+ omegaContext.setParameters(newLocalTxId, user);
+ omegaContext.setParameters(anotherLocalTxId, illegalUser);
+ coordinateMessageHandler.onReceive(globalTxId, newLocalTxId, globalTxId, cancelMethod);
+ assertArrayEquals(
+ new String[] {
+ new TccStartedEvent(globalTxId, globalTxId).toString(),
+ new ParticipatedEvent(globalTxId, newLocalTxId, globalTxId, confirmMethod, cancelMethod, TransactionStatus.Succeed).toString(),
+ new ParticipatedEvent(globalTxId, anotherLocalTxId, globalTxId, confirmMethod, cancelMethod, TransactionStatus.Failed).toString(),
+ new TccEndedEvent(globalTxId, globalTxId, TransactionStatus.Failed).toString(),
+ new CoordinatedEvent(globalTxId, newLocalTxId, globalTxId, cancelMethod, TransactionStatus.Succeed).toString()
+ },
+ toArray(messages)
+ );
+
+ User result = userRepository.findByUsername(user.username());
+ assertThat(result, is(nullValue()));
+
+ result = userRepository.findByUsername(jack.username());
+ assertThat(result, is(nullValue()));
+ }
+}
diff --git a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TccUserService.java
similarity index 51%
copy from omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java
copy to omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TccUserService.java
index e8b4b77..eefe749 100644
--- a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TccUserService.java
@@ -17,24 +17,44 @@
package org.apache.servicecomb.saga.omega.transaction.spring;
-import java.lang.reflect.Method;
-import org.apache.servicecomb.saga.omega.context.CallbackContext;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+
import org.apache.servicecomb.saga.omega.transaction.annotations.Participate;
-import org.springframework.util.ReflectionUtils.MethodCallback;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+class TccUserService {
+ static final String ILLEGAL_USER = "Illegal User";
+ private final UserRepository userRepository;
-public class ParticipateMethodCheckingCallback extends MethodCheckingCallback implements MethodCallback {
+ private int count = 0;
- public ParticipateMethodCheckingCallback(Object bean, CallbackContext callbackContext) {
- super(bean, callbackContext, "coordinate");
+ @Autowired
+ TccUserService(UserRepository userRepository) {
+ this.userRepository = userRepository;
}
- @Override
- public void doWith(Method method) throws IllegalArgumentException {
- if (!method.isAnnotationPresent(Participate.class)) {
- return;
+ void resetCount() {
+ this.count = 0;
+ }
+
+ @Participate(confirmMethod = "confirm", cancelMethod = "cancel")
+ User add(User user) {
+ if (ILLEGAL_USER.equals(user.username())) {
+ throw new IllegalArgumentException("User is illegal");
}
- String confirmMethod = method.getAnnotation(Participate.class).confirmMethod();
- String cancelMethod = method.getAnnotation(Participate.class).cancelMethod();
- loadMethodContext(method, confirmMethod, cancelMethod);
+ return userRepository.save(user);
+ }
+
+ void confirm(User user) {
+ userRepository.findByUsername(user.username());
+ }
+
+ void cancel(User user) {
+ userRepository.delete(user);
}
+
+
}
diff --git a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TccUserServiceMain.java
similarity index 50%
copy from omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java
copy to omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TccUserServiceMain.java
index e8b4b77..94e5232 100644
--- a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/ParticipateMethodCheckingCallback.java
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TccUserServiceMain.java
@@ -17,24 +17,23 @@
package org.apache.servicecomb.saga.omega.transaction.spring;
-import java.lang.reflect.Method;
-import org.apache.servicecomb.saga.omega.context.CallbackContext;
-import org.apache.servicecomb.saga.omega.transaction.annotations.Participate;
-import org.springframework.util.ReflectionUtils.MethodCallback;
+import org.apache.servicecomb.saga.omega.context.annotations.TccStart;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
-public class ParticipateMethodCheckingCallback extends MethodCheckingCallback implements MethodCallback {
+@Component
+public class TccUserServiceMain {
- public ParticipateMethodCheckingCallback(Object bean, CallbackContext callbackContext) {
- super(bean, callbackContext, "coordinate");
+ @Autowired
+ private TccUserService tccUserService;
+
+ void resetCount() {
+ tccUserService.resetCount();
}
- @Override
- public void doWith(Method method) throws IllegalArgumentException {
- if (!method.isAnnotationPresent(Participate.class)) {
- return;
- }
- String confirmMethod = method.getAnnotation(Participate.class).confirmMethod();
- String cancelMethod = method.getAnnotation(Participate.class).cancelMethod();
- loadMethodContext(method, confirmMethod, cancelMethod);
+ @TccStart
+ void add(User A, User B) {
+ tccUserService.add(A);
+ tccUserService.add(B);
}
}
diff --git a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java
index f340bfe..6b66bef 100644
--- a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionInterceptionTest.java
@@ -33,60 +33,43 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.when;
-import java.util.ArrayList;
+import akka.actor.AbstractLoggingActor;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.japi.Creator;
+import akka.japi.pf.FI.UnitApply;
+import io.reactivex.Flowable;
+import io.reactivex.functions.Consumer;
+import io.reactivex.schedulers.Schedulers;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-
-import org.apache.servicecomb.saga.omega.context.CallbackContext;
import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
-import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
-import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxAbortedEvent;
import org.apache.servicecomb.saga.omega.transaction.TxCompensatedEvent;
import org.apache.servicecomb.saga.omega.transaction.TxEndedEvent;
-import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.apache.servicecomb.saga.omega.transaction.TxStartedEvent;
-import org.apache.servicecomb.saga.omega.transaction.spring.TransactionInterceptionTest.MessageConfig;
import org.apache.servicecomb.saga.omega.transaction.spring.annotations.OmegaContextAware;
-import org.apache.servicecomb.saga.omega.transaction.tcc.TccEventService;
-import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccStartedEvent;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.junit4.SpringRunner;
-import io.reactivex.Flowable;
-import io.reactivex.functions.Consumer;
-import io.reactivex.schedulers.Schedulers;
-import akka.actor.AbstractLoggingActor;
-import akka.actor.ActorRef;
-import akka.actor.ActorSystem;
-import akka.actor.Props;
-import akka.japi.Creator;
-import akka.japi.pf.FI.UnitApply;
-
@RunWith(SpringRunner.class)
@SpringBootTest(classes = {TransactionTestMain.class, MessageConfig.class})
@AutoConfigureMockMvc
public class TransactionInterceptionTest {
@SuppressWarnings("unchecked")
- private static final IdGenerator<String> idGenerator = Mockito.mock(IdGenerator.class);
private static final String globalTxId = UUID.randomUUID().toString();
private final String newLocalTxId = UUID.randomUUID().toString();
private final String anotherLocalTxId = UUID.randomUUID().toString();
@@ -103,6 +86,9 @@ public class TransactionInterceptionTest {
private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
@Autowired
+ private IdGenerator<String> idGenerator;
+
+ @Autowired
private List<String> messages;
@Autowired
@@ -367,6 +353,11 @@ public class TransactionInterceptionTest {
actorSystem.terminate();
}
+ @Test
+ public void tccWorkflowTest() {
+
+ }
+
private void waitTillSavedUser(final String username) {
await().atMost(1000, MILLISECONDS).until(new Callable<Boolean>() {
@Override
@@ -407,103 +398,4 @@ public class TransactionInterceptionTest {
private String[] toArray(List<String> messages) {
return messages.toArray(new String[messages.size()]);
}
-
- @Configuration
- static class MessageConfig {
- private final List<String> messages = new ArrayList<>();
-
- @Bean(name = "compensationContext")
- CallbackContext recoveryCompensationContext(OmegaContext omegaContext) {
- return new CallbackContext(omegaContext);
- }
-
- @Bean(name = "coordinateContext")
- CallbackContext recoveryCoordinateContext(OmegaContext omegaContext) {
- return new CallbackContext(omegaContext);
- }
-
- @Bean
- OmegaContext omegaContext() {
- return new OmegaContext(idGenerator);
- }
-
- @Bean
- List<String> messages() {
- return messages;
- }
-
- @Bean
- MessageSender sender() {
- return new MessageSender() {
- @Override
- public void onConnected() {
-
- }
-
- @Override
- public void onDisconnected() {
-
- }
-
- @Override
- public void close() {
-
- }
-
- @Override
- public String target() {
- return "UNKNOW";
- }
-
- @Override
- public AlphaResponse send(TxEvent event) {
- messages.add(event.toString());
- return new AlphaResponse(false);
- }
- };
- }
-
- @Bean
- TccEventService tccEventService() {
- return new TccEventService() {
- @Override
- public void onConnected() {
- }
-
- @Override
- public void onDisconnected() {
-
- }
-
- @Override
- public void close() {
- }
-
- @Override
- public String target() {
- return "UNKNOWN";
- }
-
- @Override
- public AlphaResponse participate(ParticipatedEvent participateEvent) {
- return null;
- }
-
- @Override
- public AlphaResponse tccTransactionStart(TccStartedEvent tccStartEvent) {
- return null;
- }
-
- @Override
- public AlphaResponse tccTransactionStop(TccEndedEvent tccEndEvent) {
- return null;
- }
-
- @Override
- public AlphaResponse coordinate(CoordinatedEvent coordinatedEvent) {
- return null;
- }
- };
- }
- }
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandler.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandler.java
index ca39ccd..65f3cd5 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandler.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandler.java
@@ -39,7 +39,7 @@ public class CoordinateMessageHandler implements MessageHandler {
@Override
public void onReceive(String globalTxId, String localTxId, String parentTxId, String methodName) {
- callbackContext.apply(globalTxId, localTxId, methodName, omegaContext.parameters());
+ callbackContext.apply(globalTxId, localTxId, methodName, omegaContext.parameters(localTxId));
tccEventService.coordinate(new CoordinatedEvent(globalTxId, localTxId, parentTxId, methodName, TransactionStatus.Succeed));
}
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java
index a78d819..6ad9253 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java
@@ -19,14 +19,8 @@ package org.apache.servicecomb.saga.omega.transaction.tcc;
import java.lang.invoke.MethodHandles;
import java.lang.reflect.Method;
-
import org.apache.servicecomb.saga.common.TransactionStatus;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
-import org.apache.servicecomb.saga.omega.transaction.MessageSender;
-import org.apache.servicecomb.saga.omega.transaction.OmegaException;
-import org.apache.servicecomb.saga.omega.transaction.RecoveryPolicy;
-import org.apache.servicecomb.saga.omega.transaction.RecoveryPolicyFactory;
-import org.apache.servicecomb.saga.omega.transaction.annotations.Compensable;
import org.apache.servicecomb.saga.omega.transaction.annotations.Participate;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
import org.aspectj.lang.ProceedingJoinPoint;
@@ -52,8 +46,8 @@ public class TccParticipatorAspect {
Object advise(ProceedingJoinPoint joinPoint, Participate participate) throws Throwable {
Method method = ((MethodSignature) joinPoint.getSignature()).getMethod();
String localTxId = context.localTxId();
- String cancelMethod = participate.cancelMethod();
- String confirmMethod = participate.confirmMethod();
+ String cancelMethod = callbackMethodSignature(joinPoint, participate.cancelMethod(), method);
+ String confirmMethod = callbackMethodSignature(joinPoint, participate.confirmMethod(), method);
context.newLocalTxId();
LOG.debug("Updated context {} for participate method {} ", context, method.toString());
@@ -75,4 +69,12 @@ public class TccParticipatorAspect {
context.setLocalTxId(localTxId);
}
}
+
+ String callbackMethodSignature(ProceedingJoinPoint joinPoint, String callbackMethod, Method tryMethod) throws NoSuchMethodException {
+ return callbackMethod.isEmpty() ? "" :
+ joinPoint.getTarget()
+ .getClass()
+ .getDeclaredMethod(callbackMethod, tryMethod.getParameterTypes())
+ .toString();
+ }
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/CoordinatedEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/CoordinatedEvent.java
index b770ff3..2a80153 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/CoordinatedEvent.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/CoordinatedEvent.java
@@ -54,4 +54,15 @@ public class CoordinatedEvent {
public TransactionStatus getStatus() {
return status;
}
+
+ @Override
+ public String toString() {
+ return "CoordinatedEvent{" +
+ "globalTxId='" + globalTxId + '\'' +
+ ", localTxId='" + localTxId + '\'' +
+ ", parentTxId='" + parentTxId + '\'' +
+ ", methodName='" + methodName + '\'' +
+ ", status=" + status +
+ '}';
+ }
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/ParticipatedEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/ParticipatedEvent.java
index ac2dcb9..f55fd43 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/ParticipatedEvent.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/ParticipatedEvent.java
@@ -62,4 +62,16 @@ public class ParticipatedEvent {
public TransactionStatus getStatus() {
return status;
}
+
+ @Override
+ public String toString() {
+ return "ParticipatedEvent{" +
+ "globalTxId='" + globalTxId + '\'' +
+ ", localTxId='" + localTxId + '\'' +
+ ", parentTxId='" + parentTxId + '\'' +
+ ", confirmMethod='" + confirmMethod + '\'' +
+ ", cancelMethod='" + cancelMethod + '\'' +
+ ", status=" + status +
+ '}';
+ }
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccEndedEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccEndedEvent.java
index 31a187f..3944fb1 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccEndedEvent.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccEndedEvent.java
@@ -42,4 +42,13 @@ public class TccEndedEvent {
this.localTxId = localTxId;
this.status = status;
}
+
+ @Override
+ public String toString() {
+ return "TccEndedEvent{" +
+ "globalTxId='" + globalTxId + '\'' +
+ ", localTxId='" + localTxId + '\'' +
+ ", status=" + status +
+ '}';
+ }
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccStartedEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccStartedEvent.java
index 8449cb4..7a1e7b3 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccStartedEvent.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccStartedEvent.java
@@ -31,4 +31,12 @@ public class TccStartedEvent {
this.globalTxId = globalTxId;
this.localTxId = localTxId;
}
+
+ @Override
+ public String toString() {
+ return "TccStartedEvent{" +
+ "globalTxId='" + globalTxId + '\'' +
+ ", localTxId='" + localTxId + '\'' +
+ '}';
+ }
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandlerTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandlerTest.java
index b260295..fb71505 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandlerTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandlerTest.java
@@ -26,6 +26,8 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.servicecomb.saga.common.TransactionStatus;
+import org.apache.servicecomb.saga.omega.context.CallbackContext;
+import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
@@ -33,6 +35,8 @@ import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccStartedEvent;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
public class CoordinateMessageHandlerTest {
private final List<CoordinatedEvent> coordinatedEvents = new ArrayList<>();
@@ -85,7 +89,9 @@ public class CoordinateMessageHandlerTest {
private final String parentTxId = uniquify("parentTxId");
private final String methodName= uniquify("Method");
- private final CoordinateMessageHandler handler = new CoordinateMessageHandler(eventService);
+ private final CallbackContext callbackContext = Mockito.mock(CallbackContext.class);
+ private final OmegaContext omegaContext = Mockito.mock(OmegaContext.class);
+ private final CoordinateMessageHandler handler = new CoordinateMessageHandler(eventService, callbackContext, omegaContext);
@Before
public void setUp() {