forked from MyCATApache/abandomed
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request MyCATApache#6 from ynfeng/state_sync
前端状态同步到后端基础完成
- Loading branch information
Showing
6 changed files
with
219 additions
and
44 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
111 changes: 111 additions & 0 deletions
111
source/src/main/java/io/mycat/mycat2/tasks/BackendSynchronzationTask.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| package io.mycat.mycat2.tasks; | ||
|
|
||
| import io.mycat.mycat2.MySQLSession; | ||
| import io.mycat.mysql.packet.ErrorPacket; | ||
| import io.mycat.mysql.packet.MySQLPacket; | ||
| import io.mycat.mysql.packet.QueryPacket; | ||
| import io.mycat.proxy.ProxyBuffer; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| /** | ||
| * Created by ynfeng on 2017/8/13. | ||
| * <p> | ||
| * 同步状态至后端数据库，包括：字符集，事务，隔离级别等 | ||
| */ | ||
| public class BackendSynchronzationTask implements BackendIOTask<MySQLSession> { | ||
| private static Logger logger = LoggerFactory.getLogger(BackendSynchronzationTask.class); | ||
| private AsynTaskCallBack callBack; | ||
| private MySQLSession session; | ||
| private static QueryPacket[] CMDS = new QueryPacket[3]; | ||
| private int processCmd = 0; | ||
| private ErrorPacket errPkg; | ||
|
|
||
| static { | ||
| QueryPacket isolationSynCmd = new QueryPacket(); | ||
| isolationSynCmd.packetId = 0; | ||
|
|
||
| QueryPacket charsetSynCmd = new QueryPacket(); | ||
| charsetSynCmd.packetId = 0; | ||
|
|
||
| QueryPacket transactionSynCmd = new QueryPacket(); | ||
| transactionSynCmd.packetId = 0; | ||
|
|
||
| CMDS[0] = isolationSynCmd; | ||
| CMDS[1] = charsetSynCmd; | ||
| CMDS[2] = transactionSynCmd; | ||
| } | ||
|
|
||
| public BackendSynchronzationTask(MySQLSession session) throws IOException { | ||
| this.processCmd = 0; | ||
| this.session = session; | ||
| syncState(session); | ||
| } | ||
|
|
||
| private void syncState(MySQLSession session) throws IOException { | ||
| logger.info("synchronzation state to bakcend.session=" + session.toString()); | ||
| ProxyBuffer frontBuffer = session.frontBuffer; | ||
| frontBuffer.reset(); | ||
| //TODO 字符集映射和前端事务设置还未完成，这里只用隔离级别模拟实现(其实都是SET xxx效果一样)，回头补充 | ||
| switch (processCmd) { | ||
| case 1: | ||
| case 2: | ||
| case 0: | ||
| CMDS[processCmd].sql = session.isolation.getCmd(); | ||
| CMDS[processCmd].write(frontBuffer); | ||
|
|
||
| frontBuffer.flip(); | ||
| session.writeToChannel(frontBuffer, session.backendChannel); | ||
| processCmd++; | ||
| break; | ||
| default: | ||
| this.finished(true); | ||
| break; | ||
| } | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public void onBackendConnect(MySQLSession userSession, boolean success, String msg) throws IOException { | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public void onBackendRead(MySQLSession session) throws IOException { | ||
| session.frontBuffer.reset(); | ||
| if (!session.readSocket(false) | ||
| || !session.resolveMySQLPackage(session.frontBuffer, session.curBackendMSQLPackgInf, false)) {// 没有读到数据或者报文不完整 | ||
| return; | ||
| } | ||
| if (session.curBackendMSQLPackgInf.pkgType == MySQLPacket.OK_PACKET) { | ||
| syncState(session); | ||
| } else { | ||
| //TODO 同步失败如何处理？？是否应该关闭此连接？？ | ||
| errPkg = new ErrorPacket(); | ||
| errPkg.read(session.frontBuffer); | ||
| logger.warn("backend state sync Error.Err No. " + errPkg.errno + "," + errPkg.message); | ||
| this.finished(false); | ||
| } | ||
| } | ||
|
|
||
| private void finished(boolean success) throws IOException { | ||
| callBack.finished(session, this, success, this.errPkg); | ||
| } | ||
|
|
||
| @Override | ||
| public void onBackendWrite(MySQLSession session) throws IOException { | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public void onBackendSocketClosed(MySQLSession userSession, boolean normal) { | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public void setCallback(AsynTaskCallBack callBack) { | ||
| this.callBack = callBack; | ||
| } | ||
| } |
90 changes: 52 additions & 38 deletions
90
.../main/java/io/mycat/mysql/Isolations.java → ...c/main/java/io/mycat/mysql/Isolation.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,38 +1,52 @@ | ||
| /* | ||
| * Copyright (c) 2013, OpenCloudDB/MyCAT and/or its affiliates. All rights reserved. | ||
| * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. | ||
| * | ||
| * This code is free software;Designed and Developed mainly by many Chinese | ||
| * opensource volunteers. you can redistribute it and/or modify it under the | ||
| * terms of the GNU General Public License version 2 only, as published by the | ||
| * Free Software Foundation. | ||
| * | ||
| * This code is distributed in the hope that it will be useful, but WITHOUT | ||
| * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or | ||
| * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License | ||
| * version 2 for more details (a copy is included in the LICENSE file that | ||
| * accompanied this code). | ||
| * | ||
| * You should have received a copy of the GNU General Public License version | ||
| * 2 along with this work; if not, write to the Free Software Foundation, | ||
| * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. | ||
| * | ||
| * Any questions about this component can be directed to it's project Web address | ||
| * https://code.google.com/p/opencloudb/. | ||
| * | ||
| */ | ||
| package io.mycat.mysql; | ||
|
|
||
| /** | ||
| * 事务隔离级别定义 | ||
| * | ||
| * @author mycat | ||
| */ | ||
| public interface Isolations { | ||
|
|
||
| public static final int READ_UNCOMMITTED = 1; | ||
| public static final int READ_COMMITTED = 2; | ||
| public static final int REPEATED_READ = 3; | ||
| public static final int SERIALIZABLE = 4; | ||
|
|
||
| } | ||
| /* | ||
| * Copyright (c) 2013, OpenCloudDB/MyCAT and/or its affiliates. All rights reserved. | ||
| * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. | ||
| * | ||
| * This code is free software;Designed and Developed mainly by many Chinese | ||
| * opensource volunteers. you can redistribute it and/or modify it under the | ||
| * terms of the GNU General Public License version 2 only, as published by the | ||
| * Free Software Foundation. | ||
| * | ||
| * This code is distributed in the hope that it will be useful, but WITHOUT | ||
| * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or | ||
| * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License | ||
| * version 2 for more details (a copy is included in the LICENSE file that | ||
| * accompanied this code). | ||
| * | ||
| * You should have received a copy of the GNU General Public License version | ||
| * 2 along with this work; if not, write to the Free Software Foundation, | ||
| * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. | ||
| * | ||
| * Any questions about this component can be directed to it's project Web address | ||
| * https://code.google.com/p/opencloudb/. | ||
| * | ||
| */ | ||
| package io.mycat.mysql; | ||
|
|
||
|
|
||
| /** | ||
| * 事务隔离级别定义 | ||
| * | ||
| * @author mycat, ynfeng | ||
| */ | ||
| public enum Isolation { | ||
| READ_UNCOMMITTED("SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;"), | ||
| READ_COMMITTED("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;"), | ||
| REPEATED_READ("SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;"), | ||
| SERIALIZABLE("SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;"); | ||
|
|
||
| private String cmd; | ||
|
|
||
| Isolation(String cmd) { | ||
| this.cmd = cmd; | ||
| } | ||
|
|
||
| public String getCmd() { | ||
| return cmd; | ||
| } | ||
|
|
||
| public void setCmd(String cmd) { | ||
| this.cmd = cmd; | ||
| } | ||
| } | ||
|
|
29 changes: 29 additions & 0 deletions
29
source/src/main/java/io/mycat/mysql/packet/QueryPacket.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| package io.mycat.mysql.packet; | ||
|
|
||
| import io.mycat.proxy.ProxyBuffer; | ||
|
|
||
| /** | ||
| * Created by ynfeng on 2017/8/13. | ||
| */ | ||
| public class QueryPacket extends MySQLPacket { | ||
| public String sql; | ||
| private byte pkgType = MySQLPacket.COM_QUERY; | ||
|
|
||
| @Override | ||
| public int calcPacketSize() { | ||
| return sql.length() + 1; | ||
| } | ||
|
|
||
| @Override | ||
| protected String getPacketInfo() { | ||
| return "A COM_QUERY packet:" + sql; | ||
| } | ||
|
|
||
| @Override | ||
| public void write(ProxyBuffer buffer) { | ||
| buffer.writeFixInt(3, calcPacketSize()); | ||
| buffer.writeByte(packetId); | ||
| buffer.writeByte(pkgType); | ||
| buffer.writeFixString(sql); | ||
| } | ||
| } |