From 9aa8b74e5ced030319590bc89f631e2351a130cf Mon Sep 17 00:00:00 2001 From: pigxcloud <2270033969@qq.com> Date: Wed, 5 Jun 2019 17:37:05 +0800 Subject: [PATCH] :bookmark: Releasing / Version tags. 3.1.0 --- pigx-visual/pigx-seata-server/pom.xml | 48 -- .../server/AbstractTCInboundHandler.java | 240 ------- .../src/main/java/io/seata/server/Server.java | 78 --- .../java/io/seata/server/UUIDGenerator.java | 96 --- .../io/seata/server/coordinator/Core.java | 55 -- .../seata/server/coordinator/CoreFactory.java | 46 -- .../coordinator/DefaultCoordinator.java | 362 ----------- .../seata/server/coordinator/DefaultCore.java | 310 --------- .../server/lock/DefaultLockManagerImpl.java | 186 ------ .../io/seata/server/lock/LockManager.java | 63 -- .../seata/server/lock/LockManagerFactory.java | 46 -- .../session/AbstractSessionManager.java | 191 ------ .../seata/server/session/BranchSession.java | 395 ------------ .../server/session/DefaultSessionManager.java | 65 -- .../session/FileBasedSessionManager.java | 232 ------- .../seata/server/session/GlobalSession.java | 541 ---------------- .../io/seata/server/session/Lockable.java | 42 -- .../io/seata/server/session/Reloadable.java | 29 - .../server/session/SessionCondition.java | 55 -- .../seata/server/session/SessionHelper.java | 121 ---- .../seata/server/session/SessionHolder.java | 242 ------- .../server/session/SessionLifecycle.java | 89 --- .../session/SessionLifecycleListener.java | 90 --- .../seata/server/session/SessionManager.java | 109 ---- .../store/FileTransactionStoreManager.java | 590 ------------------ .../io/seata/server/store/FlushDiskMode.java | 44 -- .../seata/server/store/SessionStorable.java | 38 -- .../io/seata/server/store/StoreConfig.java | 56 -- .../server/store/TransactionStoreManager.java | 118 ---- .../server/store/TransactionWriteStore.java | 129 ---- .../src/main/resources/logback.xml | 54 -- .../src/main/resources/registry.conf | 71 --- pigx-visual/pom.xml | 1 - 33 files changed, 4832 deletions(-) delete mode 100755 pigx-visual/pigx-seata-server/pom.xml delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/AbstractTCInboundHandler.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/Server.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/UUIDGenerator.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/Core.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/CoreFactory.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCore.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/DefaultLockManagerImpl.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManager.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManagerFactory.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/AbstractSessionManager.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/BranchSession.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/DefaultSessionManager.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/FileBasedSessionManager.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/GlobalSession.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Lockable.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Reloadable.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionCondition.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHelper.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHolder.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycle.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycleListener.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionManager.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FileTransactionStoreManager.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FlushDiskMode.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/SessionStorable.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/StoreConfig.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionStoreManager.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionWriteStore.java delete mode 100755 pigx-visual/pigx-seata-server/src/main/resources/logback.xml delete mode 100755 pigx-visual/pigx-seata-server/src/main/resources/registry.conf diff --git a/pigx-visual/pigx-seata-server/pom.xml b/pigx-visual/pigx-seata-server/pom.xml deleted file mode 100755 index ce385ff70..000000000 --- a/pigx-visual/pigx-seata-server/pom.xml +++ /dev/null @@ -1,48 +0,0 @@ - - - - - com.pig4cloud - pigx-visual - 3.1.0 - - - 4.0.0 - pigx-seata-server - jar - pigx-seata-server - - - - io.seata - seata-core - 0.5.2 - - - io.seata - seata-config-all - 0.5.2 - - - io.seata - seata-discovery-all - 0.5.2 - - - diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/AbstractTCInboundHandler.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/AbstractTCInboundHandler.java deleted file mode 100755 index d6655196e..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/AbstractTCInboundHandler.java +++ /dev/null @@ -1,240 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server; - -import io.seata.common.XID; -import io.seata.core.exception.AbstractExceptionHandler; -import io.seata.core.exception.TransactionException; -import io.seata.core.model.GlobalStatus; -import io.seata.core.protocol.transaction.BranchRegisterRequest; -import io.seata.core.protocol.transaction.BranchRegisterResponse; -import io.seata.core.protocol.transaction.BranchReportRequest; -import io.seata.core.protocol.transaction.BranchReportResponse; -import io.seata.core.protocol.transaction.GlobalBeginRequest; -import io.seata.core.protocol.transaction.GlobalBeginResponse; -import io.seata.core.protocol.transaction.GlobalCommitRequest; -import io.seata.core.protocol.transaction.GlobalCommitResponse; -import io.seata.core.protocol.transaction.GlobalLockQueryRequest; -import io.seata.core.protocol.transaction.GlobalLockQueryResponse; -import io.seata.core.protocol.transaction.GlobalRollbackRequest; -import io.seata.core.protocol.transaction.GlobalRollbackResponse; -import io.seata.core.protocol.transaction.GlobalStatusRequest; -import io.seata.core.protocol.transaction.GlobalStatusResponse; -import io.seata.core.protocol.transaction.TCInboundHandler; -import io.seata.core.rpc.RpcContext; -import io.seata.server.session.GlobalSession; -import io.seata.server.session.SessionHolder; - -/** - * The type Abstract tc inbound handler. - * - * @author sharajava - */ -public abstract class AbstractTCInboundHandler extends AbstractExceptionHandler implements TCInboundHandler { - - @Override - public GlobalBeginResponse handle(GlobalBeginRequest request, final RpcContext rpcContext) { - GlobalBeginResponse response = new GlobalBeginResponse(); - exceptionHandleTemplate(new AbstractCallback() { - @Override - public void execute(GlobalBeginRequest request, GlobalBeginResponse response) throws TransactionException { - doGlobalBegin(request, response, rpcContext); - } - }, request, response); - return response; - } - - /** - * Do global begin. - * - * @param request the request - * @param response the response - * @param rpcContext the rpc context - * @throws TransactionException the transaction exception - */ - protected abstract void doGlobalBegin(GlobalBeginRequest request, GlobalBeginResponse response, - RpcContext rpcContext) throws TransactionException; - - @Override - public GlobalCommitResponse handle(GlobalCommitRequest request, final RpcContext rpcContext) { - GlobalCommitResponse response = new GlobalCommitResponse(); - exceptionHandleTemplate(new AbstractCallback() { - @Override - public void execute(GlobalCommitRequest request, GlobalCommitResponse response) - throws TransactionException { - doGlobalCommit(request, response, rpcContext); - } - }, request, response); - return response; - } - - /** - * Do global commit. - * - * @param request the request - * @param response the response - * @param rpcContext the rpc context - * @throws TransactionException the transaction exception - */ - protected abstract void doGlobalCommit(GlobalCommitRequest request, GlobalCommitResponse response, - RpcContext rpcContext) throws TransactionException; - - @Override - public GlobalRollbackResponse handle(GlobalRollbackRequest request, final RpcContext rpcContext) { - GlobalRollbackResponse response = new GlobalRollbackResponse(); - exceptionHandleTemplate(new AbstractCallback() { - @Override - public void execute(GlobalRollbackRequest request, GlobalRollbackResponse response) - throws TransactionException { - doGlobalRollback(request, response, rpcContext); - } - - @Override - public void onTransactionException(GlobalRollbackRequest request, GlobalRollbackResponse response, - TransactionException tex) { - super.onTransactionException(request, response, tex); - GlobalSession globalSession = SessionHolder.findGlobalSession( - XID.getTransactionId(request.getXid())); - if (globalSession != null) { - response.setGlobalStatus(globalSession.getStatus()); - } else { - response.setGlobalStatus(GlobalStatus.Finished); - } - } - - @Override - public void onException(GlobalRollbackRequest request, GlobalRollbackResponse response, Exception rex) { - super.onException(request, response, rex); - GlobalSession globalSession = SessionHolder.findGlobalSession( - XID.getTransactionId(request.getXid())); - if (globalSession != null) { - response.setGlobalStatus(globalSession.getStatus()); - } else { - response.setGlobalStatus(GlobalStatus.Finished); - } - } - }, request, response); - return response; - } - - /** - * Do global rollback. - * - * @param request the request - * @param response the response - * @param rpcContext the rpc context - * @throws TransactionException the transaction exception - */ - protected abstract void doGlobalRollback(GlobalRollbackRequest request, GlobalRollbackResponse response, - RpcContext rpcContext) throws TransactionException; - - @Override - public BranchRegisterResponse handle(BranchRegisterRequest request, final RpcContext rpcContext) { - BranchRegisterResponse response = new BranchRegisterResponse(); - exceptionHandleTemplate(new AbstractCallback() { - @Override - public void execute(BranchRegisterRequest request, BranchRegisterResponse response) - throws TransactionException { - doBranchRegister(request, response, rpcContext); - } - }, request, response); - return response; - } - - /** - * Do branch register. - * - * @param request the request - * @param response the response - * @param rpcContext the rpc context - * @throws TransactionException the transaction exception - */ - protected abstract void doBranchRegister(BranchRegisterRequest request, BranchRegisterResponse response, - RpcContext rpcContext) throws TransactionException; - - @Override - public BranchReportResponse handle(BranchReportRequest request, final RpcContext rpcContext) { - BranchReportResponse response = new BranchReportResponse(); - exceptionHandleTemplate(new AbstractCallback() { - @Override - public void execute(BranchReportRequest request, BranchReportResponse response) - throws TransactionException { - doBranchReport(request, response, rpcContext); - } - }, request, response); - return response; - } - - /** - * Do branch report. - * - * @param request the request - * @param rpcContext the rpc context - * @throws TransactionException the transaction exception - */ - protected abstract void doBranchReport(BranchReportRequest request, BranchReportResponse response, - RpcContext rpcContext) - throws TransactionException; - - @Override - public GlobalLockQueryResponse handle(GlobalLockQueryRequest request, final RpcContext rpcContext) { - GlobalLockQueryResponse response = new GlobalLockQueryResponse(); - exceptionHandleTemplate(new AbstractCallback() { - @Override - public void execute(GlobalLockQueryRequest request, GlobalLockQueryResponse response) - throws TransactionException { - doLockCheck(request, response, rpcContext); - } - }, request, response); - return response; - } - - /** - * Do lock check. - * - * @param request the request - * @param response the response - * @param rpcContext the rpc context - * @throws TransactionException the transaction exception - */ - protected abstract void doLockCheck(GlobalLockQueryRequest request, GlobalLockQueryResponse response, - RpcContext rpcContext) throws TransactionException; - - @Override - public GlobalStatusResponse handle(GlobalStatusRequest request, final RpcContext rpcContext) { - GlobalStatusResponse response = new GlobalStatusResponse(); - exceptionHandleTemplate(new AbstractCallback() { - @Override - public void execute(GlobalStatusRequest request, GlobalStatusResponse response) - throws TransactionException { - doGlobalStatus(request, response, rpcContext); - } - }, request, response); - return response; - } - - /** - * Do global status. - * - * @param request the request - * @param response the response - * @param rpcContext the rpc context - * @throws TransactionException the transaction exception - */ - protected abstract void doGlobalStatus(GlobalStatusRequest request, GlobalStatusResponse response, - RpcContext rpcContext) throws TransactionException; - -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/Server.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/Server.java deleted file mode 100755 index 43df651ee..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/Server.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server; - -import io.seata.common.XID; -import io.seata.common.thread.NamedThreadFactory; -import io.seata.common.util.NetUtil; -import io.seata.core.rpc.netty.RpcServer; -import io.seata.core.rpc.netty.ShutdownHook; -import io.seata.server.coordinator.DefaultCoordinator; -import io.seata.server.session.SessionHolder; - -import java.io.IOException; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; - -/** - * The type Server. - * - * @author jimin.jm @alibaba-inc.com - */ -public class Server { - - private static final int MIN_SERVER_POOL_SIZE = 10; - private static final int MAX_SERVER_POOL_SIZE = 50; - private static final int MAX_TASK_QUEUE_SIZE = 2000; - private static final int KEEP_ALIVE_TIME = 50; - private static final int SERVER_DEFAULT_PORT = 8091; - private static final ThreadPoolExecutor WORKING_THREADS = new ThreadPoolExecutor(MIN_SERVER_POOL_SIZE, - MAX_SERVER_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, - new LinkedBlockingQueue(MAX_TASK_QUEUE_SIZE), - new NamedThreadFactory("ServerHandlerThread", MAX_SERVER_POOL_SIZE), new ThreadPoolExecutor.CallerRunsPolicy()); - - /** - * The entry point of application. - * - * @param args the input arguments - * @throws IOException the io exception - */ - public static void main(String[] args) throws IOException { - RpcServer rpcServer = new RpcServer(WORKING_THREADS); - - rpcServer.setListenPort(SERVER_DEFAULT_PORT); - - //log store mode : file、db - String storeMode = "file"; - - UUIDGenerator.init(1); - SessionHolder.init(storeMode); - - DefaultCoordinator coordinator = new DefaultCoordinator(rpcServer); - coordinator.init(); - rpcServer.setHandler(coordinator); - // register ShutdownHook - ShutdownHook.getInstance().addDisposable(coordinator); - - XID.setIpAddress(NetUtil.getLocalIp()); - XID.setPort(rpcServer.getListenPort()); - - rpcServer.init(); - - System.exit(0); - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/UUIDGenerator.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/UUIDGenerator.java deleted file mode 100755 index efc55e555..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/UUIDGenerator.java +++ /dev/null @@ -1,96 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server; - -import java.text.ParseException; -import java.text.SimpleDateFormat; -import java.util.Calendar; -import java.util.Date; -import java.util.concurrent.atomic.AtomicLong; - -import io.seata.common.exception.ShouldNeverHappenException; - -/** - * The type Uuid generator. - * - * @author sharajava - */ -public class UUIDGenerator { - - private static final AtomicLong UUID = new AtomicLong(1000); - private static int serverNodeId = 1; - private static final long UUID_INTERNAL = 2000000000; - - /** - * Generate uuid long. - * - * @return the long - */ - public static long generateUUID() { - long id = UUID.incrementAndGet(); - if (id >= UUID_INTERNAL * (serverNodeId + 1)) { - synchronized (UUID) { - if (UUID.get() >= id) { - id -= UUID_INTERNAL; - UUID.set(id); - } - } - } - return id; - } - - /** - * Gets current uuid. - * - * @return the current uuid - */ - public static long getCurrentUUID() { - return UUID.get(); - } - - /** - * Sets uuid. - * - * @param expect the expect - * @param update the update - * @return the uuid - */ - public static boolean setUUID(long expect, long update) { - return UUID.compareAndSet(expect, update); - - } - - /** - * Init. - * - * @param serverNodeId the server node id - */ - public static void init(int serverNodeId) { - try { - UUIDGenerator.serverNodeId = serverNodeId; - UUID.set(UUID_INTERNAL * serverNodeId); - SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd"); - Calendar cal = Calendar.getInstance(); - Date date = format.parse("2019-01-01"); - cal.setTime(date); - long base = cal.getTimeInMillis(); - long current = System.currentTimeMillis(); - UUID.addAndGet((current - base) / 1000); - } catch (ParseException e) { - throw new ShouldNeverHappenException(e); - } - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/Core.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/Core.java deleted file mode 100755 index 7d34dfee1..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/Core.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.coordinator; - -import io.seata.core.exception.TransactionException; -import io.seata.core.model.ResourceManagerInbound; -import io.seata.core.model.ResourceManagerOutbound; -import io.seata.core.model.TransactionManager; -import io.seata.server.session.GlobalSession; - -/** - * The interface Core. - * - * @author sharajava - */ -public interface Core extends TransactionManager, ResourceManagerOutbound { - - /** - * Sets resource manager inbound. - * - * @param resourceManagerInbound the resource manager inbound - */ - void setResourceManagerInbound(ResourceManagerInbound resourceManagerInbound); - - /** - * Do global commit. - * - * @param globalSession the global session - * @param retrying the retrying - * @throws TransactionException the transaction exception - */ - void doGlobalCommit(GlobalSession globalSession, boolean retrying) throws TransactionException; - - /** - * Do global rollback. - * - * @param globalSession the global session - * @param retrying the retrying - * @throws TransactionException the transaction exception - */ - void doGlobalRollback(GlobalSession globalSession, boolean retrying) throws TransactionException; -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/CoreFactory.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/CoreFactory.java deleted file mode 100755 index 19e09f70e..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/CoreFactory.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.coordinator; - -/** - * The type Core factory. - * - * @author sharajava - */ -public class CoreFactory { - - private static class SingletonHolder { - private static Core INSTANCE = new DefaultCore(); - } - - /** - * Get core. - * - * @return the core - */ - public static final Core get() { - return SingletonHolder.INSTANCE; - } - - /** - * Just for test mocking - * - * @param core the core - */ - public static void set(Core core) { - SingletonHolder.INSTANCE = core; - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java deleted file mode 100755 index 024f741a9..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java +++ /dev/null @@ -1,362 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.coordinator; - -import io.seata.common.XID; -import io.seata.common.thread.NamedThreadFactory; -import io.seata.core.exception.TransactionException; -import io.seata.core.model.BranchStatus; -import io.seata.core.model.BranchType; -import io.seata.core.model.GlobalStatus; -import io.seata.core.model.ResourceManagerInbound; -import io.seata.core.protocol.AbstractMessage; -import io.seata.core.protocol.AbstractResultMessage; -import io.seata.core.protocol.transaction.*; -import io.seata.core.rpc.Disposable; -import io.seata.core.rpc.RpcContext; -import io.seata.core.rpc.ServerMessageSender; -import io.seata.core.rpc.TransactionMessageHandler; -import io.seata.core.rpc.netty.RpcServer; -import io.seata.server.AbstractTCInboundHandler; -import io.seata.server.session.BranchSession; -import io.seata.server.session.GlobalSession; -import io.seata.server.session.SessionHolder; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.util.Collection; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; - -import static io.seata.core.exception.TransactionExceptionCode.FailedToSendBranchCommitRequest; -import static io.seata.core.exception.TransactionExceptionCode.FailedToSendBranchRollbackRequest; - -/** - * The type Default coordinator. - * - * @author sharajava - */ -public class DefaultCoordinator extends AbstractTCInboundHandler - implements TransactionMessageHandler, ResourceManagerInbound, Disposable { - - private static final Logger LOGGER = LoggerFactory.getLogger(DefaultCoordinator.class); - - private static final int TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS = 5000; - - private ServerMessageSender messageSender; - - private Core core = CoreFactory.get(); - - /** - * Instantiates a new Default coordinator. - * - * @param messageSender the message sender - */ - public DefaultCoordinator(ServerMessageSender messageSender) { - this.messageSender = messageSender; - core.setResourceManagerInbound(this); - } - - @Override - protected void doGlobalBegin(GlobalBeginRequest request, GlobalBeginResponse response, RpcContext rpcContext) - throws TransactionException { - response.setXid(core.begin(rpcContext.getApplicationId(), rpcContext.getTransactionServiceGroup(), - request.getTransactionName(), request.getTimeout())); - } - - @Override - protected void doGlobalCommit(GlobalCommitRequest request, GlobalCommitResponse response, RpcContext rpcContext) - throws TransactionException { - response.setGlobalStatus(core.commit(request.getXid())); - - } - - @Override - protected void doGlobalRollback(GlobalRollbackRequest request, GlobalRollbackResponse response, - RpcContext rpcContext) throws TransactionException { - response.setGlobalStatus(core.rollback(request.getXid())); - - } - - @Override - protected void doGlobalStatus(GlobalStatusRequest request, GlobalStatusResponse response, RpcContext rpcContext) - throws TransactionException { - response.setGlobalStatus(core.getStatus(request.getXid())); - } - - @Override - protected void doBranchRegister(BranchRegisterRequest request, BranchRegisterResponse response, - RpcContext rpcContext) throws TransactionException { - response.setBranchId( - core.branchRegister(request.getBranchType(), request.getResourceId(), rpcContext.getClientId(), - request.getXid(), request.getApplicationData(), request.getLockKey())); - - } - - @Override - protected void doBranchReport(BranchReportRequest request, BranchReportResponse response, RpcContext rpcContext) - throws TransactionException { - core.branchReport(request.getBranchType(), request.getXid(), request.getBranchId(), - request.getStatus(), - request.getApplicationData()); - - } - - @Override - protected void doLockCheck(GlobalLockQueryRequest request, GlobalLockQueryResponse response, RpcContext rpcContext) - throws TransactionException { - response.setLockable(core.lockQuery(request.getBranchType(), request.getResourceId(), - request.getXid(), request.getLockKey())); - } - - @Override - public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId, - String applicationData) - throws TransactionException { - try { - BranchCommitRequest request = new BranchCommitRequest(); - request.setXid(xid); - request.setBranchId(branchId); - request.setResourceId(resourceId); - request.setApplicationData(applicationData); - request.setBranchType(branchType); - - GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); - if(globalSession == null){ - return BranchStatus.PhaseTwo_Committed; - } - BranchSession branchSession = globalSession.getBranch(branchId); - - BranchCommitResponse response = (BranchCommitResponse)messageSender.sendSyncRequest(resourceId, - branchSession.getClientId(), request); - return response.getBranchStatus(); - } catch (IOException e) { - throw new TransactionException(FailedToSendBranchCommitRequest, branchId + "/" + xid, e); - } catch (TimeoutException e) { - throw new TransactionException(FailedToSendBranchCommitRequest, branchId + "/" + xid, e); - } - } - - @Override - public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId, - String applicationData) - throws TransactionException { - try { - BranchRollbackRequest - request = new BranchRollbackRequest(); - request.setXid(xid); - request.setBranchId(branchId); - request.setResourceId(resourceId); - request.setApplicationData(applicationData); - request.setBranchType(branchType); - - GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); - if(globalSession == null){ - return BranchStatus.PhaseTwo_Rollbacked; - } - BranchSession branchSession = globalSession.getBranch(branchId); - - BranchRollbackResponse response = (BranchRollbackResponse)messageSender.sendSyncRequest(resourceId, - branchSession.getClientId(), request); - return response.getBranchStatus(); - } catch (IOException e) { - throw new TransactionException(FailedToSendBranchRollbackRequest, branchId + "/" + xid, e); - } catch (TimeoutException e) { - throw new TransactionException(FailedToSendBranchRollbackRequest, branchId + "/" + xid, e); - } - } - - private void timeoutCheck() throws TransactionException { - Collection allSessions = SessionHolder.getRootSessionManager().allSessions(); - if (allSessions.size() > 0 && LOGGER.isDebugEnabled()) { - LOGGER.debug("Transaction Timeout Check Begin: " + allSessions.size()); - } - for (GlobalSession globalSession : allSessions) { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug(globalSession.getTransactionId() + " " + globalSession.getStatus() + " " + - globalSession.getBeginTime() + " " + globalSession.getTimeout()); - } - boolean shouldTimeout = globalSession.lockAndExcute(() -> { - if (globalSession.getStatus() != GlobalStatus.Begin || !globalSession.isTimeout()) { - return false; - } - globalSession.close(); - globalSession.changeStatus(GlobalStatus.TimeoutRollbacking); - return true; - }); - if (!shouldTimeout) { - continue; - } - LOGGER.info( - "Global transaction[" + globalSession.getTransactionId() + "] is timeout and will be rolled back."); - - globalSession.addSessionLifecycleListener(SessionHolder.getRetryRollbackingSessionManager()); - SessionHolder.getRetryRollbackingSessionManager().addGlobalSession(globalSession); - - } - if (allSessions.size() > 0 && LOGGER.isDebugEnabled()) { - LOGGER.debug("Transaction Timeout Check End. "); - } - - } - - private void handleRetryRollbacking() { - Collection rollbackingSessions = SessionHolder.getRetryRollbackingSessionManager().allSessions(); - for (GlobalSession rollbackingSession : rollbackingSessions) { - try { - core.doGlobalRollback(rollbackingSession, true); - } catch (TransactionException ex) { - LOGGER.info("Failed to retry rollbacking [{}] {} {}", - rollbackingSession.getTransactionId(), ex.getCode(), ex.getMessage()); - } - } - } - - private void handleRetryCommitting() { - Collection committingSessions = SessionHolder.getRetryCommittingSessionManager().allSessions(); - for (GlobalSession committingSession : committingSessions) { - try { - core.doGlobalCommit(committingSession, true); - } catch (TransactionException ex) { - LOGGER.info("Failed to retry committing [{}] {} {}", - committingSession.getTransactionId(), ex.getCode(), ex.getMessage()); - } - } - } - - private void handleAsyncCommitting() { - Collection asyncCommittingSessions = SessionHolder.getAsyncCommittingSessionManager() - .allSessions(); - for (GlobalSession asyncCommittingSession : asyncCommittingSessions) { - try { - core.doGlobalCommit(asyncCommittingSession, true); - } catch (TransactionException ex) { - LOGGER.info("Failed to async committing [{}] {} {}", - asyncCommittingSession.getTransactionId(), ex.getCode(), ex.getMessage()); - } - } - } - - private ScheduledThreadPoolExecutor retryRollbacking = new ScheduledThreadPoolExecutor(1, - new NamedThreadFactory("RetryRollbacking", 1)); - - private ScheduledThreadPoolExecutor retryCommitting = new ScheduledThreadPoolExecutor(1, - new NamedThreadFactory("RetryCommitting", 1)); - - private ScheduledThreadPoolExecutor asyncCommitting = new ScheduledThreadPoolExecutor(1, - new NamedThreadFactory("AsyncCommitting", 1)); - - private ScheduledThreadPoolExecutor timeoutCheck = new ScheduledThreadPoolExecutor(1, - new NamedThreadFactory("TxTimeoutCheck", 1)); - - /** - * Init. - */ - public void init() { - retryRollbacking.scheduleAtFixedRate(new Runnable() { - - @Override - public void run() { - try { - handleRetryRollbacking(); - } catch (Exception e) { - LOGGER.info("Exception retry rollbacking ... ", e); - } - - } - }, 0, 5, TimeUnit.MILLISECONDS); - - retryCommitting.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - try { - handleRetryCommitting(); - } catch (Exception e) { - LOGGER.info("Exception retry committing ... ", e); - } - - } - }, 0, 5, TimeUnit.MILLISECONDS); - - asyncCommitting.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - try { - handleAsyncCommitting(); - } catch (Exception e) { - LOGGER.info("Exception async committing ... ", e); - } - - } - }, 0, 10, TimeUnit.MILLISECONDS); - - timeoutCheck.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - try { - timeoutCheck(); - } catch (Exception e) { - LOGGER.info("Exception timeout checking ... ", e); - } - - } - }, 0, 2, TimeUnit.MILLISECONDS); - } - - @Override - public AbstractResultMessage onRequest(AbstractMessage request, RpcContext context) { - if (!(request instanceof AbstractTransactionRequestToTC)) { - throw new IllegalArgumentException(); - } - AbstractTransactionRequestToTC transactionRequest = (AbstractTransactionRequestToTC)request; - transactionRequest.setTCInboundHandler(this); - - return transactionRequest.handle(context); - } - - @Override - public void onResponse(AbstractResultMessage response, RpcContext context) { - if (!(response instanceof AbstractTransactionResponse)) { - throw new IllegalArgumentException(); - } - - } - - @Override - public void destroy() { - // 1. first shutdown timed task - retryRollbacking.shutdown(); - retryCommitting.shutdown(); - asyncCommitting.shutdown(); - timeoutCheck.shutdown(); - try { - retryRollbacking.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS); - retryCommitting.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS); - asyncCommitting.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS); - timeoutCheck.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS); - } catch (InterruptedException ingore) { - - } - // 2. sencond close netty flow - if (messageSender instanceof RpcServer){ - ((RpcServer) messageSender).destroy(); - } - // 3. last destory SessionHolder - SessionHolder.destory(); - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCore.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCore.java deleted file mode 100755 index 1ba36763f..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/coordinator/DefaultCore.java +++ /dev/null @@ -1,310 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.coordinator; - -import io.seata.common.XID; -import io.seata.core.exception.TransactionException; -import io.seata.core.exception.TransactionExceptionCode; -import io.seata.core.model.BranchStatus; -import io.seata.core.model.BranchType; -import io.seata.core.model.GlobalStatus; -import io.seata.core.model.ResourceManagerInbound; -import io.seata.server.lock.LockManager; -import io.seata.server.lock.LockManagerFactory; -import io.seata.server.session.BranchSession; -import io.seata.server.session.GlobalSession; -import io.seata.server.session.SessionHelper; -import io.seata.server.session.SessionHolder; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import static io.seata.core.exception.TransactionExceptionCode.BranchTransactionNotExist; -import static io.seata.core.exception.TransactionExceptionCode.FailedToAddBranch; -import static io.seata.core.exception.TransactionExceptionCode.GlobalTransactionNotActive; -import static io.seata.core.exception.TransactionExceptionCode.GlobalTransactionStatusInvalid; -import static io.seata.core.exception.TransactionExceptionCode.LockKeyConflict; - -/** - * The type Default core. - * - * @author sharajava - */ -public class DefaultCore implements Core { - - private static final Logger LOGGER = LoggerFactory.getLogger(DefaultCore.class); - - private LockManager lockManager = LockManagerFactory.get(); - - private ResourceManagerInbound resourceManagerInbound; - - @Override - public void setResourceManagerInbound(ResourceManagerInbound resourceManagerInbound) { - this.resourceManagerInbound = resourceManagerInbound; - } - - @Override - public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, - String applicationData, String lockKeys) throws TransactionException { - GlobalSession globalSession = assertGlobalSessionNotNull(XID.getTransactionId(xid)); - return globalSession.lockAndExcute(() -> { - if (!globalSession.isActive()) { - throw new TransactionException(GlobalTransactionNotActive, "Current Status: " + globalSession.getStatus()); - } - if (globalSession.getStatus() != GlobalStatus.Begin) { - throw new TransactionException(GlobalTransactionStatusInvalid, - globalSession.getStatus() + " while expecting " + GlobalStatus.Begin); - } - BranchSession branchSession = SessionHelper.newBranchByGlobal(globalSession, branchType, resourceId, - applicationData, lockKeys, clientId); - if (!branchSession.lock()) { - throw new TransactionException(LockKeyConflict); - } - try { - globalSession.addBranch(branchSession); - } catch (RuntimeException ex) { - throw new TransactionException(FailedToAddBranch); - } - return branchSession.getBranchId(); - }); - } - - private GlobalSession assertGlobalSessionNotNull(long transactionId) throws TransactionException { - GlobalSession globalSession = SessionHolder.findGlobalSession(transactionId); - if (globalSession == null) { - throw new TransactionException(TransactionExceptionCode.GlobalTransactionNotExist, "" + transactionId + ""); - } - return globalSession; - } - - - @Override - public void branchReport(BranchType branchType, String xid, long branchId, BranchStatus status, - String applicationData) throws TransactionException { - GlobalSession globalSession = assertGlobalSessionNotNull(XID.getTransactionId(xid)); - BranchSession branchSession = globalSession.getBranch(branchId); - if (branchSession == null) { - throw new TransactionException(BranchTransactionNotExist); - } - globalSession.changeBranchStatus(branchSession, status); - } - - @Override - public boolean lockQuery(BranchType branchType, String resourceId, String xid, String lockKeys) - throws TransactionException { - if (branchType == BranchType.AT) { - return lockManager.isLockable(XID.getTransactionId(xid), resourceId, lockKeys); - } else { - return true; - } - - } - - @Override - public String begin(String applicationId, String transactionServiceGroup, String name, int timeout) - throws TransactionException { - GlobalSession session = GlobalSession.createGlobalSession( - applicationId, transactionServiceGroup, name, timeout); - session.addSessionLifecycleListener(SessionHolder.getRootSessionManager()); - - session.begin(); - - return XID.generateXID(session.getTransactionId()); - } - - @Override - public GlobalStatus commit(String xid) throws TransactionException { - GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); - if (globalSession == null) { - return GlobalStatus.Finished; - } - // just lock changeStatus - boolean shouldCommit = globalSession.lockAndExcute(() -> { - globalSession.closeAndClean(); // Highlight: Firstly, close the session, then no more branch can be registered. - if (globalSession.getStatus() == GlobalStatus.Begin) { - globalSession.changeStatus(GlobalStatus.Committing); - return true; - } - return false; - }); - if (!shouldCommit){ - return globalSession.getStatus(); - } - if (globalSession.canBeCommittedAsync()) { - asyncCommit(globalSession); - return GlobalStatus.Committed; - } else { - doGlobalCommit(globalSession, false); - } - return globalSession.getStatus(); - } - - @Override - public void doGlobalCommit(GlobalSession globalSession, boolean retrying) throws TransactionException { - for (BranchSession branchSession : globalSession.getSortedBranches()) { - BranchStatus currentStatus = branchSession.getStatus(); - if (currentStatus == BranchStatus.PhaseOne_Failed) { - globalSession.removeBranch(branchSession); - continue; - } - try { - BranchStatus branchStatus = resourceManagerInbound.branchCommit(branchSession.getBranchType(), - XID.generateXID(branchSession.getTransactionId()), branchSession.getBranchId(), - branchSession.getResourceId(), branchSession.getApplicationData()); - - switch (branchStatus) { - case PhaseTwo_Committed: - globalSession.removeBranch(branchSession); - continue; - case PhaseTwo_CommitFailed_Unretryable: - if (globalSession.canBeCommittedAsync()) { - LOGGER.error("By [{}], failed to commit branch {}", branchStatus, branchSession); - continue; - } else { - SessionHelper.endCommitFailed(globalSession); - LOGGER.error("Finally, failed to commit global[{}] since branch[{}] commit failed", - globalSession.getTransactionId(), branchSession.getBranchId()); - return; - } - default: - if (!retrying) { - queueToRetryCommit(globalSession); - return; - } - if (globalSession.canBeCommittedAsync()) { - LOGGER.error("By [{}], failed to commit branch {}", branchStatus, branchSession); - continue; - } else { - LOGGER.error( - "Failed to commit global[{}] since branch[{}] commit failed, will retry later.", - globalSession.getTransactionId(), branchSession.getBranchId()); - return; - } - - } - - } catch (Exception ex) { - LOGGER.error("Exception committing branch {}", branchSession, ex); - if (!retrying) { - queueToRetryCommit(globalSession); - throw new TransactionException(ex); - } - - } - - } - if (globalSession.hasBranch()) { - LOGGER.info("Global[{}] committing is NOT done.", globalSession.getTransactionId()); - return; - } - SessionHelper.endCommitted(globalSession); - LOGGER.info("Global[{}] committing is successfully done.", globalSession.getTransactionId()); - } - - private void asyncCommit(GlobalSession globalSession) throws TransactionException { - globalSession.addSessionLifecycleListener(SessionHolder.getAsyncCommittingSessionManager()); - SessionHolder.getAsyncCommittingSessionManager().addGlobalSession(globalSession); - globalSession.changeStatus(GlobalStatus.AsyncCommitting); - } - - private void queueToRetryCommit(GlobalSession globalSession) throws TransactionException { - globalSession.addSessionLifecycleListener(SessionHolder.getRetryCommittingSessionManager()); - SessionHolder.getRetryCommittingSessionManager().addGlobalSession(globalSession); - globalSession.changeStatus(GlobalStatus.CommitRetrying); - } - - private void queueToRetryRollback(GlobalSession globalSession) throws TransactionException { - globalSession.addSessionLifecycleListener(SessionHolder.getRetryRollbackingSessionManager()); - SessionHolder.getRetryRollbackingSessionManager().addGlobalSession(globalSession); - GlobalStatus currentStatus = globalSession.getStatus(); - if (SessionHelper.isTimeoutGlobalStatus(currentStatus)) { - globalSession.changeStatus(GlobalStatus.TimeoutRollbackRetrying); - } else { - globalSession.changeStatus(GlobalStatus.RollbackRetrying); - } - } - - @Override - public GlobalStatus rollback(String xid) throws TransactionException { - GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); - if (globalSession == null) { - return GlobalStatus.Finished; - } - // just lock changeStatus - boolean shouldRollBack = globalSession.lockAndExcute(() -> { - globalSession.close(); // Highlight: Firstly, close the session, then no more branch can be registered. - if (globalSession.getStatus() == GlobalStatus.Begin) { - globalSession.changeStatus(GlobalStatus.Rollbacking); - return true; - } - return false; - }); - if (!shouldRollBack){ - return globalSession.getStatus(); - } - - doGlobalRollback(globalSession, false); - return globalSession.getStatus(); - } - - @Override - public void doGlobalRollback(GlobalSession globalSession, boolean retrying) throws TransactionException { - for (BranchSession branchSession : globalSession.getReverseSortedBranches()) { - BranchStatus currentBranchStatus = branchSession.getStatus(); - if (currentBranchStatus == BranchStatus.PhaseOne_Failed) { - globalSession.removeBranch(branchSession); - continue; - } - try { - BranchStatus branchStatus = resourceManagerInbound.branchRollback(branchSession.getBranchType(), - XID.generateXID(branchSession.getTransactionId()), branchSession.getBranchId(), - branchSession.getResourceId(), branchSession.getApplicationData()); - - switch (branchStatus) { - case PhaseTwo_Rollbacked: - globalSession.removeBranch(branchSession); - LOGGER.error("Successfully rolled back branch " + branchSession); - continue; - case PhaseTwo_RollbackFailed_Unretryable: - SessionHelper.endRollbackFailed(globalSession); - LOGGER.error("Failed to rollback global[" + globalSession.getTransactionId() + "] since branch[" - + branchSession.getBranchId() + "] rollback failed"); - return; - default: - LOGGER.info("Failed to rollback branch " + branchSession); - if (!retrying) { - queueToRetryRollback(globalSession); - } - return; - - } - } catch (Exception ex) { - LOGGER.error("Exception rollbacking branch " + branchSession, ex); - if (!retrying) { - queueToRetryRollback(globalSession); - } - throw new TransactionException(ex); - } - - } - SessionHelper.endRollbacked(globalSession); - } - - @Override - public GlobalStatus getStatus(String xid) throws TransactionException { - GlobalSession globalSession = SessionHolder.findGlobalSession(XID.getTransactionId(xid)); - return globalSession.getStatus(); - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/DefaultLockManagerImpl.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/DefaultLockManagerImpl.java deleted file mode 100755 index a0c576c84..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/DefaultLockManagerImpl.java +++ /dev/null @@ -1,186 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.lock; - -import io.netty.util.internal.ConcurrentSet; -import io.seata.common.exception.ShouldNeverHappenException; -import io.seata.common.util.StringUtils; -import io.seata.core.exception.TransactionException; -import io.seata.server.session.BranchSession; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.HashMap; -import java.util.Iterator; -import java.util.Map; -import java.util.Map.Entry; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; - -/** - * The type Default lock manager. - * - * @author sharajava - */ -public class DefaultLockManagerImpl implements LockManager { - - private static final Logger LOGGER = LoggerFactory.getLogger(DefaultLockManagerImpl.class); - - private static final int BUCKET_PER_TABLE = 128; - - private static final - ConcurrentHashMap>>> - LOCK_MAP = new ConcurrentHashMap>>>(); - - @Override - public boolean acquireLock(BranchSession branchSession) throws TransactionException { - String resourceId = branchSession.getResourceId(); - long transactionId = branchSession.getTransactionId(); - ConcurrentHashMap>> dbLockMap = LOCK_MAP.get(resourceId); - if (dbLockMap == null) { - LOCK_MAP.putIfAbsent(resourceId, - new ConcurrentHashMap>>()); - dbLockMap = LOCK_MAP.get(resourceId); - } - ConcurrentHashMap, Set> bucketHolder = branchSession.getLockHolder(); - - String lockKey = branchSession.getLockKey(); - if (StringUtils.isNullOrEmpty(lockKey)) { - return true; - } - - String[] tableGroupedLockKeys = lockKey.split(";"); - for (String tableGroupedLockKey : tableGroupedLockKeys) { - int idx = tableGroupedLockKey.indexOf(":"); - if (idx < 0) { - branchSession.unlock(); - throw new ShouldNeverHappenException("Wrong format of LOCK KEYS: " + branchSession.getLockKey()); - } - String tableName = tableGroupedLockKey.substring(0, idx); - String mergedPKs = tableGroupedLockKey.substring(idx + 1); - ConcurrentHashMap> tableLockMap = dbLockMap.get(tableName); - if (tableLockMap == null) { - dbLockMap.putIfAbsent(tableName, new ConcurrentHashMap>()); - tableLockMap = dbLockMap.get(tableName); - } - String[] pks = mergedPKs.split(","); - for (String pk : pks) { - int bucketId = pk.hashCode() % BUCKET_PER_TABLE; - Map bucketLockMap = tableLockMap.get(bucketId); - if (bucketLockMap == null) { - tableLockMap.putIfAbsent(bucketId, new HashMap()); - bucketLockMap = tableLockMap.get(bucketId); - } - synchronized (bucketLockMap) { - Long lockingTransactionId = bucketLockMap.get(pk); - if (lockingTransactionId == null) { - // No existing lock - bucketLockMap.put(pk, transactionId); - Set keysInHolder = bucketHolder.get(bucketLockMap); - if (keysInHolder == null) { - bucketHolder.putIfAbsent(bucketLockMap, new ConcurrentSet()); - keysInHolder = bucketHolder.get(bucketLockMap); - } - keysInHolder.add(pk); - - } else if (lockingTransactionId.longValue() == transactionId) { - // Locked by me - continue; - } else { - LOGGER.info( - "Global lock on [" + tableName + ":" + pk + "] is holding by " + lockingTransactionId); - branchSession.unlock(); // Release all acquired locks. - return false; - } - } - } - } - return true; - } - - @Override - public boolean isLockable(long transactionId, String resourceId, String lockKey) throws TransactionException { - ConcurrentHashMap>> dbLockMap = LOCK_MAP.get(resourceId); - if (dbLockMap == null) { - return true; - } - String[] tableGroupedLockKeys = lockKey.split(";"); - for (String tableGroupedLockKey : tableGroupedLockKeys) { - int idx = tableGroupedLockKey.indexOf(":"); - if (idx < 0) { - throw new ShouldNeverHappenException("Wrong format of LOCK KEYS: " + lockKey); - } - String tableName = tableGroupedLockKey.substring(0, idx); - String mergedPKs = tableGroupedLockKey.substring(idx + 1); - ConcurrentHashMap> tableLockMap = dbLockMap.get(tableName); - if (tableLockMap == null) { - continue; - } - String[] pks = mergedPKs.split(","); - for (String pk : pks) { - int bucketId = pk.hashCode() % BUCKET_PER_TABLE; - Map bucketLockMap = tableLockMap.get(bucketId); - if (bucketLockMap == null) { - continue; - } - Long lockingTransactionId = bucketLockMap.get(pk); - if (lockingTransactionId == null || lockingTransactionId.longValue() == transactionId) { - // Locked by me - continue; - } else { - LOGGER.info("Global lock on [" + tableName + ":" + pk + "] is holding by " + lockingTransactionId); - return false; - } - } - } - return true; - } - - @Override - public void cleanAllLocks() throws TransactionException { - LOCK_MAP.clear(); - } - - @Override - public boolean releaseLock(BranchSession branchSession) throws TransactionException { - ConcurrentHashMap, Set> lockHolder = branchSession.getLockHolder(); - if (lockHolder.size() == 0) { - return true; - } - Iterator, Set>> it = lockHolder.entrySet().iterator(); - while (it.hasNext()) { - Entry, Set> entry = it.next(); - Map bucket = entry.getKey(); - Set keys = entry.getValue(); - synchronized (bucket) { - for (String key : keys) { - Long v = bucket.get(key); - if (v == null) { - continue; - } - if (v.longValue() == branchSession.getTransactionId()) { - bucket.remove(key); - } - } - } - } - lockHolder.clear(); - return true; - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManager.java deleted file mode 100755 index 54bcef3c0..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManager.java +++ /dev/null @@ -1,63 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.lock; - -import io.seata.core.exception.TransactionException; -import io.seata.server.session.BranchSession; - -/** - * The interface Lock manager. - * - * @author sharajava - */ -public interface LockManager { - - /** - * Acquire lock boolean. - * - * @param branchSession the branch session - * @return the boolean - * @throws TransactionException the transaction exception - */ - boolean acquireLock(BranchSession branchSession) throws TransactionException; - - /** - * Is lockable boolean. - * - * @param transactionId the transaction id - * @param resourceId the resource id - * @param lockKey the lock key - * @return the boolean - * @throws TransactionException the transaction exception - */ - boolean isLockable(long transactionId, String resourceId, String lockKey) throws TransactionException; - - /** - * Clean all locks. - * - * @throws TransactionException the transaction exception - */ - void cleanAllLocks() throws TransactionException; - - /** - * Release lock boolean. - * - * @param branchSession the branch session - * @return the boolean - * @throws TransactionException the transaction exception - */ - boolean releaseLock(BranchSession branchSession) throws TransactionException; -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManagerFactory.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManagerFactory.java deleted file mode 100755 index 3468cc894..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/lock/LockManagerFactory.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.lock; - -/** - * The type Lock manager factory. - * - * @author sharajava - */ -public class LockManagerFactory { - - private static class SingletonHolder { - private static LockManager INSTANCE = new DefaultLockManagerImpl(); - } - - /** - * Get lock manager. - * - * @return the lock manager - */ - public static final LockManager get() { - return SingletonHolder.INSTANCE; - } - - /** - * Just for test mocking - * - * @param lockManager the lock manager - */ - public static void set(LockManager lockManager) { - SingletonHolder.INSTANCE = lockManager; - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/AbstractSessionManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/AbstractSessionManager.java deleted file mode 100755 index d58f8b71d..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/AbstractSessionManager.java +++ /dev/null @@ -1,191 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import io.seata.core.exception.TransactionException; -import io.seata.core.exception.TransactionExceptionCode; -import io.seata.core.model.BranchStatus; -import io.seata.core.model.GlobalStatus; -import io.seata.server.store.SessionStorable; -import io.seata.server.store.TransactionStoreManager; -import io.seata.server.store.TransactionStoreManager.LogOperation; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -/** - * The type Abstract session manager. - * - * @author sharajava - */ -public abstract class AbstractSessionManager implements SessionManager, SessionLifecycleListener { - - /** - * The constant LOGGER. - */ - protected static final Logger LOGGER = LoggerFactory.getLogger(AbstractSessionManager.class); - - /** - * The Session map. - */ - protected Map sessionMap = new ConcurrentHashMap<>(); - - /** - * The Transaction store manager. - */ - protected TransactionStoreManager transactionStoreManager; - - /** - * The Name. - */ - protected String name; - - /** - * Instantiates a new Abstract session manager. - * - * @param name the name - */ - public AbstractSessionManager(String name) { - this.name = name; - } - - @Override - public void addGlobalSession(GlobalSession session) throws TransactionException { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("MANAGER[" + name + "] SESSION[" + session + "] " + LogOperation.GLOBAL_ADD); - } - writeSession(LogOperation.GLOBAL_ADD, session); - sessionMap.put(session.getTransactionId(), session); - - } - - - - @Override - public GlobalSession findGlobalSession(Long transactionId) { - return sessionMap.get(transactionId); - } - - @Override - public void updateGlobalSessionStatus(GlobalSession session, GlobalStatus status) throws TransactionException { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("MANAGER[" + name + "] SESSION[" + session + "] " + LogOperation.GLOBAL_UPDATE); - } - writeSession(LogOperation.GLOBAL_UPDATE, session); - } - - @Override - public void removeGlobalSession(GlobalSession session) throws TransactionException { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("MANAGER[" + name + "] SESSION[" + session + "] " + LogOperation.GLOBAL_REMOVE); - } - writeSession(LogOperation.GLOBAL_REMOVE, session); - sessionMap.remove(session.getTransactionId()); - - } - - @Override - public void addBranchSession(GlobalSession session, BranchSession branchSession) throws TransactionException { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("MANAGER[" + name + "] SESSION[" + branchSession + "] " + LogOperation.BRANCH_ADD); - } - writeSession(LogOperation.BRANCH_ADD, branchSession); - } - - @Override - public void updateBranchSessionStatus(BranchSession branchSession, BranchStatus status) - throws TransactionException { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("MANAGER[" + name + "] SESSION[" + branchSession + "] " + LogOperation.GLOBAL_ADD); - } - writeSession(LogOperation.BRANCH_UPDATE, branchSession); - } - - @Override - public void removeBranchSession(GlobalSession globalSession, BranchSession branchSession) - throws TransactionException { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("MANAGER[" + name + "] SESSION[" + branchSession + "] " + LogOperation.GLOBAL_ADD); - } - writeSession(LogOperation.BRANCH_REMOVE, branchSession); - - } - - @Override - public Collection allSessions() { - return sessionMap.values(); - } - - @Override - public List findGlobalSessions(SessionCondition condition) { - List found = new ArrayList<>(); - for (GlobalSession globalSession : sessionMap.values()) { - if (System.currentTimeMillis() - globalSession.getBeginTime() > condition.getOverTimeAliveMills()) { - found.add(globalSession); - } - } - return found; - } - - @Override - public void onBegin(GlobalSession globalSession) throws TransactionException { - addGlobalSession(globalSession); - } - - @Override - public void onStatusChange(GlobalSession globalSession, GlobalStatus status) throws TransactionException { - updateGlobalSessionStatus(globalSession, status); - } - - @Override - public void onBranchStatusChange(GlobalSession globalSession, BranchSession branchSession, BranchStatus status) - throws TransactionException { - updateBranchSessionStatus(branchSession, status); - } - - @Override - public void onAddBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException { - addBranchSession(globalSession, branchSession); - } - - @Override - public void onRemoveBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException { - removeBranchSession(globalSession, branchSession); - } - - @Override - public void onClose(GlobalSession globalSession) throws TransactionException { - globalSession.setActive(false); - - } - - @Override - public void onEnd(GlobalSession globalSession) throws TransactionException { - removeGlobalSession(globalSession); - - } - - private void writeSession(LogOperation logOperation, SessionStorable sessionStorable) throws TransactionException{ - if (!transactionStoreManager.writeSession(logOperation, sessionStorable)) { - throw new TransactionException(TransactionExceptionCode.FailedWriteSession); - } - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/BranchSession.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/BranchSession.java deleted file mode 100755 index 31e519612..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/BranchSession.java +++ /dev/null @@ -1,395 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; - -import io.seata.common.util.CompressUtil; -import io.seata.core.exception.TransactionException; -import io.seata.core.model.BranchStatus; -import io.seata.core.model.BranchType; -import io.seata.server.lock.LockManagerFactory; -import io.seata.server.store.SessionStorable; -import io.seata.server.store.StoreConfig; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * The type Branch session. - * - * @author sharajava - */ -public class BranchSession implements Lockable, Comparable, SessionStorable { - - private static final Logger LOGGER = LoggerFactory.getLogger(BranchSession.class); - - private static final int MAX_BRANCH_SESSION_SIZE = StoreConfig.getMaxBranchSessionSize(); - - private static ThreadLocal byteBufferThreadLocal = ThreadLocal.withInitial(() -> ByteBuffer.allocate( - MAX_BRANCH_SESSION_SIZE)); - - private long transactionId; - - private long branchId; - - private String resourceGroupId; - - private String resourceId; - - private String lockKey; - - private BranchType branchType; - - private BranchStatus status = BranchStatus.Unknown; - - private String clientId; - - private String applicationData; - - private ConcurrentHashMap, Set> lockHolder - = new ConcurrentHashMap, Set>(); - - /** - * Gets application data. - * - * @return the application data - */ - public String getApplicationData() { - return applicationData; - } - - /** - * Sets application data. - * - * @param applicationData the application data - */ - public void setApplicationData(String applicationData) { - this.applicationData = applicationData; - } - - /** - * Gets resource group id. - * - * @return the resource group id - */ - public String getResourceGroupId() { - return resourceGroupId; - } - - /** - * Sets resource group id. - * - * @param resourceGroupId the resource group id - */ - public void setResourceGroupId(String resourceGroupId) { - this.resourceGroupId = resourceGroupId; - } - - /** - * Gets client id. - * - * @return the client id - */ - public String getClientId() { - return clientId; - } - - /** - * Sets client id. - * - * @param clientId the client id - */ - public void setClientId(String clientId) { - this.clientId = clientId; - } - - /** - * Gets resource id. - * - * @return the resource id - */ - public String getResourceId() { - return resourceId; - } - - /** - * Sets resource id. - * - * @param resourceId the resource id - */ - public void setResourceId(String resourceId) { - this.resourceId = resourceId; - } - - /** - * Gets lock key. - * - * @return the lock key - */ - public String getLockKey() { - return lockKey; - } - - /** - * Sets lock key. - * - * @param lockKey the lock key - */ - public void setLockKey(String lockKey) { - this.lockKey = lockKey; - } - - /** - * Gets branch type. - * - * @return the branch type - */ - public BranchType getBranchType() { - return branchType; - } - - /** - * Sets branch type. - * - * @param branchType the branch type - */ - public void setBranchType(BranchType branchType) { - this.branchType = branchType; - } - - /** - * Gets status. - * - * @return the status - */ - public BranchStatus getStatus() { - return status; - } - - /** - * Sets status. - * - * @param status the status - */ - void setStatus(BranchStatus status) { - this.status = status; - } - - /** - * Gets transaction id. - * - * @return the transaction id - */ - public long getTransactionId() { - return transactionId; - } - - /** - * Sets transaction id. - * - * @param transactionId the transaction id - */ - public void setTransactionId(long transactionId) { - this.transactionId = transactionId; - } - - /** - * Gets branch id. - * - * @return the branch id - */ - public long getBranchId() { - return branchId; - } - - /** - * Sets branch id. - * - * @param branchId the branch id - */ - public void setBranchId(long branchId) { - this.branchId = branchId; - } - - @Override - public String toString() { - return "BR:" + branchId + "/" + transactionId; - } - - @Override - public int compareTo(BranchSession o) { - return this.branchId < o.branchId ? -1 : (this.branchId > o.branchId ? 1 : 0); - } - - /** - * Gets lock holder. - * - * @return the lock holder - */ - public ConcurrentHashMap, Set> getLockHolder() { - return lockHolder; - } - - @Override - public boolean lock() throws TransactionException { - return LockManagerFactory.get().acquireLock(this); - } - - @Override - public boolean unlock() throws TransactionException { - return LockManagerFactory.get().releaseLock(this); - } - - @Override - public byte[] encode() { - - byte[] resourceIdBytes = resourceId != null ? resourceId.getBytes() : null; - - byte[] lockKeyBytes = lockKey != null ? lockKey.getBytes() : null; - - byte[] clientIdBytes = clientId != null ? clientId.getBytes() : null; - - byte[] applicationDataBytes = applicationData != null ? applicationData.getBytes() : null; - - int size = calBranchSessionSize(resourceIdBytes, lockKeyBytes, clientIdBytes, applicationDataBytes); - - if (size > MAX_BRANCH_SESSION_SIZE) { - if (lockKeyBytes == null) { - throw new RuntimeException("branch session size exceeded, size : " + size + " maxBranchSessionSize : " - + MAX_BRANCH_SESSION_SIZE); - } - // try compress lockkey - try { - size -= lockKeyBytes.length; - lockKeyBytes = CompressUtil.compress(lockKeyBytes); - } catch (IOException e) { - LOGGER.error("compress lockKey error", e); - } finally { - size += lockKeyBytes.length; - } - - if (size > MAX_BRANCH_SESSION_SIZE) { - throw new RuntimeException( - "compress branch session size exceeded, compressSize : " + size + " maxBranchSessionSize : " - + MAX_BRANCH_SESSION_SIZE); - } - } - - ByteBuffer byteBuffer = byteBufferThreadLocal.get(); - //recycle - byteBuffer.clear(); - - byteBuffer.putLong(transactionId); - byteBuffer.putLong(branchId); - - if (null != resourceIdBytes) { - byteBuffer.putInt(resourceIdBytes.length); - byteBuffer.put(resourceIdBytes); - } else { - byteBuffer.putInt(0); - } - - if (null != lockKeyBytes) { - byteBuffer.putInt(lockKeyBytes.length); - byteBuffer.put(lockKeyBytes); - } else { - byteBuffer.putInt(0); - } - - if (null != clientIdBytes) { - byteBuffer.putShort((short)clientIdBytes.length); - byteBuffer.put(clientIdBytes); - } else { - byteBuffer.putShort((short)0); - } - - if (null != applicationDataBytes) { - byteBuffer.putInt(applicationDataBytes.length); - byteBuffer.put(applicationDataBytes); - } else { - byteBuffer.putInt(0); - } - - byteBuffer.put((byte)status.getCode()); - byteBuffer.flip(); - byte[] result = new byte[byteBuffer.limit()]; - byteBuffer.get(result); - return result; - } - - private int calBranchSessionSize(byte[] resourceIdBytes, byte[] lockKeyBytes, byte[] clientIdBytes, - byte[] applicationDataBytes) { - final int size = 8 // trascationId - + 8 // branchId - + 4 // resourceIdBytes.length - + 4 // lockKeyBytes.length - + 2 // clientIdBytes.length - + 4 // applicationDataBytes.length - + 1 // statusCode - + (resourceIdBytes == null ? 0 : resourceIdBytes.length) - + (lockKeyBytes == null ? 0 : lockKeyBytes.length) - + (clientIdBytes == null ? 0 : clientIdBytes.length) - + (applicationDataBytes == null ? 0 : applicationDataBytes.length); - return size; - } - - @Override - public void decode(byte[] a) { - ByteBuffer byteBuffer = ByteBuffer.wrap(a); - this.transactionId = byteBuffer.getLong(); - this.branchId = byteBuffer.getLong(); - int resourceLen = byteBuffer.getInt(); - if (resourceLen > 0) { - byte[] byResource = new byte[resourceLen]; - byteBuffer.get(byResource); - this.resourceId = new String(byResource); - } - int lockKeyLen = byteBuffer.getInt(); - if (lockKeyLen > 0) { - byte[] byLockKey = new byte[lockKeyLen]; - byteBuffer.get(byLockKey); - if (CompressUtil.isCompressData(byLockKey)) { - try { - this.lockKey = new String(CompressUtil.uncompress(byLockKey)); - } catch (IOException e) { - throw new RuntimeException("uncompress lockKey error", e); - } - } else { - this.lockKey = new String(byLockKey); - } - - } - short clientIdLen = byteBuffer.getShort(); - if (clientIdLen > 0) { - byte[] byClientId = new byte[clientIdLen]; - byteBuffer.get(byClientId); - this.clientId = new String(byClientId); - } - int applicationDataLen = byteBuffer.getInt(); - if (applicationDataLen > 0) { - byte[] byApplicationData = new byte[applicationDataLen]; - byteBuffer.get(byApplicationData); - this.applicationData = new String(byApplicationData); - } - this.status = BranchStatus.get(byteBuffer.get()); - - } - -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/DefaultSessionManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/DefaultSessionManager.java deleted file mode 100755 index ff24f7f6b..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/DefaultSessionManager.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import java.util.List; - -import io.seata.server.store.SessionStorable; -import io.seata.server.store.TransactionStoreManager; -import io.seata.server.store.TransactionWriteStore; - -/** - * The type Default session manager. - * - * @author sharajava - */ -public class DefaultSessionManager extends AbstractSessionManager { - - /** - * Instantiates a new Default session manager. - * - * @param name the name - */ - public DefaultSessionManager(String name) { - super(name); - transactionStoreManager = new TransactionStoreManager() { - @Override - public boolean writeSession(LogOperation logOperation, SessionStorable session) { - return true; - } - - @Override - public void shutdown() { - - } - - @Override - public List readWriteStoreFromFile(int readSize, boolean isHistory) { - return null; - } - - @Override - public boolean hasRemaining(boolean isHistory) { - return false; - } - }; - } - - @Override - public void destroy() { - transactionStoreManager.shutdown(); - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/FileBasedSessionManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/FileBasedSessionManager.java deleted file mode 100755 index 1684cccdf..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/FileBasedSessionManager.java +++ /dev/null @@ -1,232 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import java.io.File; -import java.io.IOException; -import java.util.HashMap; -import java.util.Iterator; -import java.util.List; -import java.util.Map; - -import io.seata.common.exception.ShouldNeverHappenException; -import io.seata.config.ConfigurationFactory; -import io.seata.core.constants.ConfigurationKeys; -import io.seata.core.model.GlobalStatus; -import io.seata.server.UUIDGenerator; -import io.seata.server.store.FileTransactionStoreManager; -import io.seata.server.store.SessionStorable; -import io.seata.server.store.TransactionStoreManager; -import io.seata.server.store.TransactionWriteStore; - -/** - * The type File based session manager. - * - * @author jimin.jm @alibaba-inc.com - */ -public class FileBasedSessionManager extends AbstractSessionManager implements Reloadable { - - private static final int READ_SIZE = ConfigurationFactory.getInstance().getInt( - ConfigurationKeys.SERVICE_SESSION_RELOAD_READ_SIZE, 100); - - /** - * Instantiates a new File based session manager. - * - * @param name the name - * @param sessionStoreFilePath the session store file path - * @throws IOException the io exception - */ - public FileBasedSessionManager(String name, String sessionStoreFilePath) throws IOException { - super(name); - transactionStoreManager = new FileTransactionStoreManager(sessionStoreFilePath + File.separator + name, this); - } - - @Override - public void reload() { - restoreSessions(); - washSessions(); - } - - private void restoreSessions() { - Map unhandledBranchBuffer = new HashMap<>(); - - restoreSessions(true, unhandledBranchBuffer); - restoreSessions(false, unhandledBranchBuffer); - - if (!unhandledBranchBuffer.isEmpty()) { - unhandledBranchBuffer.values().forEach(branchSession -> { - long tid = branchSession.getTransactionId(); - long bid = branchSession.getBranchId(); - GlobalSession found = sessionMap.get(tid); - if (found == null) { - // Ignore - if (LOGGER.isInfoEnabled()) { - LOGGER.info("GlobalSession Does Not Exists For BranchSession [" + bid + "/" + tid + "]"); - } - } else { - BranchSession existingBranch = found.getBranch(branchSession.getBranchId()); - if (existingBranch == null) { - found.add(branchSession); - } else { - existingBranch.setStatus(branchSession.getStatus()); - } - } - - }); - } - } - - private void washSessions() { - if (sessionMap.size() > 0) { - Iterator> iterator = sessionMap.entrySet().iterator(); - while (iterator.hasNext()) { - GlobalSession globalSession = iterator.next().getValue(); - - GlobalStatus globalStatus = globalSession.getStatus(); - switch (globalStatus) { - case UnKnown: - case Committed: - case CommitFailed: - case Rollbacked: - case RollbackFailed: - case TimeoutRollbacked: - case TimeoutRollbackFailed: - case Finished: - // Remove all sessions finished - iterator.remove(); - break; - default: - break; - } - } - } - } - - private void restoreSessions(boolean isHistory, Map unhandledBranchBuffer) { - while (transactionStoreManager.hasRemaining(isHistory)) { - List stores = transactionStoreManager.readWriteStoreFromFile(READ_SIZE, isHistory); - restore(stores, unhandledBranchBuffer); - } - } - - private void restore(List stores, Map unhandledBranchSessions) { - long maxRecoverId = UUIDGenerator.getCurrentUUID(); - for (TransactionWriteStore store : stores) { - TransactionStoreManager.LogOperation logOperation = store.getOperate(); - SessionStorable sessionStorable = store.getSessionRequest(); - maxRecoverId = getMaxId(maxRecoverId, sessionStorable); - switch (logOperation) { - case GLOBAL_ADD: - case GLOBAL_UPDATE: { - GlobalSession globalSession = (GlobalSession)sessionStorable; - long tid = globalSession.getTransactionId(); - GlobalSession foundGlobalSession = sessionMap.get(tid); - if (foundGlobalSession == null) { - sessionMap.put(globalSession.getTransactionId(), globalSession); - } else { - foundGlobalSession.setStatus(globalSession.getStatus()); - } - break; - } - case GLOBAL_REMOVE: { - GlobalSession globalSession = (GlobalSession)sessionStorable; - long tid = globalSession.getTransactionId(); - if (sessionMap.remove(tid) == null) { - if (LOGGER.isInfoEnabled()) { - LOGGER.info("GlobalSession To Be Removed Does Not Exists [" + tid + "]"); - } - } - break; - } - case BRANCH_ADD: - case BRANCH_UPDATE: { - BranchSession branchSession = (BranchSession)sessionStorable; - long tid = branchSession.getTransactionId(); - GlobalSession foundGlobalSession = sessionMap.get(tid); - if (foundGlobalSession == null) { - unhandledBranchSessions.put(branchSession.getBranchId(), branchSession); - } else { - BranchSession existingBranch = foundGlobalSession.getBranch(branchSession.getBranchId()); - if (existingBranch == null) { - foundGlobalSession.add(branchSession); - } else { - existingBranch.setStatus(branchSession.getStatus()); - } - } - break; - } - case BRANCH_REMOVE: { - BranchSession branchSession = (BranchSession)sessionStorable; - long tid = branchSession.getTransactionId(); - long bid = branchSession.getBranchId(); - GlobalSession found = sessionMap.get(tid); - if (found == null) { - if (LOGGER.isInfoEnabled()) { - LOGGER.info( - "GlobalSession To Be Updated (Remove Branch) Does Not Exists [" + bid + "/" + tid - + "]"); - } - } else { - BranchSession theBranch = found.getBranch(bid); - if (theBranch == null) { - if (LOGGER.isInfoEnabled()) { - LOGGER.info("BranchSession To Be Updated Does Not Exists [" + bid + "/" + tid + "]"); - } - } else { - found.remove(theBranch); - } - } - break; - } - - default: - throw new ShouldNeverHappenException("Unknown Operation: " + logOperation); - - } - } - setMaxId(maxRecoverId); - - } - - private long getMaxId(long maxRecoverId, SessionStorable sessionStorable) { - long currentId = 0; - if (sessionStorable instanceof GlobalSession) { - currentId = ((GlobalSession)sessionStorable).getTransactionId(); - } else if (sessionStorable instanceof BranchSession) { - currentId = ((BranchSession)sessionStorable).getBranchId(); - } - - return maxRecoverId > currentId ? maxRecoverId : currentId; - } - - private void setMaxId(long maxRecoverId) { - long currentId; - // will be recover multi-thread later - while ((currentId = UUIDGenerator.getCurrentUUID()) < maxRecoverId) { - if (UUIDGenerator.setUUID(currentId, maxRecoverId)) { - break; - } - } - } - - private void restore(TransactionWriteStore store) { - } - - @Override - public void destroy() { - transactionStoreManager.shutdown(); - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/GlobalSession.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/GlobalSession.java deleted file mode 100755 index 1c9f31d5b..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/GlobalSession.java +++ /dev/null @@ -1,541 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import io.seata.core.exception.TransactionException; -import io.seata.core.exception.TransactionExceptionCode; -import io.seata.core.model.BranchStatus; -import io.seata.core.model.BranchType; -import io.seata.core.model.GlobalStatus; -import io.seata.server.UUIDGenerator; -import io.seata.server.store.SessionStorable; -import io.seata.server.store.StoreConfig; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.Collections; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; - -/** - * The type Global session. - * - * @author sharajava - */ -public class GlobalSession implements SessionLifecycle, SessionStorable { - - private static final Logger LOGGER = LoggerFactory.getLogger(GlobalSession.class); - - private static final int MAX_GLOBAL_SESSION_SIZE = StoreConfig.getMaxGlobalSessionSize(); - - private static ThreadLocal byteBufferThreadLocal = ThreadLocal.withInitial(() -> ByteBuffer.allocate( - MAX_GLOBAL_SESSION_SIZE)); - - private long transactionId; - - private volatile GlobalStatus status; - - private String applicationId; - - private String transactionServiceGroup; - - private String transactionName; - - private int timeout; - - private long beginTime; - - private boolean active; - - private ArrayList branchSessions = new ArrayList<>(); - - private GlobalSessionLock globalSessionLock = new GlobalSessionLock(); - - /** - * Add boolean. - * - * @param branchSession the branch session - * @return the boolean - */ - public boolean add(BranchSession branchSession) { - return branchSessions.add(branchSession); - } - - /** - * Remove boolean. - * - * @param branchSession the branch session - * @return the boolean - */ - public boolean remove(BranchSession branchSession) { - return branchSessions.remove(branchSession); - } - - private ArrayList lifecycleListeners = new ArrayList<>(); - - /** - * Can be committed async boolean. - * - * @return the boolean - */ - public boolean canBeCommittedAsync() { - for (BranchSession branchSession : branchSessions) { - if (branchSession.getBranchType() == BranchType.TCC) { - return false; - } - } - return true; - } - - /** - * Is timeout boolean. - * - * @return the boolean - */ - public boolean isTimeout() { - return (System.currentTimeMillis() - beginTime) > timeout; - } - - @Override - public void begin() throws TransactionException { - this.status = GlobalStatus.Begin; - this.beginTime = System.currentTimeMillis(); - this.active = true; - for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { - lifecycleListener.onBegin(this); - } - } - - @Override - public void changeStatus(GlobalStatus status) throws TransactionException { - this.status = status; - for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { - lifecycleListener.onStatusChange(this, status); - } - - } - - @Override - public void changeBranchStatus(BranchSession branchSession, BranchStatus status) - throws TransactionException { - branchSession.setStatus(status); - for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { - lifecycleListener.onBranchStatusChange(this, branchSession, status); - } - } - - @Override - public boolean isActive() { - return active; - } - - @Override - public void close() throws TransactionException { - if (active) { - for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { - lifecycleListener.onClose(this); - } - } - } - - @Override - public void end() throws TransactionException { - // Clean locks first - clean(); - - for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { - lifecycleListener.onEnd(this); - } - - } - - private void clean() throws TransactionException { - for (BranchSession branchSession : branchSessions) { - branchSession.unlock(); - } - - } - - /** - * Close and clean. - * - * @throws TransactionException the transaction exception - */ - public void closeAndClean() throws TransactionException { - close(); - clean(); - - } - - /** - * Add session lifecycle listener. - * - * @param sessionLifecycleListener the session lifecycle listener - */ - public void addSessionLifecycleListener(SessionLifecycleListener sessionLifecycleListener) { - lifecycleListeners.add(sessionLifecycleListener); - } - - /** - * Remove session lifecycle listener. - * - * @param sessionLifecycleListener the session lifecycle listener - */ - public void removeSessionLifecycleListener(SessionLifecycleListener sessionLifecycleListener) { - lifecycleListeners.remove(sessionLifecycleListener); - } - - @Override - public void addBranch(BranchSession branchSession) throws TransactionException { - for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { - lifecycleListener.onAddBranch(this, branchSession); - } - branchSession.setStatus(BranchStatus.Registered); - add(branchSession); - } - - @Override - public void removeBranch(BranchSession branchSession) throws TransactionException { - for (SessionLifecycleListener lifecycleListener : lifecycleListeners) { - lifecycleListener.onRemoveBranch(this, branchSession); - } - branchSession.unlock(); - remove(branchSession); - } - - /** - * Gets branch. - * - * @param branchId the branch id - * @return the branch - */ - public BranchSession getBranch(long branchId) { - synchronized (branchSessions) { - for (BranchSession branchSession : branchSessions) { - if (branchSession.getBranchId() == branchId) { - return branchSession; - } - } - - return null; - } - - } - - /** - * Gets sorted branches. - * - * @return the sorted branches - */ - public ArrayList getSortedBranches() { - ArrayList sorted = new ArrayList(); - sorted.addAll(branchSessions); - return sorted; - } - - /** - * Gets reverse sorted branches. - * - * @return the reverse sorted branches - */ - public ArrayList getReverseSortedBranches() { - ArrayList reversed = new ArrayList(); - reversed.addAll(branchSessions); - Collections.reverse(reversed); - return reversed; - } - - /** - * Instantiates a new Global session. - */ - public GlobalSession() {} - - /** - * Instantiates a new Global session. - * - * @param applicationId the application id - * @param transactionServiceGroup the transaction service group - * @param transactionName the transaction name - * @param timeout the timeout - */ - public GlobalSession(String applicationId, String transactionServiceGroup, String transactionName, int timeout) { - this.transactionId = UUIDGenerator.generateUUID(); - this.status = GlobalStatus.Begin; - - this.applicationId = applicationId; - this.transactionServiceGroup = transactionServiceGroup; - this.transactionName = transactionName; - this.timeout = timeout; - } - - /** - * Gets transaction id. - * - * @return the transaction id - */ - public long getTransactionId() { - return transactionId; - } - - /** - * Gets status. - * - * @return the status - */ - public GlobalStatus getStatus() { - return status; - } - - /** - * Sets status. - * - * @param status the status - */ - void setStatus(GlobalStatus status) { - this.status = status; - } - - /** - * Gets application id. - * - * @return the application id - */ - public String getApplicationId() { - return applicationId; - } - - /** - * Gets transaction service group. - * - * @return the transaction service group - */ - public String getTransactionServiceGroup() { - return transactionServiceGroup; - } - - /** - * Gets transaction name. - * - * @return the transaction name - */ - public String getTransactionName() { - return transactionName; - } - - /** - * Gets timeout. - * - * @return the timeout - */ - public int getTimeout() { - return timeout; - } - - /** - * Gets begin time. - * - * @return the begin time - */ - public long getBeginTime() { - return beginTime; - } - - - /** - * Create global session global session. - * - * @param applicationId the application id - * @param txServiceGroup the tx service group - * @param txName the tx name - * @param timeout the timeout - * @return the global session - */ - public static GlobalSession createGlobalSession(String applicationId, String txServiceGroup, String txName, - int timeout) { - GlobalSession session = new GlobalSession(applicationId, txServiceGroup, txName, timeout); - return session; - } - - /** - * Sets active. - * - * @param active the active - */ - public void setActive(boolean active) { - this.active = active; - } - - @Override - public byte[] encode() { - - byte[] byApplicationIdBytes = applicationId != null ? applicationId.getBytes() : null; - - byte[] byServiceGroupBytes = transactionServiceGroup != null ? transactionServiceGroup.getBytes() : null; - - byte[] byTxNameBytes = transactionName != null ? transactionName.getBytes() : null; - - int size = calGlobalSessionSize(byApplicationIdBytes, byServiceGroupBytes, byTxNameBytes); - - if (size > MAX_GLOBAL_SESSION_SIZE) { - throw new RuntimeException("global session size exceeded, size : " + size + " maxBranchSessionSize : " + - MAX_GLOBAL_SESSION_SIZE); - } - ByteBuffer byteBuffer = byteBufferThreadLocal.get(); - //recycle - byteBuffer.clear(); - - byteBuffer.putLong(transactionId); - byteBuffer.putInt(timeout); - if (null != byApplicationIdBytes) { - byteBuffer.putShort((short)byApplicationIdBytes.length); - byteBuffer.put(byApplicationIdBytes); - } else { - byteBuffer.putShort((short)0); - } - if (null != byServiceGroupBytes) { - byteBuffer.putShort((short)byServiceGroupBytes.length); - byteBuffer.put(byServiceGroupBytes); - } else { - byteBuffer.putShort((short)0); - } - if (null != byTxNameBytes) { - byteBuffer.putShort((short)byTxNameBytes.length); - byteBuffer.put(byTxNameBytes); - } else { - byteBuffer.putShort((short)0); - } - byteBuffer.putLong(beginTime); - byteBuffer.put((byte)status.getCode()); - byteBuffer.flip(); - byte[] result = new byte[byteBuffer.limit()]; - byteBuffer.get(result); - return result; - } - - private int calGlobalSessionSize(byte[] byApplicationIdBytes, byte[] byServiceGroupBytes, byte[] byTxNameBytes) { - final int size = 8 // trascationId - + 4 // timeout - + 2 // byApplicationIdBytes.length - + 2 // byServiceGroupBytes.length - + 2 // byTxNameBytes.length - + 8 // beginTime - + 1 // statusCode - + (byApplicationIdBytes == null ? 0 : byApplicationIdBytes.length) - + (byServiceGroupBytes == null ? 0 : byServiceGroupBytes.length) - + (byTxNameBytes == null ? 0 : byTxNameBytes.length); - return size; - } - - @Override - public void decode(byte[] a) { - ByteBuffer byteBuffer = ByteBuffer.wrap(a); - this.transactionId = byteBuffer.getLong(); - this.timeout = byteBuffer.getInt(); - short applicationIdLen = byteBuffer.getShort(); - if (applicationIdLen > 0) { - byte[] byApplicationId = new byte[applicationIdLen]; - byteBuffer.get(byApplicationId); - this.applicationId = new String(byApplicationId); - } - short serviceGroupLen = byteBuffer.getShort(); - if (serviceGroupLen > 0) { - byte[] byServiceGroup = new byte[serviceGroupLen]; - byteBuffer.get(byServiceGroup); - this.transactionServiceGroup = new String(byServiceGroup); - } - short txNameLen = byteBuffer.getShort(); - if (txNameLen > 0) { - byte[] byTxName = new byte[txNameLen]; - byteBuffer.get(byTxName); - this.transactionName = new String(byTxName); - } - this.beginTime = byteBuffer.getLong(); - this.status = GlobalStatus.get(byteBuffer.get()); - } - - /** - * Has branch boolean. - * - * @return the boolean - */ - public boolean hasBranch() { - return branchSessions.size() > 0; - } - - public void lock() throws TransactionException { - globalSessionLock.lock(); - } - - public void unlock(){ - globalSessionLock.unlock(); - } - - public void lockAndExcute(LockRunnable excuteRunnable) throws TransactionException{ - this.lock(); - try { - excuteRunnable.run(); - }finally { - this.unlock(); - } - } - - public T lockAndExcute(LockCallable lockCallable) throws TransactionException{ - this.lock(); - try { - return lockCallable.call(); - }finally { - this.unlock(); - } - } - - private class GlobalSessionLock { - - private Lock globalSessionLock = new ReentrantLock(); - - private static final int GLOBAL_SESSOION_LOCK_TIME_OUT_MILLS = 2 * 1000; - - public void lock() throws TransactionException { - try { - if (globalSessionLock.tryLock(GLOBAL_SESSOION_LOCK_TIME_OUT_MILLS, TimeUnit.MILLISECONDS)){ - return; - } - } catch (InterruptedException e) { - LOGGER.error("Interrupted error", e); - } - throw new TransactionException(TransactionExceptionCode.FailedLockGlobalTranscation); - } - - - public void unlock() { - globalSessionLock.unlock(); - } - - } - - @FunctionalInterface - public interface LockRunnable { - - void run() throws TransactionException; - } - @FunctionalInterface - public interface LockCallable { - - V call() throws TransactionException; - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Lockable.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Lockable.java deleted file mode 100755 index d023e632f..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Lockable.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import io.seata.core.exception.TransactionException; - -/** - * The interface Lockable. - * - * @author sharajava - */ -public interface Lockable { - - /** - * Lock boolean. - * - * @return the boolean - * @throws TransactionException the transaction exception - */ - boolean lock() throws TransactionException; - - /** - * Unlock boolean. - * - * @return the boolean - * @throws TransactionException the transaction exception - */ - boolean unlock() throws TransactionException; -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Reloadable.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Reloadable.java deleted file mode 100755 index 5701c69d6..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/Reloadable.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -/** - * Service contains states which can be reloaded. - * - * @author sharajava - */ -public interface Reloadable { - - /** - * Reload states. - */ - void reload(); -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionCondition.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionCondition.java deleted file mode 100755 index 4063d7aa3..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionCondition.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -/** - * The type Session condition. - * - * @author jimin.jm @alibaba-inc.com - * @date 2018 /12/13 - */ -public class SessionCondition { - - private long overTimeAliveMills; - - /** - * Instantiates a new Session condition. - * - * @param overTimeAliveMills the over time alive mills - */ - public SessionCondition( long overTimeAliveMills) { - this.overTimeAliveMills = overTimeAliveMills; - } - - - /** - * Gets over time alive mills. - * - * @return the over time alive mills - */ - public long getOverTimeAliveMills() { - return overTimeAliveMills; - } - - /** - * Sets over time alive mills. - * - * @param overTimeAliveMills the over time alive mills - */ - public void setOverTimeAliveMills(long overTimeAliveMills) { - this.overTimeAliveMills = overTimeAliveMills; - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHelper.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHelper.java deleted file mode 100755 index b3295d24e..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHelper.java +++ /dev/null @@ -1,121 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import io.seata.core.exception.TransactionException; -import io.seata.core.model.BranchType; -import io.seata.core.model.GlobalStatus; -import io.seata.server.UUIDGenerator; - -/** - * The type Session helper. - * - * @author sharajava - */ -public class SessionHelper { - - private SessionHelper() {} - - public static BranchSession newBranchByGlobal(GlobalSession globalSession, BranchType branchType, String resourceId, String lockKeys, String clientId) { - return newBranchByGlobal(globalSession, branchType, resourceId, null, lockKeys, clientId); - } - - /** - * New branch by global branch session. - * - * @param globalSession the global session - * @param branchType the branch type - * @param resourceId the resource id - * @param lockKeys the lock keys - * @param clientId the client id - * @return the branch session - */ - public static BranchSession newBranchByGlobal(GlobalSession globalSession, BranchType branchType, String resourceId, - String applicationData, String lockKeys, String clientId) { - BranchSession branchSession = new BranchSession(); - - branchSession.setTransactionId(globalSession.getTransactionId()); - branchSession.setBranchId(UUIDGenerator.generateUUID()); - branchSession.setBranchType(branchType); - branchSession.setResourceId(resourceId); - branchSession.setLockKey(lockKeys); - branchSession.setClientId(clientId); - branchSession.setApplicationData(applicationData); - - return branchSession; - } - - /** - * End committed. - * - * @param globalSession the global session - * @throws TransactionException the transaction exception - */ - public static void endCommitted(GlobalSession globalSession) throws TransactionException { - globalSession.changeStatus(GlobalStatus.Committed); - globalSession.end(); - } - - /** - * End commit failed. - * - * @param globalSession the global session - * @throws TransactionException the transaction exception - */ - public static void endCommitFailed(GlobalSession globalSession) throws TransactionException { - globalSession.changeStatus(GlobalStatus.CommitFailed); - globalSession.end(); - } - - /** - * End rollbacked. - * - * @param globalSession the global session - * @throws TransactionException the transaction exception - */ - public static void endRollbacked(GlobalSession globalSession) throws TransactionException { - GlobalStatus currentStatus = globalSession.getStatus(); - if (isTimeoutGlobalStatus(currentStatus)) { - globalSession.changeStatus(GlobalStatus.TimeoutRollbacked); - } else { - globalSession.changeStatus(GlobalStatus.Rollbacked); - } - globalSession.end(); - } - - /** - * End rollback failed. - * - * @param globalSession the global session - * @throws TransactionException the transaction exception - */ - public static void endRollbackFailed(GlobalSession globalSession) throws TransactionException { - GlobalStatus currentStatus = globalSession.getStatus(); - if (isTimeoutGlobalStatus(currentStatus)) { - globalSession.changeStatus(GlobalStatus.TimeoutRollbackFailed); - } else { - globalSession.changeStatus(GlobalStatus.RollbackFailed); - } - globalSession.end(); - } - - public static boolean isTimeoutGlobalStatus(GlobalStatus status) { - return status == GlobalStatus.TimeoutRollbacked - || status == GlobalStatus.TimeoutRollbackFailed - || status == GlobalStatus.TimeoutRollbacking - || status == GlobalStatus.TimeoutRollbackRetrying; - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHolder.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHolder.java deleted file mode 100755 index a46e7001d..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionHolder.java +++ /dev/null @@ -1,242 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import io.seata.common.exception.ShouldNeverHappenException; -import io.seata.common.exception.StoreException; -import io.seata.common.util.StringUtils; -import io.seata.config.Configuration; -import io.seata.config.ConfigurationFactory; -import io.seata.core.constants.ConfigurationKeys; -import io.seata.core.exception.TransactionException; -import io.seata.core.model.GlobalStatus; -import io.seata.core.store.StoreMode; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.Collection; - -/** - * The type Session holder. - * - * @author sharajava - */ -public class SessionHolder { - - private static final Logger LOGGER = LoggerFactory.getLogger(SessionHolder.class); - - /** - * The constant CONFIG. - */ - protected static final Configuration CONFIG = ConfigurationFactory.getInstance(); - - /** - * The constant ROOT_SESSION_MANAGER_NAME. - */ - public static final String ROOT_SESSION_MANAGER_NAME = "root.data"; - /** - * The constant ASYNC_COMMITTING_SESSION_MANAGER_NAME. - */ - public static final String ASYNC_COMMITTING_SESSION_MANAGER_NAME = "async.commit.data"; - /** - * The constant RETRY_COMMITTING_SESSION_MANAGER_NAME. - */ - public static final String RETRY_COMMITTING_SESSION_MANAGER_NAME = "retry.commit.data"; - /** - * The constant RETRY_ROLLBACKING_SESSION_MANAGER_NAME. - */ - public static final String RETRY_ROLLBACKING_SESSION_MANAGER_NAME = "retry.rollback.data"; - - private static SessionManager ROOT_SESSION_MANAGER; - private static SessionManager ASYNC_COMMITTING_SESSION_MANAGER; - private static SessionManager RETRY_COMMITTING_SESSION_MANAGER; - private static SessionManager RETRY_ROLLBACKING_SESSION_MANAGER; - - /** - * Init. - * - * @param mode the store mode: file、db - * @throws IOException the io exception - */ - public static void init(String mode) throws IOException { - if (StringUtils.isBlank(mode)) { - //use default - mode = CONFIG.getConfig(ConfigurationKeys.STORE_MODE); - } - //the store mode - StoreMode storeMode = StoreMode.valueof(mode); - if (StoreMode.DB.equals(storeMode)) { - //TODO database store - - } else if (StoreMode.FILE.equals(storeMode)) { - //file store - String sessionStorePath = CONFIG.getConfig(ConfigurationKeys.STORE_FILE_DIR); - if (sessionStorePath == null) { - throw new StoreException("the {store.file.dir} is empty."); - } - ROOT_SESSION_MANAGER = new FileBasedSessionManager(ROOT_SESSION_MANAGER_NAME, sessionStorePath); - ASYNC_COMMITTING_SESSION_MANAGER = new DefaultSessionManager(ASYNC_COMMITTING_SESSION_MANAGER_NAME); - RETRY_COMMITTING_SESSION_MANAGER = new DefaultSessionManager(RETRY_COMMITTING_SESSION_MANAGER_NAME); - RETRY_ROLLBACKING_SESSION_MANAGER = new DefaultSessionManager(RETRY_ROLLBACKING_SESSION_MANAGER_NAME); - } else { - //unknown store - throw new IllegalArgumentException("unknown store mode:" + mode); - } - - if (ROOT_SESSION_MANAGER instanceof Reloadable) { - ((Reloadable) ROOT_SESSION_MANAGER).reload(); - - Collection reloadedSessions = ROOT_SESSION_MANAGER.allSessions(); - if (reloadedSessions != null && !reloadedSessions.isEmpty()) { - reloadedSessions.forEach(globalSession -> { - GlobalStatus globalStatus = globalSession.getStatus(); - switch (globalStatus) { - case UnKnown: - case Committed: - case CommitFailed: - case Rollbacked: - case RollbackFailed: - case TimeoutRollbacked: - case TimeoutRollbackFailed: - case Finished: - throw new ShouldNeverHappenException("Reloaded Session should NOT be " + globalStatus); - case AsyncCommitting: - try { - globalSession.addSessionLifecycleListener(getAsyncCommittingSessionManager()); - getAsyncCommittingSessionManager().addGlobalSession(globalSession); - } catch (TransactionException e) { - throw new ShouldNeverHappenException(e); - } - break; - default: { - ArrayList branchSessions = globalSession.getSortedBranches(); - // Lock - branchSessions.forEach(branchSession -> { - try { - branchSession.lock(); - } catch (TransactionException e) { - throw new ShouldNeverHappenException(e); - } - }); - - switch (globalStatus) { - case Committing: - case CommitRetrying: - try { - globalSession.addSessionLifecycleListener( - getRetryCommittingSessionManager()); - getRetryCommittingSessionManager().addGlobalSession(globalSession); - } catch (TransactionException e) { - throw new ShouldNeverHappenException(e); - } - break; - case Rollbacking: - case RollbackRetrying: - case TimeoutRollbacking: - case TimeoutRollbackRetrying: - try { - globalSession.addSessionLifecycleListener( - getRetryRollbackingSessionManager()); - getRetryRollbackingSessionManager().addGlobalSession(globalSession); - } catch (TransactionException e) { - throw new ShouldNeverHappenException(e); - } - break; - case Begin: - globalSession.setActive(true); - break; - default: - throw new ShouldNeverHappenException("NOT properly handled " + globalStatus); - } - - break; - - } - } - - }); - } - } - - } - - /** - * Gets root session manager. - * - * @return the root session manager - */ - public static final SessionManager getRootSessionManager() { - if (ROOT_SESSION_MANAGER == null) { - throw new ShouldNeverHappenException("SessionManager is NOT init!"); - } - return ROOT_SESSION_MANAGER; - } - - /** - * Gets async committing session manager. - * - * @return the async committing session manager - */ - public static final SessionManager getAsyncCommittingSessionManager() { - if (ASYNC_COMMITTING_SESSION_MANAGER == null) { - throw new ShouldNeverHappenException("SessionManager is NOT init!"); - } - return ASYNC_COMMITTING_SESSION_MANAGER; - } - - /** - * Gets retry committing session manager. - * - * @return the retry committing session manager - */ - public static final SessionManager getRetryCommittingSessionManager() { - if (RETRY_COMMITTING_SESSION_MANAGER == null) { - throw new ShouldNeverHappenException("SessionManager is NOT init!"); - } - return RETRY_COMMITTING_SESSION_MANAGER; - } - - /** - * Gets retry rollbacking session manager. - * - * @return the retry rollbacking session manager - */ - public static final SessionManager getRetryRollbackingSessionManager() { - if (RETRY_ROLLBACKING_SESSION_MANAGER == null) { - throw new ShouldNeverHappenException("SessionManager is NOT init!"); - } - return RETRY_ROLLBACKING_SESSION_MANAGER; - } - - /** - * Find global session. - * - * @param transactionId the transaction id - * @return the global session - */ - public static GlobalSession findGlobalSession(Long transactionId) { - return getRootSessionManager().findGlobalSession(transactionId); - } - - public static void destory() { - ROOT_SESSION_MANAGER.destroy(); - ASYNC_COMMITTING_SESSION_MANAGER.destroy(); - RETRY_COMMITTING_SESSION_MANAGER.destroy(); - RETRY_ROLLBACKING_SESSION_MANAGER.destroy(); - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycle.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycle.java deleted file mode 100755 index 9c46a2fd7..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycle.java +++ /dev/null @@ -1,89 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import io.seata.core.exception.TransactionException; -import io.seata.core.model.BranchStatus; -import io.seata.core.model.GlobalStatus; - -/** - * The interface Session lifecycle. - * - * @author sharajava - */ -public interface SessionLifecycle { - - /** - * Begin. - * - * @throws TransactionException the transaction exception - */ - void begin() throws TransactionException; - - /** - * Change status. - * - * @param status the status - * @throws TransactionException the transaction exception - */ - void changeStatus(GlobalStatus status) throws TransactionException; - - /** - * Change branch status. - * - * @param branchSession the branch session - * @param status the status - * @throws TransactionException the transaction exception - */ - void changeBranchStatus(BranchSession branchSession, BranchStatus status) throws TransactionException; - - /** - * Add branch. - * - * @param branchSession the branch session - * @throws TransactionException the transaction exception - */ - void addBranch(BranchSession branchSession) throws TransactionException; - - /** - * Remove branch. - * - * @param branchSession the branch session - * @throws TransactionException the transaction exception - */ - void removeBranch(BranchSession branchSession) throws TransactionException; - - /** - * Is active boolean. - * - * @return the boolean - */ - boolean isActive(); - - /** - * Close. - * - * @throws TransactionException the transaction exception - */ - void close() throws TransactionException; - - /** - * End. - * - * @throws TransactionException the transaction exception - */ - void end() throws TransactionException; -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycleListener.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycleListener.java deleted file mode 100755 index 54c415e92..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionLifecycleListener.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import io.seata.core.exception.TransactionException; -import io.seata.core.model.BranchStatus; -import io.seata.core.model.GlobalStatus; - -/** - * The interface Session lifecycle listener. - * - * @author sharajava - */ -public interface SessionLifecycleListener { - - /** - * On begin. - * - * @param globalSession the global session - * @throws TransactionException the transaction exception - */ - void onBegin(GlobalSession globalSession) throws TransactionException; - - /** - * On status change. - * - * @param globalSession the global session - * @param status the status - * @throws TransactionException the transaction exception - */ - void onStatusChange(GlobalSession globalSession, GlobalStatus status) throws TransactionException; - - /** - * On branch status change. - * - * @param globalSession the global session - * @param branchSession the branch session - * @param status the status - * @throws TransactionException the transaction exception - */ - void onBranchStatusChange(GlobalSession globalSession, BranchSession branchSession, BranchStatus status) - throws TransactionException; - - /** - * On add branch. - * - * @param globalSession the global session - * @param branchSession the branch session - * @throws TransactionException the transaction exception - */ - void onAddBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException; - - /** - * On remove branch. - * - * @param globalSession the global session - * @param branchSession the branch session - * @throws TransactionException the transaction exception - */ - void onRemoveBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException; - - /** - * On close. - * - * @param globalSession the global session - * @throws TransactionException the transaction exception - */ - void onClose(GlobalSession globalSession) throws TransactionException; - - /** - * On end. - * - * @param globalSession the global session - * @throws TransactionException the transaction exception - */ - void onEnd(GlobalSession globalSession) throws TransactionException; -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionManager.java deleted file mode 100755 index 2883d4404..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/session/SessionManager.java +++ /dev/null @@ -1,109 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.session; - -import java.util.Collection; -import java.util.List; - -import io.seata.core.exception.TransactionException; -import io.seata.core.model.BranchStatus; -import io.seata.core.model.GlobalStatus; -import io.seata.core.rpc.Disposable; - -/** - * The interface Session manager. - * - * @author sharajava - */ -public interface SessionManager extends SessionLifecycleListener, Disposable { - - /** - * Add global session. - * - * @param session the session - * @throws TransactionException the transaction exception - */ - void addGlobalSession(GlobalSession session) throws TransactionException; - - /** - * Find global session global session. - * - * @param transactionId the transaction id - * @return the global session - * @throws TransactionException the transaction exception - */ - GlobalSession findGlobalSession(Long transactionId); - - /** - * Update global session status. - * - * @param session the session - * @param status the status - * @throws TransactionException the transaction exception - */ - void updateGlobalSessionStatus(GlobalSession session, GlobalStatus status) throws TransactionException; - - /** - * Remove global session. - * - * @param session the session - * @throws TransactionException the transaction exception - */ - void removeGlobalSession(GlobalSession session) throws TransactionException; - - /** - * Add branch session. - * - * @param globalSession the global session - * @param session the session - * @throws TransactionException the transaction exception - */ - void addBranchSession(GlobalSession globalSession, BranchSession session) throws TransactionException; - - /** - * Update branch session status. - * - * @param session the session - * @param status the status - * @throws TransactionException the transaction exception - */ - void updateBranchSessionStatus(BranchSession session, BranchStatus status) throws TransactionException; - - /** - * Remove branch session. - * - * @param globalSession the global session - * @param session the session - * @throws TransactionException the transaction exception - */ - void removeBranchSession(GlobalSession globalSession, BranchSession session) throws TransactionException; - - /** - * All sessions collection. - * - * @return the collection - */ - Collection allSessions(); - - /** - * Find global sessions list. - * - * @param condition the condition - * @return the list - */ - List findGlobalSessions(SessionCondition condition); - -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FileTransactionStoreManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FileTransactionStoreManager.java deleted file mode 100755 index 9b3a32229..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FileTransactionStoreManager.java +++ /dev/null @@ -1,590 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.store; - -import io.seata.common.thread.NamedThreadFactory; -import io.seata.common.util.CollectionUtils; -import io.seata.server.session.BranchSession; -import io.seata.server.session.GlobalSession; -import io.seata.server.session.SessionCondition; -import io.seata.server.session.SessionManager; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.File; -import java.io.IOException; -import java.io.RandomAccessFile; -import java.nio.ByteBuffer; -import java.nio.channels.FileChannel; -import java.nio.file.Files; -import java.nio.file.StandardCopyOption; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.locks.ReentrantLock; - -/** - * The type File transaction store manager. - * - * @author jimin.jm @alibaba-inc.com - */ -public class FileTransactionStoreManager implements TransactionStoreManager { - - private static final Logger LOGGER = LoggerFactory.getLogger(FileTransactionStoreManager.class); - - private static final int MAX_THREAD_WRITE = 1; - - private ExecutorService fileWriteExecutor; - - private volatile boolean stopping = false; - - private static final int MAX_SHUTDOWN_RETRY = 3; - - private static final int SHUTDOWN_CHECK_INTERNAL = 1 * 1000; - - private static final int MAX_WRITE_RETRY = 5; - - private static final String HIS_DATA_FILENAME_POSTFIX = ".1"; - - private static final AtomicLong FILE_TRX_NUM = new AtomicLong(0); - - private static final AtomicLong FILE_FLUSH_NUM = new AtomicLong(0); - - private static final int MARK_SIZE = 4; - - private static final int MAX_WAIT_TIME_MILLS = 2 * 1000; - - private static final int MAX_FLUSH_TIME_MILLS = 2 * 1000; - - private static final int MAX_FLUSH_NUM = 10; - - private static int PER_FILE_BLOCK_SIZE = 65535 * 8; - - private static long MAX_TRX_TIMEOUT_MILLS = 30 * 60 * 1000; - - private static volatile long trxStartTimeMills = System.currentTimeMillis(); - - private File currDataFile; - - private RandomAccessFile currRaf; - - private FileChannel currFileChannel; - - private long recoverCurrOffset = 0; - - private long recoverHisOffset = 0; - - private SessionManager sessionManager; - - private String currFullFileName; - - private String hisFullFileName; - - private WriteDataFileRunnable writeDataFileRunnable; - - private ReentrantLock writeSessionLock = new ReentrantLock(); - - private volatile long lastModifiedTime; - - private static final int MAX_WRITE_BUFFER_SIZE = StoreConfig.getFileWriteBufferCacheSize(); - - private final ByteBuffer writeBuffer = ByteBuffer.allocateDirect(MAX_WRITE_BUFFER_SIZE); - - private static final FlushDiskMode FLUSH_DISK_MODE = StoreConfig.getFlushDiskMode(); - - private static final int MAX_WAIT_FOR_FLUSH_TIME_MILLS = 2 * 1000; - - private static final int INT_BYTE_SIZE = 4; - - /** - * Instantiates a new File transaction store manager. - * - * @param fullFileName the dir path - * @param sessionManager the session manager - * @throws IOException the io exception - */ - public FileTransactionStoreManager(String fullFileName, SessionManager sessionManager) throws IOException { - initFile(fullFileName); - fileWriteExecutor = - new ThreadPoolExecutor(MAX_THREAD_WRITE, MAX_THREAD_WRITE, Integer.MAX_VALUE, TimeUnit.MILLISECONDS, - new LinkedBlockingQueue(), - new NamedThreadFactory("fileTransactionStore", MAX_THREAD_WRITE, true)); - writeDataFileRunnable = new WriteDataFileRunnable(); - fileWriteExecutor.submit(writeDataFileRunnable); - this.sessionManager = sessionManager; - } - - private void initFile(String fullFileName) throws IOException { - this.currFullFileName = fullFileName; - this.hisFullFileName = fullFileName + HIS_DATA_FILENAME_POSTFIX; - try { - currDataFile = new File(currFullFileName); - if (!currDataFile.exists()) { - //create parent dir first - if (currDataFile.getParentFile() != null && !currDataFile.getParentFile().exists()) { - currDataFile.getParentFile().mkdirs(); - } - currDataFile.createNewFile(); - trxStartTimeMills = System.currentTimeMillis(); - } else { - trxStartTimeMills = currDataFile.lastModified(); - } - lastModifiedTime = System.currentTimeMillis(); - currRaf = new RandomAccessFile(currDataFile, "rw"); - currRaf.seek(currDataFile.length()); - currFileChannel = currRaf.getChannel(); - } catch (IOException exx) { - LOGGER.error("init file error," + exx.getMessage()); - throw exx; - } - } - - @Override - public boolean writeSession(LogOperation logOperation, SessionStorable session) { - writeSessionLock.lock(); - long curFileTrxNum; - try { - if (!writeDataFile(new TransactionWriteStore(session, logOperation).encode())) { - return false; - } - lastModifiedTime = System.currentTimeMillis(); - curFileTrxNum = FILE_TRX_NUM.incrementAndGet(); - if (curFileTrxNum % PER_FILE_BLOCK_SIZE == 0 && - (System.currentTimeMillis() - trxStartTimeMills) > MAX_TRX_TIMEOUT_MILLS) { - return saveHistory(); - } - } catch (Exception exx) { - LOGGER.error("writeSession error," + exx.getMessage()); - return false; - } finally { - writeSessionLock.unlock(); - } - flushDisk(curFileTrxNum, currFileChannel); - return true; - } - - private void flushDisk(long curFileNum, FileChannel currFileChannel) { - - if (FLUSH_DISK_MODE == FlushDiskMode.SYNC_MODEL) { - SyncFlushRequest syncFlushRequest = new SyncFlushRequest(curFileNum, currFileChannel); - writeDataFileRunnable.putRequest(syncFlushRequest); - syncFlushRequest.waitForFlush(MAX_WAIT_FOR_FLUSH_TIME_MILLS); - } else { - writeDataFileRunnable.putRequest(new AsyncFlushRequest(curFileNum, currFileChannel)); - } - } - - /** - * get all overTimeSessionStorables - * merge write file - * - * @throws IOException - */ - private boolean saveHistory() throws IOException { - boolean result; - try { - result = findTimeoutAndSave(); - writeDataFileRunnable.putRequest(new CloseFileRequest(currFileChannel, currRaf)); - Files.move(currDataFile.toPath(), new File(hisFullFileName).toPath(), StandardCopyOption.REPLACE_EXISTING); - } catch (IOException exx) { - LOGGER.error("save history data file error," + exx.getMessage()); - result = false; - } finally { - initFile(currFullFileName); - } - return result; - } - - private boolean findTimeoutAndSave() throws IOException { - List globalSessionsOverMaxTimeout = - sessionManager.findGlobalSessions(new SessionCondition(MAX_TRX_TIMEOUT_MILLS)); - if (CollectionUtils.isEmpty(globalSessionsOverMaxTimeout)) { - return true; - } - List listBytes = new ArrayList<>(); - int totalSize = 0; - // 1. find all data and merge - for (GlobalSession globalSession : globalSessionsOverMaxTimeout) { - TransactionWriteStore globalWriteStore = new TransactionWriteStore(globalSession, LogOperation.GLOBAL_ADD); - byte[] data = globalWriteStore.encode(); - listBytes.add(data); - totalSize += data.length + INT_BYTE_SIZE; - List branchSessIonsOverMaXTimeout = globalSession.getSortedBranches(); - if (null != branchSessIonsOverMaXTimeout) { - for (BranchSession branchSession : branchSessIonsOverMaXTimeout) { - TransactionWriteStore branchWriteStore = - new TransactionWriteStore(branchSession, LogOperation.BRANCH_ADD); - data = branchWriteStore.encode(); - listBytes.add(data); - totalSize += data.length + INT_BYTE_SIZE; - } - } - } - // 2. batch write - ByteBuffer byteBuffer = ByteBuffer.allocateDirect(totalSize); - for (byte[] bytes : listBytes) { - byteBuffer.putInt(bytes.length); - byteBuffer.put(bytes); - } - if (writeDataFileByBuffer(byteBuffer)) { - currFileChannel.force(false); - return true; - } - return false; - } - - @Override - public void shutdown() { - if (null != fileWriteExecutor) { - fileWriteExecutor.shutdown(); - stopping = true; - int retry = 0; - while (!fileWriteExecutor.isTerminated() && retry < MAX_SHUTDOWN_RETRY) { - ++retry; - try { - Thread.sleep(SHUTDOWN_CHECK_INTERNAL); - } catch (InterruptedException exx) { - } - } - if (retry >= MAX_SHUTDOWN_RETRY) { - fileWriteExecutor.shutdownNow(); - } - } - try { - currFileChannel.force(true); - } catch (IOException e) { - LOGGER.error("filechannel force error", e); - } - closeFile(currRaf); - } - - @Override - public List readWriteStoreFromFile(int readSize, boolean isHistory) { - File file = null; - long currentOffset = 0; - if (isHistory) { - file = new File(hisFullFileName); - currentOffset = recoverHisOffset; - } else { - file = new File(currFullFileName); - currentOffset = recoverCurrOffset; - } - if (file.exists()) { - return parseDataFile(file, readSize, currentOffset); - } - return null; - } - - @Override - public boolean hasRemaining(boolean isHistory) { - File file = null; - RandomAccessFile raf = null; - long currentOffset = 0; - if (isHistory) { - file = new File(hisFullFileName); - currentOffset = recoverHisOffset; - } else { - file = new File(currFullFileName); - currentOffset = recoverCurrOffset; - } - try { - raf = new RandomAccessFile(file, "r"); - return currentOffset < raf.length(); - - } catch (IOException exx) { - } finally { - closeFile(raf); - } - return false; - } - - private List parseDataFile(File file, int readSize, long currentOffset) { - List transactionWriteStores = new ArrayList<>(readSize); - RandomAccessFile raf = null; - FileChannel fileChannel = null; - try { - raf = new RandomAccessFile(file, "r"); - raf.seek(currentOffset); - fileChannel = raf.getChannel(); - fileChannel.position(currentOffset); - long size = raf.length(); - ByteBuffer buffSize = ByteBuffer.allocate(MARK_SIZE); - while (fileChannel.position() < size) { - try { - buffSize.clear(); - int avilReadSize = fileChannel.read(buffSize); - if (avilReadSize != MARK_SIZE) { - break; - } - buffSize.flip(); - int bodySize = buffSize.getInt(); - byte[] byBody = new byte[bodySize]; - ByteBuffer buffBody = ByteBuffer.wrap(byBody); - avilReadSize = fileChannel.read(buffBody); - if (avilReadSize != bodySize) { - break; - } - TransactionWriteStore writeStore = new TransactionWriteStore(); - writeStore.decode(byBody); - transactionWriteStores.add(writeStore); - if (transactionWriteStores.size() == readSize) { - break; - } - } catch (Exception ex) { - LOGGER.error("decode data file error:" + ex.getMessage()); - break; - } - } - return transactionWriteStores; - } catch (IOException exx) { - LOGGER.error("parse data file error:" + exx.getMessage() + ",file:" + file.getName()); - return null; - } finally { - try { - if (null != fileChannel) { - if (isHisFile(file)) { - recoverHisOffset = fileChannel.position(); - } else { - recoverCurrOffset = fileChannel.position(); - } - } - closeFile(raf); - } catch (IOException exx) { - LOGGER.error("file close error," + exx.getMessage()); - } - } - - } - - private boolean isHisFile(File file) { - - return file.getName().endsWith(HIS_DATA_FILENAME_POSTFIX); - } - - private void closeFile(RandomAccessFile raf) { - try { - if (null != raf) { - raf.close(); - raf = null; - } - } catch (IOException exx) { - LOGGER.error("file close error," + exx.getMessage()); - } - } - - private boolean writeDataFile(byte[] bs) { - if (bs == null) { - return false; - } - ByteBuffer byteBuffer = null; - - if (bs.length > MAX_WRITE_BUFFER_SIZE) { - //allocateNew - byteBuffer = ByteBuffer.allocateDirect(bs.length); - } else { - byteBuffer = writeBuffer; - //recycle - byteBuffer.clear(); - } - - byteBuffer.putInt(bs.length); - byteBuffer.put(bs); - return writeDataFileByBuffer(byteBuffer); - } - - private boolean writeDataFileByBuffer(ByteBuffer byteBuffer) { - byteBuffer.flip(); - for (int retry = 0; retry < MAX_WRITE_RETRY; retry++) { - try { - while (byteBuffer.hasRemaining()) { - currFileChannel.write(byteBuffer); - } - return true; - } catch (IOException exx) { - LOGGER.error("write data file error:" + exx.getMessage()); - } - } - LOGGER.error("write dataFile failed,retry more than :" + MAX_WRITE_RETRY); - return false; - } - - interface StoreRequest { - - } - abstract class AbstractFlushRequest implements StoreRequest{ - private final long curFileTrxNum; - - private final FileChannel curFileChannel; - - protected AbstractFlushRequest(long curFileTrxNum, FileChannel curFileChannel) { - this.curFileTrxNum = curFileTrxNum; - this.curFileChannel = curFileChannel; - } - public long getCurFileTrxNum() { - return curFileTrxNum; - } - - public FileChannel getCurFileChannel() { - return curFileChannel; - } - } - class SyncFlushRequest extends AbstractFlushRequest { - - private final CountDownLatch countDownLatch = new CountDownLatch(1); - - public SyncFlushRequest(long curFileTrxNum, FileChannel curFileChannel) { - super(curFileTrxNum, curFileChannel); - } - - - public void wakeupCustomer() { - this.countDownLatch.countDown(); - } - - public void waitForFlush(long timeout) { - try { - this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS); - } catch (InterruptedException e) { - LOGGER.error("Interrupted", e); - } - } - } - - class AsyncFlushRequest extends AbstractFlushRequest { - - public AsyncFlushRequest(long curFileTrxNum, FileChannel curFileChannel) { - super(curFileTrxNum, curFileChannel); - } - - } - - class CloseFileRequest implements StoreRequest { - - private FileChannel fileChannel; - - private RandomAccessFile file; - - public CloseFileRequest(FileChannel fileChannel, RandomAccessFile file) { - this.fileChannel = fileChannel; - this.file = file; - } - - public FileChannel getFileChannel() { - return fileChannel; - } - - public RandomAccessFile getFile() { - return file; - } - } - - /** - * The type Write data file runnable. - */ - class WriteDataFileRunnable implements Runnable { - - private LinkedBlockingQueue storeRequests = new LinkedBlockingQueue<>(); - - - public void putRequest(final StoreRequest request) { - storeRequests.add(request); - } - - - @Override - public void run() { - while (!stopping) { - try { - StoreRequest storeRequest = storeRequests.poll(MAX_WAIT_TIME_MILLS, TimeUnit.MILLISECONDS); - handleStoreRequest(storeRequest); - } catch (Exception exx) { - LOGGER.error("write file error", exx.getMessage()); - } - } - handleRestRequest(); - } - - /** - * handle the rest requests when stopping is true - */ - private void handleRestRequest() { - int remainNums = storeRequests.size(); - for (int i = 0; i < remainNums; i++) { - handleStoreRequest(storeRequests.poll()); - } - } - - private void handleStoreRequest(StoreRequest storeRequest) { - if (storeRequest == null){ - flushOnCondition(currFileChannel); - } - if (storeRequest instanceof SyncFlushRequest) { - syncFlush((SyncFlushRequest) storeRequest); - } else if (storeRequest instanceof AsyncFlushRequest) { - async((AsyncFlushRequest) storeRequest); - } else if (storeRequest instanceof CloseFileRequest) { - closeAndFlush((CloseFileRequest) storeRequest); - } - } - - private void closeAndFlush(CloseFileRequest req) { - long diff = FILE_TRX_NUM.get() - FILE_FLUSH_NUM.get(); - flush(req.getFileChannel()); - FILE_FLUSH_NUM.addAndGet(diff); - closeFile(currRaf); - } - - private void async(AsyncFlushRequest req) { - if (req.getCurFileTrxNum() < FILE_FLUSH_NUM.get()) { - flushOnCondition(req.getCurFileChannel()); - } - } - - private void syncFlush(SyncFlushRequest req) { - if (req.getCurFileTrxNum() < FILE_FLUSH_NUM.get()) { - long diff = FILE_TRX_NUM.get() - FILE_FLUSH_NUM.get(); - flush(req.getCurFileChannel()); - FILE_FLUSH_NUM.addAndGet(diff); - } - // notify - req.wakeupCustomer(); - } - - private void flushOnCondition(FileChannel fileChannel) { - if (FLUSH_DISK_MODE == FlushDiskMode.SYNC_MODEL) { - return; - } - long diff = FILE_TRX_NUM.get() - FILE_FLUSH_NUM.get(); - if (diff == 0) { - return; - } - if (diff % MAX_FLUSH_NUM == 0 || - System.currentTimeMillis() - lastModifiedTime > MAX_FLUSH_TIME_MILLS) { - flush(fileChannel); - FILE_FLUSH_NUM.addAndGet(diff); - } - } - - private void flush(FileChannel fileChannel) { - try { - fileChannel.force(false); - } catch (IOException exx) { - LOGGER.error("flush error:" + exx.getMessage()); - } - } - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FlushDiskMode.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FlushDiskMode.java deleted file mode 100755 index 34f9ba331..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/FlushDiskMode.java +++ /dev/null @@ -1,44 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.store; - -/** - * - * @author lizhao - */ -public enum FlushDiskMode { - /** - * sync flush disk - */ - SYNC_MODEL("sync"), - /** - * async flush disk - */ - ASYNC_MODEL("async"); - - private String modeStr; - - FlushDiskMode(String modeStr) { - this.modeStr = modeStr; - } - - public static FlushDiskMode findDiskMode(String modeStr) { - if (SYNC_MODEL.modeStr.equals(modeStr)){ - return SYNC_MODEL; - } - return ASYNC_MODEL; - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/SessionStorable.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/SessionStorable.java deleted file mode 100755 index 23b5dce75..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/SessionStorable.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.store; - -/** - * The interface Session storable. - * - * @author jimin.jm @alibaba-inc.com - */ -public interface SessionStorable { - - /** - * Encode byte [ ]. - * - * @return the byte [ ] - */ - byte[] encode(); - - /** - * Decode. - * - * @param src the src - */ - void decode(byte[] src); -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/StoreConfig.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/StoreConfig.java deleted file mode 100755 index 37e2b0f3b..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/StoreConfig.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.store; - -import io.seata.config.Configuration; -import io.seata.config.ConfigurationFactory; - -import static io.seata.core.constants.ConfigurationKeys.STORE_FILE_PREFIX; - - -/** - * @author lizhao - */ -public class StoreConfig { - - private static final Configuration CONFIGURATION = ConfigurationFactory.getInstance(); - - - // default 16kb - private static final int DEFAULT_MAX_BRANCH_SESSION_SIZE = 1024 * 16; - - // default 512b - private static final int DEFAULT_MAX_GLOBAL_SESSION_SIZE = 512; - - // default 16kb - private static final int DEFAULT_WRITE_BUFFER_SIZE = 1024 * 16; - - public static int getMaxBranchSessionSize() { - return CONFIGURATION.getInt(STORE_FILE_PREFIX + "max-branch-session-size", DEFAULT_MAX_BRANCH_SESSION_SIZE); - } - - public static int getMaxGlobalSessionSize() { - return CONFIGURATION.getInt(STORE_FILE_PREFIX + "max-global-session-size", DEFAULT_MAX_GLOBAL_SESSION_SIZE); - } - - public static int getFileWriteBufferCacheSize() { - return CONFIGURATION.getInt(STORE_FILE_PREFIX + "file-write-buffer-cache-size", DEFAULT_WRITE_BUFFER_SIZE); - } - - public static FlushDiskMode getFlushDiskMode() { - return FlushDiskMode.findDiskMode(CONFIGURATION.getConfig(STORE_FILE_PREFIX + "flush-disk-mode")); - } -} \ No newline at end of file diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionStoreManager.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionStoreManager.java deleted file mode 100755 index da7d44a4a..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionStoreManager.java +++ /dev/null @@ -1,118 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.store; - -import java.util.List; - -/** - * The interface Transaction store manager. - * - * @author jimin.jm @alibaba-inc.com - */ -public interface TransactionStoreManager { - - /** - * Write session boolean. - * - * @param logOperation the log operation - * @param session the session - * @return the boolean - */ - boolean writeSession(LogOperation logOperation, SessionStorable session); - - /** - * Shutdown. - */ - void shutdown(); - - /** - * Read write store from file list. - * - * @param readSize the read size - * @param isHistory the is history - * @return the list - */ - List readWriteStoreFromFile(int readSize, boolean isHistory); - - /** - * Has remaining boolean. - * - * @param isHistory the is history - * @return the boolean - */ - boolean hasRemaining(boolean isHistory); - - /** - * The enum Log operation. - */ - enum LogOperation { - - /** - * Global add log operation. - */ - GLOBAL_ADD((byte)1), - /** - * Global update log operation. - */ - GLOBAL_UPDATE((byte)2), - /** - * Global remove log operation. - */ - GLOBAL_REMOVE((byte)3), - /** - * Branch add log operation. - */ - BRANCH_ADD((byte)4), - /** - * Branch update log operation. - */ - BRANCH_UPDATE((byte)5), - /** - * Branch remove log operation. - */ - BRANCH_REMOVE((byte)6); - - private byte code; - - LogOperation(byte code) { - this.code = code; - } - - /** - * Gets code. - * - * @return the code - */ - public byte getCode() { - return this.code; - } - - /** - * Gets log operation by code. - * - * @param code the code - * @return the log operation by code - */ - public static LogOperation getLogOperationByCode(byte code) { - for (LogOperation temp : values()) { - if (temp.getCode() == code) { - return temp; - } - } - throw new IllegalArgumentException("Unknown LogOperation[" + code + "]"); - } - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionWriteStore.java b/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionWriteStore.java deleted file mode 100755 index f7ab6c542..000000000 --- a/pigx-visual/pigx-seata-server/src/main/java/io/seata/server/store/TransactionWriteStore.java +++ /dev/null @@ -1,129 +0,0 @@ -/* - * Copyright 1999-2019 Seata.io Group. - * - * Licensed 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 io.seata.server.store; - -import java.nio.ByteBuffer; - -import io.seata.common.exception.ShouldNeverHappenException; -import io.seata.server.session.BranchSession; -import io.seata.server.session.GlobalSession; -import io.seata.server.store.TransactionStoreManager.LogOperation; - -/** - * The type Transaction write store. - * - * @author jimin.jm @alibaba-inc.com - * @date 2018 /12/11 - */ -public class TransactionWriteStore implements SessionStorable { - private SessionStorable sessionRequest; - private LogOperation operate; - - /** - * Instantiates a new Transaction write store. - * - * @param sessionRequest the session request - * @param operate the operate - */ - public TransactionWriteStore(SessionStorable sessionRequest, LogOperation operate) { - this.sessionRequest = sessionRequest; - this.operate = operate; - } - - /** - * Instantiates a new Transaction write store. - */ - public TransactionWriteStore() {} - - /** - * Gets session request. - * - * @return the session request - */ - public SessionStorable getSessionRequest() { - return sessionRequest; - } - - /** - * Sets session request. - * - * @param sessionRequest the session request - */ - public void setSessionRequest(SessionStorable sessionRequest) { - this.sessionRequest = sessionRequest; - } - - /** - * Gets operate. - * - * @return the operate - */ - public LogOperation getOperate() { - return operate; - } - - /** - * Sets operate. - * - * @param operate the operate - */ - public void setOperate(LogOperation operate) { - this.operate = operate; - } - - @Override - public byte[] encode() { - byte[] bySessionRequest = this.sessionRequest.encode(); - byte byOpCode = this.getOperate().getCode(); - int len = bySessionRequest.length + 1; - byte[] byResult = new byte[len]; - ByteBuffer byteBuffer = ByteBuffer.wrap(byResult); - byteBuffer.put(bySessionRequest); - byteBuffer.put(byOpCode); - return byResult; - } - - @Override - public void decode(byte[] src) { - ByteBuffer byteBuffer = ByteBuffer.wrap(src); - byte[] bySessionRequest = new byte[src.length - 1]; - byteBuffer.get(bySessionRequest); - byte byOpCode = byteBuffer.get(); - this.operate = LogOperation.getLogOperationByCode(byOpCode); - SessionStorable tmpSessionStorable = getSessionInstanceByOperation(this.operate); - tmpSessionStorable.decode(bySessionRequest); - this.sessionRequest = tmpSessionStorable; - } - - private SessionStorable getSessionInstanceByOperation(LogOperation logOperation) { - SessionStorable sessionStorable = null; - switch (logOperation) { - case GLOBAL_ADD: - case GLOBAL_UPDATE: - case GLOBAL_REMOVE: - sessionStorable = new GlobalSession(); - break; - case BRANCH_ADD: - case BRANCH_UPDATE: - case BRANCH_REMOVE: - sessionStorable = new BranchSession(); - break; - default: - throw new ShouldNeverHappenException("incorrect logOperation"); - } - return sessionStorable; - } -} diff --git a/pigx-visual/pigx-seata-server/src/main/resources/logback.xml b/pigx-visual/pigx-seata-server/src/main/resources/logback.xml deleted file mode 100755 index 14508c9dd..000000000 --- a/pigx-visual/pigx-seata-server/src/main/resources/logback.xml +++ /dev/null @@ -1,54 +0,0 @@ - - - - - - - - - %date{yyyy-MM-dd HH:mm:ss.SSS} %-5level[%thread]%logger{56}.%method:%L -%msg%n - - - - - ${LOG_HOME}/seata-server.log - true - - ${LOG_HOME}/seata-server.log.%d{yyyy-MM-dd}.%i - 2GB - 7 - 7GB - true - - - %date %level %msg%n%n - UTF-8 - - - - - - - - - - - - - - diff --git a/pigx-visual/pigx-seata-server/src/main/resources/registry.conf b/pigx-visual/pigx-seata-server/src/main/resources/registry.conf deleted file mode 100755 index 79b92577a..000000000 --- a/pigx-visual/pigx-seata-server/src/main/resources/registry.conf +++ /dev/null @@ -1,71 +0,0 @@ -registry { - # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa - type = "nacos" - - nacos { - serverAddr = "localhost" - namespace = "public" - cluster = "default" - } - eureka { - serviceUrl = "http://localhost:1001/eureka" - application = "default" - weight = "1" - } - redis { - serverAddr = "localhost:6379" - db = "0" - } - zk { - cluster = "default" - serverAddr = "127.0.0.1:2181" - session.timeout = 6000 - connect.timeout = 2000 - } - consul { - cluster = "default" - serverAddr = "127.0.0.1:8500" - } - etcd3 { - cluster = "default" - serverAddr = "http://localhost:2379" - } - sofa { - serverAddr = "127.0.0.1:9603" - application = "default" - region = "DEFAULT_ZONE" - datacenter = "DefaultDataCenter" - cluster = "default" - group = "SEATA_GROUP" - addressWaitTime = "3000" - } - file { - name = "file.conf" - } -} - -config { - # file、nacos 、apollo、zk、consul - type = "nacos" - - nacos { - serverAddr = "pigx-nacos" - namespace = "public" - cluster = "default" - } - consul { - serverAddr = "127.0.0.1:8500" - } - apollo { - app.id = "seata-server" - apollo.meta = "http://192.168.1.204:8801" - } - zk { - serverAddr = "127.0.0.1:2181" - session.timeout = 6000 - connect.timeout = 2000 - } - file { - name = "file.conf" - } -} diff --git a/pigx-visual/pom.xml b/pigx-visual/pom.xml index b9787c19b..4241dbdbf 100644 --- a/pigx-visual/pom.xml +++ b/pigx-visual/pom.xml @@ -38,7 +38,6 @@ pigx-daemon-quartz pigx-monitor pigx-mp-manager - pigx-seata-server pigx-sso-client-demo pigx-tx-manager