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 PacketManager.java

Browse Source
This commit is contained in:
luoyaogui
2017-06-27 15:03:53 +08:00
committed by GitHub
parent 9ea88c7863
commit b6090da0e5
1 changed files with 8 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java
View File

@@ -1,7 +1,6 @@
package com.alibaba.otter.canal.parse.driver.mysql.utils; package com.alibaba.otter.canal.parse.driver.mysql.utils;
import java.io.IOException; import java.io.IOException;
import java.nio.ByteBuffer;
import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
@@ -10,48 +9,26 @@ public abstract class PacketManager {
public static HeaderPacket readHeader(SocketChannel ch, int len) throws IOException { public static HeaderPacket readHeader(SocketChannel ch, int len) throws IOException {
HeaderPacket header = new HeaderPacket(); HeaderPacket header = new HeaderPacket();
header.fromBytes(readBytesAsBuffer(ch, len).array()); header.fromBytes(ch.read(len));
return header; return header;
} }
public static ByteBuffer readBytesAsBuffer(SocketChannel ch, int len) throws IOException {
ByteBuffer buffer = ByteBuffer.allocate(len);
while (buffer.hasRemaining()) {
int readNum = ch.read(buffer);
if (readNum == -1) {
throw new IOException("Unexpected End Stream");
}
}
return buffer;
}
public static byte[] readBytes(SocketChannel ch, int len) throws IOException { public static byte[] readBytes(SocketChannel ch, int len) throws IOException {
return readBytesAsBuffer(ch, len).array(); return ch.read(len);
} }
/** public static void writePkg(SocketChannel ch, byte[]... srcs) throws IOException {
* Since We r using blocking IO, so we will just write once and assert the
* length to simplify the read operation.<br>
* If the block write doesn't work as we expected, we will change this
* implementation as per the result.
*
* @param ch
* @param len
* @return
* @throws IOException
*/
public static void write(SocketChannel ch, byte[]... srcs) throws IOException {
ch.writeChannel(srcs); ch.writeChannel(srcs);
} }
public static void write(SocketChannel ch, byte[] body) throws IOException { public static void writeBody(SocketChannel ch, byte[] body) throws IOException {
write(ch, body, (byte) 0); writeBody0(ch, body, (byte) 0);
} }
public static void write(SocketChannel ch, byte[] body, byte packetSeqNumber) throws IOException { public static void writeBody0(SocketChannel ch, byte[] body, byte packetSeqNumber) throws IOException {
HeaderPacket header = new HeaderPacket(); HeaderPacket header = new HeaderPacket();
header.setPacketBodyLength(body.length); header.setPacketBodyLength(body.length);
header.setPacketSequenceNumber(packetSeqNumber); header.setPacketSequenceNumber(packetSeqNumber);
write(ch, header.toBytes(),body); ch.writeChannel(header.toBytes(),body);
} }
} }