package com.alibaba.otter.canal.parse.driver.mysql.socket;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;

/**
 * @author luoyaogui
 * 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。
 * 2016-12-28
 */
public class SocketChannel {
	private Channel channel = null;
	private Object lock = new Object();
	private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(1024*1024*5);//缓存大小

	public Channel getChannel() {
		return channel;
	}
	public void setChannel(Channel channel,boolean notify) {
		this.channel = channel;
		if(notify){//是否需要通知，主要时channel不可用时
			synchronized(this){
				notifyAll();
			}
		}
	}

	public void writeCache(ByteBuf buf){
		synchronized (lock) {
			cache.discardReadBytes();//回收内存
			cache.writeBytes(buf);
		}
		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 int read(ByteBuffer buffer) throws IOException {
		if(null == channel)
			throw new IOException("socket has Interrupted !");
		if(cache.readableBytes() <  buffer.remaining()){
			synchronized(this){
				try {wait();} catch (InterruptedException e) {}
			}
		}else{
			synchronized (lock){
				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;
	}
}

@@ -270,37 +270,8 @@
 ss;%0D
-%0Aimport java.nio.ByteBuffer;%0D
 %0A%0D%0A/
@@ -544,10 +544,8 @@
 1024
-*5
 );//
@@ -650,23 +650,8 @@
 nnel
-,boolean notify
 ) %7B%0D
@@ -682,99 +682,8 @@
 l;%0D%0A
-%09%09if(notify)%7B//%E6%98%AF%E5%90%A6%E9%9C%80%E8%A6%81%E9%80%9A%E7%9F%A5%EF%BC%8C%E4%B8%BB%E8%A6%81%E6%97%B6channel%E4%B8%8D%E5%8F%AF%E7%94%A8%E6%97%B6%0D%0A%09%09%09synchronized(this)%7B%0D%0A%09%09%09%09notifyAll();%0D%0A%09%09%09%7D%0D%0A%09%09%7D%0D%0A
 %09%7D%0D%0A
@@ -820,53 +820,8 @@
 %09%7D%0D%0A
-%09%09synchronized(this)%7B%0D%0A%09%09%09notifyAll();%0D%0A%09%09%7D%0D%0A
 %09%7D%0D%0A
@@ -1075,58 +1075,107 @@
 lic 
-int read(ByteBuffer buffer) throws IOException %7B%0D%0A
+byte%5B%5D read(int readSize) throws IOException %7B%0D%0A%09%09do%7B%0D%0A%09%09%09if(readSize %3E cache.readableBytes())%7B%0D%0A%09%09
 %09%09if
@@ -1193,16 +1193,18 @@
 annel)%0D%0A
+%09%09
 %09%09%09throw
@@ -1255,59 +1255,8 @@
 ;%0D%0A%09
-%09if(cache.readableBytes() %3C  buffer.remaining())%7B%0D%0A
 %09%09%09s
@@ -1270,33 +1270,41 @@
 ized
+ 
 (this)
+ 
 %7B%0D%0A%09%09%09%09
+%09
 try %7B
+ 
 wait(
+100
 );
+ 
 %7D ca
@@ -1336,24 +1336,25 @@
  e) %7B%7D%0D%0A
+%09
 %09%09%09%7D%0D%0A%09%09
 %7Delse%7B%0D%0A
@@ -1349,16 +1349,59 @@
 %0D%0A%09%09
-%7D
+%09%7D 
 else
+ 
 %7B%0D%0A
+%09%09%09%09byte%5B%5D back = new byte%5BreadSize%5D;%0D%0A%09
 %09%09%09s
@@ -1418,19 +1418,21 @@
 d (lock)
+ 
 %7B%0D%0A
+%09
 %09%09%09%09cach
@@ -1448,25 +1448,24 @@
 es(b
-uffer
+ack
 );%0D%0A
+%09
 %09%09%09%7D%0D%0A%09%09
 %7D%0D%0A%09
@@ -1460,19 +1460,16 @@
 %09%09%09%7D%0D%0A%09%09
-%7D%0D%0A
 %09%09return
@@ -1469,17 +1469,43 @@
 %09return 
-0
+back;%0D%0A%09%09%09%7D%0D%0A%09%09%7Dwhile(true)
 ;%0D%0A%09%7D%0D%0A%09

Update SocketChannel.java