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

format & import

Browse Source
This commit is contained in:
agapple
2017-01-19 13:07:53 +08:00
parent 92c1a51726
commit ff97dcbbc9
41 changed files with 22383 additions and 21999 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterNodeAccessStrategy.java
View File

@@ -114,5 +114,4 @@ public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy {
return zkClient; return zkClient;
} }
} }

3
common/pom.xml
View File

@@ -16,9 +16,7 @@
<dependency> <dependency>
<groupId>io.netty</groupId> <groupId>io.netty</groupId>
<artifactId>netty-all</artifactId> <artifactId>netty-all</artifactId>
<version>4.1.6.Final</version>
</dependency> </dependency>
<!-- zk --> <!-- zk -->
<dependency> <dependency>
<groupId>org.apache.zookeeper</groupId> <groupId>org.apache.zookeeper</groupId>
@@ -36,7 +34,6 @@
<dependency> <dependency>
<groupId>commons-lang</groupId> <groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId> <artifactId>commons-lang</artifactId>
<version>2.6</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.springframework</groupId> <groupId>org.springframework</groupId>

4
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java
View File

@@ -168,8 +168,8 @@ public class LogBuffer {
* @param newLimit The new limit value; must be non-negative and no larger * @param newLimit The new limit value; must be non-negative and no larger
* than this buffer's capacity * than this buffer's capacity
* @return This buffer * @return This buffer
* @throws IllegalArgumentException If the preconditions on <tt>newLimit</tt> * @throws IllegalArgumentException If the preconditions on
* do not hold * <tt>newLimit</tt> do not hold
*/ */
public final LogBuffer limit(int newLimit) { public final LogBuffer limit(int newLimit) {
if (origin + newLimit > buffer.length || newLimit < 0) throw new IllegalArgumentException("capacity excceed: " if (origin + newLimit > buffer.length || newLimit < 0) throw new IllegalArgumentException("capacity excceed: "

6
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/FormatDescriptionLogEvent.java
View File

@@ -141,9 +141,9 @@ public final class FormatDescriptionLogEvent extends StartLogEventV3 {
} }
} }
public FormatDescriptionLogEvent(final int binlogVersion,int binlogChecksum){ public FormatDescriptionLogEvent(final int binlogVersion, int binlogChecksum){
this(binlogVersion); this(binlogVersion);
this.header.checksumAlg = binlogChecksum; this.header.checksumAlg = binlogChecksum;
} }
public FormatDescriptionLogEvent(final int binlogVersion){ public FormatDescriptionLogEvent(final int binlogVersion){

4
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java
View File

@@ -140,7 +140,7 @@ public class MysqlConnector {
HeaderPacket quitHeader = new HeaderPacket(); HeaderPacket quitHeader = new HeaderPacket();
quitHeader.setPacketBodyLength(cmdBody.length); quitHeader.setPacketBodyLength(cmdBody.length);
quitHeader.setPacketSequenceNumber((byte) 0x00); quitHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.write(channel,quitHeader.toBytes(),cmdBody); PacketManager.write(channel, quitHeader.toBytes(), cmdBody);
} }
private void negotiate(SocketChannel channel) throws IOException { private void negotiate(SocketChannel channel) throws IOException {
@@ -177,7 +177,7 @@ public class MysqlConnector {
h.setPacketBodyLength(clientAuthPkgBody.length); h.setPacketBodyLength(clientAuthPkgBody.length);
h.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1)); h.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
PacketManager.write(channel,h.toBytes(), clientAuthPkgBody); PacketManager.write(channel, h.toBytes(), clientAuthPkgBody);
logger.info("client authentication packet is sent out."); logger.info("client authentication packet is sent out.");
// check auth result // check auth result

4
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java
View File

@@ -82,8 +82,8 @@ public class MysqlQueryExecutor {
rowDataPacket.fromBytes(body); rowDataPacket.fromBytes(body);
rowData.add(rowDataPacket); rowData.add(rowDataPacket);
} }
//未知，不知道是否需要锁定 // 未知，不知道是否需要锁定
//channel.lock();//锁定读 // channel.lock();//锁定读
ResultSetPacket resultSet = new ResultSetPacket(); ResultSetPacket resultSet = new ResultSetPacket();
resultSet.getFieldDescriptors().addAll(fields); resultSet.getFieldDescriptors().addAll(fields);

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

@@ -48,7 +48,7 @@ public class MysqlUpdateExecutor {
packet.fromBytes(body); packet.fromBytes(body);
throw new IOException(packet + "\n with command: " + updateString); throw new IOException(packet + "\n with command: " + updateString);
} }
//channel.lock();//锁定读 // channel.lock();//锁定读
OKPacket packet = new OKPacket(); OKPacket packet = new OKPacket();
packet.fromBytes(body); packet.fromBytes(body);

2
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/FieldPacket.java
View File

@@ -74,7 +74,7 @@ public class FieldPacket extends PacketWithHeaderPacket {
this.decimals = data[index]; this.decimals = data[index];
index++; index++;
// //
index += 2;//skip filter index += 2;// skip filter
// //
if (index < data.length) { if (index < data.length) {
reader.setIndex(index); reader.setIndex(index);

130
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java
View File

@@ -10,69 +10,77 @@ import java.net.SocketAddress;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
/** /**
* @author luoyaogui * @author luoyaogui 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。 2016-12-28
* 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。
* 2016-12-28
*/ */
public class SocketChannel { public class SocketChannel {
private Channel channel = null;
private Object lock = new Object();
private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(1024*1024*5);//缓存大小
public Channel getChannel() { private Channel channel = null;
return channel; private Object lock = new Object();
} private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(1024 * 1024 * 5); // 缓存大小
public void setChannel(Channel channel,boolean notify) {
this.channel = channel;
if(notify){//是否需要通知，主要时channel不可用时
synchronized(this){
notifyAll();
}
}
}
public void writeCache(ByteBuf buf){ public Channel getChannel() {
synchronized (lock) { return channel;
cache.discardReadBytes();//回收内存 }
cache.writeBytes(buf);
} public void setChannel(Channel channel, boolean notify) {
synchronized(this){ this.channel = channel;
notifyAll(); if (notify) {// 是否需要通知，主要时channel不可用时
} synchronized (this) {
} notifyAll();
public void writeChannel(byte[]... buf) throws IOException { }
if(channel != null && channel.isWritable()) }
channel.writeAndFlush(Unpooled.copiedBuffer(buf)); }
else
throw new IOException("write failed ! please checking !"); public void writeCache(ByteBuf buf) {
} synchronized (lock) {
public int read(ByteBuffer buffer) throws IOException { cache.discardReadBytes();// 回收内存
if(null == channel) cache.writeBytes(buf);
throw new IOException("socket has Interrupted !"); }
if(cache.readableBytes() < buffer.remaining()){ synchronized (this) {
synchronized(this){ notifyAll();
try {wait();} catch (InterruptedException e) {} }
} }
}else{
synchronized (lock){ public void writeChannel(byte[]... buf) throws IOException {
cache.readBytes(buffer); if (channel != null && channel.isWritable()) {
} channel.writeAndFlush(Unpooled.copiedBuffer(buf));
} } else {
return 0; throw new IOException("write failed ! please checking !");
} }
public boolean isConnected() { }
return channel!=null?true:false;
} public int read(ByteBuffer buffer) throws IOException {
public SocketAddress getRemoteSocketAddress(){ if (null == channel) throw new IOException("socket has Interrupted !");
return channel!=null?channel.remoteAddress():null; if (cache.readableBytes() < buffer.remaining()) {
} synchronized (this) {
public void close(){ try {
if(channel != null){ wait();
channel.close(); } catch (InterruptedException e) {
} }
channel = null; }
cache.discardReadBytes();//回收已占用的内存 } else {
cache.release();//释放整个内存 synchronized (lock) {
cache = null; cache.readBytes(buffer);
} }
}
return 0;
}
public boolean isConnected() {
return channel != null ? true : false;
}
public SocketAddress getRemoteSocketAddress() {
return channel != null ? channel.remoteAddress() : null;
}
public void close() {
if (channel != null) {
channel.close();
}
channel = null;
cache.discardReadBytes();// 回收已占用的内存
cache.release();// 释放整个内存
cache = null;
}
} }

132
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java
View File

@@ -20,76 +20,80 @@ import java.io.IOException;
import java.net.SocketAddress; import java.net.SocketAddress;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
/** /**
* @author luoyaogui * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
* 实现channel的管理（监听连接、读数据、回收）
* 2016-12-28
*/ */
@SuppressWarnings("rawtypes") @SuppressWarnings("rawtypes")
public abstract class SocketChannelPool { public abstract class SocketChannelPool {
private static EventLoopGroup group = new NioEventLoopGroup();//非阻塞IO线程组 private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组
private static Bootstrap boot = new Bootstrap();//主 private static Bootstrap boot = new Bootstrap(); // 主
private static Map<Channel,SocketChannel> chManager = new ConcurrentHashMap<Channel,SocketChannel>(); private static Map<Channel, SocketChannel> chManager = new ConcurrentHashMap<Channel, SocketChannel>();
static{ static {
boot.group(group).channel(NioSocketChannel.class) boot.group(group)
.option(ChannelOption.SO_RCVBUF, 32*1024) .channel(NioSocketChannel.class)
.option(ChannelOption.SO_SNDBUF, 32*1024) .option(ChannelOption.SO_RCVBUF, 32 * 1024)
.option(ChannelOption.TCP_NODELAY, true)//如果是延时敏感型应用，建议关闭Nagle算法 .option(ChannelOption.SO_SNDBUF, 32 * 1024)
.option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT) // 如果是延时敏感型应用，建议关闭Nagle算法
.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) // .option(ChannelOption.SO_KEEPALIVE, true)
.handler(new ChannelInitializer(){ .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
@Override .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
protected void initChannel(Channel arg0) throws Exception { //
arg0.pipeline().addLast(new BusinessHandler());//命令过滤和handler添加管理 .handler(new ChannelInitializer() {
}
});
}
public static SocketChannel open(SocketAddress address) throws Exception { @Override
final SocketChannel socket = new SocketChannel(); protected void initChannel(Channel arg0) throws Exception {
boot.connect(address).addListener(new ChannelFutureListener(){ arg0.pipeline().addLast(new BusinessHandler());// 命令过滤和handler添加管理
@Override }
public void operationComplete(ChannelFuture arg0) throws Exception { });
if(arg0.isSuccess()) }
socket.setChannel(arg0.channel(),false);
synchronized (socket) {
socket.notify();
}
}});
synchronized (socket) {
socket.wait();
}
if(null == socket.getChannel()){
throw new IOException("can't create socket!");
}
chManager.put(socket.getChannel(), socket);
return socket;
}
public static class BusinessHandler extends ChannelInboundHandlerAdapter { public static SocketChannel open(SocketAddress address) throws Exception {
private SocketChannel socket=null; final SocketChannel socket = new SocketChannel();
@Override boot.connect(address).addListener(new ChannelFutureListener() {
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
socket.setChannel(null,true); @Override
chManager.remove(ctx.channel());//移除 public void operationComplete(ChannelFuture arg0) throws Exception {
} if (arg0.isSuccess()) socket.setChannel(arg0.channel(), false);
@Override synchronized (socket) {
public void channelRead(ChannelHandlerContext ctx, Object msg) socket.notify();
throws Exception { }
if(null == socket) }
socket = chManager.get(ctx.channel()); });
if(socket != null){ synchronized (socket) {
socket.writeCache((ByteBuf) msg); socket.wait();
} }
ReferenceCountUtil.release(msg);//添加防止内存泄漏的 if (null == socket.getChannel()) {
} throw new IOException("can't create socket!");
@Override }
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) chManager.put(socket.getChannel(), socket);
throws Exception { return socket;
ctx.close(); }
}
} public static class BusinessHandler extends ChannelInboundHandlerAdapter {
private SocketChannel socket = null;
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
socket.setChannel(null, true);
chManager.remove(ctx.channel());// 移除
}
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (null == socket) {
socket = chManager.get(ctx.channel());
}
if (socket != null) {
socket.writeCache((ByteBuf) msg);
}
ReferenceCountUtil.release(msg);// 添加防止内存泄漏的
}
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
} }

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

