jefe 0619

.idea/compiler.xml

@@ -10,6 +10,7 @@
         <module name="taurus-core" />
         <module name="taurus-permanent" />
         <module name="taurus-permanent (1)" />
+        <module name="event_mgr" />
         <module name="game_common" />
         <module name="taurus-web" />
         <module name="game_zp_fangpaofa" />

.idea/encodings.xml

@@ -3,6 +3,8 @@
   <component name="Encoding">
     <file url="file://$PROJECT_DIR$/data_cache/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/data_cache/src/main/resources" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/event_mgr/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/event_mgr/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/game_common/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/game_common/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/game_mj_hongzhong/src/main/java" charset="UTF-8" />

.idea/misc.xml

@@ -12,6 +12,7 @@
         <option value="$PROJECT_DIR$/web_group/pom.xml" />
         <option value="$PROJECT_DIR$/game_zp_fangpaofa/pom.xml" />
         <option value="$PROJECT_DIR$/game_mj_hongzhong/pom.xml" />
+        <option value="$PROJECT_DIR$/event_mgr/pom.xml" />
       </list>
     </option>
   </component>

web_group/pom.xml

@@ -14,6 +14,13 @@
 	</properties>
 		
 	<dependencies>
+
+		<dependency>
+			<groupId>org.apache.httpcomponents.client5</groupId>
+			<artifactId>httpclient5</artifactId>
+			<version>5.4.4</version>
+		</dependency>
+
 		<dependency>
 			<groupId>junit</groupId>
 			<artifactId>junit</artifactId>
@@ -21,6 +28,12 @@
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>com.alibaba.fastjson2</groupId>
+			<artifactId>fastjson2</artifactId>
+			<version>2.0.43</version>
+		</dependency>
+
 		<!--依赖 data_cache -->
 		<dependency>
 			<groupId>com.data</groupId>

web_group/src/main/java/com/group/service/GroupRoomService.java

@@ -1,5 +1,11 @@
 package com.group.service;
 
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -1063,6 +1069,8 @@ public class GroupRoomService {
 			String key = createGroupRoom(session, groupId, pid);
 			resData = publicJoinRoom(groupId, session, key, platform);
 		}
+
+		/* robot jefe
 		Jedis jedis11 = Redis.use("group1_db11").getJedis();
 		try {
 			jedis11.zremrangeByScore(grooms_key, 0, 0);
@@ -1093,6 +1101,8 @@ public class GroupRoomService {
 		} finally {
 			jedis11.close();
 		}
+
+		 */
 		return resData;
 	}
 
