Merge pull request #445 from ZzzCrazyPig/mydev

修复前端connection关闭后抛出空指针异常的bug，实现mycat服务端预处理（支持mysql_stmt_prepare等的处理及响应）

pom.xml

@@ -168,6 +168,14 @@
 			<artifactId>log4j-core</artifactId>
 			<version>2.3</version>
 		</dependency>
+
+        <!-- joda日期处理工具 -->
+        <dependency>
+			<groupId>joda-time</groupId>
+			<artifactId>joda-time</artifactId>
+			<version>2.8.2</version>
+		</dependency>
+
 	</dependencies>
 
 

src/main/java/io/mycat/net/Connection.java

@@ -233,6 +233,12 @@ public void onReadData(int got) throws IOException {
 				// handle this package
 				readBuffer.position(offset);
 				handle(readBuffer, offset, length);
+
+				// maybe handle stmt_close
+				if(isClosed()) {
+					return ;
+				}
+
 				// offset to next position
 				offset += length;
 				// reached end

src/main/java/io/mycat/server/FrontendPrepareHandler.java

@@ -34,6 +34,7 @@ public interface FrontendPrepareHandler {
 
     void execute(byte[] data);
 
-    void close();
+    void close(byte[] data);
 
+    void clear();
 }

src/main/java/io/mycat/server/MySQLFrontConnection.java

@@ -14,6 +14,7 @@
 import io.mycat.server.sqlhandler.SavepointHandler;
 import io.mycat.server.sqlhandler.SelectHandler;
 import io.mycat.server.sqlhandler.ServerLoadDataInfileHandler;
+import io.mycat.server.sqlhandler.ServerPrepareHandler;
 import io.mycat.server.sqlhandler.SetHandler;
 import io.mycat.server.sqlhandler.ShowHandler;
 import io.mycat.server.sqlhandler.StartHandler;
@@ -79,7 +80,7 @@ public MySQLFrontConnection(SocketChannel channel) throws IOException {
 		this.port = localAddr.getPort();
 		this.localPort = remoteAddr.getPort();
 		loadDataInfileHandler = new ServerLoadDataInfileHandler(this);
-
+		prepareHandler = new ServerPrepareHandler(this);
 	}
 
 	public void sendAuthPackge() throws IOException {
@@ -235,7 +236,12 @@ public void query(byte[] data) {
 					"Unknown charset '" + charset + "'");
 			return;
 		}
+
+		query(sql);
 
+	}
+
+	public void query(String sql) {
 		if (sql == null || sql.length() == 0) {
 			writeErrMessage(ErrorCode.ER_NOT_ALLOWED_COMMAND, "Empty SQL");
 			return;
@@ -415,7 +421,7 @@ public void stmtExecute(byte[] data) {
 
 	public void stmtClose(byte[] data) {
 		if (prepareHandler != null) {
-			prepareHandler.close();
+			prepareHandler.close(data);
 		} else {
 			writeErrMessage(ErrorCode.ER_UNKNOWN_COM_ERROR,
 					"Prepare unsupported!");
@@ -508,11 +514,18 @@ public void close(String reason) {
 		if (getLoadDataInfileHandler() != null) {
 			getLoadDataInfileHandler().clear();
 		}
+		if(getPrepareHandler() != null) {
+			getPrepareHandler().clear();
+		}
 	}
 
 	public LoadDataInfileHandler getLoadDataInfileHandler() {
 		return loadDataInfileHandler;
 	}
+
+	public FrontendPrepareHandler getPrepareHandler() {
+		return prepareHandler;
+	}
 
 	public void ping() {
 		write(OkPacket.OK);

src/main/java/io/mycat/server/NonBlockingSession.java

@@ -61,6 +61,8 @@ public class NonBlockingSession{
 	private final MultiNodeCoordinator multiNodeCoordinator;
 	private final CommitNodeHandler commitHandler;
 	private volatile String xaTXID;
+
+	private boolean prepared;
 
 	public NonBlockingSession(MySQLFrontConnection source) {
 		this.source = source;
@@ -117,6 +119,9 @@ public void execute(RouteResultset rrs, int type) {
 
 		if (nodes.length == 1) {
 			singleNodeHandler = new SingleNodeHandler(rrs, this);
+			if(this.isPrepared()) {
+				singleNodeHandler.setPrepared(true);
+			}
 			try {
 				singleNodeHandler.execute();
 			} catch (Exception e) {
@@ -125,18 +130,25 @@ public void execute(RouteResultset rrs, int type) {
 			}
 		} else {
 			boolean autocommit = source.isAutocommit();
-			SystemConfig sysConfig = MycatServer.getInstance().getConfig()
-					.getSystem();
+//			SystemConfig sysConfig = MycatServer.getInstance().getConfig()
+//					.getSystem();
 			multiNodeHandler = new MultiNodeQueryHandler(type, rrs, autocommit,
 					this);
-
+			if(this.isPrepared()) {
+				multiNodeHandler.setPrepared(true);
+			}
 			try {
 				multiNodeHandler.execute();
 			} catch (Exception e) {
 				LOGGER.warn("{} {}", source, rrs, e);
 				source.writeErrMessage(ErrorCode.ERR_HANDLE_DATA, e.toString());
 			}
 		}
+
+		if(this.isPrepared()) {
+			this.setPrepared(false);
+		}
+
 	}
 
 	public void commit() {
@@ -374,4 +386,14 @@ public String getXaTXID() {
 		return xaTXID;
 	}
 
+
+	public boolean isPrepared() {
+		return prepared;
+	}
+
+
+	public void setPrepared(boolean prepared) {
+		this.prepared = prepared;
+	}
+
 }

src/main/java/io/mycat/server/executors/MultiNodeQueryHandler.java

@@ -34,6 +34,7 @@
 import io.mycat.route.RouteResultsetNode;
 import io.mycat.server.MySQLFrontConnection;
 import io.mycat.server.NonBlockingSession;
+import io.mycat.server.packet.BinaryRowDataPacket;
 import io.mycat.server.config.node.MycatConfig;
 import io.mycat.server.packet.FieldPacket;
 import io.mycat.server.packet.OkPacket;
@@ -73,6 +74,8 @@ public class MultiNodeQueryHandler extends MultiNodeHandler implements
 	private volatile boolean fieldsReturned;
 	private int okCount;
 	private final boolean isCallProcedure;
+	private boolean prepared;
+	private List<FieldPacket> fieldPackets = new ArrayList<FieldPacket>();
 
 	public MultiNodeQueryHandler(int sqlType, RouteResultset rrs,
 			boolean autocommit, NonBlockingSession session) {
@@ -329,8 +332,15 @@ public void outputMergeResult(final MySQLFrontConnection source,
 //			}
 			for (int i = start; i < end; i++) {
 				RowDataPacket row = results.get(i);
-				row.packetId = ++packetId;
-				row.write(bufferArray);
+				if(prepared) {
+					BinaryRowDataPacket binRowDataPk = new BinaryRowDataPacket();
+					binRowDataPk.read(fieldPackets, row);
+					binRowDataPk.packetId = ++packetId;
+					binRowDataPk.write(bufferArray);
+				} else {
+					row.packetId = ++packetId;
+					row.write(bufferArray);
+				}
 			}
 
 			eof[3] = ++packetId;
@@ -416,6 +426,7 @@ public void fieldEofResponse(byte[] header, List<byte[]> fields,
 				if (needMerg) {
 					FieldPacket fieldPkg = new FieldPacket();
 					fieldPkg.read(field);
+					fieldPackets.add(fieldPkg);
 					String fieldName = new String(fieldPkg.name).toUpperCase();
 					if (columToIndx != null
 							&& !columToIndx.containsKey(fieldName)) {
@@ -429,7 +440,6 @@ public void fieldEofResponse(byte[] header, List<byte[]> fields,
 							fieldPkg.packetId = ++packetId;
 							shouldSkip = true;
 							fieldPkg.write(bufferArray);
-
 						}
 
 						columToIndx.put(fieldName,
@@ -439,6 +449,7 @@ public void fieldEofResponse(byte[] header, List<byte[]> fields,
 					// find primary key index
 					FieldPacket fieldPkg = new FieldPacket();
 					fieldPkg.read(field);
+					fieldPackets.add(fieldPkg);
 					String fieldName = new String(fieldPkg.name);
 					if (primaryKey.equalsIgnoreCase(fieldName)) {
 						primaryKeyIndex = i;
@@ -525,4 +536,12 @@ public void requestDataResponse(byte[] data, BackendConnection conn) {
 				(MySQLBackendConnection) conn);
 	}
 
+	public boolean isPrepared() {
+		return prepared;
+	}
+
+	public void setPrepared(boolean prepared) {
+		this.prepared = prepared;
+	}
+
 }

src/main/java/io/mycat/server/executors/SingleNodeHandler.java

@@ -35,9 +35,11 @@
 import io.mycat.server.ErrorCode;
 import io.mycat.server.MySQLFrontConnection;
 import io.mycat.server.NonBlockingSession;
+import io.mycat.server.packet.BinaryRowDataPacket;
 import io.mycat.server.config.node.MycatConfig;
 import io.mycat.server.config.node.SchemaConfig;
 import io.mycat.server.packet.ErrorPacket;
+import io.mycat.server.packet.FieldPacket;
 import io.mycat.server.packet.OkPacket;
 import io.mycat.server.packet.RowDataPacket;
 import io.mycat.server.packet.util.LoadDataUtil;
@@ -48,6 +50,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
@@ -65,6 +68,9 @@ public class SingleNodeHandler implements ResponseHandler, Terminatable,
 	private volatile byte packetId;
 	private volatile boolean isRunning;
 	private Runnable terminateCallBack;
+	private boolean prepared;
+	private int fieldCount;
+	private List<FieldPacket> fieldPackets = new ArrayList<FieldPacket>();
     private volatile boolean isDefaultNodeShowTable;
     private Set<String> shardingTablesSet;
 
@@ -256,8 +262,13 @@ public void fieldEofResponse(byte[] header, List<byte[]> fields,
 		for (int i = 0, len = fields.size(); i < len; ++i) {
 			byte[] field = fields.get(i);
 			field[3] = ++packetId;
+			// 保存field信息
+			FieldPacket fieldPk = new FieldPacket();
+			fieldPk.read(field);
+			fieldPackets.add(fieldPk);
 			bufferArray.write(field);
 		}
+		fieldCount = fieldPackets.size();
 		eof[3] = ++packetId;
 		bufferArray.write(eof);
 
@@ -283,7 +294,16 @@ public void rowResponse(byte[] row, BackendConnection conn) {
             if(shardingTablesSet.contains(table.toUpperCase())) return;
         }
 		row[3] = ++packetId;
-		session.getSource().write(row);
+		if(prepared) {
+			RowDataPacket rowDataPk = new RowDataPacket(fieldCount);
+			rowDataPk.read(row);
+			BinaryRowDataPacket binRowDataPk = new BinaryRowDataPacket();
+			binRowDataPk.read(fieldPackets, rowDataPk);
+			binRowDataPk.packetId = rowDataPk.packetId;
+			binRowDataPk.write(session.getSource());
+		} else {
+			session.getSource().write(row);
+		}
 	}
 
 	@Override
@@ -313,4 +333,12 @@ public String toString() {
 				+ "]";
 	}
 
+	public boolean isPrepared() {
+		return prepared;
+	}
+
+	public void setPrepared(boolean prepared) {
+		this.prepared = prepared;
+	}
+
 }