mysqlio.java

SSD9 练习9

    final ResultSet sqlQueryDirect(Buffer QueryPacket, int max_rows) throws Exception
    {
	long updateCount = -1;
	long updateID = -1;
  
	// Send query command and sql query string
	clearAllReceive();
		
	Buffer Packet = sendCommand(MysqlDefs.QUERY, null, QueryPacket);
	Packet.pos--;

	long columnCount = Packet.readLength();

	if (Driver.trace) {
	    Debug.msg(this, "Column count: " + columnCount);
	}

	if (columnCount == 0) {
	    try {

		if (_server_major_version >= 3 &&
		    _server_minor_version >= 22 &&
		    _server_sub_minor_version >= 5) {
		    updateCount = Packet.newReadLength();
		    updateID = Packet.newReadLength();
		}
		else {
		    updateCount = (long)Packet.readLength();
		    updateID = (long)Packet.readLength();
		}
	    }
	    catch (Exception E) {
		throw new java.sql.SQLException(SQLError.get("S1000") + 
						": " + E.getClass().getName(),
						"S1000", -1);
	    }
		  
	    if (Driver.trace) {
		Debug.msg(this, "Update Count = " + updateCount);
	    }
		 
	    return new ResultSet(updateCount, updateID);
	}
	else {
	    Field[] Fields = new Field[(int)columnCount];
	    // Read in the column information
	    for (int i = 0; i < columnCount; i++) {
		Packet = readPacket();
		String TableName = Packet.readLenString();
		String ColName   = Packet.readLenString();
		int colLength    = Packet.readnBytes();
		int colType      = Packet.readnBytes();
		Packet.readByte(); // We know it's currently 2
		short colFlag    = (short)Buffer.ub(Packet.readByte());
		int colDecimals  = Buffer.ub(Packet.readByte());
		    
		Fields[i] = new Field(TableName, ColName, colLength, 
				      colType, colFlag, colDecimals);
	    }

	    Packet = readPacket();
    
	    Vector Rows = new Vector();

	    // Now read the data
	    byte[][] Row = nextRow((int)columnCount);
		 
	    Rows.addElement(Row);

	    int row_count = 1;

	    while (Row != null && row_count < max_rows) {
		Row = nextRow((int)columnCount);

		if (Row != null) {
		    Rows.addElement(Row);
		    row_count++;
		}
		else {
		    if (Driver.trace) {
			Debug.msg(this, "* NULL Row *");
		    }
		}
	    }

	    if (Driver.trace) {
		Debug.msg(this, "* Fetched " + Rows.size() + " rows from server *");
	    }

	    return new ResultSet(Fields, Rows);
	}
    }

    /**
     * Send a query specified in the String "Query" to the MySQL server.
     *
     * This method uses the specified character encoding to get the
     * bytes from the query string.
     */

    final ResultSet sqlQuery(String Query, int max_rows, String Encoding)
	throws Exception
    {
	// We don't know exactly how many bytes we're going to get 
	// from the query. Since we're dealing with Unicode, the 
        // max is 2, so pad it (2 * query) + space for headers 

	int pack_length = HEADER_LENGTH + 1 + (Query.length() * 2) + 2;
	Buffer Packet = new Buffer(pack_length);
	Packet.writeByte((byte)MysqlDefs.QUERY);

	if (Encoding != null) {
	    Packet.writeStringNoNull(Query, Encoding); 
        }
        else {
            Packet.writeStringNoNull(Query);
	}

	return sqlQueryDirect(Packet, max_rows);
    }
	
    final ResultSet sqlQuery(String Query, int max_rows) throws Exception
    {
	long updateCount = -1;
	long updateID = -1;
  
	// Send query command and sql query string
	clearAllReceive();
	Buffer Packet = sendCommand(MysqlDefs.QUERY, Query, null); //, (byte)0);
	Packet.pos--;

	long columnCount = Packet.readLength();
		
	if (Driver.trace) {
	    Debug.msg(this, "Column count: " + columnCount);
	}
    
	if (columnCount == 0) {
	    try {

		if (_server_major_version >= 3 &&
		    _server_minor_version >= 22 &&
		    _server_sub_minor_version >= 5) {
		    updateCount = Packet.newReadLength();
		    updateID = Packet.newReadLength();
		}
		else {
		    updateCount = (long)Packet.readLength();
		    updateID = (long)Packet.readLength();
		}
	    }
	    catch (Exception E) {
		throw new java.sql.SQLException(SQLError.get("S1000") + ": " + E.getClass().getName(),
						"S1000", -1);
	    }
		  
	    if (Driver.trace) {
		Debug.msg(this, "Update Count = " + updateCount);
	    }
		 
	    return new ResultSet(updateCount, updateID);
	}
	else {
	    Field[] Fields = new Field[(int)columnCount];
	    // Read in the column information
	    for (int i = 0; i < columnCount; i++) {
		Packet = readPacket();
		String TableName = Packet.readLenString();
		String ColName   = Packet.readLenString();
		int colLength    = Packet.readnBytes();
		int colType      = Packet.readnBytes();
		Packet.readByte(); // We know it's currently 2
		short colFlag    = (short)Buffer.ub(Packet.readByte());
		int colDecimals  = Buffer.ub(Packet.readByte());
		    
		Fields[i] = new Field(TableName, ColName, colLength, colType, colFlag, 
				      colDecimals);
	    }
	    Packet = readPacket();
    
	    Vector Rows = new Vector();
	    // Now read the data
	    byte[][] Row = nextRow((int)columnCount);

		 
	    Rows.addElement(Row);

	    int row_count = 1;

	    while (Row != null && row_count < max_rows) {
		Row = nextRow((int)columnCount);
		if (Row != null) {
		    Rows.addElement(Row);
		    row_count++;
		}
		else {
		    if (Driver.trace) {
			Debug.msg(this, "* NULL Row *");
		    }
		}
	    }
	    if (Driver.trace) {
		Debug.msg(this, "* Fetched " + Rows.size() + " rows from server *");
	    }
	    return new ResultSet(Fields, Rows);
	}
    }
    
    /**
     * Retrieve one row from the MySQL server.
     *
     * Note: this method is not thread-safe, but it is only called
     * from methods that are guarded by synchronizing on this object.
     */

    private final byte[][] nextRow(int columnCount) throws Exception
    {
	// Get the next incoming packet, re-using the packet because
	// all the data we need gets copied out of it.

	Buffer Packet = reuseAndReadPacket(_ReusablePacket);
    
	// check for errors.
    
	if (Packet.readByte() == (byte)0xff) {
	    String ErrorMessage;
	    int errno = 2000;
		  
	    if (_protocol_V > 9) {
		errno = Packet.readInt();
		ErrorMessage = Packet.readString();
		String XOpen = SQLError.mysqlToXOpen(errno);
		clearReceive();
		throw new java.sql.SQLException(SQLError.get(SQLError.get(XOpen)) + ": " 
						+ ErrorMessage, XOpen, errno);
	    }
	    else {
		ErrorMessage = Packet.readString();
		clearReceive();
		throw new java.sql.SQLException(ErrorMessage, SQLError.mysqlToXOpen(errno), errno);
	    }
	}
    
	// Away we go....
    
	Packet.pos--;
    
	int[] dataStart = new int[columnCount];
	byte[][] Row = new byte[columnCount][];

	if (!Packet.isLastDataPacket()) {
		 
	    for (int i = 0; i < columnCount; i++) {
		int p = Packet.pos;
		dataStart[i] = p;
		Packet.pos = (int)Packet.readLength() + Packet.pos;
	    }
	    for (int i = 0; i < columnCount; i++) {
		Packet.pos = dataStart[i];
		Row[i] = Packet.readLenByteArray();
	
		if (Driver.trace) {
		    if (Row[i] == null) {
			Debug.msg(this, "Field value: NULL");
		    }
		    else {
			Debug.msg(this, "Field value: " + Row[i].toString());
		    }
		}
	    }
	    return Row;
	}
	return null;
    }

    /**
     * Log-off of the MySQL server and close the socket.
     */

    final void quit() throws IOException
    {
	Buffer Packet = new Buffer(6);
	_packetSequence = -1;
	Packet.writeByte((byte)MysqlDefs.QUIT);
	send(Packet);
	_Mysql_Conn.close();
    }

    /**
     * Get the major version of the MySQL server we are
     * talking to.
     */

    final int getServerMajorVersion()
    {
	return _server_major_version;
    }

    /**
     * Get the minor version of the MySQL server we are
     * talking to.
     */

    final int getServerMinorVersion()
    {
	return _server_minor_version;
    }

    /**
     * Get the sub-minor version of the MySQL server we are
     * talking to.
     */

    final int getServerSubMinorVersion()
    {
	return _server_sub_minor_version;
    }

    /**
     * Read one packet from the MySQL server
     */

    private final Buffer readPacket() throws IOException
    {


	byte b0, b1, b2;

	b0 = _Mysql_Input.readByte();
	b1 = _Mysql_Input.readByte();
	b2 = _Mysql_Input.readByte();
    
	int packetLength = (int)(Buffer.ub(b0) + (256*Buffer.ub(b1)) + (256*256*Buffer.ub(b2)));

	byte packetSeq = _Mysql_Input.readByte();

	// Read data
	byte[] buffer = new byte[packetLength + 1];
	_Mysql_Input.readFully(buffer, 0, packetLength);

	buffer[packetLength] = 0;

	return new Buffer(buffer);
    }

    /**
     * Re-use a packet to read from the MySQL server
     */

    private final Buffer reuseAndReadPacket(Buffer Reuse) throws IOException
    {


	byte b0, b1, b2;

	b0 = _Mysql_Input.readByte();
	b1 = _Mysql_Input.readByte();
	b2 = _Mysql_Input.readByte();
    
	int packetLength = (int)(Buffer.ub(b0) + (256*Buffer.ub(b1)) + (256*256*Buffer.ub(b2)));
	

	byte packetSeq = _Mysql_Input.readByte();

	// Set the Buffer to it's original state

	Reuse.pos = 0;
	Reuse.send_length = 0;
	
	// Do we need to re-alloc the byte buffer?
	//
	// Note: We actually check the length of the buffer,
	// rather than buf_length, because buf_length is not
	// necesarily the actual length of the byte array
	// used as the buffer

	if (Reuse.buf.length < packetLength) {
	    Reuse.buf = new byte[packetLength];
	}

	// Set the new length

	Reuse.buf_length = packetLength;

	// Read the data from the server

	_Mysql_Input.readFully(Reuse.buf, 0, packetLength);

	Reuse.buf[packetLength] = 0; // Null-termination

       	return Reuse;
    }
	

    /**
     * Send a packet to the MySQL server
     */

    private final void send(Buffer Packet) throws IOException
    {
	int l = Packet.pos;
	_packetSequence++;
	Packet.pos = 0;
	Packet.writeLongInt(l - HEADER_LENGTH);
	Packet.writeByte(_packetSequence);
	_Mysql_Output.write(Packet.buf, 0, l);
	_Mysql_Output.flush();
	_Mysql_Buf_Output.flush();

	int total_header_length = HEADER_LENGTH;
    }

    /**
     * Clear waiting data in the InputStream
     */

    private final void clearReceive() throws IOException
    {

	int len = _Mysql_Buf_Input.available();

	if (len > 0) {
	    _Mysql_Input.skipBytes(len);
	}  
    }

    /**
     * Clear all data in the InputStream that is being
     * sent by the MySQL server.
     */

    private final void clearAllReceive() throws java.sql.SQLException
    {
	try {
	    int len = _Mysql_Buf_Input.available();

	    if (len > 0) {
		Buffer Packet = readPacket();
		if (Packet.buf[0] == (byte)0xff){
		    clearReceive();
		    return;
		}
		while (!Packet.isLastDataPacket()) {            
		    // larger than the socket buffer.
		    Packet = readPacket();
		    if (Packet.buf[0] == (byte)0xff)
			break;
		}
	    }
	    clearReceive();
	}
	catch (IOException E) {
	    throw new SQLException("Communication link failure: " + 
				   E.getClass().getName(), "08S01");
	}
    }
  
    static int getMaxBuf()
    {
        return MAXBUF;
    }
};