@@ -1106,80 +1116,161 @@ public class GroupRoomService {
 
 
 		log.info("sendRobotData :"+roomid+"groupid:"+groupid+"robot_host:"+robot_host);
+		Jedis js1 = Redis.use("group1_db3").getJedis();
+		try {
 
-		CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
+			String flag = js1.get("roomsend" + roomid);
-			// 执行任务
+			if (flag != null) {
-			Jedis js1 = Redis.use("group1_db3").getJedis();
-			String flag =  js1.get("roomsend"+roomid);
-			if(flag!=null){
-				js1.close();
 				return null;
 			}
-			js1.set("roomsend"+roomid,1+"");
+			js1.set("roomsend" + roomid, 1 + "");
-			js1.expire("roomsend"+roomid,5);
+			js1.expire("roomsend" + roomid, 5);
-			try {
-				Thread eventThread = new Thread(() -> {
-					while (true) {
-						NetManager.processEvents();
-						try {
-							Thread.sleep(100);
-						} catch (InterruptedException e) {
-							break;
-						} catch (Exception e) {
-						}
-					}
-				}, "tstse");
 
-				eventThread.setDaemon(true); // 设置为守护线程
+			int robotid = 0;
-				eventThread.start();
+			Set<String> jsRobot = js1.smembers("robots");
-				int robotid = 0;
+			log.info("jsRobot" + jsRobot);
-				//
+			String wokelock = "wokelock";
-				Set<String> jsRobot = js1.smembers("robots");
+			for (String key : jsRobot) {
-
+				String tmp = js1.get(wokelock + key);
-				String wokelock = "wokelock";
+				if (StringUtil.isEmpty(tmp)) {
-				for (String key : jsRobot) {
+					robotid = Integer.parseInt(key);
-					String tmp = js1.get(wokelock+key);
-					if(StringUtil.isEmpty(tmp)){
-						robotid = Integer.parseInt(key);
-					}
 				}
-
-				if(robotid!=0){
-					js1.set(wokelock+robotid, String.valueOf(1));
-					js1.expire(wokelock+robotid,2);
-					TaurusClient client = new TaurusClient(robot_host, "games"+robotid, TaurusClient.ConnectionProtocol.Tcp);
-					client.setSession("abbccccd"+robotid+roomid);
-					client.connect();
-					sleep(1500);
-					ITObject params = TObject.newInstance();
-					params.putString("roomid",roomid);
-					params.putInt("groupid",groupid);
-					params.putInt("robotid",robotid);
-					System.out.println("ssss");
-					System.out.println(client.isConnected());
-					log.info("=====================sendRobotData发送前 :"+roomid+"robotid:"+robotid);
-					Message msg= client.send("225", params, response -> {
-						System.out.println("abcd");
-					});
-					log.info("=====================sendRobotData发送了 :"+roomid+"robotid:"+robotid);
-					sleep(10000);
-					eventThread.setDaemon(false);
-					client.killConnection();
-				}
-				eventThread.yield();
-				log.info("sendRobotData :"+roomid+"robotid:"+robotid);
-				js1.close();
-			}catch (Exception e){
-				js1.close();
-			}finally {
-				js1.close();
 			}
-			return "执行结果";
+
-		});
+
+//            int robotid = 0;
+//            Map<String, String> jsRobot = js1.hgetAll("{robot426149}" + gameId);
+//            String wokelock = "wokelock";
+//            for (String key : jsRobot.keySet()) {
+//                String tmp = js1.get(wokelock + key);
+//                if (StringUtil.isEmpty(tmp)) {
+//                    robotid = Integer.parseInt(key);
+//                    break;
+//                }
+//            }
+
+			if (robotid == 0) {
+				log.info("sendRobotData 没有可用机器人 roomid:" + roomid);
+				return null;
+			}
+
+			//设置机器人占用标记
+			js1.set(wokelock + robotid, String.valueOf(1));
+			js1.expire(wokelock + robotid, 2);
+
+//            js1.expire(wokelock + robotid, 900);
+
+			//通过 HTTP 同步调用机器人服务的 225 协议入口
+			sleep(1000);
+			String url = buildRobotHttpUrl(robot_host);
+			String body = buildJoinRoomBody(robotid, roomid, groupid);
+			String resp = httpPostJson(url, body, 5000);
+			log.info("sendRobotData roomid:" + roomid + " robotid:" + robotid + " resp:" + resp);
+		} catch (Exception e) {
+			log.error("sendRobotData 异常 roomid:" + roomid, e);
+		} finally {
+			js1.close();
+		}
 		//client.killConnection();
 		return null;
 	}
 
+
+	/**
+	 * 根据机器人 TCP 地址推导 HTTP URL
+	 * 规则：HTTP 端口 = TCP 端口 + 1000（与 robot_mj_hz 的 Config.HTTP_SERVER_PORT 一致）
+	 * 例：robot_host="192.168.0.32:8722" → "http://192.168.0.32:9722/robot/joinRoom"
+	 */
+	private static String buildRobotHttpUrl(String robot_host) {
+		String host;
+		int tcpPort;
+		int colonIdx = robot_host.indexOf(':');
+		if (colonIdx > 0) {
+			host = robot_host.substring(0, colonIdx);
+			tcpPort = Integer.parseInt(robot_host.substring(colonIdx + 1));
+		} else {
+			host = robot_host;
+			tcpPort = 8722;
+		}
+		int httpPort = tcpPort + 1000;
+		return "http://" + host + ":" + httpPort + "/robot/joinRoom";
+	}
+
+	/**
+	 * 构造 225 协议请求体 JSON
+	 */
+	private static String buildJoinRoomBody(int robotid, String roomid, int groupid) {
+		StringBuilder sb = new StringBuilder(64);
+		sb.append("{\"robotid\":").append(robotid)
+				.append(",\"roomid\":\"").append(roomid).append("\"")
+				.append(",\"groupid\":").append(groupid)
+				.append("}");
+		return sb.toString();
+	}
+
+	/**
+	 * 通过 JDK HttpURLConnection 发送 POST JSON 请求（同步）
+	 * 替代原 TaurusClient + CompletableFuture + eventThread 异步 TCP 方式
+	 *
+	 * @param urlStr    完整 URL
+	 * @param jsonBody  请求体 JSON 字符串
+	 * @param timeoutMs 连接与读取超时（毫秒）
+	 * @return 响应体字符串；失败时返回错误描述
+	 */
+	private static String httpPostJson(String urlStr, String jsonBody, int timeoutMs) {
+		HttpURLConnection conn = null;
+		try {
+			URL url = new URL(urlStr);
+			conn = (HttpURLConnection) url.openConnection();
+			conn.setRequestMethod("POST");
+			conn.setRequestProperty("Content-Type", "application/json; charset=UTF-8");
+			conn.setConnectTimeout(timeoutMs);
+			conn.setReadTimeout(timeoutMs);
+			conn.setDoOutput(true);
+			conn.setDoInput(true);
+
+			//写入请求体
+			byte[] bodyBytes = jsonBody.getBytes(StandardCharsets.UTF_8);
+			try (OutputStream os = conn.getOutputStream()) {
+				os.write(bodyBytes);
+				os.flush();
+			}
+
+			//读取响应
+			int code = conn.getResponseCode();
+			InputStream is = (code >= 200 && code < 300) ? conn.getInputStream() : conn.getErrorStream();
+			String resp = readAll(is);
+			if (code < 200 || code >= 300) {
+				log.warn("httpPostJson 失败 url:" + urlStr + " code:" + code + " resp:" + resp);
+				return "HTTP " + code + ": " + resp;
+			}
+			return resp;
+		} catch (Exception e) {
+			log.error("httpPostJson 异常 url:" + urlStr, e);
+			return "ERROR:" + e.getMessage();
+		} finally {
+			if (conn != null) {
+				conn.disconnect();
+			}
+		}
+	}
+
+	/**
+	 * 读取输入流为字符串
+	 */
+	private static String readAll(InputStream in) throws Exception {
+		if (in == null) return "";
+		ByteArrayOutputStream baos = new ByteArrayOutputStream();
+		byte[] buf = new byte[1024];
+		int n;
+		while ((n = in.read(buf)) > 0) {
+			baos.write(buf, 0, n);
+		}
+		in.close();
+		return new String(baos.toByteArray(), StandardCharsets.UTF_8);
+	}
+
+
 	/**
 	 * 删除房间
 	 * 

web_group/src/main/java/com/group/service/HttpClient.java

@@ -1,220 +0,0 @@
-package com.group.service;
-
-import com.group.service.TaurusClient.ConnectionProtocol;
-import com.taurus.core.entity.ITObject;
-import com.taurus.core.entity.TObject;
-import com.taurus.core.util.*;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.ConnectException;
-import java.net.HttpURLConnection;
-import java.net.SocketTimeoutException;
-import java.net.URL;
-import java.util.LinkedList;
-import java.util.Queue;
-
-
-/**
- * Http缃戠粶璇锋眰澶勭悊绫?
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public class HttpClient implements ISocketLayer {
-	private static final Logger	logger			= Logger.getLogger(HttpClient.class);
-	private static final byte[] NULL_BYTES = new byte[0];
-	private static final String	_Method			= "POST";
-	private static final String _ContentTypeName = "Content-Type";
-	private static final String _ContentType = "application/octet-stream";
-	private static final String	_Result = "$r";
-	private static final String _Param = "$p";
-	private static final String _Session = "$s";
-	private static final String _Version = "$v";
-	String						_url;
-	String						_session_id		= "";
-
-	final Queue<Runnable>		_actionQueue	= new LinkedList<Runnable>();
-	
-
-	public HttpClient(String url) {
-		_url = url;
-	}
-
-	private void __sendMsg(Message msg) {
-		byte[] bytes = NULL_BYTES;
-		ITObject _data = TObject.newInstance();
-		_data.putString(_Session, _session_id);
-		_data.putInt(_Version, NetManager.VERSION);
-		if (msg.useText && msg.param == null) {
-			if (StringUtil.isEmpty(msg.textParam)) {
-				msg.param = TObject.newInstance();
-			} else {
-				try {
-					msg.param = TObject.newFromJsonData(msg.textParam);
-				} catch (Exception e) {
-					msg.param = TObject.newInstance();
-				}
-			}
-		}
-		_data.putTObject(_Param, msg.param);
-		ByteArray buff = new ByteArray(_data.toBinary());
-		try {
-			buff.compress();
-		} catch (Exception e) {
-			logger.error(e);
-		}
-		bytes = buff.bytes();
-
-		__requestAsync(msg, bytes);
-	}
-
-	private void __requestAsync(final Message msg,byte[] bytes) {
-		OutputStream out = null;
-		InputStream in = null;
-		int code = 500;
-		HttpURLConnection conn = null;
-		try {
-
-			URL realUrl = new URL(_url + msg.command);
-			conn = (HttpURLConnection) realUrl.openConnection();
-			conn.setDoOutput(true);
-			conn.setDoInput(true);
-			conn.setRequestProperty(_ContentTypeName, _ContentType); 
-			conn.setUseCaches(false);// Post 璇锋眰涓嶈兘浣跨敤缂撳瓨
-			conn.setRequestMethod(_Method);
-			conn.setConnectTimeout(NetManager.TIMEOUT_TIME * 1000);
-			conn.setReadTimeout(NetManager.TIMEOUT_TIME * 1000);
-			conn.connect();
-			
-			
-			out = conn.getOutputStream();
-			out.write(bytes);
-			if (conn.getResponseCode() == HttpURLConnection.HTTP_OK) {
-				in = conn.getInputStream(); 
-				bytes = Utils.uncompress(in);
-				if (bytes!=null) {
-					
-					ITObject tem = TObject.newFromBinaryData(bytes);
-					int result = tem.getInt(_Result);
-
-					if (result == 0) {
-						ITObject _param = tem.getTObject(_Param);
-						msg.param = _param;
-						if(msg.useText) {
-							msg.textParam = _param.toJson();
-						}
-					}
-					code = result;
-					
-				}
-			}
-		} catch (SocketTimeoutException e) {
-			logger.error(e);
-			code = 102;
-		} catch (ConnectException ex1) {
-			logger.error(ex1);
-			code = 101;
-		} catch (Exception ex2) {
-			logger.error(ex2);
-			code = 500;
-		}finally {
-			try {
-				if (out != null) {
-					out.close();
-				}
-			} catch (IOException ex) {
-				logger.error(ex);
-			}
-			try {
-				if (in != null) {
-					in.close();
-				}
-			}catch (IOException ex) {
-				logger.error(ex);
-			}
-			if(conn!=null) {
-				conn.disconnect();
-			}
-		}
-		if (msg.callback != null) {
-			final MessageResponse response = new MessageResponse();
-			response.messageData = msg;
-			response.returnCode = code;
-			enqueueActionForDispatch(new Runnable() {
-				@Override
-				public void run() {
-					msg.callback.action(response);
-				}
-			});
-		}
-
-	}
-
-	@Override
-	public void connect(String ip, String game) {
-
-	}
-
-	@Override
-	public void disconnect() {
-
-	}
-
-	private void enqueueActionForDispatch(Runnable action) {
-		synchronized (_actionQueue) {
-			_actionQueue.offer(action);
-		}
-	}
-
-	@Override
-	public void processEvent() {
-		while (true) {
-			Runnable myAction;
-			synchronized (_actionQueue) {
-				if (_actionQueue.size() <= 0) {
-					break;
-				}
-				myAction = _actionQueue.poll();
-			}
-			myAction.run();
-		}
-	}
-
-	public void write(final Message msg) {
-		new Thread(new Runnable() {
-			@Override
-			public void run() {
-				__sendMsg(msg);
-			}
-		}).start();
-	}
-
-	public int getAveragePingTime() {
-		return 1000;
-	}
-
-	public void setSession(String session) {
-		_session_id = session == null ? StringUtil.Empty : session;
-	}
-
-	public String getSession() {
-		return _session_id;
-
-	}
-
-	@Override
-	public void setOnConnect(ICallback<SocketCode> onConnect) {
-
-	}
-
-	@Override
-	public void setOnEvent(ICallback<Message> onEvent) {
-
-	}
-
-	@Override
-	public ConnectionProtocol getProtocol() {
-		return ConnectionProtocol.Web;
-	}
-}

web_group/src/main/java/com/group/service/ISocketLayer.java

@@ -1,67 +0,0 @@
-package com.group.service;
-
-import com.group.service.TaurusClient.ConnectionProtocol;
-import com.taurus.core.util.ICallback;
-
-/**
- * 网络协议处理通用接口
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public interface ISocketLayer {
-	/**
-	 * 连接服务器
-	 * @param ip
-	 * @param game
-	 */
-	public void connect(String ip, String game);
-
-	/**
-	 *  断开连接
-	 */
-	public void disconnect();
-
-	/**
-	 * 处理事件
-	 */
-	public void processEvent();
-
-	/**
-	 * 发送消息
-	 * @param msg
-	 */
-	public void write(Message msg);
-	
-	/**
-	 * get Protocol type
-	 * @return
-	 */
-	public ConnectionProtocol getProtocol();
-	
-	/**
-	 * 设置连接状态回调
-	 */
-	public void setOnConnect(ICallback<SocketCode> onConnect);
-
-
-	/**
-	 * 设置事件消息回调E
-	 */
-	public void setOnEvent(ICallback<Message> onEvent);
-
-	/**
-	 * 网络延时
-	 */
-	public int getAveragePingTime();
-
-	/**
-	 * 获取服务器Session字符
-	 */
-	public String getSession();
-
-	/**
-	 * 设置服务器Session字符
-	 * @param session
-	 */
-	public void setSession(String session);
-}

