优化android和java sdk代码

cim-client-sdk/cim-android-sdk/src/com/farsunset/cim/sdk/android/CIMConnectorManager.java

@@ -24,17 +24,15 @@ package com.farsunset.cim.sdk.android;
 import java.io.IOException;
 import java.net.ConnectException;
 import java.net.InetSocketAddress;
+import java.net.SocketTimeoutException;
 import java.nio.ByteBuffer;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
 import java.util.Random;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.ReentrantLock;
 
 import com.farsunset.cim.sdk.android.coder.CIMLogger;
 import com.farsunset.cim.sdk.android.coder.ClientMessageDecoder;
@@ -68,29 +66,38 @@ class CIMConnectorManager{
 
 	private final int READ_IDLE_TIME = 120 * 1000;
 	
+	private final int CONNECT_TIME_OUT = 10 * 1000;
+
+	
 	private final int CONNECT_ALIVE_TIME_OUT = 150 * 1000;
 	
     private final AtomicLong LAST_READ_TIME = new AtomicLong(0);
-
-    private final AtomicBoolean CONNECTING_FLAG = new AtomicBoolean(false) ;
     
     private final CIMLogger LOGGER = CIMLogger.getLogger();
     
     private static final HandlerThread IDLE_HANDLER_THREAD = new HandlerThread("READ-IDLE", Process.THREAD_PRIORITY_BACKGROUND);
     
-    private LinkedBlockingQueue<Protobufable> sendFailBodyQueue = new LinkedBlockingQueue<Protobufable>();
-    private final ReentrantLock IOLOCK = new ReentrantLock(); 
-    
-	private Selector selector;
+	private Semaphore semaphore = new Semaphore(1, true);
+	
 	private SocketChannel socketChannel ;
 	private Context context;
 	
 	
     private ByteBuffer readBuffer = ByteBuffer.allocate(READ_BUFFER_SIZE);
     
-	private ExecutorService workerExecutor = Executors.newFixedThreadPool(1);
-	private ExecutorService bossExecutor = Executors.newFixedThreadPool(1);
-
+	private ExecutorService workerExecutor = Executors.newFixedThreadPool(1,new ThreadFactory() {
+		@Override
+		public Thread newThread(Runnable r) {
+			return new Thread(r,"worker-");
+		}
+	});
+	
+	private ExecutorService bossExecutor = Executors.newFixedThreadPool(1,new ThreadFactory() {
+		@Override
+		public Thread newThread(Runnable r) {
+			return new Thread(r,"boss-");
+		}
+	});
 	
 	private ClientMessageEncoder messageEncoder = new  ClientMessageEncoder();
 	private ClientMessageDecoder messageDecoder = new  ClientMessageDecoder();
@@ -107,19 +114,11 @@ class CIMConnectorManager{
 		try {
 			 if(socketChannel == null || !socketChannel.isOpen()) {
 				 socketChannel = SocketChannel.open();
-		         socketChannel.configureBlocking(false);
 	             socketChannel.socket().setTcpNoDelay(true);
 	             socketChannel.socket().setKeepAlive(true);
 	             socketChannel.socket().setReceiveBufferSize(READ_BUFFER_SIZE);
 	             socketChannel.socket().setSendBufferSize(WRITE_BUFFER_SIZE);
 			 }
-			
-             if(selector == null || !selector.isOpen()) {
-            	 selector = Selector.open();
-             }
-             
-             selector.wakeup(); 
-		     socketChannel.register(selector, SelectionKey.OP_CONNECT);
 
 		}catch(Exception ignore) {}
 	 
@@ -148,22 +147,22 @@ class CIMConnectorManager{
 
 			return;
 		}
-		 
-		if (CONNECTING_FLAG.get() || isConnected()) {
+		
+		if (isConnected()) {
 			return;
 		}
 		
-		CONNECTING_FLAG.set(true);
-		
-		if(!socketChannel.isOpen() ||!selector.isOpen()) {
+		if(!socketChannel.isOpen()) {
 			makeNioConnector();
 		} 
 		
-		workerExecutor.execute(new Runnable() {
+		bossExecutor.execute(new Runnable() {
 			@Override
 			public void run() {
 				
-				IOLOCK.lock();
+				if (isConnected()) {
+					return;
+				}
 				
 	            LOGGER.startConnect(host, port);
 				
@@ -171,38 +170,40 @@ class CIMConnectorManager{
 				
  				try {
  					
- 					socketChannel.connect(new InetSocketAddress(host, port));
+ 					semaphore.acquire();
  					
- 					while (socketChannel.isOpen()) {
- 						
- 						selector.select();
- 						
- 						if(!selector.isOpen()) {
- 							break;
- 						}
- 						
- 						for(SelectionKey key : selector.selectedKeys()){
- 							
- 							if((key.interestOps() & SelectionKey.OP_CONNECT) == SelectionKey.OP_CONNECT && socketChannel.finishConnect()) {
- 								IOLOCK.unlock();
- 								handelConnectedEvent();
- 		                		continue;
- 							}
- 							
- 							if((key.interestOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
- 								handelSocketReadEvent();
- 							}
- 						}
- 		            }
+ 					socketChannel.socket().connect(new InetSocketAddress(host, port),CONNECT_TIME_OUT);
  					
+ 					semaphore.release();
+ 					
+ 					handelConnectedEvent();
+ 					
+ 				
+ 					int result = -1;
+						
+					while((result = socketChannel.read(readBuffer)) > 0) {
+							
+						if(readBuffer.position() == readBuffer.capacity()) {
+							extendByteBuffer();
+						}
+							
+						handelSocketReadEvent(result);
+						
+					}
+					
+					handelSocketReadEvent(result);
+					
 				}catch(ConnectException ignore){
-					IOLOCK.unlock();
+					semaphore.release();
 					handleConnectAbortedEvent();
-				}catch(IllegalArgumentException ignore){
-					IOLOCK.unlock();
+				}catch(SocketTimeoutException ignore){
+					semaphore.release();
 					handleConnectAbortedEvent();
 				}catch(IOException ignore) {
+					semaphore.release();
 					handelDisconnectedEvent();
+				}catch(InterruptedException ignore) {
+					semaphore.release();
 				}
 			}
 		});
@@ -212,12 +213,16 @@ class CIMConnectorManager{
     public void destroy() {
 		
 		closeSession();
-		closeSelector();
 
 	}
 	
 
 	public void closeSession() {
+		
+		if(!isConnected()) {
+			return;
+		}
+		
 		try {
 			socketChannel.close();
 		} catch (IOException ignore) {
@@ -233,28 +238,31 @@ class CIMConnectorManager{
 	
 	public void send(final Protobufable body) {
 		
-		bossExecutor.execute(new Runnable() {
+		if(!isConnected()) {
+			return;
+		}
+		
+		workerExecutor.execute(new Runnable() {
 			
 			@Override
 			public void run() {
 				int result = 0;
 				try {
 					
-					IOLOCK.lock();
+					semaphore.acquire();
 					
 					ByteBuffer buffer =  messageEncoder.encode(body);
 					while(buffer.hasRemaining()){
 						result += socketChannel.write(buffer);   
 					}
 					 
-				} catch (IOException e) {
+				} catch (Exception e) {
 					result = -1;
 				}finally {
 					
-					IOLOCK.unlock();
+					semaphore.release();
 					
 					if(result <= 0) {
-						sendFailBodyQueue.offer(body);
 						closeSession();
 					}else {
 						messageSent(body);
@@ -264,18 +272,8 @@ class CIMConnectorManager{
 		});
 		
 	}
-
-   private void sendFaildQueueBody() {
-	   Protobufable body = sendFailBodyQueue.poll();
-	   if(body == null) {
-		   return;
-	   }
-	   
-	   send(body);
-	   
-	   sendFaildQueueBody();
-	   
-   }
+ 
+   
    private void sessionCreated() {
 		LOGGER.sessionCreated(socketChannel);
 		
@@ -286,8 +284,6 @@ class CIMConnectorManager{
 		intent.setAction(CIMConstant.IntentAction.ACTION_CONNECTION_SUCCESSED);
 		context.sendBroadcast(intent);
 		
-		sendFaildQueueBody();
-
 	}
 
 	private void sessionClosed() {
@@ -304,8 +300,6 @@ class CIMConnectorManager{
 			readBuffer = ByteBuffer.allocate(READ_BUFFER_SIZE);
 		}
 		
-		closeSelector();
-		
 		Intent intent = new Intent();
 		intent.setPackage(context.getPackageName());
 		intent.setAction(CIMConstant.IntentAction.ACTION_CONNECTION_CLOSED);
@@ -369,13 +363,10 @@ class CIMConnectorManager{
 	};
 	
 	private void handelDisconnectedEvent() {
-		CONNECTING_FLAG.set(false);
 		closeSession();
 	}
 	
     private void handleConnectAbortedEvent() {
-		
-		CONNECTING_FLAG.set(false);
 
 		long interval = CIMConstant.RECONN_INTERVAL_TIME - (5 * 1000 - new Random().nextInt(15 * 1000));
 		
@@ -389,33 +380,22 @@ class CIMConnectorManager{
 
 	}
 	 
-	private void handelConnectedEvent() throws IOException {
+	private void handelConnectedEvent()   {
 		
-		CONNECTING_FLAG.set(false);
- 		socketChannel.register(selector, SelectionKey.OP_READ);
 		sessionCreated();
 		
 		idleHandler.sendEmptyMessageDelayed(0, READ_IDLE_TIME);
 	}
 	
-	private void handelSocketReadEvent() throws IOException   {
-		
-		int result = 0;
-		
-		while((result = socketChannel.read(readBuffer)) > 0) {
-			if(readBuffer.position() == readBuffer.capacity()) {
-				extendByteBuffer();
-			}
-		}
-		
-		if(result == -1 && !readBuffer.hasRemaining()) {
+	private void handelSocketReadEvent(int result) throws IOException   {
+	 
+		if(result == -1) {
 		    closeSession();
 		    return;
 		}
 	
 	    markLastReadTime();
 		
-		
 	    readBuffer.position(0);
 	     
 		Object message = messageDecoder.doDecode(readBuffer);
@@ -424,8 +404,6 @@ class CIMConnectorManager{
 			return;
 		}
 		
-		
-		
 		LOGGER.messageReceived(socketChannel,message);
 
 		if(isHeartbeatRequest(message)) {
@@ -436,11 +414,7 @@ class CIMConnectorManager{
 		}
 		
 		this.messageReceived(message);
-		
-		if(result == -1) {
-		    closeSession();
-		    return;
-		}
+	 
 	}
 	
 	
@@ -473,13 +447,6 @@ class CIMConnectorManager{
     private boolean isHeartbeatRequest(Object data) {
 		return data instanceof HeartbeatRequest;
 	}
-	
-
-
-	private void closeSelector() {
-		try {
-			selector.close();
-		} catch (IOException ignore) {}
-	}
+	 
 
 }

cim-client-sdk/cim-android-sdk/src/com/farsunset/cim/sdk/android/CIMPushManager.java

@@ -49,8 +49,6 @@ public class CIMPushManager {
 
 	static String ACTION_SET_LOGGER_EANABLE = "ACTION_SET_LOGGER_EANABLE";
 
-	static String ACTION_DESTORY = "ACTION_DESTORY";
-
 	static String KEY_SEND_BODY = "KEY_SEND_BODY";
 
 	static String KEY_CIM_CONNECTION_STATUS = "KEY_CIM_CONNECTION_STATUS";
@@ -221,9 +219,7 @@ public class CIMPushManager {
 		CIMCacheManager.putBoolean(context, CIMCacheManager.KEY_CIM_DESTROYED, true);
 		CIMCacheManager.putString(context, CIMCacheManager.KEY_ACCOUNT, null);
 
-		Intent serviceIntent = new Intent(context, CIMPushService.class);
-		serviceIntent.setAction(ACTION_DESTORY);
-		startServiceCompat(context,serviceIntent);
+		context.stopService(new Intent(context, CIMPushService.class));
 
 	}
 

cim-client-sdk/cim-android-sdk/src/com/farsunset/cim/sdk/android/CIMPushService.java

@@ -36,8 +36,6 @@ import android.os.Handler;
 import android.os.IBinder;
 import android.os.Message;
 
-import java.util.concurrent.Semaphore;
-
 import com.farsunset.cim.sdk.android.coder.CIMLogger;
 import com.farsunset.cim.sdk.android.constant.CIMConstant;
 import com.farsunset.cim.sdk.android.model.SentBody;
@@ -48,14 +46,13 @@ import com.farsunset.cim.sdk.android.model.SentBody;
  * @author 3979434
  *
  */
-public class CIMPushService extends Service {
+public class CIMPushService extends Service  {
 	public final static String KEY_DELAYED_TIME = "KEY_DELAYED_TIME";
 	public final static String KEY_LOGGER_ENABLE = "KEY_LOGGER_ENABLE";
 
 	private CIMConnectorManager manager;
 	private KeepAliveBroadcastReceiver keepAliveReceiver;
-	private Semaphore semaphore = new Semaphore(1, true);
-
+    private ConnectivityManager connectivityManager;
 	@Override
 	public void onCreate() {
 		manager = CIMConnectorManager.getManager(this.getApplicationContext());
@@ -68,35 +65,37 @@ public class CIMPushService extends Service {
 		
 		if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) {
 
-			ConnectivityManager connectivityManager = (ConnectivityManager) getSystemService(Context.CONNECTIVITY_SERVICE);
-			connectivityManager.registerDefaultNetworkCallback(new ConnectivityManager.NetworkCallback() {
-				@Override
-				public void onAvailable(Network network) {
-					Intent intent = new Intent();
-					intent.setPackage(getPackageName());
-					intent.setAction(CIMConstant.IntentAction.ACTION_NETWORK_CHANGED);
-					sendBroadcast(intent);
-				}
-				@Override
-				public void onUnavailable() {
-					Intent intent = new Intent();
-					intent.setPackage(getPackageName());
-					intent.setAction(CIMConstant.IntentAction.ACTION_NETWORK_CHANGED);
-					sendBroadcast(intent);
-				}
-				
-			});
+			connectivityManager = (ConnectivityManager) getSystemService(Context.CONNECTIVITY_SERVICE);
+			
+			connectivityManager.registerDefaultNetworkCallback(networkCallback);
 
 		}
 	}
+	
+	ConnectivityManager.NetworkCallback networkCallback = new ConnectivityManager.NetworkCallback() {
+		@Override
+		public void onAvailable(Network network) {
+			Intent intent = new Intent();
+			intent.setPackage(getPackageName());
+			intent.setAction(CIMConstant.IntentAction.ACTION_NETWORK_CHANGED);
+			sendBroadcast(intent);
+		}
+		@Override
+		public void onUnavailable() {
+			Intent intent = new Intent();
+			intent.setPackage(getPackageName());
+			intent.setAction(CIMConstant.IntentAction.ACTION_NETWORK_CHANGED);
+			sendBroadcast(intent);
+		}
+		
+	};
 
 	Handler connectionHandler = new Handler() {
 		@Override
 		public void handleMessage(android.os.Message message) {
 			String host = message.getData().getString(CIMCacheManager.KEY_CIM_SERVIER_HOST);
 			int port = message.getData().getInt(CIMCacheManager.KEY_CIM_SERVIER_PORT, 0);
-			manager.connect(host, port);
-			semaphore.release();
+			handleConnection(host, port);
 		}
 	};
 
@@ -118,7 +117,7 @@ public class CIMPushService extends Service {
 		String action = intent.getAction();
 
 		if (CIMPushManager.ACTION_CREATE_CIM_CONNECTION.equals(action)) {
-			handleConnection(intent);
+			handleDelayConnection(intent);
 		}
 
 		if (CIMPushManager.ACTION_SEND_REQUEST_BODY.equals(action)) {
@@ -129,11 +128,6 @@ public class CIMPushService extends Service {
 			manager.closeSession();
 		}
 
-		if (CIMPushManager.ACTION_DESTORY.equals(action)) {
-			manager.destroy();
-			this.stopSelf();
-		}
-
 		if (CIMPushManager.ACTION_ACTIVATE_PUSH_SERVICE.equals(action)) {
 			handleKeepAlive();
 		}
@@ -150,41 +144,47 @@ public class CIMPushService extends Service {
 		return super.onStartCommand(intent, flags, startId);
 	}
 
-	private void handleConnection(Intent intent) {
+	private void handleDelayConnection(Intent intent) {
 
 		long delayMillis = intent.getLongExtra(KEY_DELAYED_TIME, 0);
 
 		if (delayMillis <= 0) {
 			String host = intent.getStringExtra(CIMCacheManager.KEY_CIM_SERVIER_HOST);
 			int port = intent.getIntExtra(CIMCacheManager.KEY_CIM_SERVIER_PORT, 0);
-			manager.connect(host, port);
-			return;
-		}
-
-		if (!semaphore.tryAcquire()) {
+			handleConnection(host, port);
 			return;
 		}
+		
 
 		Message msg = connectionHandler.obtainMessage();
 		msg.what = 0;
 		msg.setData(intent.getExtras());
+		connectionHandler.removeMessages(0);
 		connectionHandler.sendMessageDelayed(msg, delayMillis);
 
 	}
 
+	private void handleConnection(String host,int port) {
+
+		boolean isManualStop = CIMCacheManager.getBoolean(getApplicationContext(), CIMCacheManager.KEY_MANUAL_STOP);
+		boolean isDestroyed = CIMCacheManager.getBoolean(getApplicationContext(), CIMCacheManager.KEY_CIM_DESTROYED);
+		if(isManualStop || isDestroyed) {
+			return;
+		}
+		manager.connect(host, port);
+	}
+	
 	private void handleKeepAlive() {
+		
 		if (manager.isConnected()) {
 			CIMLogger.getLogger().connectState(true);
 			return;
 		}
-
-		boolean isManualStop = CIMCacheManager.getBoolean(getApplicationContext(), CIMCacheManager.KEY_MANUAL_STOP);
-		boolean isDestroyed = CIMCacheManager.getBoolean(getApplicationContext(), CIMCacheManager.KEY_CIM_DESTROYED);
-
-		CIMLogger.getLogger().connectState(false, isManualStop, isDestroyed);
-
-		CIMPushManager.connect(this, 0);
-
+		
+		String host = CIMCacheManager.getString(this, CIMCacheManager.KEY_CIM_SERVIER_HOST);
+		int port = CIMCacheManager.getInt(this, CIMCacheManager.KEY_CIM_SERVIER_PORT);
+		 
+		handleConnection(host,port);
 	}
 
 	@Override
@@ -195,9 +195,15 @@ public class CIMPushService extends Service {
 	@Override
 	public void onDestroy() {
 		super.onDestroy();
+		manager.destroy();
+		connectionHandler.removeMessages(0);
 		if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
 			unregisterReceiver(keepAliveReceiver);
 		}
+		
+		if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) {
+			connectivityManager.unregisterNetworkCallback(networkCallback);
+		}
 	}
 
 	public class KeepAliveBroadcastReceiver extends BroadcastReceiver {

cim-client-sdk/cim-java-sdk/src/com/farsunset/cim/sdk/client/CIMConnectorManager.java

@@ -24,26 +24,24 @@ package com.farsunset.cim.sdk.client;
 import java.io.IOException;
 import java.net.ConnectException;
 import java.net.InetSocketAddress;
-import java.net.StandardSocketOptions;
+import java.net.SocketTimeoutException;
 import java.nio.ByteBuffer;
-import java.nio.channels.ClosedSelectorException;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
 import java.util.Random;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
 
 import com.farsunset.cim.sdk.client.coder.CIMLogger;
 import com.farsunset.cim.sdk.client.coder.ClientMessageDecoder;
 import com.farsunset.cim.sdk.client.coder.ClientMessageEncoder;
 import com.farsunset.cim.sdk.client.constant.CIMConstant;
-import com.farsunset.cim.sdk.client.exception.SessionClosedException;
 import com.farsunset.cim.sdk.client.model.HeartbeatRequest;
 import com.farsunset.cim.sdk.client.model.HeartbeatResponse;
 import com.farsunset.cim.sdk.client.model.Intent;
 import com.farsunset.cim.sdk.client.model.Message;
+import com.farsunset.cim.sdk.client.model.Protobufable;
 import com.farsunset.cim.sdk.client.model.ReplyBody;
 import com.farsunset.cim.sdk.client.model.SentBody;
 
@@ -59,20 +57,34 @@ class CIMConnectorManager {
 	
 	private final int READ_BUFFER_SIZE = 2048;
 	private final int WRITE_BUFFER_SIZE = 1024;
+	private final int CONNECT_TIME_OUT = 10 * 1000;
 
-    private final AtomicBoolean CONNECTING_FLAG = new AtomicBoolean(false) ;
-    
     private final CIMLogger LOGGER = CIMLogger.getLogger();
-    
-	private Selector selector;
+
 	private SocketChannel socketChannel ;
 	
-	
     private ByteBuffer readBuffer = ByteBuffer.allocate(READ_BUFFER_SIZE);
-    
-	private ExecutorService workerExecutor = Executors.newFixedThreadPool(1);
-	private ExecutorService bossExecutor = Executors.newFixedThreadPool(1);
-	private ExecutorService eventExecutor = Executors.newFixedThreadPool(1);
+
+	private ExecutorService workerExecutor = Executors.newFixedThreadPool(1,new ThreadFactory() {
+		@Override
+		public Thread newThread(Runnable r) {
+			return new Thread(r,"worker-");
+		}
+	});
+	private ExecutorService bossExecutor = Executors.newFixedThreadPool(1,new ThreadFactory() {
+		@Override
+		public Thread newThread(Runnable r) {
+			return new Thread(r,"boss-");
+		}
+	});
+	private ExecutorService eventExecutor = Executors.newFixedThreadPool(1,new ThreadFactory() {
+		@Override
+		public Thread newThread(Runnable r) {
+			return new Thread(r,"event-");
+		}
+	});
+	
+	private Semaphore semaphore = new Semaphore(1, true);
 
 	
 	private ClientMessageEncoder messageEncoder = new  ClientMessageEncoder();
@@ -83,21 +95,14 @@ class CIMConnectorManager {
 	}
 	private void makeNioConnector() {
 		try {
+
 			 if(socketChannel == null || !socketChannel.isOpen()) {
 				 socketChannel = SocketChannel.open();
-		         socketChannel.configureBlocking(false);
-	             socketChannel.setOption(StandardSocketOptions.SO_RCVBUF,READ_BUFFER_SIZE);
-	             socketChannel.setOption(StandardSocketOptions.SO_SNDBUF, WRITE_BUFFER_SIZE);
-	             socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
-	             socketChannel.setOption(StandardSocketOptions.TCP_NODELAY,true);
+	             socketChannel.socket().setTcpNoDelay(true);
+	             socketChannel.socket().setKeepAlive(true);
+	             socketChannel.socket().setReceiveBufferSize(READ_BUFFER_SIZE);
+	             socketChannel.socket().setSendBufferSize(WRITE_BUFFER_SIZE);
 			 }
-			
-             if(selector == null || !selector.isOpen()) {
-            	 selector = Selector.open();
-             }
-             
-             selector.wakeup(); 
-		     socketChannel.register(selector, SelectionKey.OP_CONNECT);
 
 		}catch(Exception ignore) {}
 	 
@@ -116,20 +121,21 @@ class CIMConnectorManager {
 	public void connect(final String host, final int port) {
  
 		 
-		boolean isConnected = isConnected();
-		if (CONNECTING_FLAG.get() || isConnected) {
+		if (isConnected()) {
 			return;
 		}
 		
-		CONNECTING_FLAG.set(true);
-		
-		if(!socketChannel.isOpen() ||!selector.isOpen()) {
+		if(!socketChannel.isOpen()) {
 			makeNioConnector();
 		} 
 		
-		workerExecutor.execute(new Runnable() {
+		bossExecutor.execute(new Runnable() {
 			@Override
 			public void run() {
+
+				if (isConnected()) {
+					return;
+				}
 				
 	            LOGGER.startConnect(host, port);
 				
@@ -137,36 +143,41 @@ class CIMConnectorManager {
 				
  				try {
  					
- 					socketChannel.connect(new InetSocketAddress(host, port));
+ 					semaphore.acquire();
+
+ 					socketChannel.socket().connect(new InetSocketAddress(host, port),CONNECT_TIME_OUT);
  					
- 					while (socketChannel.isOpen()) {
- 						
- 						selector.select();
- 						
- 						if(!selector.isOpen()) {
- 							break;
- 						}
- 						
- 						for(SelectionKey key : selector.selectedKeys()){
- 							
- 							if((key.interestOps() & SelectionKey.OP_CONNECT) == SelectionKey.OP_CONNECT && socketChannel.finishConnect()) {
- 								handelConnectedEvent();
- 		                		continue;
- 							}
- 							
- 							if((key.interestOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
- 								handelSocketReadEvent();
- 							}
- 						}
- 		            }
+ 					semaphore.release();
  					
+ 					handelConnectedEvent();
+ 					
+ 				
+ 					int result = -1;
+						
+					while((result = socketChannel.read(readBuffer)) > 0) {
+							
+						if(readBuffer.position() == readBuffer.capacity()) {
+							extendByteBuffer();
+						}
+							
+						handelSocketReadEvent(result);
+						
+					}
+					
+					handelSocketReadEvent(result);
+					
 				}catch(ConnectException ignore){
+					semaphore.release();
 					handleConnectAbortedEvent();
-				}catch(IllegalArgumentException ignore){
+				}catch(SocketTimeoutException ignore){
+					semaphore.release();
 					handleConnectAbortedEvent();
 				}catch(IOException ignore) {
+					semaphore.release();
 					handelDisconnectedEvent();
-				}catch(ClosedSelectorException ignore) {}
+				}catch (InterruptedException ignore) {
+					semaphore.release();
+				} 
 			}
 		});
 	}
@@ -174,13 +185,10 @@ class CIMConnectorManager {
  
 	
 	private void handelDisconnectedEvent() {
-		CONNECTING_FLAG.set(false);
 		closeSession();
 	}
 	
     private void handleConnectAbortedEvent() {
-		
-		CONNECTING_FLAG.set(false);
 
 		long interval = CIMConstant.RECONN_INTERVAL_TIME - (5 * 1000 - new Random().nextInt(15 * 1000));
 		
@@ -195,26 +203,16 @@ class CIMConnectorManager {
 	 
 	private void handelConnectedEvent() throws IOException {
 		
-		CONNECTING_FLAG.set(false);
- 		socketChannel.register(selector, SelectionKey.OP_READ);
 		sessionCreated();
-		
 	}
 	
-	private void handelSocketReadEvent() throws IOException   {
+	private void handelSocketReadEvent(int result) throws IOException   {
 		
-		int result = 0;
-		
-		while((result = socketChannel.read(readBuffer)) > 0) {
-			if(readBuffer.position() == readBuffer.capacity()) {
-				extendByteBuffer();
-			}
+		if(result == -1) {
+		    closeSession();
+		    return;
 		}
-		
-	    if(result == -1) {
-	    	closeSession();
-	    	return;
-	    }
+	
 		
 	    readBuffer.position(0);
 	     
@@ -246,63 +244,44 @@ class CIMConnectorManager {
 		readBuffer.clear();
 		readBuffer = newBuffer;
 	}
-    
 	
-	public void send(final SentBody body) {
+	
+   public void send(final Protobufable body) {
 		
-		bossExecutor.execute(new Runnable() {
+	   if(!isConnected()) {
+			return;
+		}
+		
+		workerExecutor.execute(new Runnable() {
 			
 			@Override
 			public void run() {
-				boolean isSuccessed = false;
-				String exceptionName = SessionClosedException.class.getSimpleName();
-				if (isConnected()) {
+				int result = 0;
+				try {
 					
-					try {
-						ByteBuffer buffer =  messageEncoder.encode(body);
-						int result = 0;
-						while(buffer.hasRemaining()){
-							result += socketChannel.write(buffer);   
-						}
-						isSuccessed = result > 0;
-					} catch (IOException e) {
-						exceptionName = e.getClass().getSimpleName();
-						closeSession();
+					semaphore.acquire();
+					
+					ByteBuffer buffer =  messageEncoder.encode(body);
+					while(buffer.hasRemaining()){
+						result += socketChannel.write(buffer);   
 					}
 					 
-				}
-				
-				if (!isSuccessed) {
-					Intent intent = new Intent();
-					intent.setAction(CIMConstant.IntentAction.ACTION_SENT_FAILED);
-					intent.putExtra(Exception.class.getName(), exceptionName);
-					intent.putExtra(SentBody.class.getName(), body);
-					sendBroadcast(intent);
-				}else {
-					messageSent(body);
-				}
+				} catch (Exception e) {
+					result = -1;
+				}finally {
+					
+					semaphore.release();
+					
+					if(result <= 0) {
+						closeSession();
+					}else {
+						messageSent(body);
+					}
+				} 
 			}
 		});
-		
 	}
-	
-	
-   public void send(final HeartbeatResponse body) {
-		
-		bossExecutor.execute(new Runnable() {
-			@Override
-			public void run() {
-				try {
-					 socketChannel.write(messageEncoder.encode(body));
-					 messageSent(body);
-				} catch (IOException ignore) {
-					 closeSession();
-				}
-			}
-		});
-		
-	}
-
+  
 
 	public void sessionCreated() {
 		
@@ -311,7 +290,7 @@ class CIMConnectorManager {
 		Intent intent = new Intent();
 		intent.setAction(CIMConstant.IntentAction.ACTION_CONNECTION_SUCCESSED);
 		sendBroadcast(intent);
-
+		
 	}
 
 	public void sessionClosed() {
@@ -324,8 +303,6 @@ class CIMConnectorManager {
 			readBuffer = ByteBuffer.allocate(READ_BUFFER_SIZE);
 		}
 		
-		closeSelector();
-		
 		Intent intent = new Intent();
 		intent.setAction(CIMConstant.IntentAction.ACTION_CONNECTION_CLOSED);
 		sendBroadcast(intent);
@@ -373,10 +350,7 @@ class CIMConnectorManager {
 	}
  
     public void destroy() {
-		
 		closeSession();
-		closeSelector();
-
 	}
 
 	public boolean isConnected() {
@@ -384,6 +358,11 @@ class CIMConnectorManager {
 	}
 
 	public void closeSession() {
+
+		if(!isConnected()) {
+			return;
+		}
+		
 		try {
 			socketChannel.close();
 		} catch (IOException ignore) {
@@ -391,13 +370,7 @@ class CIMConnectorManager {
 			 this.sessionClosed();
 		}
 	}
-
-	public void closeSelector() {
-		try {
-			selector.close();
-		} catch (IOException ignore) {}
-	}
-
+ 
 	
 	private void sendBroadcast(final Intent intent) {
 		eventExecutor.execute(new Runnable() {

cim-client-sdk/cim-java-sdk/src/com/farsunset/cim/sdk/client/CIMEventBroadcastReceiver.java

@@ -26,7 +26,6 @@ import java.util.Timer;
 import java.util.TimerTask;
 
 import com.farsunset.cim.sdk.client.constant.CIMConstant;
-import com.farsunset.cim.sdk.client.exception.SessionClosedException;
 import com.farsunset.cim.sdk.client.model.Intent;
 import com.farsunset.cim.sdk.client.model.Message;
 import com.farsunset.cim.sdk.client.model.ReplyBody;
@@ -90,14 +89,6 @@ public class CIMEventBroadcastReceiver {
 			listener.onReplyReceived((ReplyBody) intent.getExtra(ReplyBody.class.getName()));
 		}
 
-		/*
-		 * 获取sendbody发送失败事件
-		 */
-		if (intent.getAction().equals(CIMConstant.IntentAction.ACTION_SENT_FAILED)) {
-			onSentFailed((Exception) intent.getExtra(Exception.class.getName()),
-					(SentBody) intent.getExtra(SentBody.class.getName()));
-		}
-
 		/*
 		 * 获取sendbody发送成功事件
 		 */
@@ -158,19 +149,7 @@ public class CIMEventBroadcastReceiver {
 	private boolean isForceOfflineMessage(String action) {
 		return CIMConstant.MessageAction.ACTION_999.equals(action);
 	}
-
-	private void onSentFailed(Exception e, SentBody body) {
-
-		e.printStackTrace();
-		// 与服务端端开链接，重新连接
-		if (e instanceof SessionClosedException) {
-			CIMPushManager.connect();
-		} else {
-			// 发送失败 重新发送
-			CIMPushManager.sendRequest(body);
-		}
-
-	}
+ 
 
 	private void onSentSucceed(SentBody body) {
 	}

cim-client-sdk/cim-java-sdk/src/com/farsunset/cim/sdk/client/coder/CIMLogger.java

@@ -21,7 +21,6 @@
  */
 package com.farsunset.cim.sdk.client.coder;
 
-import java.io.IOException;
 import java.nio.channels.SocketChannel;
 
 import org.slf4j.Logger;
@@ -83,6 +82,7 @@ public class CIMLogger  {
 	public void connectState(boolean isConnected,boolean isManualStop,boolean isDestroyed)  {
 		LOGGER.debug("CONNECTED:" + isConnected + " STOPED:"+isManualStop+ " DESTROYED:"+isDestroyed);
 	}
+	
 	private String getSessionInfo(SocketChannel session) {
 		StringBuilder builder = new StringBuilder();
 		if (session == null) {
@@ -92,24 +92,23 @@ public class CIMLogger  {
 		builder.append("id:").append(session.hashCode());
 		
 		try {
-			if (session.getLocalAddress() != null) {
-				builder.append(" L:").append(session.getLocalAddress().toString());
+			if (session.socket().getLocalAddress() != null) {
+				builder.append(" L:").append(session.socket().getLocalAddress()+":"+session.socket().getLocalPort());
 			}
-		} catch (IOException e) {
-			e.printStackTrace();
+		} catch (Exception ignore) {
 		}
 		
 		
 		try {
-			if (session.getRemoteAddress() != null) {
-				builder.append(" R:").append(session.getRemoteAddress().toString());
+			if (session.socket().getRemoteSocketAddress() != null) {
+				builder.append(" R:").append(session.socket().getRemoteSocketAddress().toString());
 			}
-		} catch (IOException e) {
-			e.printStackTrace();
+		} catch (Exception ignore) {
 		}
 		builder.append("]");
 		return builder.toString();
 	}
 	 
+	 
 	
 }

cim-client-sdk/cim-java-sdk/src/com/farsunset/cim/sdk/client/constant/CIMConstant.java

@@ -87,9 +87,6 @@ public interface CIMConstant {
 		// 消息广播action
 		String ACTION_MESSAGE_RECEIVED = "com.farsunset.cim.MESSAGE_RECEIVED";
 
-		// 发送sendbody失败广播
-		String ACTION_SENT_FAILED = "com.farsunset.cim.SENT_FAILED";
-
 		// 发送sendbody成功广播
 		String ACTION_SENT_SUCCESSED = "com.farsunset.cim.SENT_SUCCESSED";