@@ -52,6 +52,6 @@ public abstract class PacketManager {
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); write(ch, header.toBytes(), body);
} }
} }

1
driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java
View File

@@ -4,7 +4,6 @@ import java.io.IOException;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;

1
filter/src/test/java/com/alibaba/otter/canal/filter/AviaterFilterTest.java
View File

@@ -1,7 +1,6 @@
package com.alibaba.otter.canal.filter; package com.alibaba.otter.canal.filter;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
import com.alibaba.otter.canal.filter.aviater.AviaterELFilter; import com.alibaba.otter.canal.filter.aviater.AviaterELFilter;

3
filter/src/test/java/com/alibaba/otter/canal/filter/MutliAviaterFilterTest.java
View File

@@ -5,9 +5,8 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Assert;
import org.apache.commons.lang.RandomStringUtils; import org.apache.commons.lang.RandomStringUtils;
import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter; import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;

33
instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java
View File

@@ -1,5 +1,11 @@
package com.alibaba.otter.canal.instance.core; package com.alibaba.otter.canal.instance.core;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler; import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter; import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
@@ -16,30 +22,22 @@ import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.sink.CanalEventSink; import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.store.CanalEventStore; import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.model.Event; import com.alibaba.otter.canal.store.model.Event;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
/** /**
* Created with Intellig IDEA. * Created with Intellig IDEA. Author: yinxiu Date: 2016-01-07 Time: 22:26
* Author: yinxiu
* Date: 2016-01-07
* Time: 22:26
*/ */
public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance { public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance {
private static final Logger logger = LoggerFactory.getLogger(AbstractCanalInstance.class); private static final Logger logger = LoggerFactory.getLogger(AbstractCanalInstance.class);
protected Long canalId; // 和manager交互唯一标示 protected Long canalId; // 和manager交互唯一标示
protected String destination; // 队列名字 protected String destination; // 队列名字
protected CanalEventStore<Event> eventStore; // 有序队列 protected CanalEventStore<Event> eventStore; // 有序队列
protected CanalEventParser eventParser; // 解析对应的数据信息 protected CanalEventParser eventParser; // 解析对应的数据信息
protected CanalEventSink<List<CanalEntry.Entry>> eventSink; // 链接parse和store的桥接器 protected CanalEventSink<List<CanalEntry.Entry>> eventSink; // 链接parse和store的桥接器
protected CanalMetaManager metaManager; // 消费信息管理器 protected CanalMetaManager metaManager; // 消费信息管理器
protected CanalAlarmHandler alarmHandler; // alarm报警机制 protected CanalAlarmHandler alarmHandler; // alarm报警机制
@Override @Override
public boolean subscribeChange(ClientIdentity identity) { public boolean subscribeChange(ClientIdentity identity) {
@@ -121,7 +119,6 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can
alarmHandler.stop(); alarmHandler.stop();
} }
logger.info("stop successful...."); logger.info("stop successful....");
} }