web_group/src/main/java/com/group/service/Message.java

@@ -1,47 +0,0 @@
-package com.group.service;
-
-import com.taurus.core.entity.ITObject;
-import com.taurus.core.util.ICallback;
-
-/**
- * 网络协议消息包
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public class Message {
-     /**
-      * 使用文本，不转换ITObject
-      */
-     public boolean useText;
-	/**
-	 * 网络协议指令Key
-	 */
-	public String					command;
-
-	/**
-	 * 参数
-	 */
-	public ITObject					param;
-    /**
-     * 文本数据
-     */
-    public String textParam;
-    
-	/**
-	 * 网络消息回调
-	 */
-	public ICallback<MessageResponse>	callback;
-	/**
-	 * 自定义参数
-	 */
-	public Object					data;
-	/**
-	 * 网络包最后写入时间
-	 */
-	public long						lastTime;
-	/**
-	 * 网络包返回匹配的GID
-	 */
-	public int						gid;
-
-}

web_group/src/main/java/com/group/service/MessageResponse.java

@@ -1,17 +0,0 @@
-package com.group.service;
-
-/**
- * 缃戠粶娑堟伅鍥炶皟鏁版嵁瀵硅薄
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public class MessageResponse {
-	/**
-	 * 娑堟伅鏁版嵁
-	 */
-	public Message	messageData;
-	/**
-	 * 杩斿洖缁撴瀯鍊?
-	 */
-	public int		returnCode;
-}

web_group/src/main/java/com/group/service/NetManager.java

@@ -1,107 +0,0 @@
-package com.group.service;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * 网络管理对象
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public class NetManager {
-
-	private static List<TaurusClient>			_clientList		= new ArrayList<TaurusClient>();
-	private static Map<Integer, TaurusClient>	_clientMap		= new HashMap<Integer, TaurusClient>();
-
-	/**
-	 * 网络超时时间
-	 */
-	public static int						TIMEOUT_TIME	= 10;
-	/**
-	 * 网络版本号
-	 */
-    public static int 						VERSION			= 10;
-	public static boolean 					debug_print 	= false;
-
-	static void addNetClient(TaurusClient net) {
-		if (net == null) {
-			throw new RuntimeException("TaurusClient is null, it can't be added.");
-		}
-		if (_clientMap.containsKey(net.getId())) {
-			throw new RuntimeException("A client with id: " + net.getId() + " already exists! Controller can't be added: " + net);
-		}
-		synchronized (_clientList) {
-			_clientMap.put(net.getId(), net);
-			_clientList.add(net);
-		}
-
-	}
-
-	/**
-	 * The process events
-	 */
-	public static void processEvents() {
-		for (int i = 0; i < _clientList.size(); ++i) {
-			TaurusClient nc = _clientList.get(i);
-			if (nc._kill){
-				synchronized (_clientList) {
-	                _clientMap.remove(nc.getId());
-	                _clientList.remove(nc);
-				}
-                i--;
-                continue;
-            }else{
-            	nc.processEvent();
-            }
-		}
-
-	}
-
-	/**
-	 * 返回指定的NetClient
-	 * @param id
-	 * @return
-	 */
-	public static TaurusClient getNetClient(int id) {
-		for (int i = 0; i < _clientList.size(); ++i) {
-			TaurusClient nc = _clientList.get(i);
-			if (nc.getId() == id)
-				return nc;
-		}
-		return null;
-	}
-
-	/**
-	 * 断开指定连接
-	 * @param id
-	 */
-	public static void killConnection(int id) {
-		for (int i = 0; i < _clientList.size(); ++i) {
-			TaurusClient nc = _clientList.get(i);
-			if (nc.getId() == id) {
-				nc.killConnection();
-				synchronized (_clientList) {
-					_clientMap.remove(id);
-					_clientList.remove(nc);
-				}
-				break;
-			}
-		}
-	}
-
-	/**
-	 * 断开所有连接
-	 */
-	public static void killAllConnection() {
-		synchronized (_clientList) {
-			for (TaurusClient nc : _clientList) {
-				nc.killConnection();
-			}
-			_clientList.clear();
-			_clientMap.clear();
-		}
-
-	}
-}

web_group/src/main/java/com/group/service/SocketCode.java

@@ -1,46 +0,0 @@
-package com.group.service;
-
-/**
- * Socket code
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public enum SocketCode {
-    /**
-     * 杩炴帴鎴愬姛
-     */
-    Connect,
-    Disconnect,
-    /**
-     * 鏈煡寮傚父
-     */
-    Exception,
-    /**
-     * 杩炴帴鏈嶅姟寮傚父
-     */
-    ExceptionOnConnect,
-    /**
-     * 鍙戦€佹暟鎹祦閿欒
-     */
-    SendError,
-    /**
-     * 鎺ユ敹鏈嶅姟鍣ㄦ暟鎹祦寮傚父
-     */
-    ExceptionOnReceive,
-    /**
-     * 鏈嶅姟鍣ㄨ繛鎺ヨ秴鏃?
-     */
-    TimeoutDisconnect,
-    /**
-     * 鏈嶅姟鍣ㄦ柇寮€杩炴帴
-     */
-    DisconnectByServer,
-    /**
-     * 瀹㈡埛绔綉缁滃紓甯?
-     */
-    NetworkException,
-    /**
-     * 杩炴帴瀹夊叏寮傚父锛屼竴鑸负闃茬伀澧欓樆姝?
-     */
-    SecurityExceptionOnConnect
-}

web_group/src/main/java/com/group/service/TaurusClient.java

