chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update MysqlUpdateExecutor.java

Browse Source
This commit is contained in:
luoyaogui
2017-06-27 15:05:33 +08:00
committed by GitHub
parent 523130cf8b
commit 3dba400624
1 changed files with 7 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlUpdateExecutor.java
View File

@@ -1,6 +1,7 @@
package com.alibaba.otter.canal.parse.driver.mysql; package com.alibaba.otter.canal.parse.driver.mysql;
import java.io.IOException; import java.io.IOException;
import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -8,7 +9,6 @@ import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QueryCommandPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QueryCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.OKPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.OKPacket;
import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager; import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
/** /**
@@ -21,34 +21,33 @@ public class MysqlUpdateExecutor {
private static final Logger logger = LoggerFactory.getLogger(MysqlUpdateExecutor.class); private static final Logger logger = LoggerFactory.getLogger(MysqlUpdateExecutor.class);
private SocketChannel channel; private MysqlConnector connector;
public MysqlUpdateExecutor(MysqlConnector connector){ public MysqlUpdateExecutor(MysqlConnector connector){
if (!connector.isConnected()) { if (!connector.isConnected()) {
throw new RuntimeException("should execute connector.connect() first"); throw new RuntimeException("should execute connector.connect() first");
} }
this.channel = connector.getChannel(); this.connector = connector;
} }
public MysqlUpdateExecutor(SocketChannel ch){ /* public MysqlUpdateExecutor(SocketChannel ch){
this.channel = ch; this.channel = ch;
} }*/
public OKPacket update(String updateString) throws IOException { public OKPacket update(String updateString) throws IOException {
QueryCommandPacket cmd = new QueryCommandPacket(); QueryCommandPacket cmd = new QueryCommandPacket();
cmd.setQueryString(updateString); cmd.setQueryString(updateString);
byte[] bodyBytes = cmd.toBytes(); byte[] bodyBytes = cmd.toBytes();
PacketManager.write(channel, bodyBytes); PacketManager.writeBody(connector.getChannel(), bodyBytes);
logger.debug("read update result..."); logger.debug("read update result...");
byte[] body = PacketManager.readBytes(channel, PacketManager.readHeader(channel, 4).getPacketBodyLength()); byte[] body = PacketManager.readBytes(connector.getChannel(), PacketManager.readHeader(connector.getChannel(), 4).getPacketBodyLength());
if (body[0] < 0) { if (body[0] < 0) {
ErrorPacket packet = new ErrorPacket(); ErrorPacket packet = new ErrorPacket();
packet.fromBytes(body); packet.fromBytes(body);
throw new IOException(packet + "\n with command: " + updateString); throw new IOException(packet + "\n with command: " + updateString);
} }
//channel.lock();//锁定读
OKPacket packet = new OKPacket(); OKPacket packet = new OKPacket();
packet.fromBytes(body); packet.fromBytes(body);