4
instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java
View File

@@ -2,11 +2,11 @@ package com.alibaba.otter.canal.instance.spring;
import java.util.List; import java.util.List;
import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler; import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
import com.alibaba.otter.canal.meta.CanalMetaManager; import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.parse.CanalEventParser; import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry;
@@ -23,7 +23,7 @@ import com.alibaba.otter.canal.store.model.Event;
*/ */
public class CanalInstanceWithSpring extends AbstractCanalInstance { public class CanalInstanceWithSpring extends AbstractCanalInstance {
private static final Logger logger = LoggerFactory.getLogger(CanalInstanceWithSpring.class); private static final Logger logger = LoggerFactory.getLogger(CanalInstanceWithSpring.class);
public void start() { public void start() {
logger.info("start CannalInstance for {}-{} ", new Object[] { 1, destination }); logger.info("start CannalInstance for {}-{} ", new Object[] { 1, destination });

3
meta/src/test/java/com/alibaba/otter/canal/meta/FileMixedMetaManagerTest.java
View File

@@ -5,9 +5,8 @@ import java.io.IOException;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import org.junit.Assert;
import org.apache.commons.io.FileUtils; import org.apache.commons.io.FileUtils;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;

1
meta/src/test/java/com/alibaba/otter/canal/meta/MemoryMetaManagerTest.java
View File

@@ -3,7 +3,6 @@ package com.alibaba.otter.canal.meta;
import java.util.Map; import java.util.Map;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
import com.alibaba.otter.canal.protocol.position.PositionRange; import com.alibaba.otter.canal.protocol.position.PositionRange;

3
meta/src/test/java/com/alibaba/otter/canal/meta/MixedMetaManagerTest.java
View File

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.meta;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import org.junit.Assert;
import org.junit.After; import org.junit.After;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;

3
meta/src/test/java/com/alibaba/otter/canal/meta/PeriodMixedMetaManagerTest.java
View File

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.meta;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import org.junit.Assert;
import org.junit.After; import org.junit.After;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;

3
meta/src/test/java/com/alibaba/otter/canal/meta/ZooKeeperMetaManagerTest.java
View File

@@ -2,9 +2,8 @@ package com.alibaba.otter.canal.meta;
import java.util.Map; import java.util.Map;
import org.junit.Assert;
import org.junit.After; import org.junit.After;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;

41
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java
View File

@@ -83,30 +83,27 @@ public class LocalBinLogConnection implements ErosaConnection {
context.setLogPosition(new LogPosition(binlogfilename, binlogPosition)); context.setLogPosition(new LogPosition(binlogfilename, binlogPosition));
while (running) { while (running) {
boolean needContinue = true; boolean needContinue = true;
LogEvent event = null; LogEvent event = null;
L:while (fetcher.fetch()) { L: while (fetcher.fetch()) {
/*event = decoder.decode(fetcher, context); /*
if (event == null) { * event = decoder.decode(fetcher, context); if (event ==
throw new CanalParseException("parse failed"); * null) { throw new CanalParseException("parse failed"); }
} * if (!func.sink(event)) { needContinue = false; break; }
*/
if (!func.sink(event)) { do {
needContinue = false; if (event == null) {
break; event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition()
}*/ .getPosition());
} else {
event = decoder.decode(fetcher, context);
}
do { if (event != null && !func.sink(event)) {
if (event == null) { needContinue = false;
event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition().getPosition()); break L;
} else { }
event = decoder.decode(fetcher, context); } while (event != null);
}
if (event != null && !func.sink(event)) {
needContinue = false;
break L;
}
} while (event != null);
} }
if (needContinue) {// 读取下一个 if (needContinue) {// 读取下一个

24
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java
View File

@@ -42,8 +42,8 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
} }
@Override @Override
protected void preDump(ErosaConnection connection) { protected void preDump(ErosaConnection connection) {
metaConnection = buildMysqlConnection(); metaConnection = buildMysqlConnection();
try { try {
metaConnection.connect(); metaConnection.connect();
} catch (IOException e) { } catch (IOException e) {
@@ -52,11 +52,11 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
tableMetaCache = new TableMetaCache(metaConnection); tableMetaCache = new TableMetaCache(metaConnection);
((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache); ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache);
} }
@Override @Override
protected void afterDump(ErosaConnection connection) { protected void afterDump(ErosaConnection connection) {
if (metaConnection != null) { if (metaConnection != null) {
try { try {
metaConnection.disconnect(); metaConnection.disconnect();
} catch (IOException e) { } catch (IOException e) {
@@ -64,9 +64,9 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
.getAddress(), e); .getAddress(), e);
} }
} }
} }
public void start() throws CanalParseException { public void start() throws CanalParseException {
if (runningInfo == null) { // 第一次链接主库 if (runningInfo == null) { // 第一次链接主库
runningInfo = masterInfo; runningInfo = masterInfo;
} }
@@ -75,8 +75,8 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
} }
@Override @Override
public void stop() { public void stop() {
if (metaConnection != null) { if (metaConnection != null) {
try { try {
metaConnection.disconnect(); metaConnection.disconnect();
} catch (IOException e) { } catch (IOException e) {
@@ -90,9 +90,9 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
} }
super.stop(); super.stop();
} }
private ErosaConnection buildLocalBinLogConnection() { private ErosaConnection buildLocalBinLogConnection() {
LocalBinLogConnection connection = new LocalBinLogConnection(); LocalBinLogConnection connection = new LocalBinLogConnection();
connection.setBufferSize(this.bufferSize); connection.setBufferSize(this.bufferSize);

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
View File

@@ -2,7 +2,6 @@ package com.alibaba.otter.canal.parse.inbound.mysql;
import java.io.IOException; import java.io.IOException;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.Charset; import java.nio.charset.Charset;
import java.util.List; import java.util.List;
@@ -148,8 +147,7 @@ public class MysqlConnection implements ErosaConnection {
HeaderPacket binlogDumpHeader = new HeaderPacket(); HeaderPacket binlogDumpHeader = new HeaderPacket();
binlogDumpHeader.setPacketBodyLength(cmdBody.length); binlogDumpHeader.setPacketBodyLength(cmdBody.length);
binlogDumpHeader.setPacketSequenceNumber((byte) 0x00); binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.write(connector.getChannel(), binlogDumpHeader.toBytes(), PacketManager.write(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
cmdBody);
connector.setDumping(true); connector.setDumping(true);
} }

6
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java
View File

@@ -190,9 +190,9 @@ public class BinLogFileQueue {
files.addAll(FileUtils.listFiles(directory, new IOFileFilter() { files.addAll(FileUtils.listFiles(directory, new IOFileFilter() {
public boolean accept(File file) { public boolean accept(File file) {
Pattern pattern = Pattern.compile("\\d+$"); Pattern pattern = Pattern.compile("\\d+$");
Matcher matcher = pattern.matcher(file.getName()); Matcher matcher = pattern.matcher(file.getName());
return file.getName().startsWith(baseName) && matcher.find(); return file.getName().startsWith(baseName) && matcher.find();
} }
public boolean accept(File dir, String name) { public boolean accept(File dir, String name) {

3
parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java
View File

@@ -2,7 +2,6 @@ package com.alibaba.otter.canal.parse;
import java.io.IOException; import java.io.IOException;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
@@ -94,6 +93,6 @@ public class DirectLogFetcherTest {
HeaderPacket binlogDumpHeader = new HeaderPacket(); HeaderPacket binlogDumpHeader = new HeaderPacket();
binlogDumpHeader.setPacketBodyLength(cmdBody.length); binlogDumpHeader.setPacketBodyLength(cmdBody.length);
binlogDumpHeader.setPacketSequenceNumber((byte) 0x00); binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.write(connector.getChannel(), binlogDumpHeader.toBytes(),cmdBody); PacketManager.write(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
} }
} }

1
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/EventTransactionBufferTest.java
View File

@@ -6,7 +6,6 @@ import java.util.Date;
import java.util.List; import java.util.List;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback; import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback;

1
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java
View File

@@ -5,7 +5,6 @@ import java.nio.charset.Charset;
import java.util.List; import java.util.List;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest; import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;

1
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java
View File

@@ -5,7 +5,6 @@ import java.nio.charset.Charset;
import java.util.List; import java.util.List;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest; import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;

2
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java
View File

@@ -164,7 +164,7 @@ public class SimpleDdlParserTest {
Assert.assertEquals("retl_mark1", result.getOriTableName()); Assert.assertEquals("retl_mark1", result.getOriTableName());
Assert.assertEquals("retl_mark3", result.getTableName()); Assert.assertEquals("retl_mark3", result.getTableName());
//正则匹配test case // 正则匹配test case
queryString = "rename table totl_mark to totl_mark2"; queryString = "rename table totl_mark to totl_mark2";
result = SimpleDdlParser.parse(queryString, "retl"); result = SimpleDdlParser.parse(queryString, "retl");

3
parse/src/test/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManagerTest.java
View File

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.parse.index;
import java.io.File; import java.io.File;
import java.io.IOException; import java.io.IOException;
import org.junit.Assert;
import org.apache.commons.io.FileUtils; import org.apache.commons.io.FileUtils;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;

3
parse/src/test/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManagerTest.java
View File

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.parse.index;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.util.Date; import java.util.Date;
import org.junit.Assert;
import org.junit.After; import org.junit.After;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;

3
parse/src/test/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManagerTest.java
View File

@@ -1,8 +1,7 @@
package com.alibaba.otter.canal.parse.index; package com.alibaba.otter.canal.parse.index;
import org.junit.Assert;
import org.junit.After; import org.junit.After;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;

3
parse/src/test/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManagerTest.java
View File

@@ -1,8 +1,7 @@
package com.alibaba.otter.canal.parse.index; package com.alibaba.otter.canal.parse.index;
import org.junit.Assert;
import org.junit.After; import org.junit.After;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;

5
pom.xml
View File

@@ -199,6 +199,11 @@
<artifactId>netty</artifactId> <artifactId>netty</artifactId>
<version>3.2.5.Final</version> <version>3.2.5.Final</version>
</dependency> </dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.6.Final</version>
</dependency>
<dependency> <dependency>
<groupId>com.google.protobuf</groupId> <groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId> <artifactId>protobuf-java</artifactId>

24903
protocol/src/main/java/com/alibaba/otter/canal/protocol/CanalEntry.java
View File

File diff suppressed because it is too large Load Diff

18995
protocol/src/main/java/com/alibaba/otter/canal/protocol/CanalPacket.java
View File

File diff suppressed because it is too large Load Diff

2
server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java
View File

@@ -131,7 +131,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
*/ */
@Override @Override
public void subscribe(ClientIdentity clientIdentity) throws CanalServerException { public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
checkStart(clientIdentity.getDestination()); checkStart(clientIdentity.getDestination());
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
if (!canalInstance.getMetaManager().isStart()) { if (!canalInstance.getMetaManager().isStart()) {

3
store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreMultiThreadTest.java
View File

@@ -7,10 +7,9 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import org.junit.Assert;
import org.apache.commons.lang.math.RandomUtils; import org.apache.commons.lang.math.RandomUtils;
import org.junit.After; import org.junit.After;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;