@@ -1,355 +0,0 @@
-package com.group.service;
-
-import com.group.service.io.SocketEngine;
-import com.taurus.core.entity.ITObject;
-import com.taurus.core.events.Event;
-import com.taurus.core.events.EventDispatcher;
-import com.taurus.core.events.IEventListener;
-import com.taurus.core.util.ICallback;
-import com.taurus.core.util.Logger;
-
-
-import java.util.HashMap;
-import java.util.Map;
-
-
-/**
- * 缃戠粶鏍稿績鍔熻兘澶勭悊绫?
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public class TaurusClient {
-	private static final Logger	logger	= Logger.getLogger(TaurusClient.class);
-	private int					_id;
-	private boolean				_connect;
-	private String				_gameId;
-	private String				_currentIp;
-
-	private ISocketLayer		_scoket;
-	private ThreadManager		_threadManager;
-
-	private EventDispatcher		_dispatcher;
-	private Map<Message, ICallback<MessageResponse>> _responseMap;
-	volatile boolean			_kill	= false;
-	private static int			_tempID;
-	private int _protocolCompression=100;
-	
-	private static int _getValidID() {
-		return ++_tempID;
-	}
-
-	/**
-	 * 
-	 * @param host
-	 * @param game
-	 * @param protocol
-	 */
-	public TaurusClient(String host, String game, ConnectionProtocol protocol) {
-		ISocketLayer socket = null;
-		switch (protocol) {
-		case Tcp:
-			socket = new SocketEngine(this);
-			break;
-		case Web:
-			socket = new HttpClient(host);
-			_connect = true;
-			break;
-		default:
-			socket = new SocketEngine(this);
-			break;
-		}
-		__initialize(host, game, socket);
-	}
-
-	/**
-	 * 
-	 * @param host
-	 * @param game
-	 * @param socket
-	 * @param protocol
-	 */
-	public TaurusClient(String host, String game, ISocketLayer socket) {
-		__initialize(host, game, socket);
-		
-	}
-
-	/**
-	 * 娣诲姞浜嬩欢
-	 * @param eventType
-	 * @param listener
-	 */
-	public void addEventListener(String eventType, IEventListener listener) {
-		_dispatcher.addEventListener(eventType, listener);
-	}
-
-	/**
-	 * 绉婚櫎鎵€鏈変簨浠?
-	 */
-	public void removeAllListener() {
-		_dispatcher.removeAllListener();
-	}
-
-	/**
-	 * 绉婚櫎鎸囧畾浜嬩欢
-	 * @param eventType
-	 * @param listener
-	 */
-	public void removeEventListener(String eventType, IEventListener listener) {
-		_dispatcher.removeEventListener(eventType, listener);
-	}
-
-	private void __initialize(String host, String game, ISocketLayer socket) {
-		_dispatcher = new EventDispatcher();
-		_responseMap = new HashMap<Message, ICallback<MessageResponse>>();
-        _id = _getValidID();
-        NetManager.addNetClient(this);
-        _currentIp = host;
-        _gameId = game;
-        _threadManager = new ThreadManager();
-        _threadManager.start();
-        _scoket = socket;
-		_scoket.setOnConnect(new ICallback<SocketCode>() {
-			@Override
-			public void action(SocketCode code) {
-				if (code == SocketCode.Connect)
-					_connect = true;
-				else
-					_connect = false;
-				Event evt = new Event(NetClientEvent.Connect);
-				evt.setParameter("code", code);
-				_dispatcher.dispatchEvent(evt);
-			}
-		});
-		_scoket.setOnEvent(new ICallback<Message>() {
-			@Override
-			public void action(Message msg) {
-				if (_connect) {
-					Event evt = new Event(NetClientEvent.OnEvent);
-					evt.setParameter("msg", msg);
-					_dispatcher.dispatchEvent(evt);
-					if (NetManager.debug_print){
-                        String str = getProtocol() == ConnectionProtocol.Web ? "WEB" : "TCP";
-                        String data_str = msg.useText ? msg.textParam : (msg.param != null ? msg.param.toJson() : "null");
-                        logger.debug("Event [" + str + "->" + msg.command + "] data:" + data_str);
-                    }
-				}
-			}
-		});
-	}
-
-	/**
-	 * 杩炴帴鏈嶅姟鍣?
-	 */
-	public void connect() {
-		_scoket.connect(_currentIp, _gameId);
-	}
-
-	/**
-	 * Kill net connection.
-	 */
-	public void killConnection() {
-		removeAllListener();
-		_responseMap.clear();
-		_threadManager.stop();
-		_scoket.disconnect();
-		_connect = false;
-		_kill = true;
-	}
-
-	/**
-	 * 鍙戦€佺綉缁滄秷鎭寘
-	 * @param msg	娑堟伅鏁版嵁瀵硅薄
-	 */
-	protected void send(Message msg) {
-		if (msg == null)
-			return;
-		if (NetManager.debug_print) {
-			String str = getProtocol() == ConnectionProtocol.Web ? "WEB" : "TCP";
-			String data_str = msg.useText ? msg.textParam : (msg.param != null ? msg.param.toJson() : "null");
-			str = "Requst [" + str + "->" + msg.command + "] ->data:" + data_str;
-			logger.debug(str);
-		}
-		if (msg.callback != null) {
-			_responseMap.put(msg, msg.callback);
-			msg.callback = new ICallback<MessageResponse>() {
-
-				@Override
-				public void action(MessageResponse response) {
-					ICallback<MessageResponse> callback = _responseMap.get(response.messageData);
-					if (callback == null) {
-						return;
-					}
-					if (NetManager.debug_print){
-                        String str = getProtocol() == ConnectionProtocol.Web ? "WEB" : "TCP";
-                        String data_str = msg.useText ? msg.textParam : (msg.param != null ? msg.param.toJson() : "null");
-                        str = "Response [" + str + "->" + msg.command + "] ->result:"+ response.returnCode + " param:" + data_str;
-                        logger.debug(str);
-                    }
-					_responseMap.remove(response.messageData);
-					callback.action(response);
-				}
-
-			};
-		}
-
-		_threadManager.enqueueSend(new ICallback<Message>() {
-			@Override
-			public void action(Message msg) {
-				sendData(msg);
-			}
-		}, msg);
-	}
-
-	
-	/**
-	 * 鍙戦€佺綉缁滄秷鎭寘
-	 * @param cmd 		缃戠粶鍗忚鎸囦护Key
-	 * @param param		鍙傛暟
-	 * @param callback	缃戠粶娑堟伅鍥炶皟
-	 * @return
-	 */
-	public Message send(String cmd, ITObject param, ICallback<MessageResponse> callback) {
-		Message msg = new Message();
-		msg.useText = false;
-		msg.command = cmd;
-		msg.param = param;
-		msg.callback = callback;
-		this.send(msg);
-		return msg;
-	}
-	
-	/**
-	 * 鍙戦€佺綉缁滄秷鎭寘
-	 * @param cmd 		缃戠粶鍗忚鎸囦护Key
-	 * @param text		JSON 鏂囨湰
-	 * @param callback	缃戠粶娑堟伅鍥炶皟
-	 * @return
-	 */
-	public Message sendText(String cmd, String text, ICallback<MessageResponse> callback) {
-		Message msg = new Message();
-		msg.command = cmd;
-		msg.textParam = text;
-		msg.callback = callback;
-		msg.useText = true;
-		this.send(msg);
-		return msg;
-	}
-
-	/**
-	 * 鍙戦€佹暟鎹嚎绋嬫柟娉?
-	 * @param message
-	 */
-	private void sendData(Message msg) {
-		_scoket.write(msg);
-	}
-
-	/**
-	 * 澶勭悊浜嬩欢
-	 */
-	void processEvent() {
-		if(_scoket!=null) {
-			_scoket.processEvent();
-		}
-	}
-
-	/**
-	 * 娓呯悊鍝嶅簲鍥炶皟
-	 */
-	public void clearResponse() {
-		_responseMap.clear();
-	}
-
-	/**
-	 * The net client id
-	 * @return
-	 */
-	public int getId() {
-
-		return _id;
-
-	}
-
-	/**
-	 * The curren ip
-	 * @return
-	 */
-	public String getCurrentIp() {
-		return _currentIp;
-	}
-
-	/**
-	 * The curren Game ID
-	 * @return
-	 */
-	public String getGameID() {
-		return _gameId;
-	}
-
-	/**
-	 * 缃戠粶鏄惁宸茶繛鎺?
-	 * @return
-	 */
-	public boolean isConnected() {
-		return _connect;
-	}
-
-	/**
-	 * 褰撳墠缃戠粶绫诲瀷
-	 * @return
-	 */
-	public ConnectionProtocol getProtocol() {
-		return _scoket.getProtocol();
-	}
-
-	/**
-	 * 缃戠粶寤舵椂
-	 * @return
-	 */
-	public int getAveragePingTime() {
-		return _scoket.getAveragePingTime();
-	}
-
-	/**
-	 * 鑾峰彇鏈嶅姟鍣⊿ession瀛楃
-	 * @return
-	 */
-	public String getSession() {
-		return _scoket.getSession();
-	}
-
-	/**
-	 * 璁剧疆鏈嶅姟鍣⊿ession瀛楃
-	 * @param session
-	 */
-	public void setSession(String session) {
-		_scoket.setSession(session);
-	}
-	
-	/**
-	 * 璁剧疆鍘嬬缉闃€鍊?
-	 * @param protocolCompression
-	 */
-	public void setProtocolCompression(int protocolCompression) {
-		_protocolCompression = protocolCompression;
-	}
-
-	/**
-	 * 鑾峰彇鍘嬬缉闃€鍊?
-	 * 
-	 * @return
-	 */
-	public int getProtocolCompression() {
-		return _protocolCompression;
-	}
-    
-
-	public static enum ConnectionProtocol {
-		Tcp, Web,
-	}
-
-	public static class NetClientEvent {
-		public static final String	Connect	= "connect";
-
-		public static final String	OnEvent	= "onevent";
-	}
-}

