You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by zh...@apache.org on 2019/06/21 10:34:16 UTC
[servicecomb-pack] branch SCB-1321 updated (6f0701d -> 731c6d9)
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a change to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git.
from 6f0701d SCB-1321 Add Alpha FSM design document
new 98bc27b SCB-1321 Add Alpha FSM prototype code
new 6e592d0 SCB-1321 Add state machine uml file
new 11936fd SCB-1321 Sub-transaction support concurrent
new 77ecd8f SCB-1321 Add alpha-fsm README.md
new 731c6d9 SCB-1321 Add license header
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
alpha/alpha-fsm/README.md | 11 +
alpha/{alpha-core => alpha-fsm}/pom.xml | 67 +-
.../servicecomb/pack/alpha/fsm/SagaActor.java | 366 +++++++++
.../servicecomb/pack/alpha/fsm/SagaActorState.java | 19 +-
.../apache/servicecomb/pack/alpha/fsm/TxState.java | 9 +-
.../pack/alpha/fsm/event/SagaAbortedEvent.java | 39 +-
.../pack/alpha/fsm/event/SagaDomainEvent.java} | 22 +-
.../pack/alpha/fsm/event/SagaEndedEvent.java | 39 +-
.../pack/alpha/fsm/event/SagaStartedEvent.java | 59 ++
.../pack/alpha/fsm/event/SagaTimeoutEvent.java | 33 +-
.../pack/alpha/fsm/event}/TxAbortedEvent.java | 45 +-
.../pack/alpha/fsm/event/TxComponsitedEvent.java | 39 +-
.../pack/alpha/fsm/event}/TxEndedEvent.java | 38 +-
.../pack/alpha/fsm/event}/TxStartedEvent.java | 38 +-
.../pack/alpha/fsm/event/base/BaseEvent.java | 23 +-
.../pack/alpha/fsm/event/base/SagaEvent.java | 14 +-
.../pack/alpha/fsm/event/base/TxEvent.java | 26 +-
.../servicecomb/pack/alpha/fsm/model/SagaData.java | 147 ++++
.../servicecomb/pack/alpha/fsm/model/TxEntity.java | 111 +++
.../servicecomb/pack/alpha/fsm/SagaActorTest.java | 838 +++++++++++++++++++++
.../alpha-fsm/src/test/resources/application.conf | 6 +-
.../alpha-fsm/src/test/resources/log4j2.xml | 0
alpha/pom.xml | 3 +-
docs/fsm/assets/saga_state_diagram.png | Bin 229442 -> 237463 bytes
docs/fsm/assets/state_table.png | Bin 286375 -> 311887 bytes
docs/fsm/plantuml/saga-state-diagram.puml | 52 ++
26 files changed, 1892 insertions(+), 152 deletions(-)
create mode 100644 alpha/alpha-fsm/README.md
copy alpha/{alpha-core => alpha-fsm}/pom.xml (70%)
create mode 100644 alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
copy omega/omega-context/src/main/java/org/apache/servicecomb/pack/omega/context/UniqueIdGenerator.java => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorState.java (72%)
copy omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/core/TransactionType.java => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/TxState.java (87%)
copy demo/saga-dubbo-demo/service-pub/src/main/java/org/apache/servicecomb/pack/demo/dubbo/pub/ServiceVO.java => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaAbortedEvent.java (57%)
copy alpha/{alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/CommandRepository.java => alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaDomainEvent.java} (56%)
copy demo/saga-dubbo-demo/service-pub/src/main/java/org/apache/servicecomb/pack/demo/dubbo/pub/ServiceVO.java => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaEndedEvent.java (58%)
create mode 100644 alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaStartedEvent.java
copy omega/omega-format/src/test/java/org/apache/servicecomb/pack/omega/format/KryoMessageFormatTest.java => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaTimeoutEvent.java (57%)
copy {omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event}/TxAbortedEvent.java (51%)
copy omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/MessageSerializer.java => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxComponsitedEvent.java (50%)
copy {omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event}/TxEndedEvent.java (52%)
copy {omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event}/TxStartedEvent.java (52%)
copy omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/TccStartedEvent.java => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/BaseEvent.java (72%)
copy omega/omega-context/src/main/java/org/apache/servicecomb/pack/omega/context/UniqueIdGenerator.java => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/SagaEvent.java (74%)
copy omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/tcc/events/TccStartedEvent.java => alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/TxEvent.java (66%)
create mode 100644 alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/SagaData.java
create mode 100644 alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java
create mode 100644 alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java
copy omega/omega-transport/omega-transport-feign/src/main/resources/META-INF/spring.factories => alpha/alpha-fsm/src/test/resources/application.conf (81%)
copy omega/omega-transaction/src/test/resources/log4j2-test.xml => alpha/alpha-fsm/src/test/resources/log4j2.xml (100%)
create mode 100644 docs/fsm/plantuml/saga-state-diagram.puml
[servicecomb-pack] 05/05: SCB-1321 Add license header
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 731c6d9b561e9d8d845ca9d429ef02c28095b72a
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Fri Jun 21 18:33:37 2019 +0800
SCB-1321 Add license header
---
.../apache/servicecomb/pack/alpha/fsm/SagaActor.java | 18 +++++++++++++++++-
.../servicecomb/pack/alpha/fsm/SagaActorState.java | 17 +++++++++++++++++
.../org/apache/servicecomb/pack/alpha/fsm/TxState.java | 17 +++++++++++++++++
.../pack/alpha/fsm/event/SagaAbortedEvent.java | 17 +++++++++++++++++
.../pack/alpha/fsm/event/SagaDomainEvent.java | 17 +++++++++++++++++
.../pack/alpha/fsm/event/SagaEndedEvent.java | 17 +++++++++++++++++
.../pack/alpha/fsm/event/SagaStartedEvent.java | 18 +++++++++++++++++-
.../pack/alpha/fsm/event/SagaTimeoutEvent.java | 17 +++++++++++++++++
.../pack/alpha/fsm/event/TxAbortedEvent.java | 17 +++++++++++++++++
.../pack/alpha/fsm/event/TxComponsitedEvent.java | 17 +++++++++++++++++
.../servicecomb/pack/alpha/fsm/event/TxEndedEvent.java | 17 +++++++++++++++++
.../pack/alpha/fsm/event/TxStartedEvent.java | 17 +++++++++++++++++
.../pack/alpha/fsm/event/base/BaseEvent.java | 17 +++++++++++++++++
.../pack/alpha/fsm/event/base/SagaEvent.java | 17 +++++++++++++++++
.../servicecomb/pack/alpha/fsm/event/base/TxEvent.java | 17 +++++++++++++++++
.../servicecomb/pack/alpha/fsm/model/SagaData.java | 18 +++++++++++++++++-
.../servicecomb/pack/alpha/fsm/model/TxEntity.java | 18 +++++++++++++++++-
.../servicecomb/pack/alpha/fsm/SagaActorTest.java | 17 +++++++++++++++++
alpha/alpha-fsm/src/test/resources/application.conf | 17 +++++++++++++++++
19 files changed, 323 insertions(+), 4 deletions(-)
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
index 2b48f45..613a23b 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
@@ -1,5 +1,21 @@
-package org.apache.servicecomb.pack.alpha.fsm;
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.pack.alpha.fsm;
import akka.actor.Props;
import akka.persistence.fsm.AbstractPersistentFSM;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorState.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorState.java
index 0c7be27..212e47b 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorState.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorState.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm;
import akka.persistence.fsm.PersistentFSM;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/TxState.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/TxState.java
index e9c54e4..0494895 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/TxState.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/TxState.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm;
public enum TxState {
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaAbortedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaAbortedEvent.java
index 351146b..b26504a 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaAbortedEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaAbortedEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event;
import org.apache.servicecomb.pack.alpha.fsm.event.base.SagaEvent;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaDomainEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaDomainEvent.java
index b421c73..b16e44a 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaDomainEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaDomainEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event;
public class SagaDomainEvent {
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaEndedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaEndedEvent.java
index 7a87d77..9ea5cf1 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaEndedEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaEndedEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event;
import org.apache.servicecomb.pack.alpha.fsm.event.base.SagaEvent;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaStartedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaStartedEvent.java
index 4831be7..56af9cd 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaStartedEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaStartedEvent.java
@@ -1,5 +1,21 @@
-package org.apache.servicecomb.pack.alpha.fsm.event;
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.pack.alpha.fsm.event;
import org.apache.servicecomb.pack.alpha.fsm.event.base.SagaEvent;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaTimeoutEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaTimeoutEvent.java
index 4de5c37..bed3264 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaTimeoutEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaTimeoutEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event;
import org.apache.servicecomb.pack.alpha.fsm.event.base.SagaEvent;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java
index 620366f..8ee61de 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event;
import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxComponsitedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxComponsitedEvent.java
index cccb69a..1230fff 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxComponsitedEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxComponsitedEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event;
import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxEndedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxEndedEvent.java
index 33ea3d1..5450664 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxEndedEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxEndedEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event;
import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxStartedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxStartedEvent.java
index 2d10462..bf259d0 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxStartedEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxStartedEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event;
import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/BaseEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/BaseEvent.java
index ea8257d..5794566 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/BaseEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/BaseEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event.base;
import java.io.Serializable;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/SagaEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/SagaEvent.java
index bf39855..686400d 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/SagaEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/SagaEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event.base;
public class SagaEvent extends BaseEvent {
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/TxEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/TxEvent.java
index 56a06e0..ceaa242 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/TxEvent.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/TxEvent.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm.event.base;
public abstract class TxEvent extends BaseEvent {
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/SagaData.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/SagaData.java
index 0cc858a..3c794fa 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/SagaData.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/SagaData.java
@@ -1,5 +1,21 @@
-package org.apache.servicecomb.pack.alpha.fsm.model;
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.pack.alpha.fsm.model;
import java.io.Serializable;
import java.util.HashMap;
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java
index f6a3222..1eb3020 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java
@@ -1,5 +1,21 @@
-package org.apache.servicecomb.pack.alpha.fsm.model;
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.pack.alpha.fsm.model;
import java.io.Serializable;
import org.apache.servicecomb.pack.alpha.fsm.TxState;
diff --git a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java
index 847d144..21c9402 100644
--- a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java
+++ b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.servicecomb.pack.alpha.fsm;
import static org.junit.Assert.assertEquals;
diff --git a/alpha/alpha-fsm/src/test/resources/application.conf b/alpha/alpha-fsm/src/test/resources/application.conf
index 1265036..8d39346 100644
--- a/alpha/alpha-fsm/src/test/resources/application.conf
+++ b/alpha/alpha-fsm/src/test/resources/application.conf
@@ -1,3 +1,20 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
akka.persistence.journal.plugin = "akka.persistence.journal.inmem"
akka.persistence.snapshot-store.plugin = "akka.persistence.snapshot-store.local"
akka.persistence.snapshot-store.local.dir = "target/example/snapshots"
\ No newline at end of file
[servicecomb-pack] 04/05: SCB-1321 Add alpha-fsm README.md
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 77ecd8f03dcaeb7435622c802f05b36e229bd29a
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Fri Jun 21 18:30:17 2019 +0800
SCB-1321 Add alpha-fsm README.md
---
alpha/alpha-fsm/README.md | 11 +++++++++++
1 file changed, 11 insertions(+)
diff --git a/alpha/alpha-fsm/README.md b/alpha/alpha-fsm/README.md
new file mode 100644
index 0000000..3a37fed
--- /dev/null
+++ b/alpha/alpha-fsm/README.md
@@ -0,0 +1,11 @@
+# Saga State Machine Module
+
+基于状态机控制事务的状态转换，**这是一个试验性模块**
+[更多详细内容](https://github.com/apache/servicecomb-pack/blob/SCB-1321/docs/fsm/design_fsm_zh.md)
+
+## Test State Machine
+
+```
+cd alpha
+mvn clean package -pl alpha-fsm
+```
\ No newline at end of file
[servicecomb-pack] 02/05: SCB-1321 Add state machine uml file
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 6e592d03b1c347a542bb6949d45f4afc4a4866f1
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Fri Jun 21 17:37:26 2019 +0800
SCB-1321 Add state machine uml file
---
docs/fsm/plantuml/saga-state-diagram.puml | 52 +++++++++++++++++++++++++++++++
1 file changed, 52 insertions(+)
diff --git a/docs/fsm/plantuml/saga-state-diagram.puml b/docs/fsm/plantuml/saga-state-diagram.puml
new file mode 100644
index 0000000..429687d
--- /dev/null
+++ b/docs/fsm/plantuml/saga-state-diagram.puml
@@ -0,0 +1,52 @@
+@startuml
+hide empty description
+
+[*] --> IDEL : SagaStartedEvent
+
+IDEL --> PARTIALLY_ACTIVE : TxStartedEvent<font color=blue>:addTxEntity</font>
+
+IDEL --> SUSPENDED : SagaEndedEvent
+
+IDEL --> SUSPENDED : SagaAbortedEvent
+
+PARTIALLY_COMMITTED --> PARTIALLY_COMMITTED : TxEndedEvent<font color=blue>:UpdateTxEntity</font>
+
+PARTIALLY_COMMITTED --> COMMITTED : SagaEndedEvent
+
+PARTIALLY_COMMITTED --> PARTIALLY_ACTIVE : TxStartedEvent<font color=blue>:AddTxEntity</font>
+
+PARTIALLY_COMMITTED --> FAILED : SagaAbortedEvent<font color=red>:doCompensation</font>
+
+PARTIALLY_COMMITTED --> SUSPENDED : SagaTimeoutEvent
+
+PARTIALLY_ACTIVE --> PARTIALLY_ACTIVE : TxStartedEvent<font color=blue>:addTxEntity</font>
+
+PARTIALLY_ACTIVE --> PARTIALLY_COMMITTED : TxEndedEvent<font color=blue>:UpdateTxEntity</font>
+
+PARTIALLY_ACTIVE --> FAILED : TxAbortedEvent<font color=blue>:UpdateTxEntity</font><font color=red>,doCompensation</font>
+
+PARTIALLY_ACTIVE --> SUSPENDED : SagaTimeoutEvent
+
+SUSPENDED --> SUSPENDED : SagaTimeoutEvent
+
+FAILED --> COMPENSATED : SagaAbortedEvent<font color=red>:doCompensation</font>
+
+FAILED --> SUSPENDED : SagaTimeoutEvent
+
+FAILED --> FAILED : TxComponsitedEvent<font color=blue>:UpdateTxEntity</font>\nTxStartedEvent<font color=blue>:AddTxEntity</font>\nTxEndedEvent<font color=red>:doCompensation</font>
+
+COMPENSATED --> [*]
+
+COMMITTED --> [*]
+
+SUSPENDED --> [*]
+
+IDEL: Map<localTxId,TxEntity> txEntityMap
+PARTIALLY_ACTIVE: Map<localTxId,TxEntity> txEntityMap
+PARTIALLY_COMMITTED: Map<localTxId,TxEntity> txEntityMap
+FAILED: Map<localTxId,TxEntity> txEntityMap
+COMMITTED: Map<localTxId,TxEntity> txEntityMap
+COMPENSATED: Map<localTxId,TxEntity> txEntityMap
+SUSPENDED: Map<localTxId,TxEntity> txEntityMap
+
+@enduml
[servicecomb-pack] 03/05: SCB-1321 Sub-transaction support
concurrent
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 11936fdc9a71224b9b7d0e5a2d55192cf735bc10
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Fri Jun 21 18:25:18 2019 +0800
SCB-1321 Sub-transaction support concurrent
---
.../servicecomb/pack/alpha/fsm/SagaActor.java | 39 +++-
.../servicecomb/pack/alpha/fsm/SagaActorTest.java | 204 ++++++++++++++++++++-
alpha/pom.xml | 3 +-
docs/fsm/assets/saga_state_diagram.png | Bin 229442 -> 237463 bytes
docs/fsm/assets/state_table.png | Bin 286375 -> 311887 bytes
5 files changed, 234 insertions(+), 12 deletions(-)
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
index 492d187..2b48f45 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
@@ -95,6 +95,16 @@ public class SagaActor extends
return goTo(SagaActorState.PARTIALLY_COMMITTED);
}
}
+ ).event(TxStartedEvent.class,
+ (event, data) -> {
+ updateTxEntity(event, data);
+ if (data.getExpirationTime() > 0) {
+ return stay()
+ .forMax(Duration.create(data.getTimeout(), TimeUnit.MILLISECONDS));
+ } else {
+ return stay();
+ }
+ }
).event(SagaTimeoutEvent.class,
(event, data) -> {
return goTo(SagaActorState.SUSPENDED)
@@ -123,6 +133,16 @@ public class SagaActor extends
return goTo(SagaActorState.PARTIALLY_ACTIVE);
}
}
+ ).event(TxEndedEvent.class,
+ (event, data) -> {
+ updateTxEntity(event, data);
+ if (data.getExpirationTime() > 0) {
+ return stay()
+ .forMax(Duration.create(data.getTimeout(), TimeUnit.MILLISECONDS));
+ } else {
+ return stay();
+ }
+ }
).event(SagaTimeoutEvent.class,
(event, data) -> {
return goTo(SagaActorState.SUSPENDED)
@@ -196,8 +216,8 @@ public class SagaActor extends
).event(TxEndedEvent.class, SagaData.class,
(event, data) -> {
updateTxEntity(event, data);
- // TODO 调用补偿方法
TxEntity txEntity = data.getTxEntityMap().get(event.getLocalTxId());
+ // TODO call compensate
compensation(txEntity, data);
return stay();
}
@@ -233,8 +253,8 @@ public class SagaActor extends
whenUnhandled(
matchAnyEvent((event, data) -> {
- LOG.error("unmatch event {}", event);
- return stay();
+ LOG.error("Unhandled event {}", event);
+ return goTo(SagaActorState.SUSPENDED).replying(data);
})
);
@@ -290,26 +310,25 @@ public class SagaActor extends
if (txEntity.getState() == TxState.ACTIVE) {
txEntity.setEndTime(System.currentTimeMillis());
txEntity.setState(TxState.FAILED);
- // TODO 调用补偿方法
data.getTxEntityMap().forEach((k, v) -> {
if (v.getState() == TxState.COMMITTED) {
- // TODO 调用补偿方法
+ // call compensate
compensation(v, data);
}
});
}
} else if (event instanceof TxComponsitedEvent) {
- //补偿中计数器减一
+ // decrement the compensation running counter by one
data.getCompensationRunningCounter().decrementAndGet();
txEntity.setState(TxState.COMPENSATED);
- LOG.info("完成补偿 {}",txEntity.getLocalTxId());
+ LOG.info("compensation is completed {}",txEntity.getLocalTxId());
}
}
} else if (event instanceof SagaEvent) {
if (event instanceof SagaAbortedEvent) {
data.getTxEntityMap().forEach((k, v) -> {
if (v.getState() == TxState.COMMITTED) {
- // TODO 调用补偿方法
+ // call compensate
compensation(v, data);
}
});
@@ -324,8 +343,8 @@ public class SagaActor extends
}
private void compensation(TxEntity txEntity, SagaData data) {
- //补偿中计数器加一
+ // increments the compensation running counter by one
data.getCompensationRunningCounter().incrementAndGet();
- LOG.info("调用补偿方法 {}", txEntity.getLocalTxId());
+ LOG.info("compensate {}", txEntity.getLocalTxId());
}
}
diff --git a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java
index 156d67e..847d144 100644
--- a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java
+++ b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java
@@ -8,6 +8,7 @@ import akka.actor.Terminated;
import akka.persistence.fsm.PersistentFSM;
import akka.persistence.fsm.PersistentFSM.CurrentState;
import akka.testkit.javadsl.TestKit;
+import java.lang.invoke.MethodHandles;
import java.time.Duration;
import java.util.UUID;
import org.apache.servicecomb.pack.alpha.fsm.event.SagaAbortedEvent;
@@ -22,9 +23,13 @@ import org.apache.servicecomb.pack.alpha.fsm.model.SagaData;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class SagaActorTest {
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
static ActorSystem system;
@BeforeClass
@@ -320,7 +325,7 @@ public class SagaActorTest {
* 10. SagaAbortedEvent-1
*/
@Test
- public void receivedRemainingEventAfterfirstTxAbortedEventTest() {
+ public void receivedRemainingEventAfterFirstTxAbortedEventTest() {
new TestKit(system) {{
final String globalTxId = UUID.randomUUID().toString();
final String localTxId_1 = UUID.randomUUID().toString();
@@ -609,6 +614,203 @@ public class SagaActorTest {
}};
}
+ // tx concurrent execution
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 3. TxStartedEvent-12
+ * 4. TxStartedEvent-13
+ * 5. TxEndedEvent-11
+ * 6. TxEndedEvent-12
+ * 7. TxEndedEvent-13
+ * 8. SagaEndedEvent-1
+ */
+ @Test
+ public void successfulTestWithTxConcurrent() throws InterruptedException {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(SagaEndedEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 3);
+ sagaData.getTxEntityMap().forEach((k, v) -> {
+ assertEquals(v.getState(), TxState.COMMITTED);
+ });
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.COMMITTED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 3. TxStartedEvent-12
+ * 5. TxEndedEvent-11
+ * 4. TxStartedEvent-13
+ * 6. TxEndedEvent-12
+ * 7. TxEndedEvent-13
+ * 8. SagaEndedEvent-1
+ */
+ @Test
+ public void successfulTestWithTxConcurrentCross() throws InterruptedException {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(SagaEndedEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 3);
+ sagaData.getTxEntityMap().forEach((k, v) -> {
+ assertEquals(v.getState(), TxState.COMMITTED);
+ });
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.COMMITTED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 4. TxStartedEvent-12
+ * 6. TxStartedEvent-13
+ * 3. TxEndedEvent-11
+ * 5. TxEndedEvent-12
+ * 7. TxAbortedEvent-13
+ * 8. TxComponsitedEvent-11
+ * 9. TxComponsitedEvent-12
+ * 10. SagaAbortedEvent-1
+ */
+ @Test
+ public void lastTxAbortedEventWithTxConcurrentTest() {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxAbortedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(SagaAbortedEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.FAILED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 3);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_2).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_3).getState(), TxState.FAILED);
+ assertEquals(sagaData.getCompensationRunningCounter().intValue(), 0);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.FAILED, SagaActorState.COMPENSATED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
private static void assertSagaTransition(PersistentFSM.Transition transition, ActorRef actorRef,
SagaActorState from, SagaActorState to) {
assertEquals(transition.fsmRef(), actorRef);
diff --git a/alpha/pom.xml b/alpha/pom.xml
index 6921f5b..3dfaf13 100644
--- a/alpha/pom.xml
+++ b/alpha/pom.xml
@@ -32,11 +32,12 @@
<packaging>pom</packaging>
<modules>
<module>alpha-core</module>
+ <module>alpha-fsm</module>
<module>alpha-spring-boot-compatibility</module>
<module>alpha-spring-cloud-starter-eureka</module>
<module>alpha-spring-cloud-starter-consul</module>
<module>alpha-spring-cloud-starter-zookeeper</module>
- <module>alpha-server</module>
+ <module>alpha-server</module>
</modules>
<build>
diff --git a/docs/fsm/assets/saga_state_diagram.png b/docs/fsm/assets/saga_state_diagram.png
index 96cd6c9..349fb72 100644
Binary files a/docs/fsm/assets/saga_state_diagram.png and b/docs/fsm/assets/saga_state_diagram.png differ
diff --git a/docs/fsm/assets/state_table.png b/docs/fsm/assets/state_table.png
index 5005597..569b2ce 100644
Binary files a/docs/fsm/assets/state_table.png and b/docs/fsm/assets/state_table.png differ
[servicecomb-pack] 01/05: SCB-1321 Add Alpha FSM prototype code
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhanglei pushed a commit to branch SCB-1321
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 98bc27bba72437e10bdc74d850559058f6441d7a
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Fri Jun 21 17:07:56 2019 +0800
SCB-1321 Add Alpha FSM prototype code
---
alpha/alpha-fsm/pom.xml | 123 ++++
.../servicecomb/pack/alpha/fsm/SagaActor.java | 331 +++++++++++
.../servicecomb/pack/alpha/fsm/SagaActorState.java | 19 +
.../apache/servicecomb/pack/alpha/fsm/TxState.java | 8 +
.../pack/alpha/fsm/event/SagaAbortedEvent.java | 28 +
.../pack/alpha/fsm/event/SagaDomainEvent.java | 14 +
.../pack/alpha/fsm/event/SagaEndedEvent.java | 28 +
.../pack/alpha/fsm/event/SagaStartedEvent.java | 43 ++
.../pack/alpha/fsm/event/SagaTimeoutEvent.java | 27 +
.../pack/alpha/fsm/event/TxAbortedEvent.java | 39 ++
.../pack/alpha/fsm/event/TxComponsitedEvent.java | 38 ++
.../pack/alpha/fsm/event/TxEndedEvent.java | 39 ++
.../pack/alpha/fsm/event/TxStartedEvent.java | 38 ++
.../pack/alpha/fsm/event/base/BaseEvent.java | 26 +
.../pack/alpha/fsm/event/base/SagaEvent.java | 14 +
.../pack/alpha/fsm/event/base/TxEvent.java | 31 ++
.../servicecomb/pack/alpha/fsm/model/SagaData.java | 131 +++++
.../servicecomb/pack/alpha/fsm/model/TxEntity.java | 95 ++++
.../servicecomb/pack/alpha/fsm/SagaActorTest.java | 619 +++++++++++++++++++++
.../alpha-fsm/src/test/resources/application.conf | 3 +
alpha/alpha-fsm/src/test/resources/log4j2.xml | 30 +
21 files changed, 1724 insertions(+)
diff --git a/alpha/alpha-fsm/pom.xml b/alpha/alpha-fsm/pom.xml
new file mode 100644
index 0000000..faa4e05
--- /dev/null
+++ b/alpha/alpha-fsm/pom.xml
@@ -0,0 +1,123 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>alpha</artifactId>
+ <groupId>org.apache.servicecomb.pack</groupId>
+ <version>0.5.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>alpha-fsm</artifactId>
+ <name>Pack::Alpha::Fsm</name>
+
+ <properties>
+ <leveldbjni-all.version>1.8</leveldbjni-all.version>
+ </properties>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>com.typesafe.akka</groupId>
+ <artifactId>akka-persistence_2.12</artifactId>
+ <version>${akka.version}</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.servicecomb.pack</groupId>
+ <artifactId>pack-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+
+ <!-- log -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>javax.persistence</groupId>
+ <artifactId>javax.persistence-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- akka -->
+ <dependency>
+ <groupId>com.typesafe.akka</groupId>
+ <artifactId>akka-actor_2.12</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.typesafe.akka</groupId>
+ <artifactId>akka-persistence_2.12</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.fusesource.leveldbjni</groupId>
+ <artifactId>leveldbjni-all</artifactId>
+ <version>${leveldbjni-all.version}</version>
+ </dependency>
+
+ <!-- For testing the artifacts scope are test-->
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.github.seanyinx</groupId>
+ <artifactId>unit-scaffolding</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-all</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.typesafe.akka</groupId>
+ <artifactId>akka-testkit_2.12</artifactId>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
new file mode 100644
index 0000000..492d187
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActor.java
@@ -0,0 +1,331 @@
+package org.apache.servicecomb.pack.alpha.fsm;
+
+
+import akka.actor.Props;
+import akka.persistence.fsm.AbstractPersistentFSM;
+import java.lang.invoke.MethodHandles;
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaAbortedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaDomainEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaDomainEvent.DomainEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaEndedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaStartedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaTimeoutEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.TxAbortedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.TxComponsitedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.TxEndedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.TxStartedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.base.BaseEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.base.SagaEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
+import org.apache.servicecomb.pack.alpha.fsm.model.SagaData;
+import org.apache.servicecomb.pack.alpha.fsm.model.TxEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.concurrent.duration.Duration;
+
+public class SagaActor extends
+ AbstractPersistentFSM<SagaActorState, SagaData, DomainEvent> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static Props props(String persistenceId) {
+ return Props.create(SagaActor.class, persistenceId);
+ }
+
+ private final String persistenceId;
+
+ public SagaActor(String persistenceId) {
+ this.persistenceId = persistenceId;
+
+ startWith(SagaActorState.IDEL, SagaData.builder().build());
+
+ when(SagaActorState.IDEL,
+ matchEvent(SagaStartedEvent.class,
+ (event, data) -> {
+ data.setGlobalTxId(event.getGlobalTxId());
+ data.setBeginTime(System.currentTimeMillis());
+ if (event.getTimeout() > 0) {
+ data.setExpirationTime(data.getBeginTime() + event.getTimeout() * 1000);
+ return goTo(SagaActorState.READY)
+ .forMax(Duration.create(data.getTimeout(), TimeUnit.MILLISECONDS));
+ } else {
+ return goTo(SagaActorState.READY);
+ }
+ }
+
+ )
+ );
+
+ when(SagaActorState.READY,
+ matchEvent(TxStartedEvent.class, SagaData.class,
+ (event, data) -> {
+ updateTxEntity(event, data);
+ if (data.getExpirationTime() > 0) {
+ return goTo(SagaActorState.PARTIALLY_ACTIVE)
+ .forMax(Duration.create(data.getTimeout(), TimeUnit.MILLISECONDS));
+ } else {
+ return goTo(SagaActorState.PARTIALLY_ACTIVE);
+ }
+ }
+ ).event(SagaEndedEvent.class,
+ (event, data) -> {
+ return goTo(SagaActorState.SUSPENDED).replying(data);
+ }
+ ).event(SagaAbortedEvent.class,
+ (event, data) -> {
+ return goTo(SagaActorState.SUSPENDED).replying(data);
+ }
+ ).event(Arrays.asList(StateTimeout()), SagaData.class,
+ (event, data) -> {
+ return goTo(SagaActorState.SUSPENDED)
+ .replying(data);
+ })
+ );
+
+ when(SagaActorState.PARTIALLY_ACTIVE,
+ matchEvent(TxEndedEvent.class, SagaData.class,
+ (event, data) -> {
+ updateTxEntity(event, data);
+ if (data.getExpirationTime() > 0) {
+ return goTo(SagaActorState.PARTIALLY_COMMITTED)
+ .forMax(Duration.create(data.getTimeout(), TimeUnit.MILLISECONDS));
+ } else {
+ return goTo(SagaActorState.PARTIALLY_COMMITTED);
+ }
+ }
+ ).event(SagaTimeoutEvent.class,
+ (event, data) -> {
+ return goTo(SagaActorState.SUSPENDED)
+ .replying(data)
+ .forMax(Duration.create(1, TimeUnit.MILLISECONDS));
+ }
+ ).event(TxAbortedEvent.class,
+ (event, data) -> {
+ updateTxEntity(event, data);
+ return goTo(SagaActorState.FAILED);
+ }
+ ).event(Arrays.asList(StateTimeout()), SagaData.class,
+ (event, data) -> {
+ return goTo(SagaActorState.SUSPENDED).replying(data);
+ })
+ );
+
+ when(SagaActorState.PARTIALLY_COMMITTED,
+ matchEvent(TxStartedEvent.class,
+ (event, data) -> {
+ updateTxEntity(event, data);
+ if (data.getExpirationTime() > 0) {
+ return goTo(SagaActorState.PARTIALLY_ACTIVE)
+ .forMax(Duration.create(data.getTimeout(), TimeUnit.MILLISECONDS));
+ } else {
+ return goTo(SagaActorState.PARTIALLY_ACTIVE);
+ }
+ }
+ ).event(SagaTimeoutEvent.class,
+ (event, data) -> {
+ return goTo(SagaActorState.SUSPENDED)
+ .replying(data)
+ .forMax(Duration.create(1, TimeUnit.MILLISECONDS));
+ }
+ ).event(SagaEndedEvent.class,
+ (event, data) -> {
+ data.setEndTime(System.currentTimeMillis());
+ return goTo(SagaActorState.COMMITTED)
+ .replying(data)
+ .forMax(Duration.create(1, TimeUnit.MILLISECONDS));
+ }
+ ).event(SagaAbortedEvent.class,
+ (event, data) -> {
+ data.setEndTime(System.currentTimeMillis());
+ updateTxEntity(event, data);
+ return goTo(SagaActorState.FAILED);
+ }
+ ).event(TxAbortedEvent.class,
+ (event, data) -> {
+ updateTxEntity(event, data);
+ return goTo(SagaActorState.FAILED);
+ }
+ ).event(Arrays.asList(StateTimeout()), SagaData.class,
+ (event, data) -> {
+ return goTo(SagaActorState.SUSPENDED).replying(data);
+ })
+ );
+
+ when(SagaActorState.FAILED,
+ matchEvent(SagaTimeoutEvent.class, SagaData.class,
+ (event, data) -> {
+ data.setEndTime(System.currentTimeMillis());
+ return goTo(SagaActorState.SUSPENDED)
+ .replying(data)
+ .forMax(Duration.create(1, TimeUnit.MILLISECONDS));
+ }
+ ).event(TxComponsitedEvent.class, SagaData.class,
+ (event, data) -> {
+ data.setEndTime(System.currentTimeMillis());
+ updateTxEntity(event, data);
+ if ((!data.isTerminated() && data.getCompensationRunningCounter().intValue() > 0)
+ || hasCommittedTx(data)) {
+ return stay();
+ } else {
+ return goTo(SagaActorState.COMPENSATED)
+ .replying(data)
+ .forMax(Duration.create(1, TimeUnit.MILLISECONDS));
+ }
+ }
+ ).event(SagaAbortedEvent.class, SagaData.class,
+ (event, data) -> {
+ data.setEndTime(System.currentTimeMillis());
+ updateTxEntity(event, data);
+ data.setTerminated(true);
+ if ((!data.isTerminated() && data.getCompensationRunningCounter().intValue() > 0)
+ || hasCommittedTx(data)) {
+ return stay();
+ }else{
+ return goTo(SagaActorState.COMPENSATED)
+ .replying(data)
+ .forMax(Duration.create(1, TimeUnit.MILLISECONDS));
+ }
+ }
+ ).event(TxStartedEvent.class, SagaData.class,
+ (event, data) -> {
+ updateTxEntity(event, data);
+ return stay();
+ }
+ ).event(TxEndedEvent.class, SagaData.class,
+ (event, data) -> {
+ updateTxEntity(event, data);
+ // TODO 调用补偿方法
+ TxEntity txEntity = data.getTxEntityMap().get(event.getLocalTxId());
+ compensation(txEntity, data);
+ return stay();
+ }
+ ).event(Arrays.asList(StateTimeout()), SagaData.class,
+ (event, data) -> {
+ return goTo(SagaActorState.SUSPENDED).replying(data);
+ })
+ );
+
+ when(SagaActorState.COMMITTED,
+ matchAnyEvent(
+ (event, data) -> {
+ return stop();
+ }
+ )
+ );
+
+ when(SagaActorState.SUSPENDED,
+ matchAnyEvent(
+ (event, data) -> {
+ return stop();
+ }
+ )
+ );
+
+ when(SagaActorState.COMPENSATED,
+ matchAnyEvent(
+ (event, data) -> {
+ return stop();
+ }
+ )
+ );
+
+ whenUnhandled(
+ matchAnyEvent((event, data) -> {
+ LOG.error("unmatch event {}", event);
+ return stay();
+ })
+ );
+
+ onTransition(
+ matchState(null, null, (from, to) -> {
+ LOG.info("transition {} {} -> {}", getSelf(), from, to);
+ })
+ );
+
+ }
+
+ @Override
+ public void onRecoveryCompleted() {
+ LOG.info("onRecoveryCompleted: {} {}", stateName(), stateData());
+ }
+
+ @Override
+ public Class domainEventClass() {
+ return SagaDomainEvent.DomainEvent.class;
+ }
+
+
+ @Override
+ public String persistenceId() {
+ return persistenceId;
+ }
+
+ @Override
+ public SagaData applyEvent(DomainEvent domainEvent, SagaData currentData) {
+ return currentData;
+ }
+
+ private void updateTxEntity(BaseEvent event, SagaData data) {
+ if (event instanceof TxEvent) {
+ TxEvent txEvent = (TxEvent) event;
+ if (!data.getTxEntityMap().containsKey(txEvent.getLocalTxId())) {
+ if (event instanceof TxStartedEvent) {
+ TxEntity txEntity = TxEntity.builder()
+ .localTxId(txEvent.getLocalTxId())
+ .parentTxId(txEvent.getParentTxId())
+ .state(TxState.ACTIVE)
+ .build();
+ data.getTxEntityMap().put(txEntity.getLocalTxId(), txEntity);
+ }
+ } else {
+ TxEntity txEntity = data.getTxEntityMap().get(txEvent.getLocalTxId());
+ if (event instanceof TxEndedEvent) {
+ if (txEntity.getState() == TxState.ACTIVE) {
+ txEntity.setEndTime(System.currentTimeMillis());
+ txEntity.setState(TxState.COMMITTED);
+ }
+ } else if (event instanceof TxAbortedEvent) {
+ if (txEntity.getState() == TxState.ACTIVE) {
+ txEntity.setEndTime(System.currentTimeMillis());
+ txEntity.setState(TxState.FAILED);
+ // TODO 调用补偿方法
+ data.getTxEntityMap().forEach((k, v) -> {
+ if (v.getState() == TxState.COMMITTED) {
+ // TODO 调用补偿方法
+ compensation(v, data);
+ }
+ });
+ }
+ } else if (event instanceof TxComponsitedEvent) {
+ //补偿中计数器减一
+ data.getCompensationRunningCounter().decrementAndGet();
+ txEntity.setState(TxState.COMPENSATED);
+ LOG.info("完成补偿 {}",txEntity.getLocalTxId());
+ }
+ }
+ } else if (event instanceof SagaEvent) {
+ if (event instanceof SagaAbortedEvent) {
+ data.getTxEntityMap().forEach((k, v) -> {
+ if (v.getState() == TxState.COMMITTED) {
+ // TODO 调用补偿方法
+ compensation(v, data);
+ }
+ });
+ }
+ }
+ }
+
+ private boolean hasCommittedTx(SagaData data) {
+ return data.getTxEntityMap().entrySet().stream()
+ .filter(map -> map.getValue().getState() == TxState.COMMITTED)
+ .count() > 0;
+ }
+
+ private void compensation(TxEntity txEntity, SagaData data) {
+ //补偿中计数器加一
+ data.getCompensationRunningCounter().incrementAndGet();
+ LOG.info("调用补偿方法 {}", txEntity.getLocalTxId());
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorState.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorState.java
new file mode 100644
index 0000000..0c7be27
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorState.java
@@ -0,0 +1,19 @@
+package org.apache.servicecomb.pack.alpha.fsm;
+
+import akka.persistence.fsm.PersistentFSM;
+
+public enum SagaActorState implements PersistentFSM.FSMState {
+ IDEL,
+ READY,
+ PARTIALLY_ACTIVE,
+ PARTIALLY_COMMITTED,
+ FAILED,
+ COMMITTED,
+ COMPENSATED,
+ SUSPENDED;
+
+ @Override
+ public String identifier() {
+ return name();
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/TxState.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/TxState.java
new file mode 100644
index 0000000..e9c54e4
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/TxState.java
@@ -0,0 +1,8 @@
+package org.apache.servicecomb.pack.alpha.fsm;
+
+public enum TxState {
+ ACTIVE,
+ FAILED,
+ COMMITTED,
+ COMPENSATED;
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaAbortedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaAbortedEvent.java
new file mode 100644
index 0000000..351146b
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaAbortedEvent.java
@@ -0,0 +1,28 @@
+package org.apache.servicecomb.pack.alpha.fsm.event;
+
+import org.apache.servicecomb.pack.alpha.fsm.event.base.SagaEvent;
+
+public class SagaAbortedEvent extends SagaEvent {
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private SagaAbortedEvent sagaAbortedEvent;
+
+ private Builder() {
+ sagaAbortedEvent = new SagaAbortedEvent();
+ }
+
+ public Builder globalTxId(String globalTxId) {
+ sagaAbortedEvent.setGlobalTxId(globalTxId);
+ return this;
+ }
+
+ public SagaAbortedEvent build() {
+ return sagaAbortedEvent;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaDomainEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaDomainEvent.java
new file mode 100644
index 0000000..b421c73
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaDomainEvent.java
@@ -0,0 +1,14 @@
+package org.apache.servicecomb.pack.alpha.fsm.event;
+
+public class SagaDomainEvent {
+ public interface DomainEvent {}
+
+ public enum SagaStartedEvent implements DomainEvent {INSTANCE}
+ public enum SagaEndedEvent implements DomainEvent {INSTANCE}
+ public enum SagaAbortedEvent implements DomainEvent {INSTANCE}
+ public enum SagaTimeoutEvent implements DomainEvent {INSTANCE}
+ public enum TxStartedEvent implements DomainEvent {INSTANCE}
+ public enum TxEndedEvent implements DomainEvent {INSTANCE}
+ public enum TxAbortedEvent implements DomainEvent {INSTANCE}
+ public enum TxComponsitedEvent implements DomainEvent {INSTANCE}
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaEndedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaEndedEvent.java
new file mode 100644
index 0000000..7a87d77
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaEndedEvent.java
@@ -0,0 +1,28 @@
+package org.apache.servicecomb.pack.alpha.fsm.event;
+
+import org.apache.servicecomb.pack.alpha.fsm.event.base.SagaEvent;
+
+public class SagaEndedEvent extends SagaEvent {
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private SagaEndedEvent sagaEndedEvent;
+
+ private Builder() {
+ sagaEndedEvent = new SagaEndedEvent();
+ }
+
+ public Builder globalTxId(String globalTxId) {
+ sagaEndedEvent.setGlobalTxId(globalTxId);
+ return this;
+ }
+
+ public SagaEndedEvent build() {
+ return sagaEndedEvent;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaStartedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaStartedEvent.java
new file mode 100644
index 0000000..4831be7
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaStartedEvent.java
@@ -0,0 +1,43 @@
+package org.apache.servicecomb.pack.alpha.fsm.event;
+
+
+import org.apache.servicecomb.pack.alpha.fsm.event.base.SagaEvent;
+
+public class SagaStartedEvent extends SagaEvent {
+ private int timeout; //second
+
+ public int getTimeout() {
+ return timeout;
+ }
+
+ public void setTimeout(int timeout) {
+ this.timeout = timeout;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private SagaStartedEvent sagaStartedEvent;
+
+ private Builder() {
+ sagaStartedEvent = new SagaStartedEvent();
+ }
+
+ public Builder globalTxId(String globalTxId) {
+ sagaStartedEvent.setGlobalTxId(globalTxId);
+ return this;
+ }
+
+ public Builder timeout(int timeout) {
+ sagaStartedEvent.setTimeout(timeout);
+ return this;
+ }
+
+ public SagaStartedEvent build() {
+ return sagaStartedEvent;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaTimeoutEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaTimeoutEvent.java
new file mode 100644
index 0000000..4de5c37
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/SagaTimeoutEvent.java
@@ -0,0 +1,27 @@
+package org.apache.servicecomb.pack.alpha.fsm.event;
+
+import org.apache.servicecomb.pack.alpha.fsm.event.base.SagaEvent;
+
+public class SagaTimeoutEvent extends SagaEvent {
+
+ public static Builder builder() {
+ return new Builder();
+ }
+ public static final class Builder {
+
+ private SagaTimeoutEvent sagaTimeoutEvent;
+
+ private Builder() {
+ sagaTimeoutEvent = new SagaTimeoutEvent();
+ }
+
+ public Builder globalTxId(String globalTxId) {
+ sagaTimeoutEvent.setGlobalTxId(globalTxId);
+ return this;
+ }
+
+ public SagaTimeoutEvent build() {
+ return sagaTimeoutEvent;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java
new file mode 100644
index 0000000..620366f
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxAbortedEvent.java
@@ -0,0 +1,39 @@
+package org.apache.servicecomb.pack.alpha.fsm.event;
+
+import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
+
+public class TxAbortedEvent extends TxEvent {
+
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private TxAbortedEvent txAbortedEvent;
+
+ private Builder() {
+ txAbortedEvent = new TxAbortedEvent();
+ }
+
+ public Builder parentTxId(String parentTxId) {
+ txAbortedEvent.setParentTxId(parentTxId);
+ return this;
+ }
+
+ public Builder localTxId(String localTxId) {
+ txAbortedEvent.setLocalTxId(localTxId);
+ return this;
+ }
+
+ public Builder globalTxId(String globalTxId) {
+ txAbortedEvent.setGlobalTxId(globalTxId);
+ return this;
+ }
+
+ public TxAbortedEvent build() {
+ return txAbortedEvent;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxComponsitedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxComponsitedEvent.java
new file mode 100644
index 0000000..cccb69a
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxComponsitedEvent.java
@@ -0,0 +1,38 @@
+package org.apache.servicecomb.pack.alpha.fsm.event;
+
+import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
+
+public class TxComponsitedEvent extends TxEvent {
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private TxComponsitedEvent txComponsitedEvent;
+
+ private Builder() {
+ txComponsitedEvent = new TxComponsitedEvent();
+ }
+
+ public Builder parentTxId(String parentTxId) {
+ txComponsitedEvent.setParentTxId(parentTxId);
+ return this;
+ }
+
+ public Builder localTxId(String localTxId) {
+ txComponsitedEvent.setLocalTxId(localTxId);
+ return this;
+ }
+
+ public Builder globalTxId(String globalTxId) {
+ txComponsitedEvent.setGlobalTxId(globalTxId);
+ return this;
+ }
+
+ public TxComponsitedEvent build() {
+ return txComponsitedEvent;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxEndedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxEndedEvent.java
new file mode 100644
index 0000000..33ea3d1
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxEndedEvent.java
@@ -0,0 +1,39 @@
+package org.apache.servicecomb.pack.alpha.fsm.event;
+
+import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
+
+public class TxEndedEvent extends TxEvent {
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+
+ public static final class Builder {
+
+ private TxEndedEvent txEndedEvent;
+
+ private Builder() {
+ txEndedEvent = new TxEndedEvent();
+ }
+
+ public Builder parentTxId(String parentTxId) {
+ txEndedEvent.setParentTxId(parentTxId);
+ return this;
+ }
+
+ public Builder localTxId(String localTxId) {
+ txEndedEvent.setLocalTxId(localTxId);
+ return this;
+ }
+
+ public Builder globalTxId(String globalTxId) {
+ txEndedEvent.setGlobalTxId(globalTxId);
+ return this;
+ }
+
+ public TxEndedEvent build() {
+ return txEndedEvent;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxStartedEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxStartedEvent.java
new file mode 100644
index 0000000..2d10462
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/TxStartedEvent.java
@@ -0,0 +1,38 @@
+package org.apache.servicecomb.pack.alpha.fsm.event;
+
+import org.apache.servicecomb.pack.alpha.fsm.event.base.TxEvent;
+
+public class TxStartedEvent extends TxEvent {
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private TxStartedEvent txStartedEvent;
+
+ private Builder() {
+ txStartedEvent = new TxStartedEvent();
+ }
+
+ public Builder parentTxId(String parentTxId) {
+ txStartedEvent.setParentTxId(parentTxId);
+ return this;
+ }
+
+ public Builder localTxId(String localTxId) {
+ txStartedEvent.setLocalTxId(localTxId);
+ return this;
+ }
+
+ public Builder globalTxId(String globalTxId) {
+ txStartedEvent.setGlobalTxId(globalTxId);
+ return this;
+ }
+
+ public TxStartedEvent build() {
+ return txStartedEvent;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/BaseEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/BaseEvent.java
new file mode 100644
index 0000000..ea8257d
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/BaseEvent.java
@@ -0,0 +1,26 @@
+package org.apache.servicecomb.pack.alpha.fsm.event.base;
+
+import java.io.Serializable;
+
+public abstract class BaseEvent implements Serializable {
+ private String globalTxId;
+
+ public BaseEvent() {
+
+ }
+
+ public String getGlobalTxId() {
+ return globalTxId;
+ }
+
+ public void setGlobalTxId(String globalTxId) {
+ this.globalTxId = globalTxId;
+ }
+
+ @Override
+ public String toString() {
+ return "BaseEvent{" +
+ "globalTxId='" + globalTxId + '\'' +
+ '}';
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/SagaEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/SagaEvent.java
new file mode 100644
index 0000000..bf39855
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/SagaEvent.java
@@ -0,0 +1,14 @@
+package org.apache.servicecomb.pack.alpha.fsm.event.base;
+
+public class SagaEvent extends BaseEvent {
+
+ public SagaEvent() {
+ }
+
+ @Override
+ public String toString() {
+ return this.getClass().getSimpleName() + "{" +
+ "globalTxId='" + this.getGlobalTxId() + '\'' +
+ '}';
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/TxEvent.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/TxEvent.java
new file mode 100644
index 0000000..56a06e0
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/event/base/TxEvent.java
@@ -0,0 +1,31 @@
+package org.apache.servicecomb.pack.alpha.fsm.event.base;
+
+public abstract class TxEvent extends BaseEvent {
+ private String parentTxId;
+ private String localTxId;
+
+ public String getParentTxId() {
+ return parentTxId;
+ }
+
+ public void setParentTxId(String parentTxId) {
+ this.parentTxId = parentTxId;
+ }
+
+ public String getLocalTxId() {
+ return localTxId;
+ }
+
+ public void setLocalTxId(String localTxId) {
+ this.localTxId = localTxId;
+ }
+
+ @Override
+ public String toString() {
+ return this.getClass().getSimpleName() + "{" +
+ "globalTxId='" + this.getGlobalTxId() + '\'' +
+ "parentTxId='" + parentTxId + '\'' +
+ ", localTxId='" + localTxId + '\'' +
+ '}';
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/SagaData.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/SagaData.java
new file mode 100644
index 0000000..0cc858a
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/SagaData.java
@@ -0,0 +1,131 @@
+package org.apache.servicecomb.pack.alpha.fsm.model;
+
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class SagaData implements Serializable {
+ private long beginTime = System.currentTimeMillis();
+ private long endTime;
+ private String globalTxId;
+ private long expirationTime;
+ private boolean terminated;
+ private AtomicLong compensationRunningCounter = new AtomicLong();
+ private Map<String,TxEntity> txEntityMap = new HashMap<>();
+
+ public long getBeginTime() {
+ return beginTime;
+ }
+
+ public void setBeginTime(long beginTime) {
+ this.beginTime = beginTime;
+ }
+
+ public long getEndTime() {
+ return endTime;
+ }
+
+ public void setEndTime(long endTime) {
+ this.endTime = endTime;
+ }
+
+ public String getGlobalTxId() {
+ return globalTxId;
+ }
+
+ public void setGlobalTxId(String globalTxId) {
+ this.globalTxId = globalTxId;
+ }
+
+ public long getExpirationTime() {
+ return expirationTime;
+ }
+
+ public void setExpirationTime(long expirationTime) {
+ this.expirationTime = expirationTime;
+ }
+
+ public boolean isTerminated() {
+ return terminated;
+ }
+
+ public void setTerminated(boolean terminated) {
+ this.terminated = terminated;
+ }
+
+ public AtomicLong getCompensationRunningCounter() {
+ return compensationRunningCounter;
+ }
+
+ public void setCompensationRunningCounter(
+ AtomicLong compensationRunningCounter) {
+ this.compensationRunningCounter = compensationRunningCounter;
+ }
+
+ public Map<String, TxEntity> getTxEntityMap() {
+ return txEntityMap;
+ }
+
+ public void setTxEntityMap(
+ Map<String, TxEntity> txEntityMap) {
+ this.txEntityMap = txEntityMap;
+ }
+
+ public long getTimeout(){
+ return expirationTime-System.currentTimeMillis();
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private SagaData sagaData;
+
+ private Builder() {
+ sagaData = new SagaData();
+ }
+
+ public Builder beginTime(long beginTime) {
+ sagaData.setBeginTime(beginTime);
+ return this;
+ }
+
+ public Builder endTime(long endTime) {
+ sagaData.setEndTime(endTime);
+ return this;
+ }
+
+ public Builder globalTxId(String globalTxId) {
+ sagaData.setGlobalTxId(globalTxId);
+ return this;
+ }
+
+ public Builder expirationTime(long expirationTime) {
+ sagaData.setExpirationTime(expirationTime);
+ return this;
+ }
+
+ public Builder terminated(boolean terminated) {
+ sagaData.setTerminated(terminated);
+ return this;
+ }
+
+ public Builder compensationRunningCounter(AtomicLong compensationRunningCounter) {
+ sagaData.setCompensationRunningCounter(compensationRunningCounter);
+ return this;
+ }
+
+ public Builder txEntityMap(Map<String, TxEntity> txEntityMap) {
+ sagaData.setTxEntityMap(txEntityMap);
+ return this;
+ }
+
+ public SagaData build() {
+ return sagaData;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java
new file mode 100644
index 0000000..f6a3222
--- /dev/null
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/model/TxEntity.java
@@ -0,0 +1,95 @@
+package org.apache.servicecomb.pack.alpha.fsm.model;
+
+
+import java.io.Serializable;
+import org.apache.servicecomb.pack.alpha.fsm.TxState;
+
+public class TxEntity implements Serializable {
+ private long beginTime = System.currentTimeMillis();
+ private long endTime;
+ private String parentTxId;
+ private String localTxId;
+ private TxState state;
+
+ public long getBeginTime() {
+ return beginTime;
+ }
+
+ public void setBeginTime(long beginTime) {
+ this.beginTime = beginTime;
+ }
+
+ public long getEndTime() {
+ return endTime;
+ }
+
+ public void setEndTime(long endTime) {
+ this.endTime = endTime;
+ }
+
+ public String getParentTxId() {
+ return parentTxId;
+ }
+
+ public void setParentTxId(String parentTxId) {
+ this.parentTxId = parentTxId;
+ }
+
+ public String getLocalTxId() {
+ return localTxId;
+ }
+
+ public void setLocalTxId(String localTxId) {
+ this.localTxId = localTxId;
+ }
+
+ public TxState getState() {
+ return state;
+ }
+
+ public void setState(TxState state) {
+ this.state = state;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private TxEntity txEntity;
+
+ private Builder() {
+ txEntity = new TxEntity();
+ }
+
+ public Builder beginTime(long beginTime) {
+ txEntity.setBeginTime(beginTime);
+ return this;
+ }
+
+ public Builder endTime(long endTime) {
+ txEntity.setEndTime(endTime);
+ return this;
+ }
+
+ public Builder parentTxId(String parentTxId) {
+ txEntity.setParentTxId(parentTxId);
+ return this;
+ }
+
+ public Builder localTxId(String localTxId) {
+ txEntity.setLocalTxId(localTxId);
+ return this;
+ }
+
+ public Builder state(TxState state) {
+ txEntity.setState(state);
+ return this;
+ }
+
+ public TxEntity build() {
+ return txEntity;
+ }
+ }
+}
diff --git a/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java
new file mode 100644
index 0000000..156d67e
--- /dev/null
+++ b/alpha/alpha-fsm/src/test/java/org/apache/servicecomb/pack/alpha/fsm/SagaActorTest.java
@@ -0,0 +1,619 @@
+package org.apache.servicecomb.pack.alpha.fsm;
+
+import static org.junit.Assert.assertEquals;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Terminated;
+import akka.persistence.fsm.PersistentFSM;
+import akka.persistence.fsm.PersistentFSM.CurrentState;
+import akka.testkit.javadsl.TestKit;
+import java.time.Duration;
+import java.util.UUID;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaAbortedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaEndedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaStartedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.SagaTimeoutEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.TxAbortedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.TxComponsitedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.TxEndedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.event.TxStartedEvent;
+import org.apache.servicecomb.pack.alpha.fsm.model.SagaData;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class SagaActorTest {
+
+ static ActorSystem system;
+
+ @BeforeClass
+ public static void setup() {
+ system = ActorSystem.create("SagaActorTest");
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ TestKit.shutdownActorSystem(system);
+ system = null;
+ }
+
+ public String genPersistenceId() {
+ return UUID.randomUUID().toString();
+ }
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 3. TxEndedEvent-11
+ * 4. TxStartedEvent-12
+ * 5. TxEndedEvent-12
+ * 4. TxStartedEvent-13
+ * 5. TxEndedEvent-13
+ * 6. SagaEndedEvent-1
+ */
+ @Test
+ public void successfulTest() {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(SagaEndedEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 3);
+ sagaData.getTxEntityMap().forEach((k, v) -> {
+ assertEquals(v.getState(), TxState.COMMITTED);
+ });
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.COMMITTED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 3. TxAbortedEvent-11
+ * 7. SagaAbortedEvent-1
+ */
+ @Test
+ public void firstTxAbortedEventTest() {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxAbortedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(SagaAbortedEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.FAILED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 1);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.FAILED);
+ assertEquals(sagaData.getCompensationRunningCounter().intValue(), 0);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.FAILED, SagaActorState.COMPENSATED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 3. TxEndedEvent-11
+ * 4. TxStartedEvent-12
+ * 5. TxAbortedEvent-12
+ * 6. TxComponsitedEvent-11
+ * 7. SagaAbortedEvent-1
+ */
+ @Test
+ public void middleTxAbortedEventTest() {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxAbortedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(SagaAbortedEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.FAILED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 2);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_2).getState(), TxState.FAILED);
+ assertEquals(sagaData.getCompensationRunningCounter().intValue(), 0);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.FAILED, SagaActorState.COMPENSATED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 3. TxEndedEvent-11
+ * 4. TxStartedEvent-12
+ * 5. TxEndedEvent-12
+ * 6. TxStartedEvent-13
+ * 7. TxAbortedEvent-13
+ * 8. TxComponsitedEvent-11
+ * 9. TxComponsitedEvent-12
+ * 10. SagaAbortedEvent-1
+ */
+ @Test
+ public void lastTxAbortedEventTest() {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxAbortedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(SagaAbortedEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.FAILED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 3);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_2).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_3).getState(), TxState.FAILED);
+ assertEquals(sagaData.getCompensationRunningCounter().intValue(), 0);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.FAILED, SagaActorState.COMPENSATED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 3. TxAbortedEvent-11
+ * 4. TxStartedEvent-12
+ * 5. TxEndedEvent-12
+ * 6. TxStartedEvent-13
+ * 7. TxEndedEvent-13
+ * 8. TxComponsitedEvent-12
+ * 9. TxComponsitedEvent-13
+ * 10. SagaAbortedEvent-1
+ */
+ @Test
+ public void receivedRemainingEventAfterfirstTxAbortedEventTest() {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxAbortedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(SagaAbortedEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.FAILED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 3);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_2).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_3).getState(), TxState.FAILED);
+ assertEquals(sagaData.getCompensationRunningCounter().intValue(), 0);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.FAILED, SagaActorState.COMPENSATED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 3. TxEndedEvent-11
+ * 4. TxStartedEvent-12
+ * 5. TxEndedEvent-12
+ * 6. TxStartedEvent-13
+ * 7. TxEndedEvent-13
+ * 8. SagaAbortedEvent-1
+ * 9. TxComponsitedEvent-11
+ * 8. TxComponsitedEvent-12
+ * 9. TxComponsitedEvent-13
+ */
+ @Test
+ public void sagaAbortedEventAfterAllTxEndedTest() {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(SagaAbortedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxComponsitedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.FAILED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 3);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_2).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_3).getState(), TxState.COMPENSATED);
+ assertEquals(sagaData.getCompensationRunningCounter().intValue(), 0);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.FAILED, SagaActorState.COMPENSATED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+
+ /**
+ * 1. SagaStartedEvent-1
+ * 2. TxStartedEvent-11
+ * 3. TxEndedEvent-11
+ * 4. TxStartedEvent-12
+ * 5. TxEndedEvent-12
+ * 4. TxStartedEvent-13
+ * 5. TxEndedEvent-13
+ * 5. SagaTimeoutEvent-1
+ */
+ @Test
+ public void omegaSendSagaTimeoutEventTest() {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(SagaTimeoutEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ SagaData sagaData = expectMsgClass(SagaData.class);
+ assertEquals(sagaData.getGlobalTxId(), globalTxId);
+ assertEquals(sagaData.getTxEntityMap().size(), 3);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_1).getState(), TxState.COMMITTED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_2).getState(), TxState.COMMITTED);
+ assertEquals(sagaData.getTxEntityMap().get(localTxId_3).getState(), TxState.COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.SUSPENDED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+ /**
+ * 1. SagaStartedEvent(5s)-1
+ * 2. TxStartedEvent-11
+ * 3. TxEndedEvent-11
+ * 4. TxStartedEvent-12
+ * 5. TxEndedEvent-12
+ * 4. TxStartedEvent-13
+ * 5. TxEndedEvent-13
+ * 6. SagaEndedEvent-1
+ */
+ @Test
+ public void sagaActorTriggerTimeoutTest() throws InterruptedException {
+ new TestKit(system) {{
+ final String globalTxId = UUID.randomUUID().toString();
+ final String localTxId_1 = UUID.randomUUID().toString();
+ final String localTxId_2 = UUID.randomUUID().toString();
+ final String localTxId_3 = UUID.randomUUID().toString();
+ final int timeout = 5;
+
+ ActorRef saga = system.actorOf(SagaActor.props(genPersistenceId()), "saga");
+ watch(saga);
+ saga.tell(new PersistentFSM.SubscribeTransitionCallBack(getRef()), getRef());
+ saga.tell(SagaStartedEvent.builder().globalTxId(globalTxId).timeout(timeout).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_1).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_2).build(), getRef());
+ saga.tell(TxStartedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ saga.tell(TxEndedEvent.builder().globalTxId(globalTxId).parentTxId(globalTxId).localTxId(localTxId_3).build(), getRef());
+ Thread.sleep(timeout*2000);
+ saga.tell(SagaEndedEvent.builder().globalTxId(globalTxId).build(), getRef());
+
+ //expect
+ CurrentState currentState = expectMsgClass(PersistentFSM.CurrentState.class);
+ assertEquals(SagaActorState.IDEL, currentState.state());
+ PersistentFSM.Transition transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.IDEL, SagaActorState.READY);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.READY, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.PARTIALLY_ACTIVE);
+
+ transition = expectMsgClass(PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_ACTIVE, SagaActorState.PARTIALLY_COMMITTED);
+
+ transition = expectMsgClass(Duration.ofSeconds(timeout+2),PersistentFSM.Transition.class);
+ assertSagaTransition(transition, saga, SagaActorState.PARTIALLY_COMMITTED, SagaActorState.SUSPENDED);
+
+ Terminated terminated = expectMsgClass(Terminated.class);
+ assertEquals(terminated.getActor(), saga);
+
+ system.stop(saga);
+ }};
+ }
+
+ private static void assertSagaTransition(PersistentFSM.Transition transition, ActorRef actorRef,
+ SagaActorState from, SagaActorState to) {
+ assertEquals(transition.fsmRef(), actorRef);
+ assertEquals(transition.from(), from);
+ assertEquals(transition.to(), to);
+ }
+
+}
diff --git a/alpha/alpha-fsm/src/test/resources/application.conf b/alpha/alpha-fsm/src/test/resources/application.conf
new file mode 100644
index 0000000..1265036
--- /dev/null
+++ b/alpha/alpha-fsm/src/test/resources/application.conf
@@ -0,0 +1,3 @@
+akka.persistence.journal.plugin = "akka.persistence.journal.inmem"
+akka.persistence.snapshot-store.plugin = "akka.persistence.snapshot-store.local"
+akka.persistence.snapshot-store.local.dir = "target/example/snapshots"
\ No newline at end of file
diff --git a/alpha/alpha-fsm/src/test/resources/log4j2.xml b/alpha/alpha-fsm/src/test/resources/log4j2.xml
new file mode 100644
index 0000000..58924c6
--- /dev/null
+++ b/alpha/alpha-fsm/src/test/resources/log4j2.xml
@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+
+<Configuration status="WARN">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Root level="info">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</Configuration>