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/09/20 15:21:26 UTC
[incubator-servicecomb-saga] 02/02: SCB-923 Refactored Memory and
RDB TccEventRepository
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch SCB-923
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 6e76608d5451a9f262b2d73158c281a9477f425c
Author: Willem Jiang <ji...@huawei.com>
AuthorDate: Thu Sep 20 23:11:44 2018 +0800
SCB-923 Refactored Memory and RDB TccEventRepository
---
.../servicecomb/saga/alpha/server/AlphaConfig.java | 14 +-
.../alpha/server/tcc/AlphaTccEventController.java | 81 ++++++++++
.../server/tcc/DefaultTccTxEventFacadeImpl.java | 104 -------------
.../saga/alpha/server/tcc/GrpcTccEventService.java | 17 ++-
.../alpha/server/tcc/RdbTccTxEventFacadeImpl.java | 86 -----------
.../saga/alpha/server/tcc/TccTxEventFacade.java | 11 +-
.../server/tcc/callback/MemoryCallbackEngine.java | 32 ----
.../server/tcc/callback/RdbCallbackEngine.java | 37 -----
.../server/tcc/callback/TccCallbackEngine.java | 25 ++-
.../saga/alpha/server/tcc/jpa/GlobalTxEvent.java | 19 ++-
.../alpha/server/tcc/jpa/GlobalTxEventHistory.java | 93 -----------
.../tcc/jpa/GlobalTxEventHistoryRepository.java | 24 ---
.../alpha/server/tcc/jpa/ParticipatedEvent.java | 38 +++--
.../server/tcc/jpa/ParticipatedEventHistory.java | 61 --------
.../jpa/ParticipatedEventHistoryRepository.java | 23 ---
.../saga/alpha/server/tcc/jpa/TccTxEvent.java | 170 +++++++++++++++++++++
.../server/tcc/jpa/TccTxEventDBRepository.java | 37 +++++
.../saga/alpha/server/tcc/jpa/TccTxType.java | 6 +-
.../saga/alpha/server/tcc/jpa/TxEventFactory.java | 119 +++++++++++++--
.../server/tcc/service/GlobalTxEventService.java | 31 ----
.../tcc/service/GlobalTxEventServiceImpl.java | 82 ----------
.../server/tcc/service/MemoryEventRegistry.java | 79 ----------
.../tcc/service/MemoryTxEventRepository.java | 110 +++++++++++++
.../tcc/service/ParticipatedEventService.java | 32 ----
.../tcc/service/ParticipatedEventServiceImpl.java | 83 ----------
.../server/tcc/service/RDBTxEventRepository.java | 93 +++++++++++
.../server/tcc/service/TccTxEventRepository.java | 46 ++++++
.../server/tcc/service/TccTxEventService.java | 113 ++++++++++++++
.../src/main/resources/schema-mysql.sql | 29 +---
.../src/main/resources/schema-postgresql.sql | 29 +---
.../alpha/server/AlphaEventControllerTest.java | 4 +-
.../AlphaTccEventControllerTest.java} | 44 +++---
.../tcc}/AlphaTccServerTestBase.java | 47 +++---
.../tcc}/MemoryAlphaTccServerTest.java | 25 ++-
.../tcc}/RdbAlphaTccServerTest.java | 27 ++--
.../tcc}/TccCoordinateCommandStreamObserver.java | 14 +-
.../alpha/server/tcc/jpa/TxEventFactoryTest.java | 37 +++++
.../tcc/service/MemoryEventRegistryTest.java | 115 ++++++++++++++
alpha/alpha-server/src/test/resources/schema.sql | 25 +--
39 files changed, 1097 insertions(+), 965 deletions(-)
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
index 4d72467..9b3f833 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
@@ -38,6 +38,8 @@ import org.apache.servicecomb.saga.alpha.server.tcc.GrpcTccEventService;
import org.apache.servicecomb.saga.alpha.server.tcc.callback.OmegaCallbackWrapper;
import org.apache.servicecomb.saga.alpha.server.tcc.callback.TccCallbackEngine;
import org.apache.servicecomb.saga.alpha.server.tcc.TccTxEventFacade;
+import org.apache.servicecomb.saga.alpha.server.tcc.service.TccTxEventRepository;
+import org.apache.servicecomb.saga.alpha.server.tcc.service.TccTxEventService;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.domain.EntityScan;
@@ -102,16 +104,8 @@ class AlphaConfig {
}
@Bean
- TccTxEventFacade tccTxEventFacade(
- @Value("${alpha.server.storage:rdb}") String storage,
- @Qualifier("defaultTccTxEventFacade") TccTxEventFacade defaultTccTxEventFacade,
- @Qualifier("rdbTccTxEventFacade") TccTxEventFacade rdbTccTxEventFacade) {
- return "rdb".equals(storage) ? rdbTccTxEventFacade : defaultTccTxEventFacade;
- }
-
- @Bean
- GrpcTccEventService grpcTccEventService(TccTxEventFacade tccTxEventFacade) {
- return new GrpcTccEventService(tccTxEventFacade);
+ GrpcTccEventService grpcTccEventService(TccTxEventService tccTxEventService) {
+ return new GrpcTccEventService(tccTxEventService);
}
@Bean
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/AlphaTccEventController.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/AlphaTccEventController.java
new file mode 100644
index 0000000..8c62f45
--- /dev/null
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/AlphaTccEventController.java
@@ -0,0 +1,81 @@
+/*
+ * 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.alpha.server.tcc;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEventDBRepository;
+import org.apache.servicecomb.saga.alpha.server.tcc.service.TccTxEventRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
+
+import kamon.annotation.EnableKamon;
+
+@EnableKamon
+@Controller
+@RequestMapping("/tcc")
+@Profile("test")
+public class AlphaTccEventController {
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ //As we doesn't enable migration for test, we don't need to access the history table here
+ private final TccTxEventRepository tccTxEventRepository;
+
+ public AlphaTccEventController(TccTxEventRepository tccTxEventRepository) {
+ this.tccTxEventRepository = tccTxEventRepository;
+ }
+
+ @GetMapping(value = "/events")
+ ResponseEntity<Collection<TccTxEventVo>> events() {
+ LOG.info("Get the events request");
+ Iterable<TccTxEvent> events = tccTxEventRepository.findAll();
+
+ List<TccTxEventVo> eventVos = new LinkedList<>();
+ events.forEach(event -> eventVos.add(new TccTxEventVo(event)));
+ LOG.info("Get the event size " + eventVos.size());
+
+ return ResponseEntity.ok(eventVos);
+ }
+
+ @DeleteMapping("/events")
+ ResponseEntity<String> clear() {
+ tccTxEventRepository.deleteAll();
+ return ResponseEntity.ok("All events deleted");
+ }
+
+ @JsonAutoDetect(fieldVisibility = Visibility.ANY)
+ private static class TccTxEventVo extends TccTxEvent {
+ private TccTxEventVo(TccTxEvent event) {
+ super(event);
+ }
+ }
+
+}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/DefaultTccTxEventFacadeImpl.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/DefaultTccTxEventFacadeImpl.java
deleted file mode 100644
index e554fc7..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/DefaultTccTxEventFacadeImpl.java
+++ /dev/null
@@ -1,104 +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.saga.alpha.server.tcc;
-
-import java.lang.invoke.MethodHandles;
-import java.util.List;
-import org.apache.servicecomb.saga.alpha.server.tcc.callback.TccCallbackEngine;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.service.MemoryEventRegistry;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.stereotype.Component;
-
-/**
- * Manage TCC transaction event.
- */
-@Component("defaultTccTxEventFacade")
-public final class DefaultTccTxEventFacadeImpl implements TccTxEventFacade {
-
- private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- @Autowired
- @Qualifier("memoryCallbackEngine")
- private TccCallbackEngine tccCallbackEngine;
-
- @Override
- public boolean onTccStartEvent(GlobalTxEvent globalTxEvent) {
- MemoryEventRegistry.addGlobalTxEvent(globalTxEvent);
- LOG.info("Registered participated event, global tx: {}, local tx: {}, parent id: {}, "
- + "txType: {}, service [{}] instanceId [{}]",
- globalTxEvent.getGlobalTxId(), globalTxEvent.getLocalTxId(), globalTxEvent.getParentTxId(),
- globalTxEvent.getTxType(), globalTxEvent.getServiceName(), globalTxEvent.getInstanceId());
- return true;
- }
-
- /**
- * Register participate event.
- *
- * @param participatedEvent participated event
- */
-
- @Override
- public boolean onParticipateEvent(ParticipatedEvent participatedEvent) {
- MemoryEventRegistry.addParticipateEvent(participatedEvent);
- LOG.info("Registered participated event, global tx: {}, local tx: {}, parent id: {}, "
- + "confirm: {}, cancel: {}, status: {}, service [{}] instanceId [{}]",
- participatedEvent.getGlobalTxId(), participatedEvent.getLocalTxId(), participatedEvent.getParentTxId(),
- participatedEvent.getConfirmMethod(), participatedEvent.getCancelMethod(), participatedEvent.getStatus(),
- participatedEvent.getServiceName(), participatedEvent.getInstanceId());
-
- // TODO We need to updated the event which transactionStatus is failed.
- return true;
- }
-
- @Override
- public boolean onTccEndEvent(GlobalTxEvent globalTxEvent) {
- MemoryEventRegistry.addGlobalTxEvent(globalTxEvent);
- return tccCallbackEngine.execute(globalTxEvent);
- }
-
- @Override
- public void onCoordinatedEvent(String globalTxId, String localTxId) {
- MemoryEventRegistry.migrateParticipate(globalTxId, localTxId);
- }
-
- @Override
- public List<GlobalTxEvent> getGlobalTxEventByGlobalTxId(String globalTxId) {
- return MemoryEventRegistry.getGlobalTxEventByGlobalTxId(globalTxId);
- }
-
- /**
- * Retrieve participate event from registry.
- *
- * @param globalTxId global transaction id
- * @return participate events
- */
- @Override
- public List<ParticipatedEvent> getParticipateEventByGlobalTxId(String globalTxId) {
- return MemoryEventRegistry.getParticipateEventByGlobalTxId(globalTxId);
- }
-
- @Override
- public void migrationGlobalTxEvent(String globalTxId, String localTxId) {
- MemoryEventRegistry.migrationGlobalTxEvent(globalTxId, localTxId);
- }
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/GrpcTccEventService.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/GrpcTccEventService.java
index 9fe2f2c..2a7b745 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/GrpcTccEventService.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/GrpcTccEventService.java
@@ -22,6 +22,7 @@ import java.lang.invoke.MethodHandles;
import org.apache.servicecomb.saga.alpha.server.tcc.callback.OmegaCallback;
import org.apache.servicecomb.saga.alpha.server.tcc.callback.OmegaCallbacksRegistry;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TxEventFactory;
+import org.apache.servicecomb.saga.alpha.server.tcc.service.TccTxEventService;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcAck;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcServiceConfig;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccCoordinateCommand;
@@ -44,10 +45,10 @@ public class GrpcTccEventService extends TccEventServiceGrpc.TccEventServiceImpl
private static final GrpcAck REJECT = GrpcAck.newBuilder().setAborted(true).build();
- private final TccTxEventFacade tccTxEventFacade;
+ private final TccTxEventService tccTxEventService;
- public GrpcTccEventService(TccTxEventFacade tccTxEventFacade) {
- this.tccTxEventFacade = tccTxEventFacade;
+ public GrpcTccEventService(TccTxEventService tccTxEventService) {
+ this.tccTxEventService = tccTxEventService;
}
@Override
@@ -59,7 +60,7 @@ public class GrpcTccEventService extends TccEventServiceGrpc.TccEventServiceImpl
@Override
public void onTccTransactionStarted(GrpcTccTransactionStartedEvent request, StreamObserver<GrpcAck> responseObserver) {
LOG.info("Received transaction start event, global tx id: {}", request.getGlobalTxId());
- responseObserver.onNext(tccTxEventFacade.onTccStartEvent(TxEventFactory.create(request)) ? ALLOW : REJECT);
+ responseObserver.onNext(tccTxEventService.onTccStartedEvent(TxEventFactory.convertToGlobalTxEvent(request)) ? ALLOW : REJECT);
responseObserver.onCompleted();
}
@@ -67,14 +68,15 @@ public class GrpcTccEventService extends TccEventServiceGrpc.TccEventServiceImpl
public void participate(GrpcTccParticipatedEvent request, StreamObserver<GrpcAck> responseObserver) {
LOG.info("Received participated event from service {} , global tx id: {}, local tx id: {}", request.getServiceName(),
request.getGlobalTxId(), request.getLocalTxId()) ;
- responseObserver.onNext(tccTxEventFacade.onParticipateEvent(TxEventFactory.create(request)) ? ALLOW : REJECT);
+ responseObserver.onNext(
+ tccTxEventService.onParticipatedEvent(TxEventFactory.convertToParticipatedEvent(request)) ? ALLOW : REJECT);
responseObserver.onCompleted();
}
@Override
public void onTccTransactionEnded(GrpcTccTransactionEndedEvent request, StreamObserver<GrpcAck> responseObserver) {
LOG.info("Received transaction end event, global tx id: {}", request.getGlobalTxId());
- responseObserver.onNext(tccTxEventFacade.onTccEndEvent(TxEventFactory.create(request)) ? ALLOW : REJECT);
+ responseObserver.onNext(tccTxEventService.onTccEndedEvent(TxEventFactory.convertToGlobalTxEvent(request)) ? ALLOW : REJECT);
responseObserver.onCompleted();
}
@@ -84,8 +86,7 @@ public class GrpcTccEventService extends TccEventServiceGrpc.TccEventServiceImpl
+ "method: {}, status: {}, service [{}] instanceId [{}]",
request.getGlobalTxId(), request.getLocalTxId(), request.getParentTxId(),
request.getMethodName(), request.getStatus(), request.getServiceName(), request.getInstanceId());
- tccTxEventFacade.onCoordinatedEvent(request.getGlobalTxId(), request.getLocalTxId());
- responseObserver.onNext(ALLOW);
+ responseObserver.onNext(tccTxEventService.onCoordinatedEvent(TxEventFactory.convertToTccTxEvent(request)) ? ALLOW : REJECT);
responseObserver.onCompleted();
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbTccTxEventFacadeImpl.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbTccTxEventFacadeImpl.java
deleted file mode 100644
index 5df17da..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbTccTxEventFacadeImpl.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.saga.alpha.server.tcc;
-
-import com.google.common.collect.Lists;
-import java.lang.invoke.MethodHandles;
-import java.util.List;
-import org.apache.servicecomb.saga.alpha.server.tcc.callback.TccCallbackEngine;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.service.GlobalTxEventService;
-import org.apache.servicecomb.saga.alpha.server.tcc.service.ParticipatedEventService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.stereotype.Component;
-
-@Component("rdbTccTxEventFacade")
-public class RdbTccTxEventFacadeImpl implements TccTxEventFacade {
-
- @Autowired
- private ParticipatedEventService participatedEventService;
-
- @Autowired
- private GlobalTxEventService globalTxEventService;
-
- @Autowired
- @Qualifier("rdbCallbackEngine")
- private TccCallbackEngine tccCallbackEngine;
-
- private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- @Override
- public boolean onTccStartEvent(GlobalTxEvent globalTxEvent) {
- return globalTxEventService.addEvent(globalTxEvent);
- }
-
- @Override
- public boolean onParticipateEvent(ParticipatedEvent event) {
- return participatedEventService.addEvent(event);
- }
-
- @Override
- public boolean onTccEndEvent(GlobalTxEvent globalTxEvent) {
- if (globalTxEventService.addEvent(globalTxEvent)) {
- return tccCallbackEngine.execute(globalTxEvent);
- }
- return false;
- }
-
- @Override
- public void onCoordinatedEvent(String globalTxId, String localTxId) {
- participatedEventService.migration(globalTxId, localTxId);
- }
-
- @Override
- public List<GlobalTxEvent> getGlobalTxEventByGlobalTxId(String globalTxId) {
- return globalTxEventService.getEventByGlobalTxId(globalTxId).orElse(Lists.newArrayList());
- }
-
- @Override
- public List<ParticipatedEvent> getParticipateEventByGlobalTxId(String globalTxId) {
- return participatedEventService.getEventByGlobalTxId(globalTxId).orElse(Lists.newArrayList());
- }
-
- @Override
- public void migrationGlobalTxEvent(String globalTxId, String localTxId) {
- globalTxEventService.migration(globalTxId, localTxId);
- }
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TccTxEventFacade.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TccTxEventFacade.java
index 64209fe..2eb1409 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TccTxEventFacade.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TccTxEventFacade.java
@@ -17,9 +17,9 @@
package org.apache.servicecomb.saga.alpha.server.tcc;
-import java.util.List;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
public interface TccTxEventFacade {
@@ -29,13 +29,6 @@ public interface TccTxEventFacade {
boolean onTccEndEvent(GlobalTxEvent globalTxEvent);
- void onCoordinatedEvent(String globalTxId, String localTxId);
+ void onCoordinatedEvent(TccTxEvent event);
- List<GlobalTxEvent> getGlobalTxEventByGlobalTxId(String globalTxId);
-
- List<ParticipatedEvent> getParticipateEventByGlobalTxId(String globalTxId);
-
-// void migrationParticipateEvent(String globalTxId, String localTxId);
-//
- void migrationGlobalTxEvent(String globalTxId, String localTxId);
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/MemoryCallbackEngine.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/MemoryCallbackEngine.java
deleted file mode 100644
index 5a570c6..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/MemoryCallbackEngine.java
+++ /dev/null
@@ -1,32 +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.saga.alpha.server.tcc.callback;
-
-import java.util.List;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.service.MemoryEventRegistry;
-import org.springframework.stereotype.Component;
-
-@Component("memoryCallbackEngine")
-public class MemoryCallbackEngine extends TccCallbackEngine {
-
- @Override
- protected List<ParticipatedEvent> findParticipate(String globalTxId) {
- return MemoryEventRegistry.getParticipateEventByGlobalTxId(globalTxId);
- }
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/RdbCallbackEngine.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/RdbCallbackEngine.java
deleted file mode 100644
index c84340e..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/RdbCallbackEngine.java
+++ /dev/null
@@ -1,37 +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.saga.alpha.server.tcc.callback;
-
-import com.google.common.collect.Lists;
-import java.util.List;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.service.ParticipatedEventService;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-@Component("rdbCallbackEngine")
-public class RdbCallbackEngine extends TccCallbackEngine {
-
- @Autowired
- private ParticipatedEventService participatedEventService;
-
- @Override
- protected List<ParticipatedEvent> findParticipate(String globalTxId) {
- return participatedEventService.getEventByGlobalTxId(globalTxId).orElse(Lists.newArrayList());
- }
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/TccCallbackEngine.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/TccCallbackEngine.java
index 9cb29dd..f981c8d 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/TccCallbackEngine.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/TccCallbackEngine.java
@@ -19,40 +19,51 @@ package org.apache.servicecomb.saga.alpha.server.tcc.callback;
import java.lang.invoke.MethodHandles;
import java.util.List;
+
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxType;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TxEventFactory;
+import org.apache.servicecomb.saga.alpha.server.tcc.service.TccTxEventRepository;
import org.apache.servicecomb.saga.common.TransactionStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
-public abstract class TccCallbackEngine implements CallbackEngine {
+import com.google.common.collect.Lists;
+@Component
+public class TccCallbackEngine implements CallbackEngine {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@Autowired
private OmegaCallbackWrapper omegaCallbackWrapper;
+ @Autowired
+ private TccTxEventRepository tccTxEventRepository;
+
@Override
public boolean execute(GlobalTxEvent request) {
boolean result = true;
- List<ParticipatedEvent> list = findParticipate(request.getGlobalTxId());
- for (ParticipatedEvent event : list) {
+ List<TccTxEvent> events = tccTxEventRepository.findByGlobalTxIdAndTxType(request.getGlobalTxId(), TccTxType.PARTICIPATED).orElse(
+ Lists.newArrayList());
+ for (TccTxEvent event : events) {
+ ParticipatedEvent participatedEvent = TxEventFactory.convertToParticipatedEvent(event);
try {
// only invoke the event is succeed
if (event.getStatus().equals(TransactionStatus.Succeed.toString())) {
- omegaCallbackWrapper.invoke(event, TransactionStatus.valueOf(request.getStatus()));
+ omegaCallbackWrapper.invoke(participatedEvent, TransactionStatus.valueOf(request.getStatus()));
}
} catch (Exception ex) {
- logError(event, ex);
+ logError(participatedEvent, ex);
result = false;
}
}
return result;
}
- protected abstract List<ParticipatedEvent> findParticipate(String globalTxId);
-
private void logError(ParticipatedEvent event, Exception ex) {
LOG.error(
"Failed to invoke service [{}] instance [{}] with method [{}], global tx id [{}] and local tx id [{}]",
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEvent.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEvent.java
index cc6e155..451e88e 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEvent.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEvent.java
@@ -45,8 +45,7 @@ public class GlobalTxEvent {
private GlobalTxEvent() {
}
- public GlobalTxEvent(String globalTxId, String localTxId, String parentTxId, String serviceName,
- String instanceId, String txType, String status) {
+ public GlobalTxEvent(String serviceName, String instanceId, String globalTxId, String localTxId, String parentTxId, String txType, String status) {
this.globalTxId = globalTxId;
this.localTxId = localTxId;
this.parentTxId = parentTxId;
@@ -121,4 +120,20 @@ public class GlobalTxEvent {
return Objects.hash(globalTxId, localTxId, parentTxId, serviceName, instanceId, txType, status);
}
+
+ @Override
+ public String toString() {
+ return "GlobalTxEvent{" +
+ "surrogateId=" + surrogateId +
+ ", globalTxId='" + globalTxId + '\'' +
+ ", localTxId='" + localTxId + '\'' +
+ ", parentTxId='" + parentTxId + '\'' +
+ ", serviceName='" + serviceName + '\'' +
+ ", instanceId='" + instanceId + '\'' +
+ ", txType='" + txType + '\'' +
+ ", status='" + status + '\'' +
+ ", creationTime=" + creationTime +
+ ", lastModified=" + lastModified +
+ '}';
+ }
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistory.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistory.java
deleted file mode 100644
index f0179ff..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistory.java
+++ /dev/null
@@ -1,93 +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.saga.alpha.server.tcc.jpa;
-
-import java.util.Date;
-import javax.persistence.Entity;
-import javax.persistence.GeneratedValue;
-import javax.persistence.GenerationType;
-import javax.persistence.Id;
-import javax.persistence.Table;
-
-@Entity
-@Table(name = "tcc_global_tx_event_history")
-public class GlobalTxEventHistory {
-
- @Id
- @GeneratedValue(strategy = GenerationType.IDENTITY)
- private Long id;
- private String globalTxId;
- private String localTxId;
- private String parentTxId;
- private String serviceName;
- private String instanceId;
- private String txType;
- private Date creationTime;
- private Date lastModified;
-
- private GlobalTxEventHistory() {
- }
-
- public GlobalTxEventHistory(String globalTxId, String localTxId, String parentTxId, String serviceName,
- String instanceId, String txType, Date creationTime, Date lastModified) {
- this.globalTxId = globalTxId;
- this.localTxId = localTxId;
- this.parentTxId = parentTxId;
- this.serviceName = serviceName;
- this.instanceId = instanceId;
- this.txType = txType;
- this.creationTime = creationTime;
- this.lastModified = lastModified;
- }
-
- public Long getId() {
- return id;
- }
-
- public String getGlobalTxId() {
- return globalTxId;
- }
-
- public String getLocalTxId() {
- return localTxId;
- }
-
- public String getParentTxId() {
- return parentTxId;
- }
-
- public String getServiceName() {
- return serviceName;
- }
-
- public String getInstanceId() {
- return instanceId;
- }
-
- public String getTxType() {
- return txType;
- }
-
- public Date getCreationTime() {
- return creationTime;
- }
-
- public Date getLastModified() {
- return lastModified;
- }
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistoryRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistoryRepository.java
deleted file mode 100644
index a0affa3..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistoryRepository.java
+++ /dev/null
@@ -1,24 +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.saga.alpha.server.tcc.jpa;
-
-import org.springframework.data.repository.CrudRepository;
-
-public interface GlobalTxEventHistoryRepository extends CrudRepository<GlobalTxEventHistory, Long> {
-
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEvent.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEvent.java
index b9fc49f..b0868c5 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEvent.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEvent.java
@@ -46,8 +46,8 @@ public class ParticipatedEvent {
private ParticipatedEvent() {
}
- public ParticipatedEvent(String globalTxId, String localTxId, String parentTxId, String serviceName,
- String instanceId, String confirmMethod, String cancelMethod, String status) {
+ public ParticipatedEvent(String serviceName, String instanceId, String globalTxId, String localTxId,
+ String parentTxId, String confirmMethod, String cancelMethod, String status, Date creationTime, Date lastModified) {
this.globalTxId = globalTxId;
this.localTxId = localTxId;
this.parentTxId = parentTxId;
@@ -56,8 +56,13 @@ public class ParticipatedEvent {
this.confirmMethod = confirmMethod;
this.cancelMethod = cancelMethod;
this.status = status;
- this.creationTime = new Date();
- this.lastModified = new Date();
+ this.creationTime = creationTime;
+ this.lastModified = lastModified;
+ }
+
+ public ParticipatedEvent(String serviceName, String instanceId, String globalTxId, String localTxId,
+ String parentTxId, String confirmMethod, String cancelMethod, String status) {
+ this(serviceName, instanceId, globalTxId, localTxId, parentTxId, confirmMethod, cancelMethod, status, new Date(), new Date());
}
public Long getId() {
@@ -100,18 +105,10 @@ public class ParticipatedEvent {
return creationTime;
}
- public void setCreationTime(Date creationTime) {
- this.creationTime = creationTime;
- }
-
public Date getLastModified() {
return lastModified;
}
- public void setLastModified(Date lastModified) {
- this.lastModified = lastModified;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -136,4 +133,21 @@ public class ParticipatedEvent {
return Objects
.hash(globalTxId, localTxId, parentTxId, serviceName, instanceId, confirmMethod, cancelMethod, status);
}
+
+ @Override
+ public String toString() {
+ return "ParticipatedEvent{" +
+ "surrogateId=" + surrogateId +
+ ", globalTxId='" + globalTxId + '\'' +
+ ", localTxId='" + localTxId + '\'' +
+ ", parentTxId='" + parentTxId + '\'' +
+ ", serviceName='" + serviceName + '\'' +
+ ", instanceId='" + instanceId + '\'' +
+ ", confirmMethod='" + confirmMethod + '\'' +
+ ", cancelMethod='" + cancelMethod + '\'' +
+ ", status='" + status + '\'' +
+ ", creationTime=" + creationTime +
+ ", lastModified=" + lastModified +
+ '}';
+ }
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistory.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistory.java
deleted file mode 100644
index f7f560c..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistory.java
+++ /dev/null
@@ -1,61 +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.saga.alpha.server.tcc.jpa;
-
-import java.util.Date;
-import javax.persistence.Entity;
-import javax.persistence.GeneratedValue;
-import javax.persistence.GenerationType;
-import javax.persistence.Id;
-import javax.persistence.Table;
-
-@Entity
-@Table(name = "tcc_participate_event_history")
-public class ParticipatedEventHistory {
-
- @Id
- @GeneratedValue(strategy = GenerationType.IDENTITY)
- private Long surrogateId;
- private String globalTxId;
- private String localTxId;
- private String parentTxId;
- private String serviceName;
- private String instanceId;
- private String confirmMethod;
- private String cancelMethod;
- private String status;
- private Date creationTime;
- private Date lastModified;
-
- private ParticipatedEventHistory() {
- }
-
- public ParticipatedEventHistory(String globalTxId, String localTxId, String parentTxId, String serviceName,
- String instanceId, String confirmMethod, String cancelMethod, String status) {
- this.globalTxId = globalTxId;
- this.localTxId = localTxId;
- this.parentTxId = parentTxId;
- this.serviceName = serviceName;
- this.instanceId = instanceId;
- this.confirmMethod = confirmMethod;
- this.cancelMethod = cancelMethod;
- this.status = status;
- this.creationTime = new Date();
- this.lastModified = new Date();
- }
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistoryRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistoryRepository.java
deleted file mode 100644
index 01bce29..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistoryRepository.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.saga.alpha.server.tcc.jpa;
-
-import org.springframework.data.repository.CrudRepository;
-
-public interface ParticipatedEventHistoryRepository extends CrudRepository<ParticipatedEventHistory, Long> {
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxEvent.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxEvent.java
new file mode 100644
index 0000000..e1d9654
--- /dev/null
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxEvent.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.alpha.server.tcc.jpa;
+
+
+import java.util.Date;
+import java.util.Objects;
+
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.Table;
+
+@Entity
+@Table(name = "tcc_tx_event")
+public class TccTxEvent {
+ @Id
+ @GeneratedValue(strategy = GenerationType.IDENTITY)
+ private Long surrogateId;
+ private String globalTxId;
+ private String localTxId;
+ private String parentTxId;
+ private String serviceName;
+ private String instanceId;
+ private String methodInfo;
+ private String txType;
+ private String status;
+ private Date creationTime;
+ private Date lastModified;
+
+ private TccTxEvent(){
+ // this construction is for JPA
+ }
+
+ public TccTxEvent(String serviceName, String instanceId, String globalTxId, String localTxId, String parentTxId, String txType, String status) {
+ this(serviceName, instanceId, globalTxId, localTxId, parentTxId, txType, "", status);
+ }
+
+ public TccTxEvent(String serviceName, String instanceId, String globalTxId, String localTxId, String parentTxId, String txType, String methodInfo, String status) {
+ this.globalTxId = globalTxId;
+ this.localTxId = localTxId;
+ this.parentTxId = parentTxId;
+ this.serviceName = serviceName;
+ this.instanceId = instanceId;
+ this.txType = txType;
+ this.status = status;
+ this.methodInfo = methodInfo;
+ creationTime = new Date();
+ lastModified = new Date();
+ }
+
+ public TccTxEvent(TccTxEvent event) {
+ this.globalTxId = event.globalTxId;
+ this.localTxId = event.localTxId;
+ this.parentTxId = event.parentTxId;
+ this.creationTime = event.creationTime;
+ this.serviceName = event.serviceName;
+ this.instanceId = event.instanceId;
+ this.lastModified = event.lastModified;
+ this.methodInfo = event.methodInfo;
+ this.status = event.status;
+ this.txType = event.txType;
+ }
+
+ public Long getId() {
+ return surrogateId;
+ }
+
+ public String getLocalTxId() {
+ return localTxId;
+ }
+
+ public String getGlobalTxId() {
+ return globalTxId;
+ }
+
+ public String getParentTxId() {
+ return parentTxId;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public String getInstanceId() {
+ return instanceId;
+ }
+
+ public String getTxType() {
+ return txType;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public String getMethodInfo() {
+ return methodInfo;
+ }
+
+ public Date getCreationTime() {
+ return creationTime;
+ }
+
+ public Date getLastModified() {
+ return lastModified;
+ }
+
+ @Override
+ public String toString() {
+ return "TccTxEvent{" +
+ "surrogateId=" + surrogateId +
+ ", globalTxId='" + globalTxId + '\'' +
+ ", localTxId='" + localTxId + '\'' +
+ ", parentTxId='" + parentTxId + '\'' +
+ ", serviceName='" + serviceName + '\'' +
+ ", instanceId='" + instanceId + '\'' +
+ ", methodInfo='" + methodInfo + '\'' +
+ ", txType='" + txType + '\'' +
+ ", status='" + status + '\'' +
+ ", creationTime=" + creationTime +
+ ", lastModified=" + lastModified +
+ '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TccTxEvent that = (TccTxEvent) o;
+ return
+ Objects.equals(globalTxId, that.globalTxId) &&
+ Objects.equals(localTxId, that.localTxId) &&
+ Objects.equals(parentTxId, that.parentTxId) &&
+ Objects.equals(serviceName, that.serviceName) &&
+ Objects.equals(instanceId, that.instanceId) &&
+ Objects.equals(methodInfo, that.methodInfo) &&
+ Objects.equals(txType, that.txType) &&
+ Objects.equals(status, that.status) &&
+ Objects.equals(creationTime, that.creationTime) &&
+ Objects.equals(lastModified, that.lastModified);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects
+ .hash(surrogateId, globalTxId, localTxId, parentTxId, serviceName, instanceId, methodInfo, txType, status,
+ creationTime, lastModified);
+ }
+}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxEventDBRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxEventDBRepository.java
new file mode 100644
index 0000000..172a554
--- /dev/null
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxEventDBRepository.java
@@ -0,0 +1,37 @@
+/*
+ * 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.alpha.server.tcc.jpa;
+
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.servicecomb.saga.alpha.server.tcc.service.TccTxEventRepository;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.CrudRepository;
+
+public interface TccTxEventDBRepository extends CrudRepository<TccTxEvent, Long> {
+
+ @Query(value = "SELECT t FROM TccTxEvent AS t WHERE t.globalTxId = ?1")
+ Optional<List<TccTxEvent>> findByGlobalTxId(String globalTxId);
+
+ @Query(value = "SELECT t FROM TccTxEvent AS t WHERE t.globalTxId = ?1 and t.localTxId = ?2 and t.txType = ?3")
+ Optional<TccTxEvent> findByUniqueKey(String globalTxId, String localTxId, String txType);
+
+ @Query(value = "SELECT t FROM TccTxEvent AS t WHERE t.globalTxId = ?1 and t.txType = ?2")
+ Optional<List<TccTxEvent>> findByGlobalTxIdAndTxType(String globalTxId, String txType);
+}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxType.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxType.java
index 55369eb..b1527d3 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxType.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxType.java
@@ -18,5 +18,9 @@
package org.apache.servicecomb.saga.alpha.server.tcc.jpa;
public enum TccTxType {
- TCC_START, TCC_END, TCC_END_TIMEOUT
+ STARTED,
+ ENDED,
+ END_TIMEOUT,
+ PARTICIPATED,
+ COORDINATED
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactory.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactory.java
index eec4587..19d6e9a 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactory.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactory.java
@@ -18,46 +18,143 @@
package org.apache.servicecomb.saga.alpha.server.tcc.jpa;
import org.apache.servicecomb.saga.common.TransactionStatus;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccCoordinatedEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccParticipatedEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccTransactionEndedEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccTransactionStartedEvent;
public class TxEventFactory {
- public static ParticipatedEvent create(GrpcTccParticipatedEvent request) {
+ public static ParticipatedEvent convertToParticipatedEvent(GrpcTccParticipatedEvent request) {
return new ParticipatedEvent(
+ request.getServiceName(),
+ request.getInstanceId(),
request.getGlobalTxId(),
request.getLocalTxId(),
request.getParentTxId(),
- request.getServiceName(),
- request.getInstanceId(),
request.getConfirmMethod(),
request.getCancelMethod(),
request.getStatus()
);
}
- public static GlobalTxEvent create(GrpcTccTransactionStartedEvent request) {
+ public static GlobalTxEvent convertToGlobalTxEvent(GrpcTccTransactionStartedEvent request) {
return new GlobalTxEvent(
+ request.getServiceName(),
+ request.getInstanceId(),
request.getGlobalTxId(),
request.getLocalTxId(),
request.getParentTxId(),
- request.getServiceName(),
- request.getInstanceId(),
- TccTxType.TCC_START.name(),
+ TccTxType.STARTED.name(),
TransactionStatus.Succeed.name()
);
}
- public static GlobalTxEvent create(GrpcTccTransactionEndedEvent request) {
+ public static GlobalTxEvent convertToGlobalTxEvent(GrpcTccTransactionEndedEvent request) {
return new GlobalTxEvent(
+ request.getServiceName(),
+ request.getInstanceId(),
request.getGlobalTxId(),
request.getLocalTxId(),
request.getParentTxId(),
- request.getServiceName(),
- request.getInstanceId(),
- TccTxType.TCC_END.name(),
+ TccTxType.ENDED.name(),
request.getStatus()
);
}
+
+ public static TccTxEvent convertToTccTxEvent(GrpcTccTransactionStartedEvent event) {
+ return new TccTxEvent(
+ event.getServiceName(),
+ event.getInstanceId(),
+ event.getGlobalTxId(),
+ event.getLocalTxId(),
+ event.getParentTxId(),
+ TccTxType.STARTED.name(),
+ TransactionStatus.Succeed.name());
+ }
+
+ public static TccTxEvent convertToTccTxEvent(GrpcTccTransactionEndedEvent event) {
+ return new TccTxEvent(event.getServiceName(),
+ event.getInstanceId(),
+ event.getGlobalTxId(),
+ event.getLocalTxId(),
+ event.getParentTxId(),
+ TccTxType.STARTED.name(),
+ event.getStatus());
+ }
+
+ public static TccTxEvent convertToTccTxEvent(GrpcTccParticipatedEvent event) {
+ return new TccTxEvent(event.getServiceName(),
+ event.getInstanceId(),
+ event.getGlobalTxId(),
+ event.getLocalTxId(),
+ event.getParentTxId(),
+ toMethodInfo(event.getCancelMethod(), event.getConfirmMethod()),
+ TccTxType.PARTICIPATED.name(),
+ event.getStatus());
+ }
+
+ public static TccTxEvent convertToTccTxEvent(GrpcTccCoordinatedEvent event) {
+ return new TccTxEvent( event.getServiceName(),
+ event.getInstanceId(),
+ event.getGlobalTxId(),
+ event.getLocalTxId(),
+ event.getParentTxId(),
+ event.getMethodName(),
+ TccTxType.COORDINATED.name(),
+ event.getStatus());
+ }
+
+ public static TccTxEvent convertToTccTxEvent(GlobalTxEvent event) {
+ return new TccTxEvent(event.getServiceName(),
+ event.getInstanceId(),
+ event.getGlobalTxId(),
+ event.getLocalTxId(),
+ event.getParentTxId(),
+ event.getTxType(),
+ event.getStatus());
+ }
+
+ public static TccTxEvent convertToTccTxEvent(ParticipatedEvent event) {
+ return new TccTxEvent(event.getServiceName(),
+ event.getInstanceId(),
+ event.getGlobalTxId(),
+ event.getLocalTxId(),
+ event.getParentTxId(),
+ TccTxType.PARTICIPATED.name(),
+ toMethodInfo(event.getConfirmMethod(), event.getCancelMethod()),
+ event.getStatus());
+ }
+
+ public static ParticipatedEvent convertToParticipatedEvent(TccTxEvent event) {
+ return new ParticipatedEvent(
+ event.getServiceName(),
+ event.getInstanceId(),
+ event.getGlobalTxId(),
+ event.getLocalTxId(),
+ event.getParentTxId(),
+ getMethodName(event.getMethodInfo(),true),
+ getMethodName(event.getMethodInfo(), false),
+ event.getStatus(),
+ event.getCreationTime(),
+ event.getLastModified()
+ );
+ }
+
+ public static String toMethodInfo(String confirmMethod, String cancelMethod) {
+ return "confirm=" + confirmMethod + ",cancel=" + cancelMethod;
+ }
+
+ public static String getMethodName(String methodInfo, boolean isConfirm) {
+ String[] methods = methodInfo.split(",");
+ if (methods.length == 2 && methods[0].startsWith("confirm=") && methods[1].startsWith("cancel=")) {
+ String confirmMethod = methods[0].substring(8);
+ String cancelMethod = methods[1].substring(7);
+ if (isConfirm) return confirmMethod;
+ else return cancelMethod;
+ } else {
+ throw new IllegalArgumentException("MethodInfo: " + methodInfo + ", has some bad formats.");
+ }
+
+ }
}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventService.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventService.java
deleted file mode 100644
index 2d4444c..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventService.java
+++ /dev/null
@@ -1,31 +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.saga.alpha.server.tcc.service;
-
-import java.util.List;
-import java.util.Optional;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
-
-public interface GlobalTxEventService {
-
- boolean addEvent(GlobalTxEvent globalTxEvent);
-
- Optional<List<GlobalTxEvent>> getEventByGlobalTxId(String globalTxId);
-
- void migration(String globalTxId, String localTxId);
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventServiceImpl.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventServiceImpl.java
deleted file mode 100644
index 5033ee7..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventServiceImpl.java
+++ /dev/null
@@ -1,82 +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.saga.alpha.server.tcc.service;
-
-import java.lang.invoke.MethodHandles;
-import java.util.List;
-import java.util.Optional;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEventHistory;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEventHistoryRepository;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEventRepository;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-@Component
-public class GlobalTxEventServiceImpl implements GlobalTxEventService {
-
- @Autowired
- private GlobalTxEventRepository hotRepository;
-
- @Autowired
- private GlobalTxEventHistoryRepository coldRepository;
-
- private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- @Override
- public boolean addEvent(GlobalTxEvent event) {
- try {
- if (!hotRepository.findByUniqueKey(event.getGlobalTxId(),
- event.getLocalTxId(), event.getTxType()).isPresent()) {
- hotRepository.save(event);
- }
- } catch (Exception ex) {
- LOG.warn("Add globalTxEvent triggered exception, globalTxId:{}, localTxId:{}, txType:{}, ",
- event.getGlobalTxId(), event.getLocalTxId(), event.getTxType(), ex);
- return false;
- }
- return true;
- }
-
- @Override
- public Optional<List<GlobalTxEvent>> getEventByGlobalTxId(String globalTxId) {
- return hotRepository.findByGlobalTxId(globalTxId);
- }
-
- @Override
- public void migration(String globalTxId, String localTxId) {
- hotRepository.findByGlobalTxId(globalTxId).ifPresent(list ->
- list.forEach((e) -> {
- hotRepository.delete(e);
- GlobalTxEventHistory finishedEvent = new GlobalTxEventHistory(
- e.getGlobalTxId(),
- e.getLocalTxId(),
- e.getParentTxId(),
- e.getServiceName(),
- e.getInstanceId(),
- e.getTxType(),
- e.getCreationTime(),
- e.getLastModified()
- );
- coldRepository.save(finishedEvent);
- })
- );
- }
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryEventRegistry.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryEventRegistry.java
deleted file mode 100644
index 615db50..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryEventRegistry.java
+++ /dev/null
@@ -1,79 +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.saga.alpha.server.tcc.service;
-
-import com.google.common.collect.Lists;
-import java.lang.invoke.MethodHandles;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class MemoryEventRegistry {
-
- private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- private final static Map<String, Set<GlobalTxEvent>> globalTxMap = new ConcurrentHashMap<>();
-
- private final static Map<String, Set<ParticipatedEvent>> participateMap = new ConcurrentHashMap<>();
-
- public static void addGlobalTxEvent(GlobalTxEvent globalTxEvent) {
- globalTxMap
- .computeIfAbsent(globalTxEvent.getGlobalTxId(), key -> new LinkedHashSet<>())
- .add(globalTxEvent);
- }
-
- public static void addParticipateEvent(ParticipatedEvent participatedEvent) {
- participateMap
- .computeIfAbsent(participatedEvent.getGlobalTxId(), key -> new LinkedHashSet<>())
- .add(participatedEvent);
- }
-
- public static void migrateParticipate(String globalTxId, String localTxId) {
- Set<ParticipatedEvent> needRemoveSet = participateMap.get(globalTxId).stream()
- .filter((e) -> globalTxId.equals(e.getGlobalTxId()) && localTxId.equals(e.getLocalTxId()))
- .collect(Collectors.toSet());
- participateMap.get(globalTxId).removeAll(needRemoveSet);
- }
-
- public static List<GlobalTxEvent> getGlobalTxEventByGlobalTxId(String globalTxId) {
- Set<GlobalTxEvent> events = globalTxMap.get(globalTxId);
- return null == events ? Lists.newArrayList() : Lists.newArrayList(events);
- }
-
- public static List<ParticipatedEvent> getParticipateEventByGlobalTxId(String globalTxId) {
- Set<ParticipatedEvent> events = participateMap.get(globalTxId);
- return null == events ? Lists.newArrayList() : Lists.newArrayList(events);
- }
-
- public static void migrationGlobalTxEvent(String globalTxId, String localTxId) {
- Set<GlobalTxEvent> needRemoveSet = globalTxMap.get(globalTxId).stream()
- .filter((e) -> globalTxId.equals(e.getGlobalTxId()) && localTxId.equals(e.getLocalTxId()))
- .collect(Collectors.toSet());
- globalTxMap.get(globalTxId).removeAll(needRemoveSet);
- }
-
-
-
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryTxEventRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryTxEventRepository.java
new file mode 100644
index 0000000..fa50fec
--- /dev/null
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryTxEventRepository.java
@@ -0,0 +1,110 @@
+/*
+ * 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.alpha.server.tcc.service;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxType;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TxEventFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+@Component
+@Profile("memory")
+public class MemoryTxEventRepository implements TccTxEventRepository {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private final Map<String, Set<TccTxEvent>> tccEventMap = new ConcurrentHashMap<>();
+
+ @Override
+ public void saveGlobalTxEvent(GlobalTxEvent event) {
+ save(TxEventFactory.convertToTccTxEvent(event));
+ }
+
+ @Override
+ public void saveParticipatedEvent(ParticipatedEvent event) {
+ save(TxEventFactory.convertToTccTxEvent(event));
+ }
+
+ @Override
+ public void save(TccTxEvent event) {
+ tccEventMap
+ .computeIfAbsent(event.getGlobalTxId(), key-> new LinkedHashSet<>())
+ .add(event);
+ }
+
+ @Override
+ public Optional<List<TccTxEvent>> findByGlobalTxId(String globalTxId) {
+ Set<TccTxEvent> events = tccEventMap.get(globalTxId);
+ if ( events != null) {
+ return Optional.of(new ArrayList<>(events));
+ } else {
+ return Optional.empty();
+ }
+ }
+
+ @Override
+ public Optional<List<TccTxEvent>> findByGlobalTxIdAndTxType(String globalTxId, TccTxType tccTxType) {
+ Set<TccTxEvent> events = tccEventMap.get(globalTxId);
+ if ( events != null) {
+ return Optional.of(events.stream().filter(e -> tccTxType.name().equals(e.getTxType())).collect(Collectors.toList()));
+ } else {
+ return Optional.empty();
+ }
+ }
+
+ @Override
+ public Optional<TccTxEvent> findByUniqueKey(String globalTxId, String localTxId, TccTxType tccTxType) {
+ Set<TccTxEvent> events = tccEventMap.get(globalTxId);
+ if (events != null) {
+ return events.stream().filter(e ->
+ tccTxType.name().equals(e.getTxType())
+ && localTxId.equals(e.getLocalTxId())).findAny();
+ } else {
+ return Optional.empty();
+ }
+ }
+
+ @Override
+ public Iterable<TccTxEvent> findAll() {
+ List<TccTxEvent> events = new ArrayList<>();
+ for (String golableTxId : tccEventMap.keySet()) {
+ events.addAll(tccEventMap.get(golableTxId));
+ }
+ return events;
+ }
+
+ @Override
+ public void deleteAll() {
+ tccEventMap.clear();
+ }
+}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipatedEventService.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipatedEventService.java
deleted file mode 100644
index a932ebf..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipatedEventService.java
+++ /dev/null
@@ -1,32 +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.saga.alpha.server.tcc.service;
-
-import java.util.List;
-import java.util.Optional;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-
-public interface ParticipatedEventService {
-
- boolean addEvent(ParticipatedEvent participateEvent);
-
- Optional<List<ParticipatedEvent>> getEventByGlobalTxId(String globalTxId);
-
- void migration(String globalTxId, String localTxId);
-
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipatedEventServiceImpl.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipatedEventServiceImpl.java
deleted file mode 100644
index 662b918..0000000
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipatedEventServiceImpl.java
+++ /dev/null
@@ -1,83 +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.saga.alpha.server.tcc.service;
-
-import java.lang.invoke.MethodHandles;
-import java.util.List;
-import java.util.Optional;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEventHistory;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEventHistoryRepository;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEventRepository;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-import org.springframework.transaction.annotation.Transactional;
-
-@Component
-public class ParticipatedEventServiceImpl implements ParticipatedEventService {
-
- @Autowired
- private ParticipatedEventRepository hotRepository;
-
- @Autowired
- private ParticipatedEventHistoryRepository coldRepository;
-
- private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- @Override
- @Transactional
- public boolean addEvent(ParticipatedEvent event) {
- try {
- if (!hotRepository.findByUniqueKey(
- event.getGlobalTxId(), event.getLocalTxId()).isPresent()) {
- hotRepository.save(event);
- }
- } catch (Exception ex) {
- LOG.warn("Add participateEvent triggered exception, globalTxId:{}, localTxId:{}, ",
- event.getGlobalTxId(), event.getLocalTxId(), ex);
- return false;
- }
- return true;
- }
-
- @Override
- public Optional<List<ParticipatedEvent>> getEventByGlobalTxId(String globalTxId) {
- return hotRepository.findByGlobalTxId(globalTxId);
- }
-
- @Override
- @Transactional
- public void migration(String globalTxId, String localTxId) {
- hotRepository.findByUniqueKey(globalTxId, localTxId).ifPresent( e -> {
- hotRepository.delete(e);
- ParticipatedEventHistory finishedEvent = new ParticipatedEventHistory(
- e.getGlobalTxId(),
- e.getLocalTxId(),
- e.getParentTxId(),
- e.getServiceName(),
- e.getInstanceId(),
- e.getConfirmMethod(),
- e.getCancelMethod(),
- e.getStatus()
- );
- coldRepository.save(finishedEvent);
- });
- }
-}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/RDBTxEventRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/RDBTxEventRepository.java
new file mode 100644
index 0000000..4c704ea
--- /dev/null
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/RDBTxEventRepository.java
@@ -0,0 +1,93 @@
+/*
+ * 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.alpha.server.tcc.service;
+
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEventRepository;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEventRepository;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEventDBRepository;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxType;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TxEventFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+@Component
+@Profile("!memory")
+public class RDBTxEventRepository implements TccTxEventRepository {
+
+ @Autowired
+ GlobalTxEventRepository globalTxEventRepository;
+
+ @Autowired
+ ParticipatedEventRepository participatedEventRepository;
+
+ @Autowired
+ TccTxEventDBRepository tccTxEventDBRepository;
+
+ @Override
+ public void saveGlobalTxEvent(GlobalTxEvent event) {
+ globalTxEventRepository.save(event);
+ // saveTccEventHere
+ tccTxEventDBRepository.save(TxEventFactory.convertToTccTxEvent(event));
+ }
+
+ @Override
+ public void saveParticipatedEvent(ParticipatedEvent event) {
+ participatedEventRepository.save(event);
+ // saveTccEventHere
+ tccTxEventDBRepository.save(TxEventFactory.convertToTccTxEvent(event));
+ }
+
+ @Override
+ public void save(TccTxEvent event) {
+ tccTxEventDBRepository.save(event);
+ }
+
+ @Override
+ public Optional<List<TccTxEvent>> findByGlobalTxId(String globalTxId) {
+ return tccTxEventDBRepository.findByGlobalTxId(globalTxId);
+ }
+
+ @Override
+ public Optional<List<TccTxEvent>> findByGlobalTxIdAndTxType(String globalTxId, TccTxType tccTxType) {
+ return tccTxEventDBRepository.findByGlobalTxIdAndTxType(globalTxId, tccTxType.name());
+ }
+
+ @Override
+ public Optional<TccTxEvent> findByUniqueKey(String globalTxId, String localTxId, TccTxType tccTxType) {
+ return tccTxEventDBRepository.findByUniqueKey(globalTxId, localTxId, tccTxType.name());
+ }
+
+ @Override
+ public Iterable<TccTxEvent> findAll() {
+ return tccTxEventDBRepository.findAll();
+ }
+
+ @Override
+ public void deleteAll() {
+ globalTxEventRepository.deleteAll();
+ participatedEventRepository.deleteAll();
+ tccTxEventDBRepository.deleteAll();
+ }
+}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/TccTxEventRepository.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/TccTxEventRepository.java
new file mode 100644
index 0000000..3e7ba7e
--- /dev/null
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/TccTxEventRepository.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.saga.alpha.server.tcc.service;
+
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxType;
+
+public interface TccTxEventRepository {
+
+ void saveGlobalTxEvent(GlobalTxEvent event);
+
+ void saveParticipatedEvent(ParticipatedEvent event);
+
+ void save(TccTxEvent event);
+
+ Optional<List<TccTxEvent>> findByGlobalTxId(String globalTxId);
+
+ Optional<List<TccTxEvent>> findByGlobalTxIdAndTxType(String globalTxId, TccTxType tccTxType);
+
+ Optional<TccTxEvent> findByUniqueKey(String globalTxId, String localTxId, TccTxType tccTxType);
+
+ Iterable<TccTxEvent> findAll();
+
+ void deleteAll();
+
+}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/TccTxEventService.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/TccTxEventService.java
new file mode 100644
index 0000000..af9a529
--- /dev/null
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/TccTxEventService.java
@@ -0,0 +1,113 @@
+/*
+ * 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.alpha.server.tcc.service;
+
+import java.lang.invoke.MethodHandles;
+
+import org.apache.servicecomb.saga.alpha.server.tcc.callback.TccCallbackEngine;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+@Component
+public class TccTxEventService {
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private final TccTxEventRepository tccTxEventRepository;
+ private final TccCallbackEngine tccCallbackEngine;
+
+ public TccTxEventService(
+ TccTxEventRepository tccTxEventRepository,
+ TccCallbackEngine tccCallbackEngine) {
+ this.tccTxEventRepository = tccTxEventRepository;
+ this.tccCallbackEngine = tccCallbackEngine;
+ }
+
+ public boolean onTccStartedEvent(GlobalTxEvent globalTxEvent) {
+ LOG.info("Registered TccStarted event, global tx: {}, local tx: {}, parent id: {}, "
+ + "txType: {}, service [{}] instanceId [{}]",
+ globalTxEvent.getGlobalTxId(), globalTxEvent.getLocalTxId(), globalTxEvent.getParentTxId(),
+ globalTxEvent.getTxType(), globalTxEvent.getServiceName(), globalTxEvent.getInstanceId());
+ try {
+ if (!tccTxEventRepository.findByUniqueKey(globalTxEvent.getGlobalTxId(), globalTxEvent.getLocalTxId(),
+ TccTxType.valueOf(globalTxEvent.getTxType())).isPresent()) {
+ tccTxEventRepository.saveGlobalTxEvent(globalTxEvent);
+ }
+ } catch (Exception ex) {
+ LOG.warn("Add globalTxEvent triggered exception, globalTxId:{}, localTxId:{}, txType:{}, ",
+ globalTxEvent.getGlobalTxId(), globalTxEvent.getLocalTxId(), globalTxEvent.getTxType(), ex);
+ return false;
+ }
+ return true;
+ }
+
+ public boolean onParticipatedEvent(ParticipatedEvent participatedEvent) {
+ LOG.info("Registered participated event, global tx: {}, local tx: {}, parent id: {}, "
+ + "confirm: {}, cancel: {}, status: {}, service [{}] instanceId [{}]",
+ participatedEvent.getGlobalTxId(), participatedEvent.getLocalTxId(), participatedEvent.getParentTxId(),
+ participatedEvent.getConfirmMethod(), participatedEvent.getCancelMethod(), participatedEvent.getStatus(),
+ participatedEvent.getServiceName(), participatedEvent.getInstanceId());
+ try {
+ if (!tccTxEventRepository.findByUniqueKey(participatedEvent.getGlobalTxId(), participatedEvent.getLocalTxId(), TccTxType.PARTICIPATED).isPresent()) {
+ tccTxEventRepository.saveParticipatedEvent(participatedEvent);
+ }
+ } catch (Exception ex) {
+ LOG.warn("Add participateEvent triggered exception, globalTxId:{}, localTxId:{}, ",
+ participatedEvent.getGlobalTxId(), participatedEvent.getLocalTxId(), ex);
+ return false;
+ }
+ return true;
+ }
+
+ public boolean onTccEndedEvent(GlobalTxEvent globalTxEvent) {
+ LOG.info("Registered TccEnded event, global tx: {}, local tx: {}, parent id: {}, "
+ + "txType: {}, service [{}] instanceId [{}]",
+ globalTxEvent.getGlobalTxId(), globalTxEvent.getLocalTxId(), globalTxEvent.getParentTxId(),
+ globalTxEvent.getTxType(), globalTxEvent.getServiceName(), globalTxEvent.getInstanceId());
+ try {
+ tccTxEventRepository.saveGlobalTxEvent(globalTxEvent);
+ } catch (Exception ex) {
+ LOG.warn("Add globalTxEvent triggered exception, globalTxId:{}, localTxId:{}, txType:{}, ",
+ globalTxEvent.getGlobalTxId(), globalTxEvent.getLocalTxId(), globalTxEvent.getTxType(), ex);
+ return false;
+ }
+ // Just return the excution result back
+ return tccCallbackEngine.execute(globalTxEvent);
+
+ }
+
+ public boolean onCoordinatedEvent(TccTxEvent tccTxEvent) {
+ LOG.info("Registered TccStarted event, global tx: {}, local tx: {}, parent id: {}, "
+ + "txType: {}, service [{}] instanceId [{}]",
+ tccTxEvent.getGlobalTxId(), tccTxEvent.getLocalTxId(), tccTxEvent.getParentTxId(),
+ tccTxEvent.getTxType(), tccTxEvent.getServiceName(), tccTxEvent.getInstanceId());
+ try {
+ tccTxEventRepository.save(tccTxEvent);
+ } catch (Exception ex) {
+ LOG.warn("Add coordinatedEvent triggered exception, globalTxId:{}, localTxId:{} ",
+ tccTxEvent.getGlobalTxId(), tccTxEvent.getLocalTxId(), ex);
+ return false;
+ }
+ return true;
+ }
+
+}
diff --git a/alpha/alpha-server/src/main/resources/schema-mysql.sql b/alpha/alpha-server/src/main/resources/schema-mysql.sql
index c134456..776fb07 100644
--- a/alpha/alpha-server/src/main/resources/schema-mysql.sql
+++ b/alpha/alpha-server/src/main/resources/schema-mysql.sql
@@ -82,21 +82,6 @@ CREATE TABLE IF NOT EXISTS tcc_global_tx_event (
UNIQUE INDEX tcc_global_tx_event_index (globalTxId, localTxId, parentTxId, txType)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-CREATE TABLE IF NOT EXISTS tcc_global_tx_event_history (
- surrogateId bigint NOT NULL AUTO_INCREMENT,
- globalTxId varchar(36) NOT NULL,
- localTxId varchar(36) NOT NULL,
- parentTxId varchar(36) DEFAULT NULL,
- serviceName varchar(36) NOT NULL,
- instanceId varchar(36) NOT NULL,
- txType varchar(12),
- status varchar(12),
- creationTime datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
- lastModified datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
- PRIMARY KEY (surrogateId),
- UNIQUE INDEX tcc_global_tx_event_history_index (globalTxId, localTxId, parentTxId, txType)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-
CREATE TABLE IF NOT EXISTS tcc_participate_event (
surrogateId bigint NOT NULL AUTO_INCREMENT,
serviceName varchar(36) NOT NULL,
@@ -113,18 +98,18 @@ CREATE TABLE IF NOT EXISTS tcc_participate_event (
UNIQUE INDEX tcc_participate_event_index (globalTxId, localTxId, parentTxId)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-CREATE TABLE IF NOT EXISTS tcc_participate_event_history (
+CREATE TABLE IF NOT EXISTS tcc_tx_event (
surrogateId bigint NOT NULL AUTO_INCREMENT,
- serviceName varchar(36) NOT NULL,
- instanceId varchar(36) NOT NULL,
globalTxId varchar(36) NOT NULL,
localTxId varchar(36) NOT NULL,
parentTxId varchar(36) DEFAULT NULL,
- confirmMethod varchar(256) NOT NULL,
- cancelMethod varchar(256) NOT NULL,
- status varchar(50) NOT NULL,
+ serviceName varchar(36) NOT NULL,
+ instanceId varchar(36) NOT NULL,
+ methodInfo varchar(512) NOT NULL,
+ txType varchar(12),
+ status varchar(12),
creationTime datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
lastModified datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (surrogateId),
- UNIQUE INDEX tcc_finished_event_index (globalTxId, localTxId, parentTxId)
+ UNIQUE INDEX tcc_tx_event_index (globalTxId, localTxId, parentTxId, txType)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
diff --git a/alpha/alpha-server/src/main/resources/schema-postgresql.sql b/alpha/alpha-server/src/main/resources/schema-postgresql.sql
index 786e922..39cdf82 100644
--- a/alpha/alpha-server/src/main/resources/schema-postgresql.sql
+++ b/alpha/alpha-server/src/main/resources/schema-postgresql.sql
@@ -84,21 +84,6 @@ CREATE TABLE IF NOT EXISTS tcc_global_tx_event (
CREATE UNIQUE INDEX IF NOT EXISTS tcc_global_tx_event_index ON tcc_global_tx_event (globalTxId, localTxId, parentTxId, txType);
-CREATE TABLE IF NOT EXISTS tcc_global_tx_event_history (
- surrogateId BIGSERIAL PRIMARY KEY,
- globalTxId varchar(36) NOT NULL,
- localTxId varchar(36) NOT NULL,
- parentTxId varchar(36) DEFAULT NULL,
- serviceName varchar(36) NOT NULL,
- instanceId varchar(36) NOT NULL,
- txType varchar(12),
- status varchar(12),
- creationTime timestamp(6) NOT NULL DEFAULT CURRENT_DATE,
- lastModified timestamp(6) NOT NULL DEFAULT CURRENT_DATE
-);
-
-CREATE UNIQUE INDEX IF NOT EXISTS tcc_global_tx_event_history_index ON tcc_global_tx_event_history (globalTxId, localTxId, parentTxId, txType);
-
CREATE TABLE IF NOT EXISTS tcc_participate_event (
surrogateId BIGSERIAL PRIMARY KEY,
serviceName varchar(36) NOT NULL,
@@ -115,18 +100,18 @@ CREATE TABLE IF NOT EXISTS tcc_participate_event (
CREATE UNIQUE INDEX IF NOT EXISTS tcc_participate_event_index ON tcc_participate_event (globalTxId, localTxId, parentTxId);
-CREATE TABLE IF NOT EXISTS tcc_participate_event_history (
+CREATE TABLE IF NOT EXISTS tcc_tx_event (
surrogateId BIGSERIAL PRIMARY KEY,
- serviceName varchar(36) NOT NULL,
- instanceId varchar(36) NOT NULL,
globalTxId varchar(36) NOT NULL,
localTxId varchar(36) NOT NULL,
parentTxId varchar(36) DEFAULT NULL,
- confirmMethod varchar(256) NOT NULL,
- cancelMethod varchar(256) NOT NULL,
- status varchar(50) NOT NULL,
+ serviceName varchar(36) NOT NULL,
+ instanceId varchar(36) NOT NULL,
+ methodInfo varchar(512) NOT NULL,
+ txType varchar(12),
+ status varchar(12),
creationTime timestamp(6) NOT NULL DEFAULT CURRENT_DATE,
lastModified timestamp(6) NOT NULL DEFAULT CURRENT_DATE
);
-CREATE UNIQUE INDEX IF NOT EXISTS tcc_finished_event_index ON tcc_participate_event_history (globalTxId, localTxId, parentTxId);
+CREATE UNIQUE INDEX IF NOT EXISTS tcc_tx_event_index ON tcc_tx_event (globalTxId, localTxId, parentTxId, txType);
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaEventControllerTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaEventControllerTest.java
index c212b3b..ee2c444 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaEventControllerTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaEventControllerTest.java
@@ -36,11 +36,13 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.web.servlet.MockMvc;
@RunWith(SpringRunner.class)
@WebMvcTest(AlphaEventController.class)
+@ActiveProfiles("test")
public class AlphaEventControllerTest {
private final TxEvent someEvent = someEvent();
@@ -57,7 +59,7 @@ public class AlphaEventControllerTest {
@Test
public void retrievesEventsFromRepo() throws Exception {
- mockMvc.perform(get("/events"))
+ mockMvc.perform(get("/saga/events"))
.andExpect(status().isOk())
.andExpect(jsonPath("$", hasSize(1)))
.andExpect(jsonPath("$.[0].globalTxId", is(someEvent.globalTxId())))
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaEventControllerTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/AlphaTccEventControllerTest.java
similarity index 62%
copy from alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaEventControllerTest.java
copy to alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/AlphaTccEventControllerTest.java
index c212b3b..21d75d5 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaEventControllerTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/AlphaTccEventControllerTest.java
@@ -15,11 +15,10 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.alpha.server;
+package org.apache.servicecomb.saga.alpha.server.tcc;
import static com.seanyinx.github.unit.scaffolding.Randomness.uniquify;
import static java.util.Collections.singletonList;
-import static org.apache.servicecomb.saga.common.EventType.TxStartedEvent;
import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
import static org.hamcrest.core.Is.is;
import static org.mockito.Mockito.when;
@@ -27,28 +26,32 @@ import static org.springframework.test.web.servlet.request.MockMvcRequestBuilder
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
-import java.util.UUID;
-
-import org.apache.servicecomb.saga.alpha.core.TxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEventDBRepository;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxType;
+import org.apache.servicecomb.saga.alpha.server.tcc.service.TccTxEventRepository;
+import org.apache.servicecomb.saga.common.TransactionStatus;
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.autoconfigure.web.servlet.WebMvcTest;
import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.web.servlet.MockMvc;
@RunWith(SpringRunner.class)
-@WebMvcTest(AlphaEventController.class)
-public class AlphaEventControllerTest {
- private final TxEvent someEvent = someEvent();
+@WebMvcTest(AlphaTccEventController.class)
+@ActiveProfiles("test")
+public class AlphaTccEventControllerTest {
+ private final TccTxEvent someEvent = someEvent();
@Autowired
private MockMvc mockMvc;
@MockBean
- private TxEventEnvelopeRepository eventRepository;
+ private TccTxEventRepository eventRepository;
@Before
public void setUp() throws Exception {
@@ -57,22 +60,27 @@ public class AlphaEventControllerTest {
@Test
public void retrievesEventsFromRepo() throws Exception {
- mockMvc.perform(get("/events"))
+ mockMvc.perform(get("/tcc/events"))
.andExpect(status().isOk())
.andExpect(jsonPath("$", hasSize(1)))
- .andExpect(jsonPath("$.[0].globalTxId", is(someEvent.globalTxId())))
- .andExpect(jsonPath("$.[0].localTxId", is(someEvent.localTxId())));
+ .andExpect(jsonPath("$.[0].globalTxId", is(someEvent.getGlobalTxId())))
+ .andExpect(jsonPath("$.[0].localTxId", is(someEvent.getLocalTxId())))
+ .andExpect(jsonPath("$.[0].serviceName", is(someEvent.getServiceName())))
+ .andExpect(jsonPath("$.[0].instanceId", is(someEvent.getInstanceId())))
+ .andExpect(jsonPath("$.[0].txType", is(someEvent.getTxType())))
+ .andExpect(jsonPath("$.[0].status", is(someEvent.getStatus())));
+
}
- private TxEvent someEvent() {
- return new TxEvent(
+ private TccTxEvent someEvent() {
+ return new TccTxEvent(
uniquify("serviceName"),
uniquify("instanceId"),
uniquify("globalTxId"),
uniquify("localTxId"),
- UUID.randomUUID().toString(),
- TxStartedEvent.name(),
- this.getClass().getCanonicalName(),
- uniquify("blah").getBytes());
+ uniquify("parentTxId"),
+ TccTxType.STARTED.name(),
+ TransactionStatus.Succeed.name());
}
+
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/AlphaTccServerTestBase.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/AlphaTccServerTestBase.java
similarity index 87%
rename from alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/AlphaTccServerTestBase.java
rename to alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/AlphaTccServerTestBase.java
index c9233e1..cb0dbe3 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/AlphaTccServerTestBase.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/AlphaTccServerTestBase.java
@@ -6,16 +6,16 @@
* (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
+ * 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.
+ * 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.alpha.tcc.server;
+package org.apache.servicecomb.saga.alpha.server.tcc;
import static com.seanyinx.github.unit.scaffolding.Randomness.uniquify;
import static java.util.concurrent.TimeUnit.SECONDS;
@@ -25,18 +25,22 @@ import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
import io.grpc.ManagedChannel;
-import io.grpc.netty.NettyChannelBuilder;
+
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
-import org.apache.servicecomb.saga.alpha.server.tcc.TccTxEventFacade;
+
import org.apache.servicecomb.saga.alpha.server.tcc.callback.GrpcOmegaTccCallback;
import org.apache.servicecomb.saga.alpha.server.tcc.callback.OmegaCallbacksRegistry;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxType;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TxEventFactory;
+import org.apache.servicecomb.saga.alpha.server.tcc.service.TccTxEventRepository;
+import org.apache.servicecomb.saga.alpha.server.tcc.service.TccTxEventService;
import org.apache.servicecomb.saga.common.TransactionStatus;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcAck;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcServiceConfig;
@@ -51,6 +55,8 @@ import org.apache.servicecomb.saga.pack.contract.grpc.TccEventServiceGrpc.TccEve
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
public abstract class AlphaTccServerTestBase {
@@ -79,7 +85,9 @@ public abstract class AlphaTccServerTestBase {
.setInstanceId(instanceId)
.build();
- public abstract TccTxEventFacade getTccTxEventFacade();
+ @Autowired
+ private TccTxEventRepository tccTxEventRepository;
+
@AfterClass
public static void tearDown() {
@@ -90,6 +98,7 @@ public abstract class AlphaTccServerTestBase {
@After
public void after() {
blockingStub.onDisconnected(serviceConfig);
+ tccTxEventRepository.deleteAll();
}
@Test
@@ -123,16 +132,16 @@ public abstract class AlphaTccServerTestBase {
blockingStub.onTccTransactionStarted(newTxStart());
blockingStub.onTccTransactionStarted(newTxStart());
blockingStub.onTccTransactionEnded(newTxEnd("Succeed"));
- List<GlobalTxEvent> events = getTccTxEventFacade().getGlobalTxEventByGlobalTxId(globalTxId);
+ List<TccTxEvent> events = tccTxEventRepository.findByGlobalTxId(globalTxId).get();
assertThat(events.size(), is(2));
- Iterator<GlobalTxEvent> iterator = events.iterator();
- GlobalTxEvent event = iterator.next();
+ Iterator<TccTxEvent> iterator = events.iterator();
+ TccTxEvent event = iterator.next();
assertThat(event.getGlobalTxId(), is(globalTxId));
assertThat(event.getLocalTxId(), is(localTxId));
assertThat(event.getInstanceId(), is(instanceId));
assertThat(event.getServiceName(), is(serviceName));
- assertThat(event.getTxType(), is(TccTxType.TCC_START.name()));
+ assertThat(event.getTxType(), is(TccTxType.STARTED.name()));
assertThat(event.getStatus(), is(TransactionStatus.Succeed.name()));
event = iterator.next();
@@ -140,7 +149,7 @@ public abstract class AlphaTccServerTestBase {
assertThat(event.getLocalTxId(), is(localTxId));
assertThat(event.getInstanceId(), is(instanceId));
assertThat(event.getServiceName(), is(serviceName));
- assertThat(event.getTxType(), is(TccTxType.TCC_END.name()));
+ assertThat(event.getTxType(), is(TccTxType.ENDED.name()));
assertThat(event.getStatus(), is(TransactionStatus.Succeed.name()));
}
@@ -150,15 +159,15 @@ public abstract class AlphaTccServerTestBase {
awaitUntilConnected();
blockingStub.participate(newParticipatedEvent("Succeed"));
blockingStub.participate(newParticipatedEvent("Succeed"));
- List<ParticipatedEvent> events = getTccTxEventFacade().getParticipateEventByGlobalTxId(globalTxId);
+ List<TccTxEvent> events = tccTxEventRepository.findByGlobalTxId(globalTxId).get();
assertThat(events.size(), is(1));
- ParticipatedEvent event = events.iterator().next();
+ TccTxEvent event = events.iterator().next();
assertThat(event.getGlobalTxId(), is(globalTxId));
assertThat(event.getLocalTxId(), is(localTxId));
assertThat(event.getInstanceId(), is(instanceId));
assertThat(event.getServiceName(), is(serviceName));
- assertThat(event.getConfirmMethod(), is(confirmMethod));
- assertThat(event.getCancelMethod(), is(cancelMethod));
+ assertThat(TxEventFactory.getMethodName(event.getMethodInfo(), true), is(confirmMethod));
+ assertThat(TxEventFactory.getMethodName(event.getMethodInfo(), false), is(cancelMethod));
assertThat(event.getStatus(), is("Succeed"));
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/MemoryAlphaTccServerTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/MemoryAlphaTccServerTest.java
similarity index 69%
rename from alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/MemoryAlphaTccServerTest.java
rename to alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/MemoryAlphaTccServerTest.java
index 6dd9fcf..42bd071 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/MemoryAlphaTccServerTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/MemoryAlphaTccServerTest.java
@@ -6,35 +6,34 @@
* (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
+ * 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.
+ * 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.alpha.tcc.server;
+package org.apache.servicecomb.saga.alpha.server.tcc;
import io.grpc.netty.NettyChannelBuilder;
import org.apache.servicecomb.saga.alpha.server.AlphaApplication;
-import org.apache.servicecomb.saga.alpha.server.tcc.TccTxEventFacade;
import org.junit.BeforeClass;
-import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = {AlphaApplication.class},
properties = {
"alpha.server.host=0.0.0.0",
- "alpha.server.port=8090",
- "alpha.server.storage=local"
+ "alpha.server.port=8090"
})
+@ActiveProfiles("memory")
public class MemoryAlphaTccServerTest extends AlphaTccServerTestBase {
@BeforeClass
@@ -42,13 +41,13 @@ public class MemoryAlphaTccServerTest extends AlphaTccServerTestBase {
clientChannel = NettyChannelBuilder.forAddress("localhost", 8090).usePlaintext().build();
}
- @Autowired
+ /*@Autowired
@Qualifier("defaultTccTxEventFacade")
private TccTxEventFacade tccTxEventFacade;
@Override
public TccTxEventFacade getTccTxEventFacade() {
return tccTxEventFacade;
- }
+ }*/
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/RdbAlphaTccServerTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbAlphaTccServerTest.java
similarity index 61%
rename from alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/RdbAlphaTccServerTest.java
rename to alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbAlphaTccServerTest.java
index a81a415..0424438 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/RdbAlphaTccServerTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbAlphaTccServerTest.java
@@ -6,34 +6,32 @@
* (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
+ * 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.
+ * 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.alpha.tcc.server;
+package org.apache.servicecomb.saga.alpha.server.tcc;
import io.grpc.netty.NettyChannelBuilder;
import org.apache.servicecomb.saga.alpha.server.AlphaApplication;
-import org.apache.servicecomb.saga.alpha.server.tcc.TccTxEventFacade;
import org.junit.BeforeClass;
-import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = {AlphaApplication.class},
properties = {
"alpha.server.host=0.0.0.0",
- "alpha.server.port=8091",
- "alpha.server.storage=rdb"
+ "alpha.server.port=8091"
})
public class RdbAlphaTccServerTest extends AlphaTccServerTestBase {
@@ -42,13 +40,6 @@ public class RdbAlphaTccServerTest extends AlphaTccServerTestBase {
clientChannel = NettyChannelBuilder.forAddress("localhost", 8091).usePlaintext().build();
}
- @Autowired
- @Qualifier("rdbTccTxEventFacade")
- private TccTxEventFacade tccTxEventFacade;
- @Override
- public TccTxEventFacade getTccTxEventFacade() {
- return tccTxEventFacade;
- }
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/TccCoordinateCommandStreamObserver.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/TccCoordinateCommandStreamObserver.java
similarity index 76%
rename from alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/TccCoordinateCommandStreamObserver.java
rename to alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/TccCoordinateCommandStreamObserver.java
index a520d89..5c4143a 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/TccCoordinateCommandStreamObserver.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/TccCoordinateCommandStreamObserver.java
@@ -6,16 +6,16 @@
* (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
+ * 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.
+ * 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.alpha.tcc.server;
+package org.apache.servicecomb.saga.alpha.server.tcc;
import io.grpc.stub.StreamObserver;
import java.util.Queue;
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactoryTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactoryTest.java
new file mode 100644
index 0000000..d026f14
--- /dev/null
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactoryTest.java
@@ -0,0 +1,37 @@
+/*
+ * 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.alpha.server.tcc.jpa;
+
+import org.junit.Test;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+
+
+public class TxEventFactoryTest {
+
+ @Test
+ public void getMethodInfo() {
+ assertThat(TxEventFactory.toMethodInfo("test1", "test2"), is("confirm=test1,cancel=test2"));
+ }
+
+ @Test
+ public void getMethodName() {
+ assertThat(TxEventFactory.getMethodName("confirm=text1,cancel=text2",true), is("text1"));
+ assertThat(TxEventFactory.getMethodName("confirm=text1,cancel=text2",false), is("text2"));
+ }
+}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryEventRegistryTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryEventRegistryTest.java
new file mode 100644
index 0000000..c1d929f
--- /dev/null
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryEventRegistryTest.java
@@ -0,0 +1,115 @@
+/*
+ * 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.alpha.server.tcc.service;
+
+import static com.seanyinx.github.unit.scaffolding.Randomness.uniquify;
+import static org.hamcrest.core.Is.is;
+import static org.hamcrest.Matchers.contains;
+import static org.junit.Assert.assertThat;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxType;
+import org.apache.servicecomb.saga.common.TransactionStatus;
+import org.junit.Before;
+import org.junit.Test;
+
+
+public class MemoryEventRegistryTest {
+ private MemoryTxEventRepository registry = new MemoryTxEventRepository();
+ private String gloableTxId = uniquify("globalTxId");
+ private String serviceName = uniquify("serviceName");
+ private String instanceId = uniquify("instanceId");
+ private String localTxId1 = uniquify("localTxId");
+ private String localTxId2 = uniquify("localTxId");
+
+
+ @Before
+ public void setup() {
+ registry.save(someTccTxEvent(localTxId1, TccTxType.STARTED));
+ registry.save(someTccTxEvent(localTxId1, TccTxType.PARTICIPATED));
+ registry.save(someTccTxEvent(localTxId2, TccTxType.PARTICIPATED));
+ }
+
+ private TccTxEvent someTccTxEvent(String localTxId, TccTxType tccTxType) {
+ return new TccTxEvent(serviceName, instanceId, gloableTxId, localTxId, "", tccTxType.name(), "", TransactionStatus.Succeed.name());
+ }
+
+ @Test
+ public void testFindByGlobalTxId() {
+ Optional<List<TccTxEvent>> result = registry.findByGlobalTxId(gloableTxId);
+ assertThat(result.isPresent(), is(true));
+ assertThat(result.get().size(), is(3));
+
+ result = registry.findByGlobalTxId("globalTx");
+ assertThat(result.isPresent(), is(false));
+ }
+
+ @Test
+ public void testfindByGlobalTxIdAndTxType() {
+ Optional<List<TccTxEvent>> result = registry.findByGlobalTxIdAndTxType(gloableTxId, TccTxType.PARTICIPATED);
+ assertThat(result.isPresent(), is(true));
+ assertThat(result.get().size(), is(2));
+
+ result = registry.findByGlobalTxIdAndTxType(gloableTxId, TccTxType.STARTED);
+ assertThat(result.isPresent(), is(true));
+ assertThat(result.get().size(), is(1));
+
+ result = registry.findByGlobalTxIdAndTxType("globalTx", TccTxType.ENDED);
+ assertThat(result.isPresent(), is(false));
+ }
+
+ @Test
+ public void testfindByUniqueKey() {
+ Optional<TccTxEvent> result = registry.findByUniqueKey(gloableTxId, localTxId1, TccTxType.PARTICIPATED);
+ assertThat(result.isPresent(), is(true));
+ assertThat(result.get().getLocalTxId(), is(localTxId1));
+ assertThat(result.get().getTxType(), is(TccTxType.PARTICIPATED.name()));
+
+ result = registry.findByUniqueKey(gloableTxId, localTxId2, TccTxType.PARTICIPATED);
+ assertThat(result.isPresent(), is(true));
+ assertThat(result.get().getLocalTxId(), is(localTxId2));
+ assertThat(result.get().getTxType(), is(TccTxType.PARTICIPATED.name()));
+
+ result = registry.findByUniqueKey(gloableTxId, localTxId1, TccTxType.ENDED);
+ assertThat(result.isPresent(), is(false));
+
+ }
+
+ @Test
+ public void testfindAll() {
+ Iterable<TccTxEvent> result = registry.findAll();
+ List<String> events = new ArrayList<>();
+ // Just check the event type here
+ for(TccTxEvent event : result) {
+ events.add(event.getTxType());
+ }
+ assertThat(events, contains("STARTED", "PARTICIPATED", "PARTICIPATED"));
+ }
+
+ @Test
+ public void testDeleteAll() {
+ registry.deleteAll();
+ Iterable<TccTxEvent> result = registry.findAll();
+ assertThat(result.iterator().hasNext(), is(false));
+ }
+
+}
diff --git a/alpha/alpha-server/src/test/resources/schema.sql b/alpha/alpha-server/src/test/resources/schema.sql
index d3f239d..f003d45 100644
--- a/alpha/alpha-server/src/test/resources/schema.sql
+++ b/alpha/alpha-server/src/test/resources/schema.sql
@@ -73,18 +73,6 @@ CREATE TABLE IF NOT EXISTS tcc_global_tx_event (
lastModified TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
);
-CREATE TABLE IF NOT EXISTS tcc_global_tx_event_history (
- surrogateId bigint GENERATED BY DEFAULT AS IDENTITY(START WITH 1, INCREMENT BY 1) PRIMARY KEY,
- globalTxId varchar(36) NOT NULL,
- localTxId varchar(36) NOT NULL,
- parentTxId varchar(36) DEFAULT NULL,
- serviceName varchar(36) NOT NULL,
- instanceId varchar(36) NOT NULL,
- txType varchar(12),
- creationTime TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
- lastModified TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
-);
-
CREATE TABLE IF NOT EXISTS tcc_participate_event (
surrogateId bigint GENERATED BY DEFAULT AS IDENTITY(START WITH 1, INCREMENT BY 1) PRIMARY KEY,
serviceName varchar(36) NOT NULL,
@@ -99,16 +87,17 @@ CREATE TABLE IF NOT EXISTS tcc_participate_event (
lastModified TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
);
-CREATE TABLE IF NOT EXISTS tcc_participate_event_history (
+CREATE TABLE IF NOT EXISTS tcc_tx_event (
surrogateId bigint GENERATED BY DEFAULT AS IDENTITY(START WITH 1, INCREMENT BY 1) PRIMARY KEY,
- serviceName varchar(36) NOT NULL,
- instanceId varchar(36) NOT NULL,
globalTxId varchar(36) NOT NULL,
localTxId varchar(36) NOT NULL,
parentTxId varchar(36) DEFAULT NULL,
- confirmMethod varchar(256) NOT NULL,
- cancelMethod varchar(256) NOT NULL,
- status varchar(50) NOT NULL,
+ serviceName varchar(36) NOT NULL,
+ instanceId varchar(36) NOT NULL,
+ methodInfo varchar(512) NOT NULL,
+ txType varchar(12),
+ status varchar(12),
creationTime TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
lastModified TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
);
+