web_group/src/main/java/com/group/service/ThreadManager.java

@@ -1,123 +0,0 @@
-package com.group.service;
-
-import com.taurus.core.util.ICallback;
-
-import java.util.LinkedList;
-import java.util.Queue;
-import java.util.Stack;
-
-
-
-/**
- * 网络(读/写) 线程管理
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public class ThreadManager {
-	class MessgaeThread {
-		public ICallback<Message>	_dataDelegate;
-
-		public Message			_message;
-	}
-	
-	private boolean					_running			= false;
-	private Thread					_outThread;
-	private boolean					_outHasQueuedItems	= false;
-	private Queue<MessgaeThread>	_outThreadQueue		= new LinkedList<MessgaeThread>();
-	private Object					_outQueueLocker		= new Object();
-	private Stack<MessgaeThread> pool = new Stack<MessgaeThread>();
-
-	private MessgaeThread MessgaeThreadGet() {
-		if (pool.size() > 0) {
-			MessgaeThread msg = pool.pop();
-			return msg;
-		} else
-			return new MessgaeThread();
-	}
-
-	private void MessgaeThreadReturn(MessgaeThread value) {
-		pool.push(value);
-		value._dataDelegate = null;
-		value._message = null;
-	}
-
-	private static void sleep(int ms) {
-		try {
-			Thread.sleep(ms);
-		} catch (InterruptedException e) {
-			
-		}
-	}
-
-	private void outThread() {
-		while (_running) {
-			sleep(5);
-			if (_outHasQueuedItems) {
-				Object obj = _outQueueLocker;
-				synchronized (obj) {
-					while (_outThreadQueue.size() > 0) {
-						MessgaeThread item = _outThreadQueue.poll();
-						processOutItem(item);
-					}
-					_outHasQueuedItems = false;
-				}
-			}
-		}
-	}
-
-	private void processOutItem(MessgaeThread item) {
-		ICallback<Message> _dataDelegate = item._dataDelegate;
-		Message message = item._message;
-		MessgaeThreadReturn(item);
-		if (_dataDelegate != null) {
-			_dataDelegate.action(message);
-		}
-	}
-
-	public void start() {
-		if (_running) {
-			return;
-		}
-		_running = true;
-		if (_outThread == null) {
-			_outThread = new Thread(new Runnable() {
-				@Override
-				public void run() {
-					outThread();
-				}
-			});
-			_outThread.start();
-			_outThread.setName("WriteMessage");
-		}
-	}
-
-	public void stop() {
-		Thread thread = new Thread(new Runnable() {
-			@Override
-			public void run() {
-				stopThread();
-			}
-		});
-		thread.start();
-	}
-
-	private void stopThread() {
-		_running = false;
-
-		if (_outThread != null) {
-			_outThread.interrupt();
-		}
-		_outThread = null;
-	}
-
-	public void enqueueSend(ICallback<Message> callback, Message message) {
-		MessgaeThread hashtable = MessgaeThreadGet();
-		hashtable._dataDelegate = callback;
-		hashtable._message = message;
-		Object obj = _outQueueLocker;
-		synchronized (obj) {
-			_outThreadQueue.offer(hashtable);
-			_outHasQueuedItems = true;
-		}
-	}
-}

web_group/src/main/java/com/group/service/io/IOHandler.java

@@ -1,251 +0,0 @@
-package com.group.service.io;
-
-import com.group.service.SocketCode;
-import com.taurus.core.entity.ITObject;
-import com.taurus.core.entity.TObject;
-import com.taurus.core.util.ByteArray;
-import com.taurus.core.util.Logger;
-import com.taurus.core.util.Utils;
-
-import java.io.IOException;
-
-/**
- * 鍗忚鍖呭瓧鑺傛祦瑙ｆ瀽
- * 
- * @author daixiwei daixiwei15@126.com
- *
- */
-class IOHandler {
-	private final Logger			logger			= Logger.getLogger(IOHandler.class);
-	private static final String		ACTION_ID		= "a";
-	private static final String		PARAM_ID		= "p";
-	static final int				INT_BYTE_SIZE	= 4;
-
-	private static final ByteArray	EMPTY_BUFFER	= new ByteArray();
-	private int						_skipBytes		= 0;
-	private ByteArray				_buffer;
-	private int						_expectedLength	= -1;
-	private boolean					_compressed		= false;
-
-	private SocketEngine			_engine;
-	private SystemController		_controller;
-	
-	/**
-	 * 璇诲彇鐘舵€?
-	 */
-	private PacketReadState			_readState		= PacketReadState.WAIT_NEW_PACKET;
-
-	IOHandler(SocketEngine client, SystemController controller) {
-		_engine = client;
-		_controller = controller;
-	}
-
-	/**
-	 * 
-	 * @param data
-	 * @return
-	 * @throws IOException
-	 */
-	private ByteArray handleDataSize(ByteArray data) throws Exception {
-		int num = data.readInt();
-		int pos = INT_BYTE_SIZE;
-		if (num != -1) {
-			_expectedLength = num;
-			data = resizeByteArray(data, pos, data.length() - pos);
-			_readState = PacketReadState.WAIT_DATA;
-			return data;
-		}
-		_readState = PacketReadState.WAIT_DATA_SIZE_FRAGMENT;
-		_buffer.writeBytes(data.bytes());
-		data = EMPTY_BUFFER;
-		return data;
-	}
-
-	/**
-	 * 
-	 * @param data
-	 * @return
-	 * @throws IOException
-	 */
-	private ByteArray handleDataSizeFragment(ByteArray data) throws Exception {
-		int count = INT_BYTE_SIZE - _buffer.length();
-		if (data.length() >= count) {
-			_buffer.writeBytes(data.bytes(), 0, count);
-			int num2 = INT_BYTE_SIZE;
-			ByteArray array = new ByteArray();
-			array.writeBytes(_buffer.bytes(), 0, num2);
-			array.position(0);
-			int num3 = array.readInt();
-			_expectedLength = num3;
-			_buffer = new ByteArray();
-			_readState = PacketReadState.WAIT_DATA;
-			if (data.length() > count) {
-				data = resizeByteArray(data, count, data.length() - count);
-				return data;
-			}
-			data = EMPTY_BUFFER;
-			return data;
-		}
-		_buffer.writeBytes(data.bytes());
-		data = EMPTY_BUFFER;
-		return data;
-	}
-
-	/**
-	 * 
-	 * @param data
-	 * @return
-	 */
-	private ByteArray handleInvalidData(ByteArray data) {
-		if (_skipBytes == 0) {
-			_readState = PacketReadState.WAIT_NEW_PACKET;
-			return data;
-		}
-		int pos = Math.min(data.length(), _skipBytes);
-		data = resizeByteArray(data, pos, data.length() - pos);
-		_skipBytes -= pos;
-		return data;
-	}
-
-	/**
-	 * 
-	 * @param data
-	 * @return
-	 */
-	private ByteArray handleNewPacket(ByteArray data) throws Exception {
-		_buffer = new ByteArray();
-		_expectedLength = -1;
-		_compressed = data.readByte() > 0;
-		data = resizeByteArray(data, 1, data.length() - 1);
-		_readState = PacketReadState.WAIT_DATA_SIZE;
-		return data;
-	}
-
-	/**
-	 * 
-	 * @param data
-	 * @return
-	 */
-	private ByteArray handlePacketData(ByteArray data) throws Exception {
-		int count = _expectedLength - _buffer.length();
-		boolean flag = data.length() > count;
-		ByteArray array = new ByteArray(data.bytes());
-		try {
-			if (data.length() >= count) {
-				_buffer.writeBytes(data.bytes(), 0, count);
-				if (_compressed) {
-					_buffer.uncompress();
-				}
-				dispatchRequest(_buffer);
-				_readState = PacketReadState.WAIT_NEW_PACKET;
-			} else {
-				_buffer.writeBytes(data.bytes());
-			}
-			if (flag) {
-				data = resizeByteArray(data, count, data.length() - count);
-				return data;
-			}
-			data = EMPTY_BUFFER;
-		} catch (Exception exception) {
-			logger.error("Error handling data ", exception);
-			_skipBytes = count;
-			_readState = PacketReadState.INVALID_DATA;
-			return array;
-		}
-		return data;
-	}
-
-	private void dispatchRequest(ByteArray _buffer) throws IOException {
-		ITObject requestObject = TObject.newFromBinaryData(_buffer.bytes());
-		Packet packet = new Packet();
-		if (requestObject.isNull(ACTION_ID)) {
-			throw new IOException("Request rejected: No Action ID in request!");
-		}
-		packet.actionID = requestObject.getByte(ACTION_ID);
-		packet.parameters = requestObject.getTObject(PARAM_ID);
-		_controller.handlePacket(packet);
-	}
-
-	/**
-	 * 
-	 * @param data
-	 * @throws IOException
-	 */
-	public void onDataRead(ByteArray data) throws IOException {
-		if (data.length() == 0) {
-			throw new IOException("Unexpected empty packet data: no readable bytes available!");
-		}
-
-		data.position(0);
-		try {
-			while (data.length() > 0) {
-				if (_readState == PacketReadState.WAIT_NEW_PACKET) {
-					data = handleNewPacket(data);
-				} else {
-					if (_readState == PacketReadState.WAIT_DATA_SIZE) {
-						data = handleDataSize(data);
-						continue;
-					}
-					if (_readState == PacketReadState.WAIT_DATA_SIZE_FRAGMENT) {
-						data = handleDataSizeFragment(data);
-						continue;
-					}
-					if (_readState == PacketReadState.WAIT_DATA) {
-						data = handlePacketData(data);
-						continue;
-					}
-					if (_readState == PacketReadState.INVALID_DATA) {
-						data = handleInvalidData(data);
-					}
-				}
-			}
-		} catch (Exception e) {
-			this.logger.error(e);
-		}
-
-	}
-
-	/**
-	 * 
-	 * @param packet
-	 * @throws IOException
-	 */
-	public void onDataWrite(Packet packet) throws IOException {
-		ByteArray binData = new ByteArray();
-
-		ITObject obj2 = new TObject();
-		obj2.putByte(ACTION_ID, (byte) packet.actionID);
-		obj2.putTObject(PARAM_ID, packet.parameters);
-		byte[] array2 = obj2.toBinary();
-		boolean compression = array2.length > _engine.getProtocolCompression();
-		if (compression) {
-			array2 = Utils.compress(array2);
-		}
-		binData.writeByte((byte) (compression ? 1 : 0));
-		binData.writeInt(array2.length);
-		binData.writeBytes(array2);
-
-		if (_engine._socket.isConnected()) {
-			if (!_engine._socket.send(binData.bytes())) {
-				_engine.enqueueStatusCallback(SocketCode.SendError);
-			}
-		}
-	}
-
-	/**
-	 * 
-	 * @param array
-	 * @param pos
-	 * @param len
-	 * @return
-	 */
-	private ByteArray resizeByteArray(ByteArray array, int pos, int len) {
-		byte[] dst = new byte[len];
-		System.arraycopy(array.bytes(), pos, dst, 0, len);
-		return new ByteArray(dst);
-	}
-
-	static enum PacketReadState {
-		WAIT_NEW_PACKET, WAIT_DATA_SIZE, WAIT_DATA_SIZE_FRAGMENT, WAIT_DATA, INVALID_DATA
-	}
-}

