pgserverthread.java

非常棒的java数据库

        if (types.contains(ObjectUtils.getInteger(type))) {
            error("Unsupported type: " + type, null);
        }
    }

    private String getSQL(String s) {
        String lower = s.toLowerCase();
        if (lower.startsWith("show max_identifier_length")) {
            s = "CALL 63";
        } else if (lower.startsWith("set client_encoding to")) {
            s = "set DATESTYLE ISO";
        }
        // s = StringUtils.replaceAll(s, "i.indkey[ia.attnum-1]", "0");
        if (server.getLog()) {
            server.log(s + ";");
        }
        return s;
    }

    private void sendCommandComplete(String sql, int updateCount) throws IOException {
        startMessage('C');
        sql = sql.trim().toUpperCase();
        // TODO remove remarks at the beginning
        String tag;
        if (sql.startsWith("INSERT")) {
            tag = "INSERT 0 " + updateCount;
        } else if (sql.startsWith("DELETE")) {
            tag = "DELETE " + updateCount;
        } else if (sql.startsWith("UPDATE")) {
            tag = "UPDATE " + updateCount;
        } else if (sql.startsWith("SELECT") || sql.startsWith("CALL")) {
            tag = "SELECT";
        } else if (sql.startsWith("BEGIN")) {
            tag = "BEGIN";
        } else {
            error("check command tag: " + sql, null);
            tag = "UPDATE " + updateCount;
        }
        writeString(tag);
        sendMessage();
    }

    private void sendDataRow(int[] formatCodes, ResultSet rs) throws IOException {
        try {
            int columns = rs.getMetaData().getColumnCount();
            String[] values = new String[columns];
            for (int i = 0; i < columns; i++) {
                values[i] = rs.getString(i + 1);
            }
            startMessage('D');
            writeShort(columns);
            for (int i = 0; i < columns; i++) {
                String s = values[i];
                if (s == null) {
                    writeInt(-1);
                } else {
                    // TODO write Binary data
                    byte[] d2 = s.getBytes(getEncoding());
                    writeInt(d2.length);
                    write(d2);
                }
            }
            sendMessage();
        } catch (SQLException e) {
            sendErrorResponse(e);
        }
    }

    private String getEncoding() {
        if ("UNICODE".equals(clientEncoding)) {
            return "UTF-8";
        }
        return clientEncoding;
    }

    private void setParameter(PreparedStatement prep, int i, byte[] d2, int[] formatCodes) throws SQLException {
        boolean text = (i >= formatCodes.length) || (formatCodes[i] == 0);
        String s;
        try {
            if (text) {
                s = new String(d2, getEncoding());
            } else {
                server.logError(new SQLException("Binary format not supported"));
                s = new String(d2, getEncoding());
            }
        } catch (Exception e) {
            error("conversion error", e);
            s = null;
        }
        // if(server.getLog()) {
        // server.log(" " + i + ": " + s);
        // }
        prep.setString(i + 1, s);
    }

    private void sendErrorResponse(SQLException e) throws IOException {
        error("SQLException", e);
        startMessage('E');
        write('S');
        writeString("ERROR");
        write('C');
        writeString(e.getSQLState());
        write('M');
        writeString(e.getMessage());
        write('D');
        writeString(e.toString());
        write(0);
        sendMessage();
    }

    private void sendParameterDescription(Prepared p) throws IOException {
        try {
            PreparedStatement prep = p.prep;
            ParameterMetaData meta = prep.getParameterMetaData();
            int count = meta.getParameterCount();
            startMessage('t');
            writeShort(count);
            for (int i = 0; i < count; i++) {
                int type;
                if (p.paramType != null && p.paramType[i] != 0) {
                    type = p.paramType[i];
                } else {
                    type = TYPE_STRING;
                }
                checkType(type);
                writeInt(type);
            }
            sendMessage();
        } catch (SQLException e) {
            sendErrorResponse(e);
        }
    }

    private void sendNoData() throws IOException {
        startMessage('n');
        sendMessage();
    }

    private void sendRowDescription(ResultSetMetaData meta) throws IOException {
        try {
            if (meta == null) {
                sendNoData();
            } else {
                int columns = meta.getColumnCount();
                int[] types = new int[columns];
                int[] precision = new int[columns];
                String[] names = new String[columns];
                for (int i = 0; i < columns; i++) {
                    names[i] = meta.getColumnName(i + 1);
                    int type = meta.getColumnType(i + 1);
                    precision[i] = meta.getColumnDisplaySize(i + 1);
                    checkType(type);
                    types[i] = type;
                }
                startMessage('T');
                writeShort(columns);
                for (int i = 0; i < columns; i++) {
                    writeString(names[i].toLowerCase());
                    writeInt(0); // object ID
                    writeShort(0); // attribute number of the column
                    writeInt(types[i]); // data type
                    writeShort(getTypeSize(types[i], precision[i])); // pg_type.typlen
                    writeInt(getModifier(types[i])); // pg_attribute.atttypmod
                    writeShort(0); // text
                }
                sendMessage();
            }
        } catch (SQLException e) {
            sendErrorResponse(e);
        }
    }

    private int getTypeSize(int type, int precision) {
        switch (type) {
        case Types.VARCHAR:
            return Math.max(255, precision + 10);
        default:
            return precision + 4;
        }
    }

    private int getModifier(int type) {
        return -1;
    }

    private void sendErrorResponse(String message) throws IOException {
        error("Exception: " + message, null);
        startMessage('E');
        write('S');
        writeString("ERROR");
        write('C');
        writeString("08P01"); // PROTOCOL VIOLATION
        write('M');
        writeString(message);
        sendMessage();
    }

    private void sendParseComplete() throws IOException {
        startMessage('1');
        sendMessage();
    }

    private void sendBindComplete() throws IOException {
        startMessage('2');
        sendMessage();
    }

    private void initDb() throws SQLException {
        Statement stat = null;
        ResultSet rs = null;
        Reader r = null;
        try {
            rs = conn.getMetaData().getTables(null, "PG_CATALOG", "PG_VERSION", null);
            boolean tableFound = rs.next();
            stat = conn.createStatement();
            if (tableFound) {
                rs = stat.executeQuery("SELECT VERSION FROM PG_CATALOG.PG_VERSION");
                if (rs.next()) {
                    if (rs.getInt(1) == 1) {
                        // already installed
                        stat.execute("set search_path = PUBLIC, pg_catalog");
                        return;
                    }
                }
            }
            r = new InputStreamReader(PgServerThread.class.getResourceAsStream("pg_catalog.sql"));
            ScriptReader reader = new ScriptReader(new BufferedReader(r));
            while (true) {
                String sql = reader.readStatement();
                if (sql == null) {
                    break;
                }
                stat.execute(sql);
            }
            reader.close();

            rs = stat.executeQuery("SELECT OID FROM PG_CATALOG.PG_TYPE");
            while (rs.next()) {
                types.add(ObjectUtils.getInteger(rs.getInt(1)));
            }
        } finally {
            JdbcUtils.closeSilently(stat);
            JdbcUtils.closeSilently(rs);
            IOUtils.closeSilently(r);
        }
    }

    public void close() {
        try {
            stop = true;
            JdbcUtils.closeSilently(conn);
            if (socket != null) {
                socket.close();
            }
            server.log("Close");
        } catch (Exception e) {
            server.logError(e);
        }
        conn = null;
        socket = null;
        server.remove(this);
    }

    private void sendAuthenticationCleartextPassword() throws IOException {
        startMessage('R');
        writeInt(3);
        sendMessage();
    }

    private void sendAuthenticationOk() throws IOException {
        startMessage('R');
        writeInt(0);
        sendMessage();
        sendParameterStatus("client_encoding", clientEncoding);
        sendParameterStatus("DateStyle", dateStyle);
        sendParameterStatus("integer_datetimes", "off");
        sendParameterStatus("is_superuser", "off");
        sendParameterStatus("server_encoding", "SQL_ASCII");
        sendParameterStatus("server_version", "8.1.4");
        sendParameterStatus("session_authorization", userName);
        sendParameterStatus("standard_conforming_strings", "off");
        sendParameterStatus("TimeZone", "CET"); // TODO
        sendBackendKeyData();
        sendReadyForQuery();
    }

    private void sendReadyForQuery() throws IOException {
        startMessage('Z');
        char c;
        try {
            if (conn.getAutoCommit()) {
                c = 'I'; // idle
            } else {
                c = 'T'; // in a transaction block
            }
        } catch (SQLException e) {
            c = 'E'; // failed transaction block
        }
        write((byte) c);
        sendMessage();
    }

    private void sendBackendKeyData() throws IOException {
        startMessage('K');
        writeInt(processId);
        writeInt(processId);
        sendMessage();
    }

    private void writeString(String s) throws IOException {
        write(s.getBytes(getEncoding()));
        write(0);
    }

    private void writeInt(int i) throws IOException {
        dataOut.writeInt(i);
    }

    private void writeShort(int i) throws IOException {
        dataOut.writeShort(i);
    }

    private void write(byte[] data) throws IOException {
        dataOut.write(data);
    }

    private void write(int b) throws IOException {
        dataOut.write(b);
    }

    private void startMessage(int messageType) {
        this.messageType = messageType;
        outBuffer = new ByteArrayOutputStream();
        dataOut = new DataOutputStream(outBuffer);
    }

    private void sendMessage() throws IOException {
        dataOut.flush();
        byte[] buff = outBuffer.toByteArray();
        int len = buff.length;
        dataOut = new DataOutputStream(out);
        dataOut.write(messageType);
        dataOut.writeInt(len + 4);
        dataOut.write(buff);
        dataOut.flush();
    }

    private void sendParameterStatus(String param, String value) throws IOException {
        startMessage('S');
        writeString(param);
        writeString(value);
        sendMessage();
    }

    public void setThread(Thread thread) {
        this.thread = thread;
    }

    public Thread getThread() {
        return thread;
    }

    public void setProcessId(int id) {
        this.processId = id;
    }

    private static class Prepared {
        String name;
        String sql;
        PreparedStatement prep;
        int[] paramType;
    }

    private static class Portal {
        String name;
        String sql;
        int[] resultColumnFormat;
        PreparedStatement prep;
    }

}