/*
|
* 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.exception.StoreException;
|
import io.seata.core.exception.AbstractExceptionHandler;
|
import io.seata.core.exception.TransactionException;
|
import io.seata.core.exception.TransactionExceptionCode;
|
import io.seata.core.model.GlobalStatus;
|
import io.seata.core.protocol.transaction.*;
|
import io.seata.core.rpc.RpcContext;
|
import io.seata.server.session.GlobalSession;
|
import io.seata.server.session.SessionHolder;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
|
/**
|
* The type Abstract tc inbound handler.
|
*
|
* @author sharajava
|
*/
|
public abstract class AbstractTCInboundHandler extends AbstractExceptionHandler implements TCInboundHandler {
|
|
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractTCInboundHandler.class);
|
|
@Override
|
public GlobalBeginResponse handle(GlobalBeginRequest request, final RpcContext rpcContext) {
|
GlobalBeginResponse response = new GlobalBeginResponse();
|
exceptionHandleTemplate(new AbstractCallback<GlobalBeginRequest, GlobalBeginResponse>() {
|
@Override
|
public void execute(GlobalBeginRequest request, GlobalBeginResponse response) throws TransactionException {
|
try {
|
doGlobalBegin(request, response, rpcContext);
|
} catch (StoreException e) {
|
throw new TransactionException(TransactionExceptionCode.FailedStore,
|
String.format("begin global request failed. xid=%s, msg=%s", response.getXid(), e.getMessage()),
|
e);
|
}
|
}
|
}, 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();
|
response.setGlobalStatus(GlobalStatus.Committing);
|
exceptionHandleTemplate(new AbstractCallback<GlobalCommitRequest, GlobalCommitResponse>() {
|
@Override
|
public void execute(GlobalCommitRequest request, GlobalCommitResponse response)
|
throws TransactionException {
|
try {
|
doGlobalCommit(request, response, rpcContext);
|
} catch (StoreException e) {
|
throw new TransactionException(TransactionExceptionCode.FailedStore,
|
String.format("global commit request failed. xid=%s, msg=%s", request.getXid(), e.getMessage()),
|
e);
|
}
|
}
|
@Override
|
public void onTransactionException(GlobalCommitRequest request, GlobalCommitResponse response,
|
TransactionException tex) {
|
super.onTransactionException(request, response, tex);
|
checkTransactionStatus(request, response);
|
}
|
|
@Override
|
public void onException(GlobalCommitRequest request, GlobalCommitResponse response, Exception rex) {
|
super.onException(request, response, rex);
|
checkTransactionStatus(request, response);
|
}
|
|
|
}, 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();
|
response.setGlobalStatus(GlobalStatus.Rollbacking);
|
exceptionHandleTemplate(new AbstractCallback<GlobalRollbackRequest, GlobalRollbackResponse>() {
|
@Override
|
public void execute(GlobalRollbackRequest request, GlobalRollbackResponse response)
|
throws TransactionException {
|
try {
|
doGlobalRollback(request, response, rpcContext);
|
} catch (StoreException e) {
|
throw new TransactionException(TransactionExceptionCode.FailedStore, String
|
.format("global rollback request failed. xid=%s, msg=%s", request.getXid(), e.getMessage()), e);
|
}
|
}
|
|
@Override
|
public void onTransactionException(GlobalRollbackRequest request, GlobalRollbackResponse response,
|
TransactionException tex) {
|
super.onTransactionException(request, response, tex);
|
// may be appears StoreException outer layer method catch
|
checkTransactionStatus(request, response);
|
}
|
|
@Override
|
public void onException(GlobalRollbackRequest request, GlobalRollbackResponse response, Exception rex) {
|
super.onException(request, response, rex);
|
// may be appears StoreException outer layer method catch
|
checkTransactionStatus(request, response);
|
}
|
}, 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<BranchRegisterRequest, BranchRegisterResponse>() {
|
@Override
|
public void execute(BranchRegisterRequest request, BranchRegisterResponse response)
|
throws TransactionException {
|
try {
|
doBranchRegister(request, response, rpcContext);
|
} catch (StoreException e) {
|
throw new TransactionException(TransactionExceptionCode.FailedStore, String
|
.format("branch register request failed. xid=%s, msg=%s", request.getXid(), e.getMessage()), e);
|
}
|
}
|
}, 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<BranchReportRequest, BranchReportResponse>() {
|
@Override
|
public void execute(BranchReportRequest request, BranchReportResponse response)
|
throws TransactionException {
|
try {
|
doBranchReport(request, response, rpcContext);
|
} catch (StoreException e) {
|
throw new TransactionException(TransactionExceptionCode.FailedStore, String
|
.format("branch report request failed. xid=%s, branchId=%s, msg=%s", request.getXid(),
|
request.getBranchId(), e.getMessage()), e);
|
}
|
}
|
}, 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<GlobalLockQueryRequest, GlobalLockQueryResponse>() {
|
@Override
|
public void execute(GlobalLockQueryRequest request, GlobalLockQueryResponse response)
|
throws TransactionException {
|
try {
|
doLockCheck(request, response, rpcContext);
|
} catch (StoreException e) {
|
throw new TransactionException(TransactionExceptionCode.FailedStore, String
|
.format("global lock query request failed. xid=%s, msg=%s", request.getXid(), e.getMessage()),
|
e);
|
}
|
}
|
}, 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();
|
response.setGlobalStatus(GlobalStatus.UnKnown);
|
exceptionHandleTemplate(new AbstractCallback<GlobalStatusRequest, GlobalStatusResponse>() {
|
@Override
|
public void execute(GlobalStatusRequest request, GlobalStatusResponse response)
|
throws TransactionException {
|
try {
|
doGlobalStatus(request, response, rpcContext);
|
} catch (StoreException e) {
|
throw new TransactionException(TransactionExceptionCode.FailedStore,
|
String.format("global status request failed. xid=%s, msg=%s", request.getXid(), e.getMessage()),
|
e);
|
}
|
}
|
|
@Override
|
public void onTransactionException(GlobalStatusRequest request, GlobalStatusResponse response,
|
TransactionException tex) {
|
super.onTransactionException(request, response, tex);
|
checkTransactionStatus(request, response);
|
}
|
|
@Override
|
public void onException(GlobalStatusRequest request, GlobalStatusResponse response, Exception rex) {
|
super.onException(request, response, rex);
|
checkTransactionStatus(request, response);
|
}
|
}, 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;
|
|
@Override
|
public GlobalReportResponse handle(GlobalReportRequest request, final RpcContext rpcContext) {
|
GlobalReportResponse response = new GlobalReportResponse();
|
response.setGlobalStatus(request.getGlobalStatus());
|
exceptionHandleTemplate(new AbstractCallback<GlobalReportRequest, GlobalReportResponse>() {
|
@Override
|
public void execute(GlobalReportRequest request, GlobalReportResponse response)
|
throws TransactionException {
|
doGlobalReport(request, response, rpcContext);
|
}
|
}, request, response);
|
return response;
|
}
|
|
/**
|
* Do global report.
|
*
|
* @param request the request
|
* @param response the response
|
* @param rpcContext the rpc context
|
* @throws TransactionException the transaction exception
|
*/
|
protected abstract void doGlobalReport(GlobalReportRequest request, GlobalReportResponse response,
|
RpcContext rpcContext) throws TransactionException;
|
|
private void checkTransactionStatus(AbstractGlobalEndRequest request, AbstractGlobalEndResponse response) {
|
try {
|
GlobalSession globalSession = SessionHolder.findGlobalSession(request.getXid(), false);
|
if (globalSession != null) {
|
response.setGlobalStatus(globalSession.getStatus());
|
} else {
|
response.setGlobalStatus(GlobalStatus.Finished);
|
}
|
} catch (Exception exx) {
|
LOGGER.error("check transaction status error,{}]", exx.getMessage());
|
}
|
}
|
}
|