web_group/src/main/java/com/group/service/io/ISocket.java

@@ -1,113 +0,0 @@
-package com.group.service.io;
-
-import com.group.service.SocketCode;
-import com.taurus.core.util.Logger;
-import com.taurus.core.util.StringUtil;
-
-/**
- * Socket核心基类
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public abstract class ISocket {
-	protected static final Logger	logger	= Logger.getLogger(ISocket.class);
-	SocketEngine						_engine;
-	/**
-	 * 当前socket状态
-	 */
-	public volatile SocketState		state	= SocketState.Disconnected;
-	/**
-	 * 服务器地址
-	 */
-	protected String				serverAddress;
-	/**
-	 * 服务器端口
-	 */
-	protected int					serverPort;
-
-	/**
-	 * 网络链接状态
-	 * @return
-	 */
-	public boolean isConnected() {
-		return state == SocketState.Connected;
-	}
-
-	public ISocket(SocketEngine peerBase) {
-		if (peerBase == null) {
-			throw new RuntimeException("Can't init without peer");
-		}
-		this._engine = peerBase;
-	}
-
-	public boolean connect() {
-		boolean result;
-		if (state != SocketState.Disconnected) {
-			logger.error("Calling connect when the socket is not disconnected");
-			result = false;
-		} else if (_engine == null) {
-			result = false;
-		} else if (!tryParseAddress(_engine.getServerAddress())) {
-			logger.error("Failed parsing address: " + _engine.getServerAddress());
-			result = false;
-		} else {
-			result = true;
-		}
-		return result;
-	}
-
-	/**
-	 * 断开连接
-	 * @return
-	 */
-	public abstract boolean disconnect();
-
-	/**
-	 * 发送流数据
-	 * @param data
-	 * @return
-	 */
-	public abstract boolean send(byte[] data);
-
-	protected void handleException(SocketCode statusCode) {
-		state = SocketState.Disconnecting;
-		_engine.enqueueStatusCallback(statusCode);
-		_engine.enqueueActionForDispatch(new Runnable() {
-			@Override
-			public void run() {
-				disconnect();
-			}
-		});
-	}
-
-	/**
-	 * 解析网络地址
-	 * @param addressAndPort
-	 * @param address
-	 * @param port
-	 * @return
-	 */
-	protected boolean tryParseAddress(String addressAndPort) {
-		boolean result = true;
-		if (StringUtil.isEmpty(addressAndPort)) {
-			result = false;
-		} else {
-			int num = addressAndPort.lastIndexOf(':');
-			if (num <= 0) {
-				result = false;
-			} else if (addressAndPort.indexOf(':') != num && (!addressAndPort.contains("[") || !addressAndPort.contains("]"))) {
-				result = false;
-			} else {
-				serverAddress = addressAndPort.substring(0, num);
-				String s = addressAndPort.substring(num + 1);
-				try {
-					serverPort = Integer.parseInt(s);
-				} catch (Exception e) {
-					result = false;
-				}
-			}
-		}
-		return result;
-	}
-
-}

web_group/src/main/java/com/group/service/io/LagMonitor.java

@@ -1,104 +0,0 @@
-package com.group.service.io;
-
-import com.taurus.core.util.Logger;
-
-import java.io.IOException;
-import java.util.*;
-
-
-/**
- * 网络监听/心跳
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public class LagMonitor extends TimerTask {
-	private static final Logger	logger	= Logger.getLogger(LagMonitor.class);
-	private int			_averagePingTime;
-	private Timer		_pollTimer;
-	private int			_queueSize;
-	private int			_interval;
-	private boolean		_run;
-	private SocketEngine	_engine;
-	private List<Long>	_valueQueue;
-	private Queue<Long>	write_queue	= new LinkedList<Long>();
-
-	public LagMonitor(SocketEngine engine, int interval, int queueSize) {
-		if (interval < 1) {
-			interval = 1;
-		}
-		_engine = engine;
-		_valueQueue = new ArrayList<Long>();
-		_queueSize = queueSize;
-		_pollTimer = new Timer();
-		_interval = interval * 1000;
-		_averagePingTime = 1000;
-	}
-
-	public void fnPingPong() {
-		long w_time = write_queue.poll();
-		long item = System.currentTimeMillis() - w_time;
-		if (_valueQueue.size() >= _queueSize) {
-			_valueQueue.remove(0);
-		}
-		_valueQueue.add(item);
-		if (_valueQueue.size() == 0) {
-			_averagePingTime = 0;
-		}
-		int num = 0;
-		for (int i = 0; i < _valueQueue.size(); ++i) {
-			num += _valueQueue.get(i);
-		}
-		_averagePingTime = (num / _valueQueue.size());
-	}
-
-	/**
-	 * 启动监听
-	 */
-	public void start() {
-		if (!isRunning()) {
-			this._run = true;
-			_pollTimer.scheduleAtFixedRate(this, 0, _interval);
-		}
-	}
-
-	/**
-	 * 停止监听
-	 */
-	public void stop() {
-		if (isRunning()) {
-			this._run = false;
-			_pollTimer.purge();
-		}
-	}
-
-	public void destroy() {
-		stop();
-		_pollTimer.cancel();
-		_pollTimer = null;
-		_engine = null;
-	}
-
-	/**
-	 * 平均Ping的时间
-	 * @return
-	 */
-	public int getAveragePingTime() {
-		return _averagePingTime;
-
-	}
-
-	public boolean isRunning() {
-		return _run;
-
-	}
-
-	@Override
-	public void run() {
-		try {
-			_engine._controller.sendPingPong();
-			write_queue.offer(System.currentTimeMillis());
-		} catch (IOException e) {
-			logger.error(e);
-		}
-	}
-}

web_group/src/main/java/com/group/service/io/Packet.java

@@ -1,20 +0,0 @@
-package com.group.service.io;
-
-import com.taurus.core.entity.ITObject;
-
-/**
- * 网络协议数据包
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public class Packet {
-
-	/**
-	 * 协议包ID
-	 */
-	public int			actionID;
-	/**
-	 * 参数数据
-	 */
-	public ITObject	parameters;
-}

web_group/src/main/java/com/group/service/io/SocketEngine.java

@@ -1,160 +0,0 @@
-package com.group.service.io;
-
-import com.group.service.*;
-import com.group.service.TaurusClient.ConnectionProtocol;
-import com.taurus.core.util.ICallback;
-import com.taurus.core.util.Logger;
-
-import java.io.IOException;
-import java.util.LinkedList;
-import java.util.Queue;
-
-/**
- * SocketEngine
- * 
- * @author daixiwei daixiwei15@126.com
- *
- */
-public class SocketEngine implements ISocketLayer {
-	private static final Logger		logger			= Logger.getLogger(SocketEngine.class);
-
-	ISocket							_socket;
-	private String					_sessionToken;
-	private String					_serverAddress;
-
-	private ICallback<SocketCode>	_onConnect;
-	ICallback<Message>		_onEvent;
-	private Queue<Runnable>			_actionQueue	= new LinkedList<Runnable>();
-
-	SystemController				_controller;
-	LagMonitor						_lagMonitor;
-	IOHandler						_packetHandler;
-	TaurusClient _client;
-	
-	public SocketEngine(TaurusClient client) {
-		_socket = new SocketTcp(this);
-		_controller = new SystemController(this);
-		_packetHandler = new IOHandler(this, _controller);
-		_lagMonitor = new LagMonitor(this, NetManager.TIMEOUT_TIME, 10);
-	}
-
-	@Override
-	public void connect(String ip, String game) {
-		_serverAddress = ip;
-		_socket.connect();
-	}
-
-	@Override
-	public void disconnect() {
-		_socket.disconnect();
-		if (_lagMonitor != null)
-			_lagMonitor.destroy();
-		_lagMonitor = null;
-	}
-
-	@Override
-	public void processEvent() {
-		if (_socket.isConnected() && _controller.isTimeOut()) {
-			if (_onConnect != null)
-				_onConnect.action(SocketCode.TimeoutDisconnect);
-			return;
-		}
-
-		while (true) {
-			Runnable myAction;
-			synchronized (_actionQueue) {
-				if (_actionQueue.size() <= 0) {
-					break;
-				}
-				myAction = _actionQueue.poll();
-			}
-			myAction.run();
-		}
-	}
-
-	public void write(Message msg) {
-		try {
-			_controller.sendRequest(msg);
-		} catch (IOException e) {
-			logger.error(e);
-		}
-	}
-
-	void onStatusChanged(SocketCode statusCode) {
-		if (statusCode != SocketCode.Connect) {
-			_socket.disconnect();
-			this.disconnect();
-		}
-		System.out.println("status : " + statusCode);
-		logger.info("status : " + statusCode);
-		if (statusCode == SocketCode.Connect) {
-			this.onHandshake();
-		} else {
-			if (_onConnect != null)
-				_onConnect.action(statusCode);
-		}
-	}
-
-	void onHandshake() {
-		_controller.refreshTime();
-		_lagMonitor.start();
-		if (_onConnect != null)
-			_onConnect.action(SocketCode.Connect);
-	}
-
-	void enqueueActionForDispatch(Runnable action) {
-		synchronized (_actionQueue) {
-			_actionQueue.offer(action);
-		}
-	}
-
-	void enqueueStatusCallback(final SocketCode statusValue) {
-		synchronized (_actionQueue) {
-			_actionQueue.offer(new Runnable() {
-				@Override
-				public void run() {
-					onStatusChanged(statusValue);
-				}
-			});
-		}
-	}
-
-
-
-
-	public String getServerAddress() {
-		return _serverAddress;
-	}
-
-	@Override
-	public ConnectionProtocol getProtocol() {
-		return ConnectionProtocol.Tcp;
-	}
-	
-	public int getAveragePingTime() {
-		return _lagMonitor != null ? _lagMonitor.getAveragePingTime() : 1000;
-	}
-
-	public void setSession(String session) {
-		_sessionToken = session == null ? "" : session;
-	}
-
-	public String getSession() {
-		return _sessionToken;
-
-	}
-
-	@Override
-	public void setOnConnect(ICallback<SocketCode> onConnect) {
-		_onConnect = onConnect;
-	}
-
-	@Override
-	public void setOnEvent(ICallback<Message> onEvent) {
-		_onEvent = onEvent;
-	}
-
-	public int getProtocolCompression() {
-		return 100;
-	}
-}

web_group/src/main/java/com/group/service/io/SocketState.java

@@ -1,13 +0,0 @@
-package com.group.service.io;
-
-/**
- * 缃戠粶鐘舵€佹灇涓?
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public enum SocketState{
-    Disconnected,
-    Connecting,
-    Connected,
-    Disconnecting
-}

web_group/src/main/java/com/group/service/io/SocketTcp.java

@@ -1,156 +0,0 @@
-package com.group.service.io;
-
-import com.group.service.NetManager;
-import com.group.service.SocketCode;
-import com.taurus.core.util.ByteArray;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.SocketAddress;
-
-/**
- * Tcp协议 Socket核心类
- * @author daixiwei	daixiwei15@126.com
- *
- */
-class SocketTcp extends ISocket {
-	private static final int	READ_BUFFER_SIZE	= 0x1000;
-	private byte[]				_byteBuffer			= new byte[READ_BUFFER_SIZE];
-
-	private Socket				_connection;
-	private final Object		_syncer				= new Object();
-
-	public SocketTcp(SocketEngine npeer) {
-		super(npeer);
-	}
-
-	@Override
-	public boolean connect() {
-		if (!super.connect()) {
-			return false;
-		} else {
-			state = SocketState.Connecting;
-			Thread t = new Thread(new Runnable() {
-				@Override
-				public void run() {
-					connectThread();
-				}
-			});
-			t.setName("mpnet connect thread");
-			t.start();
-			return true;
-		}
-	}
-
-	private void connectThread() {
-		try {
-			SocketAddress remoteAddr = new InetSocketAddress(serverAddress, super.serverPort);
-			_connection = new Socket();
-			_connection.setTcpNoDelay(true);
-			_connection.connect(remoteAddr, NetManager.TIMEOUT_TIME * 1000);
-			state = SocketState.Connected;
-			_engine.enqueueStatusCallback(SocketCode.Connect);
-		} catch (Exception ex) {
-			try {
-				if (_connection != null)
-					_connection.close();
-			} catch (Exception e) {
-				logger.error("Connection close error", e);
-			}
-			logger.error("Connection error", ex);
-			handleException(SocketCode.ExceptionOnConnect);
-			return;
-		}
-		Thread _thrSocketReader = new Thread(new Runnable() {
-			@Override
-			public void run() {
-				__read();
-			}
-		});
-		_thrSocketReader.start();
-	}
-
-	@Override
-	public boolean disconnect() {
-		if (state == SocketState.Disconnected)
-			return false;
-		state = SocketState.Disconnecting;
-		synchronized (_syncer) {
-			if (_connection != null) {
-				try {
-					_connection.close();
-				} catch (Exception ex) {
-					logger.error("Disconnect error", ex);
-				}
-				_connection = null;
-			}
-		}
-		state = SocketState.Disconnected;
-		return true;
-	}
-
-	@Override
-	public boolean send(byte[] data) {
-		return __writeSocket(data);
-	}
-
-	private boolean __writeSocket(byte[] buf) {
-		if (_connection == null || state != SocketState.Connected) {
-			logger.error("Trying to write to disconnected socket");
-			return false;
-		} else {
-			try {
-				OutputStream outStream = _connection.getOutputStream();
-				outStream.write(buf);
-				outStream.flush();
-			} catch (Exception exception2) {
-				logger.error("General error writing to socket", exception2);
-				handleException(SocketCode.SendError);
-				return false;
-			}
-		}
-		return true;
-	}
-
-	private void __read() {
-		int size = 0;
- 		while (state == SocketState.Connected) {
-			try {
-				try {
-					InputStream inputStream = _connection.getInputStream();
-					size = inputStream.read(_byteBuffer);
-				} catch (Exception ex) {
-					throw new TSocketException(SocketCode.ExceptionOnReceive);
-				}
-				if (size == 0 || size == -1) {
-//					System.out.println("socket----报错--------111111--------");
-					throw new TSocketException(SocketCode.DisconnectByServer);
-				}
-				__handleBinaryData(_byteBuffer, size);
-
-			} catch (TSocketException ex3) {
-				if (state != SocketState.Disconnecting && state.ordinal() > SocketState.Disconnected.ordinal()) {
-					logger.error("Receiving failed.: ", ex3);
-					super.handleException(ex3.socketCode);
-				}
-			} catch (Exception ex4) {
-				if (state != SocketState.Disconnecting && state.ordinal() > SocketState.Disconnected.ordinal()) {
-					logger.error("Receiving Exception: ", ex4);
-					super.handleException(SocketCode.ExceptionOnReceive);
-				}
-			}
-		}
-
-		this.disconnect();
-	}
-
-	private void __handleBinaryData(byte[] buf, int size) throws IOException {
-		byte[] dst = new byte[size];
-		System.arraycopy(buf, 0, dst, 0, size);
-		ByteArray array = new ByteArray(dst);
-		_engine._packetHandler.onDataRead(array);
-	}
-}

web_group/src/main/java/com/group/service/io/SystemController.java

@@ -1,177 +0,0 @@
-package com.group.service.io;
-
-import com.group.service.Message;
-import com.group.service.MessageResponse;
-import com.group.service.NetManager;
-import com.taurus.core.entity.ITObject;
-import com.taurus.core.entity.TObject;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * 协议处理控制器
- * 
- * @author daixiwei daixiwei15@126.com
- *
- */
-public class SystemController {
-	public static final String				CONNECT_TOKE				= "$t";
-	public static final String				CONNECT_PROT_COMPRESSION	= "$pc";
-	public static final String				REQUEST_CMD					= "$c";
-	public static final String				REQUEST_GID					= "$gi";
-	public static final String				REQUEST_PARM				= "$p";
-	public static final String				REQUEST_RESULT				= "$r";
-
-	/**
-	 * pingpong
-	 */
-	public static final int					ACTION_PINGPONG				= 0;
-	/**
-	 * 客户端请求
-	 */
-	public static final int					ACTION_REQUST_CMD			= 1;
-	/**
-	 * 服务器事件消息
-	 */
-	public static final int					ACTION_EVENT_CMD			= 2;
-
-	private SocketEngine					_engine;
-	private long							_lastTime;
-	private List<Message>					_handleList					= new ArrayList<Message>();
-	private int								_tempID;
-
-	public SystemController(SocketEngine client) {
-		_engine = client;
-		_lastTime = System.currentTimeMillis();
-	}
-
-	private int _getValidID() {
-		return ++_tempID;
-	}
-
-
-	public void handlePacket(Packet packet) {
-		_lastTime = System.currentTimeMillis();
-		switch (packet.actionID) {
-		case ACTION_PINGPONG:
-			fnPingPong(packet);
-			break;
-		case ACTION_REQUST_CMD:
-			fnOperationResponse(packet);
-			break;
-		case ACTION_EVENT_CMD:
-			fnOperationEvent(packet);
-			break;
-		}
-	}
-
-	void sendRequest(Message msg) throws IOException {
-		Packet packet = new Packet();
-		packet.actionID = SystemController.ACTION_REQUST_CMD;
-		ITObject obj = TObject.newInstance();
-		obj.putString(SystemController.REQUEST_CMD, msg.command);
-		int gid = 0;
-		if (msg.callback != null) {
-			gid = _getValidID();
-			msg.lastTime = System.currentTimeMillis();
-			msg.gid = gid;
-			synchronized (_handleList) {
-				_handleList.add(msg);
-			}
-		}
-		obj.putInt(SystemController.REQUEST_GID, gid);
-		obj.putTObject(SystemController.REQUEST_PARM, msg.param);
-		packet.parameters = obj;
-		_engine._packetHandler.onDataWrite(packet);
-	}
-
-	void sendPingPong() throws IOException {
-		Packet packet = new Packet();
-		packet.actionID = SystemController.ACTION_PINGPONG;
-		packet.parameters = TObject.newInstance();
-		_engine._packetHandler.onDataWrite(packet);
-	}
-
-	private final void fnOperationResponse(final Packet packet) {
-		final int gid = packet.parameters.getInt(REQUEST_GID);
-		final int result = packet.parameters.getInt(REQUEST_RESULT);
-		ITObject tem = null;
-		if (packet.parameters.containsKey(REQUEST_PARM)) {
-			tem = packet.parameters.getTObject(REQUEST_PARM);
-		}
-		final ITObject param = tem;
-		_engine.enqueueActionForDispatch(new Runnable() {
-			@Override
-			public void run() {
-				handleOperationResponse(result, gid, param);
-			}
-		});
-	}
-
-	private void handleOperationResponse(int result, int gid, ITObject param) {
-		Message msg = null;
-		int index = 0;
-		int len = _handleList.size();
-		if (gid > 0) {
-			for (int i = 0; i < len; ++i) {
-				if (gid == _handleList.get(i).gid) {
-					msg = _handleList.get(i);
-					index = i;
-					break;
-				}
-			}
-		}
-
-		if (msg != null) {
-			if (msg.callback != null) {
-				MessageResponse response = new MessageResponse();
-				msg.param = param;
-				response.messageData = msg;
-				response.returnCode = result;
-				msg.callback.action(response);
-			}
-			synchronized (_handleList) {
-				_handleList.remove(index);
-			}
-		}
-	}
-
-	private final void fnOperationEvent(Packet packet) {
-		final String cmd = packet.parameters.getString(REQUEST_CMD);
-		ITObject tem = null;
-		if (packet.parameters.containsKey(REQUEST_PARM)) {
-			tem = packet.parameters.getTObject(REQUEST_PARM);
-		}
-		final ITObject param = tem;
-		_engine.enqueueActionForDispatch(new Runnable() {
-			@Override
-			public void run() {
-				handleEvent(cmd, param);
-			}
-		});
-	}
-
-	void handleEvent(String cmd, ITObject param) {
-		Message msg = new Message();
-		msg.command = cmd;
-		msg.param = param;
-		if (_engine._onEvent != null)
-			_engine._onEvent.action(msg);
-	}
-
-
-	private final void fnPingPong(Packet packet) {
-		_engine._lagMonitor.fnPingPong();
-	}
-
-	public boolean isTimeOut() {
-		return (System.currentTimeMillis() - _lastTime) / 1000 > NetManager.TIMEOUT_TIME;
-	}
-
-	public void refreshTime() {
-		_lastTime = System.currentTimeMillis();
-	}
-
-}

web_group/src/main/java/com/group/service/io/TSocketException.java

@@ -1,23 +0,0 @@
-package com.group.service.io;
-
-import com.group.service.SocketCode;
-
-/**
- * 缃戠粶閫氫俊寮傚父
- * @author daixiwei	daixiwei15@126.com
- *
- */
-public class TSocketException extends Exception{
-
-	/**
-	 * 
-	 */
-	private static final long serialVersionUID = 1L;
-	
-	public SocketCode socketCode;
-	
-	public TSocketException(SocketCode code) {
-		socketCode = code;
-	}
-
-}