- * The purpose of this message is to provide a resynchronization point for error recovery. - * When an error is detected while processing any extended-query message, the backend issues {@link ErrorResponse}, - * then reads and discards messages until this message is reached, then issues {@link ReadyForQuery} and returns to normal - * message processing. - *
- * Note that no skipping occurs if an error is detected while processing this message which ensures that there is one - * and only one {@link ReadyForQuery} sent for each of this message. - *
- * Note this message does not cause a transaction block opened with BEGIN to be closed. It is possible to detect this - * situation in {@link ReadyForQuery#txStatus()} that includes {@link TxStatus} information. - */ - void writeSync() { - ensureBuffer(); - out.writeByte(SYNC); - out.writeInt(4); - } - - /** - *
- * The message closes an existing prepared statement or portal and releases resources. - * Note that closing a prepared statement implicitly closes any open portals that were constructed from that statement. - *
- * The response is either {@link CloseComplete} or {@link ErrorResponse} - * - * @param portal - */ - void writeClosePortal(String portal) { - ensureBuffer(); + private static void renderQueryMessage(Query query, ByteBuf out) { int pos = out.writerIndex(); - out.writeByte(CLOSE); + out.writeByte(QUERY); out.writeInt(0); - out.writeByte('P'); // 'S' to close a prepared statement or 'P' to close a portal - Util.writeCStringUTF8(out, portal); + Util.writeCStringUTF8(out, query.sql); out.setInt(pos + 1, out.writerIndex() - pos - 1); } - void writeClosePreparedStatement(byte[] statementName) { - ensureBuffer(); + private static int estimateQueryMessage(Query query) { + return 1 + 4 + DataTypeEstimator.estimateCStringUTF8(query.sql); + } + + private static void renderPasswordMessage(PasswordMessage msg, ByteBuf out) { int pos = out.writerIndex(); - out.writeByte(CLOSE); + out.writeByte(PASSWORD_MESSAGE); out.writeInt(0); - out.writeByte('S'); // 'S' to close a prepared statement or 'P' to close a portal - out.writeBytes(statementName); - out.setInt(pos + 1, out.writerIndex() - pos - 1); + Util.writeCStringUTF8(out, msg.hash); + out.setInt(pos + 1, out.writerIndex() - pos- 1); } - void writeStartupMessage(StartupMessage msg) { - ensureBuffer(); + private static int estimatePasswordMessage(PasswordMessage msg) { + return 1 + 4 + DataTypeEstimator.estimateCStringUTF8(msg.hash); + } + private static void renderStartupMessage(StartupMessage msg, ByteBuf out) { int pos = out.writerIndex(); out.writeInt(0); @@ -238,17 +253,23 @@ void writeStartupMessage(StartupMessage msg) { out.setInt(pos, out.writerIndex() - pos); } - void writePasswordMessage(PasswordMessage msg) { - ensureBuffer(); - int pos = out.writerIndex(); - out.writeByte(PASSWORD_MESSAGE); - out.writeInt(0); - Util.writeCStringUTF8(out, msg.hash); - out.setInt(pos + 1, out.writerIndex() - pos- 1); + private static int estimateStartupMessage(StartupMessage msg) { + int length = 4 + + 2 + + 2 + + (StartupMessage.BUFF_USER_LENGTH + 1) + + DataTypeEstimator.estimateCStringUTF8(msg.username) + + (StartupMessage.BUFF_DATABASE_LENGTH + 1) + + DataTypeEstimator.estimateCStringUTF8(msg.database); + for (Map.Entry property : msg.properties.entrySet()) { + length += DataTypeEstimator.estimateCStringUTF8(property.getKey()); + length += DataTypeEstimator.estimateCStringUTF8(property.getValue()); + } + length++; + return length; } - void writeScramClientInitialMessage(ScramClientInitialMessage msg) { - ensureBuffer(); + private static void renderScramInitialMessage(ScramClientInitialMessage msg, ByteBuf out) { out.writeByte(PASSWORD_MESSAGE); int totalLengthPosition = out.writerIndex(); out.writeInt(0); // message length -> will be set later @@ -263,8 +284,11 @@ void writeScramClientInitialMessage(ScramClientInitialMessage msg) { out.setInt(totalLengthPosition, out.writerIndex() - totalLengthPosition); } - void writeScramClientFinalMessage(ScramClientFinalMessage msg) { - ensureBuffer(); + private static int estimateScramInitialMessage(ScramClientInitialMessage msg) { + return 1 + 4 + DataTypeEstimator.estimateCStringUTF8(msg.mechanism) + 4 + DataTypeEstimator.estimateUTF8(msg.message); + } + + private static void renderScramFinalMessage(ScramClientFinalMessage msg, ByteBuf out) { out.writeByte(PASSWORD_MESSAGE); int totalLengthPosition = out.writerIndex(); out.writeInt(0); // message length -> will be set later @@ -274,43 +298,58 @@ void writeScramClientFinalMessage(ScramClientFinalMessage msg) { out.setInt(totalLengthPosition, out.writerIndex() - totalLengthPosition); } - /** - * - * This message includes an SQL command (or commands) expressed as a text string. - * - * The possible response messages from the backend are - * {@link CommandComplete}, {@link RowDesc}, {@link DataRow}, {@link EmptyQueryResponse}, {@link ErrorResponse}, - * {@link ReadyForQuery} and {@link NoticeResponse} - */ - void writeQuery(Query query) { - ensureBuffer(); + private static int estimateScramFinalMessage(ScramClientFinalMessage msg) { + return 1 + 4 + DataTypeEstimator.estimateUTF8(msg.message); + } + + private static void renderClosePreparedStatement(byte[] statementName, ByteBuf out) { int pos = out.writerIndex(); - out.writeByte(QUERY); + out.writeByte(CLOSE); out.writeInt(0); - Util.writeCStringUTF8(out, query.sql); + out.writeByte('S'); // 'S' to close a prepared statement or 'P' to close a portal + out.writeBytes(statementName); out.setInt(pos + 1, out.writerIndex() - pos - 1); } - /** - * - * The message that using "statement" variant specifies the name of an existing prepared statement. - * - * The response is a {@link ParamDesc} message describing the parameters needed by the statement, - * followed by a {@link RowDesc} message describing the rows that will be returned when the statement is eventually - * executed or a {@link NoData} message if the statement will not return rows. - * {@link ErrorResponse} is issued if there is no such prepared statement. - * - * Note that since {@link Bind} has not yet been issued, the formats to be used for returned columns are not yet known to - * the backend; the format code fields in the {@link RowDesc} message will be zeroes in this case. - * - * The message that using "portal" variant specifies the name of an existing portal. - * - * The response is a {@link RowDesc} message describing the rows that will be returned by executing the portal; - * or a {@link NoData} message if the portal does not contain a query that will return rows; or {@link ErrorResponse} - * if there is no such portal. - */ - void writeDescribe(Describe describe) { - ensureBuffer(); + private static int estimateClosePreparedStatement(byte[] statementName) { + return 1 + 4 + 1 + DataTypeEstimator.estimateByteArray(statementName); + } + + private static void renderTerminate(ByteBuf out) { + out.writeByte(TERMINATE); + out.writeInt(4); + } + + private static int estimateTerminate() { + return 1 + 4; + } + + private static void renderSync(ByteBuf out) { + out.writeByte(SYNC); + out.writeInt(4); + } + + private static int estimateSync() { + return 1 + 4; + } + + private static void renderExecute(String portal, int rowCount, ByteBuf out) { + int pos = out.writerIndex(); + out.writeByte(EXECUTE); + out.writeInt(0); + if (portal != null) { + out.writeCharSequence(portal, UTF_8); + } + out.writeByte(0); + out.writeInt(rowCount); // Zero denotes "no limit" maybe for ReadStream + out.setInt(pos + 1, out.writerIndex() - pos - 1); + } + + private static int estimateExecute(String portal, int rowCount) { + return 1 + 4 + (portal != null ? DataTypeEstimator.estimateUTF8(portal) : 0) + 1 + 4; + } + + private static void renderDescribe(Describe describe, ByteBuf out) { int pos = out.writerIndex(); out.writeByte(DESCRIBE); out.writeInt(0); @@ -327,8 +366,19 @@ void writeDescribe(Describe describe) { out.setInt(pos + 1, out.writerIndex() - pos- 1); } - void writeParse(String sql, byte[] statement, DataType[] parameterTypes) { - ensureBuffer(); + private static int estimateDescribe(Describe describe) { + int length = 1 + 4; + if (describe.statement.length > 1) { + length += 1 + DataTypeEstimator.estimateByteArray(describe.statement); + } else if (describe.portal != null) { + length += 1 + DataTypeEstimator.estimateCStringUTF8(describe.portal); + } else { + length += 1 + DataTypeEstimator.estimateCStringUTF8(""); + } + return length; + } + + private static void renderParse(String sql, byte[] statement, DataType[] parameterTypes, ByteBuf out) { int pos = out.writerIndex(); out.writeByte(PARSE); out.writeInt(0); @@ -346,47 +396,15 @@ void writeParse(String sql, byte[] statement, DataType[] parameterTypes) { out.setInt(pos + 1, out.writerIndex() - pos - 1); } - /** - * The message specifies the portal and a maximum row count (zero meaning "fetch all rows") of the result. - * - * The row count of the result is only meaningful for portals containing commands that return row sets; - * in other cases the command is always executed to completion, and the row count of the result is ignored. - * - * The possible responses to this message are the same as {@link Query} message, except that - * it doesn't cause {@link ReadyForQuery} or {@link RowDesc} to be issued. - * - * If Execute terminates before completing the execution of a portal, it will send a {@link PortalSuspended} message; - * the appearance of this message tells the frontend that another Execute should be issued against the same portal to - * complete the operation. The {@link CommandComplete} message indicating completion of the source SQL command - * is not sent until the portal's execution is completed. Therefore, This message is always terminated by - * the appearance of exactly one of these messages: {@link CommandComplete}, - * {@link EmptyQueryResponse}, {@link ErrorResponse} or {@link PortalSuspended}. - * - * @author Emad Alblueshi - */ - void writeExecute(String portal, int rowCount) { - ensureBuffer(); - int pos = out.writerIndex(); - out.writeByte(EXECUTE); - out.writeInt(0); - if (portal != null) { - out.writeCharSequence(portal, UTF_8); - } - out.writeByte(0); - out.writeInt(rowCount); // Zero denotes "no limit" maybe for ReadStream - out.setInt(pos + 1, out.writerIndex() - pos - 1); + private static int estimateParse(String sql, byte[] statement, DataType[] parameterTypes) { + return 1 + + 4 + + DataTypeEstimator.estimateByteArray(statement) + + DataTypeEstimator.estimateCStringUTF8(sql) + + 2 + (parameterTypes == null ? 0 : parameterTypes.length * 4); } - /** - * - * The message gives the name of the prepared statement, the name of portal, - * and the values to use for any parameter values present in the prepared statement. - * The supplied parameter set must match those needed by the prepared statement. - * - * The response is either {@link BindComplete} or {@link ErrorResponse}. - */ - void writeBind(Bind bind, String portal, Tuple paramValues) { - ensureBuffer(); + private static void renderBind(Bind bind, String portal, Tuple paramValues, ByteBuf out) { int pos = out.writerIndex(); out.writeByte(BIND); out.writeInt(0); @@ -436,12 +454,218 @@ void writeBind(Bind bind, String portal, Tuple paramValues) { out.setInt(pos + 1, out.writerIndex() - pos - 1); } - private void ensureBuffer() { - if (out == null) { - out = ctx.alloc().ioBuffer(); + private static int estimateBind(Bind bind, String portal, Tuple paramValues) { + + int paramLen = paramValues.size(); + + int length = 1 + + 4 + + (portal != null ? DataTypeEstimator.estimateUTF8(portal) : 0) + + 1 + + DataTypeEstimator.estimateByteArray(bind.statement) + + 2 + + paramLen * 2 + + 2; + + for (int c = 0;c < paramLen;c++) { + Object param = paramValues.getValue(c); + if (param == null) { + length += 4; + } else { + DataType dataType = bind.paramTypes[c]; + length += 4; + if (dataType.supportsBinary) { + int estimator = dataType.lengthEstimator; + if (dataType.array) { + length += 4 + 4 + 4 + 4 + 4; + Object[] array = (Object[]) param; + for (Object elt : array) { + length += 4; + if (elt != null) { + length += DataTypeEstimator.estimate(estimator, elt); + } + } + } else { + length += DataTypeEstimator.estimate(estimator, param); + } + } else { + length += DataTypeEstimator.estimate(dataType.lengthEstimator, param); + } + } + } + + // Result columns are all in Binary format + if (bind.resultColumns.length > 0) { + length += 2 + bind.resultColumns.length * 2; + } else { + length += 2 + 2; + } + return length; + } + + private static void renderClosePortal(String portal, ByteBuf out) { + int pos = out.writerIndex(); + out.writeByte(CLOSE); + out.writeInt(0); + out.writeByte('P'); // 'S' to close a prepared statement or 'P' to close a portal + Util.writeCStringUTF8(out, portal); + out.setInt(pos + 1, out.writerIndex() - pos - 1); + } + + private static int estimateClosePortal(String portal) { + return 1 + 4 + 1 + DataTypeEstimator.estimateCStringUTF8(portal); + } + + void close() { + ByteBuf buff = renderPendingMessages(); + ctx.writeAndFlush(buff).addListener(v -> { + Channel ch = channelHandlerContext().channel(); + if (ch instanceof SocketChannel) { + SocketChannel channel = (SocketChannel) ch; + channel.shutdownOutput(); + } + }); + } + + void flush() { + ByteBuf buff = renderPendingMessages(); + if (buff == Unpooled.EMPTY_BUFFER) { + ctx.flush(); + } else { + ctx.writeAndFlush(buff, ctx.voidPromise()); } } + /** + * This message immediately closes the connection. On receipt of this message, + * the backend closes the connection and terminates. + */ + void writeTerminate() { + enqueueMessage(TerminateMessage.INSTANCE, estimateTerminate()); + } + + /** + * + * The purpose of this message is to provide a resynchronization point for error recovery. + * When an error is detected while processing any extended-query message, the backend issues {@link ErrorResponse}, + * then reads and discards messages until this message is reached, then issues {@link ReadyForQuery} and returns to normal + * message processing. + * + * Note that no skipping occurs if an error is detected while processing this message which ensures that there is one + * and only one {@link ReadyForQuery} sent for each of this message. + * + * Note this message does not cause a transaction block opened with BEGIN to be closed. It is possible to detect this + * situation in {@link ReadyForQuery#txStatus()} that includes {@link TxStatus} information. + */ + void writeSync() { + enqueueMessage(SyncMessage.INSTANCE, estimateSync()); + } + + /** + * + * The message closes an existing prepared statement or portal and releases resources. + * Note that closing a prepared statement implicitly closes any open portals that were constructed from that statement. + * + * The response is either {@link CloseComplete} or {@link ErrorResponse} + * + * @param portal + */ + void writeClosePortal(String portal) { + enqueueMessage(ClosePortalMessage.INSTANCE, portal, estimateClosePortal(portal)); + } + + void writeClosePreparedStatement(byte[] statementName) { + enqueueMessage(ClosePreparedStatementMessage.INSTANCE, statementName, estimateClosePreparedStatement(statementName)); + } + + void writeStartupMessage(StartupMessage msg) { + enqueueMessage(msg, estimateStartupMessage(msg)); + } + + void writePasswordMessage(PasswordMessage msg) { + enqueueMessage(msg, estimatePasswordMessage(msg)); + } + + void writeScramClientInitialMessage(ScramClientInitialMessage msg) { + enqueueMessage(msg, estimateScramInitialMessage(msg)); + } + + void writeScramClientFinalMessage(ScramClientFinalMessage msg) { + enqueueMessage(msg, estimateScramFinalMessage(msg)); + } + + /** + * + * This message includes an SQL command (or commands) expressed as a text string. + * + * The possible response messages from the backend are + * {@link CommandComplete}, {@link RowDesc}, {@link DataRow}, {@link EmptyQueryResponse}, {@link ErrorResponse}, + * {@link ReadyForQuery} and {@link NoticeResponse} + */ + void writeQuery(Query query) { + enqueueMessage(query, estimateQueryMessage(query)); + } + + /** + * + * The message that using "statement" variant specifies the name of an existing prepared statement. + * + * The response is a {@link ParamDesc} message describing the parameters needed by the statement, + * followed by a {@link RowDesc} message describing the rows that will be returned when the statement is eventually + * executed or a {@link NoData} message if the statement will not return rows. + * {@link ErrorResponse} is issued if there is no such prepared statement. + * + * Note that since {@link Bind} has not yet been issued, the formats to be used for returned columns are not yet known to + * the backend; the format code fields in the {@link RowDesc} message will be zeroes in this case. + * + * The message that using "portal" variant specifies the name of an existing portal. + * + * The response is a {@link RowDesc} message describing the rows that will be returned by executing the portal; + * or a {@link NoData} message if the portal does not contain a query that will return rows; or {@link ErrorResponse} + * if there is no such portal. + */ + void writeDescribe(Describe describe) { + enqueueMessage(describe, estimateDescribe(describe)); + } + + void writeParse(String sql, byte[] statement, DataType[] parameterTypes) { + enqueueMessage(ParseMessage.INSTANCE, sql, statement, parameterTypes, estimateParse(sql, statement, parameterTypes)); + } + + /** + * The message specifies the portal and a maximum row count (zero meaning "fetch all rows") of the result. + * + * The row count of the result is only meaningful for portals containing commands that return row sets; + * in other cases the command is always executed to completion, and the row count of the result is ignored. + * + * The possible responses to this message are the same as {@link Query} message, except that + * it doesn't cause {@link ReadyForQuery} or {@link RowDesc} to be issued. + * + * If Execute terminates before completing the execution of a portal, it will send a {@link PortalSuspended} message; + * the appearance of this message tells the frontend that another Execute should be issued against the same portal to + * complete the operation. The {@link CommandComplete} message indicating completion of the source SQL command + * is not sent until the portal's execution is completed. Therefore, This message is always terminated by + * the appearance of exactly one of these messages: {@link CommandComplete}, + * {@link EmptyQueryResponse}, {@link ErrorResponse} or {@link PortalSuspended}. + * + * @author Emad Alblueshi + */ + void writeExecute(String portal, int rowCount) { + enqueueMessage(ExecuteMessage.INSTANCE, portal, rowCount, estimateExecute(portal, rowCount)); + } + + /** + * + * The message gives the name of the prepared statement, the name of portal, + * and the values to use for any parameter values present in the prepared statement. + * The supplied parameter set must match those needed by the prepared statement. + * + * The response is either {@link BindComplete} or {@link ErrorResponse}. + */ + void writeBind(Bind bind, String portal, Tuple paramValues) { + enqueueMessage(bind, portal, paramValues, estimateBind(bind, portal, paramValues)); + } + byte[] nextStatementName() { return psSeq.next(); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java index 5c2d133db1..9f4fd9e98c 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java @@ -16,11 +16,14 @@ */ package io.vertx.pgclient.impl.codec; +import io.vertx.core.VertxException; import io.vertx.sqlclient.impl.ErrorMessageFactory; +import io.vertx.sqlclient.internal.ArrayTuple; import io.vertx.sqlclient.internal.ParamDesc; import io.vertx.sqlclient.internal.TupleInternal; import java.util.Arrays; +import java.util.function.Function; /** * @author Emad Alblueshi @@ -38,15 +41,16 @@ DataType[] paramDataTypes() { return paramDataTypes; } - public String prepare(TupleInternal values) { + public TupleInternal prepare(TupleInternal values) { int numberOfParams = values.size(); if (numberOfParams > 65535) { - return "The number of parameters (" + numberOfParams + ") exceeds the maximum of 65535. Use arrays or split the query."; + throw new VertxException("The number of parameters (" + numberOfParams + ") exceeds the maximum of 65535. Use arrays or split the query.", true); } int paramDescLength = paramDataTypes.length; if (numberOfParams != paramDescLength) { - return ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParams); + throw new VertxException(ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParams), true); } + TupleInternal prepared = values; for (int i = 0; i < paramDescLength; i++) { DataType paramDataType = paramDataTypes[i]; ParamExtractor extractor = paramDataType.paramExtractor; @@ -54,11 +58,29 @@ public String prepare(TupleInternal values) { try { val = extractor.get(values, i); } catch (Exception e) { - return ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(paramDataType.decodingType, i, values.getValue(i)); + throw new VertxException(ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(paramDataType.decodingType, i, values.getValue(i)), true); } - values.setValue(i, val); + if (val != null) { + Function preparator = paramDataType.preEncoder; + if (preparator != null) { + if (prepared == values) { + prepared = new ArrayTuple(prepared); + } + if (paramDataType.array) { + Object[] array = (Object[]) val; + Object[] tmp = new Object[array.length]; + for (int j = 0;j < array.length;j++) { + tmp[j] = preparator.apply(array[j]); + } + val = tmp; + } else { + val = preparator.apply(val); + } + } + } + prepared.setValue(i, val); } - return null; + return prepared; } @Override diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java index be9c5036eb..1f078cb558 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java @@ -53,7 +53,7 @@ public String sql() { } @Override - public String prepare(TupleInternal values) { + public TupleInternal prepare(TupleInternal values) { return paramDesc.prepare(values); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java index f6fe07a152..72710b9ba9 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java @@ -20,7 +20,7 @@ /** * @author Emad Alblueshi */ -class Query { +class Query extends OutboundMessage { final String sql; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java index 5861f1bd4f..adbdb5fe3b 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java @@ -19,7 +19,7 @@ /** */ -class ScramClientFinalMessage { +class ScramClientFinalMessage extends OutboundMessage { final String message; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java index 270991de88..d35930e10f 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java @@ -19,7 +19,7 @@ /** */ -public class ScramClientInitialMessage { +public class ScramClientInitialMessage extends OutboundMessage { final String mechanism; final String message; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java index a1d6ceff7e..31eb47144e 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java @@ -27,11 +27,14 @@ /** * @author Emad Alblueshi */ -class StartupMessage { +class StartupMessage extends OutboundMessage { static final ByteBuf BUFF_USER = Unpooled.copiedBuffer("user", UTF_8).asReadOnly(); static final ByteBuf BUFF_DATABASE = Unpooled.copiedBuffer("database", UTF_8).asReadOnly(); + static final int BUFF_USER_LENGTH = 4; + static final int BUFF_DATABASE_LENGTH = 8; + final String username; final String database; final Map properties; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java new file mode 100644 index 0000000000..a870d5e88d --- /dev/null +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java @@ -0,0 +1,7 @@ +package io.vertx.pgclient.impl.codec; + +class SyncMessage extends OutboundMessage { + + static final SyncMessage INSTANCE = new SyncMessage(); + +} diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java new file mode 100644 index 0000000000..abf0335e3b --- /dev/null +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java @@ -0,0 +1,7 @@ +package io.vertx.pgclient.impl.codec; + +class TerminateMessage extends OutboundMessage { + + static final TerminateMessage INSTANCE = new TerminateMessage(); + +} diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java index 188d0c0af6..87551f1b06 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java @@ -18,6 +18,7 @@ package io.vertx.pgclient.impl.util; import io.netty.buffer.ByteBuf; +import io.netty.util.CharsetUtil; import io.vertx.core.buffer.Buffer; import io.vertx.sqlclient.Tuple; diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java index cc9d6e456e..b60c1f2970 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java @@ -29,6 +29,7 @@ import io.vertx.sqlclient.internal.PreparedStatement; import io.vertx.sqlclient.internal.TupleInternal; +import java.util.ArrayList; import java.util.List; import java.util.function.Function; import java.util.stream.Collector; @@ -65,23 +66,24 @@ QueryResultBuilder executeExtendedQuery(CommandScheduler scheduler, PreparedStatement preparedStatement, PrepareOptions options, boolean autoCommit, - Tuple arguments, + Tuple values, int fetch, String cursorId, boolean suspended, PromiseInternal promise) { ContextInternal context = promise.context(); QueryResultBuilder handler = createHandler(promise); - String msg = preparedStatement.prepare((TupleInternal) arguments); - if (msg != null) { - handler.fail(msg); + try { + values = preparedStatement.prepare((TupleInternal) values); + } catch (Exception e) { + handler.fail(e); return null; } ExtendedQueryCommand cmd = ExtendedQueryCommand.createQuery( preparedStatement.sql(), options, preparedStatement, - arguments, + values, fetch, cursorId, suspended, @@ -122,12 +124,11 @@ void executeBatchQuery(CommandScheduler scheduler, PromiseInternal promise) { ContextInternal context = promise.context(); QueryResultBuilder handler = createHandler(promise); - for (Tuple args : batch) { - String msg = preparedStatement.prepare((TupleInternal)args); - if (msg != null) { - handler.fail(msg); - return; - } + try { + batch = preparedStatement.prepare((List) batch); + } catch (Exception e) { + handler.fail(e); + return; } ExtendedQueryCommand cmd = ExtendedQueryCommand.createBatch(preparedStatement.sql(), options, preparedStatement, batch, autoCommit, collector, handler); scheduler.schedule(cmd, handler); diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java index c223728359..55591917f6 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java @@ -17,6 +17,11 @@ package io.vertx.sqlclient.internal; +import io.vertx.sqlclient.Tuple; + +import java.util.ArrayList; +import java.util.List; + public interface PreparedStatement { ParamDesc paramDesc(); @@ -25,6 +30,23 @@ public interface PreparedStatement { String sql(); - String prepare(TupleInternal values); + default List prepare(List batch) throws Exception { + List actual = batch; + int size = actual.size(); + for (int idx = 0;idx < size;idx++) { + TupleInternal values = batch.get(idx); + TupleInternal prepared = prepare(values); + if (prepared != values) { + if (batch == actual) { + actual = new ArrayList<>(actual); + } + actual.set(idx, prepared); + } + } + return actual; + } + default TupleInternal prepare(TupleInternal values) throws Exception { + return values; + } } diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java index dd44fb5b02..ba3956f95a 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java @@ -72,10 +72,11 @@ public static ExtendedQueryCommand createBatch( protected final PrepareOptions options; public PreparedStatement ps; protected final boolean batch; - protected final Object tuples; + private Object tuples; protected final int fetch; protected final String cursorId; protected final boolean suspended; + private boolean prepared; private ExtendedQueryCommand(String sql, PrepareOptions options, @@ -97,6 +98,7 @@ private ExtendedQueryCommand(String sql, this.fetch = fetch; this.cursorId = cursorId; this.suspended = suspended; + this.prepared = ps != null; } public PrepareOptions options() { @@ -109,17 +111,17 @@ public PrepareOptions options() { * @return {@code null} if the tuple preparation was successfull otherwise the validation error */ public String prepare() { - if (ps != null) { - if (batch) { - for (TupleInternal tuple : (List) tuples) { - String msg = ps.prepare(tuple); - if (msg != null) { - return msg; - } + if (ps != null && !prepared) { + prepared = true; // TODO : fix this + try { + if (batch) { + tuples = ps.prepare((List) tuples); + return null; + } else { + tuples = ps.prepare((TupleInternal) tuples); } - return null; - } else { - return ps.prepare((TupleInternal) tuples); + } catch (Exception e) { + return e.getMessage(); } } return null;
- * This message includes an SQL command (or commands) expressed as a text string. - *
- * The possible response messages from the backend are - * {@link CommandComplete}, {@link RowDesc}, {@link DataRow}, {@link EmptyQueryResponse}, {@link ErrorResponse}, - * {@link ReadyForQuery} and {@link NoticeResponse} - */ - void writeQuery(Query query) { - ensureBuffer(); + private static int estimateScramFinalMessage(ScramClientFinalMessage msg) { + return 1 + 4 + DataTypeEstimator.estimateUTF8(msg.message); + } + + private static void renderClosePreparedStatement(byte[] statementName, ByteBuf out) { int pos = out.writerIndex(); - out.writeByte(QUERY); + out.writeByte(CLOSE); out.writeInt(0); - Util.writeCStringUTF8(out, query.sql); + out.writeByte('S'); // 'S' to close a prepared statement or 'P' to close a portal + out.writeBytes(statementName); out.setInt(pos + 1, out.writerIndex() - pos - 1); } - /** - *
- * The message that using "statement" variant specifies the name of an existing prepared statement. - *
- * The response is a {@link ParamDesc} message describing the parameters needed by the statement, - * followed by a {@link RowDesc} message describing the rows that will be returned when the statement is eventually - * executed or a {@link NoData} message if the statement will not return rows. - * {@link ErrorResponse} is issued if there is no such prepared statement. - *
- * Note that since {@link Bind} has not yet been issued, the formats to be used for returned columns are not yet known to - * the backend; the format code fields in the {@link RowDesc} message will be zeroes in this case. - *
- * The message that using "portal" variant specifies the name of an existing portal. - *
- * The response is a {@link RowDesc} message describing the rows that will be returned by executing the portal; - * or a {@link NoData} message if the portal does not contain a query that will return rows; or {@link ErrorResponse} - * if there is no such portal. - */ - void writeDescribe(Describe describe) { - ensureBuffer(); + private static int estimateClosePreparedStatement(byte[] statementName) { + return 1 + 4 + 1 + DataTypeEstimator.estimateByteArray(statementName); + } + + private static void renderTerminate(ByteBuf out) { + out.writeByte(TERMINATE); + out.writeInt(4); + } + + private static int estimateTerminate() { + return 1 + 4; + } + + private static void renderSync(ByteBuf out) { + out.writeByte(SYNC); + out.writeInt(4); + } + + private static int estimateSync() { + return 1 + 4; + } + + private static void renderExecute(String portal, int rowCount, ByteBuf out) { + int pos = out.writerIndex(); + out.writeByte(EXECUTE); + out.writeInt(0); + if (portal != null) { + out.writeCharSequence(portal, UTF_8); + } + out.writeByte(0); + out.writeInt(rowCount); // Zero denotes "no limit" maybe for ReadStream + out.setInt(pos + 1, out.writerIndex() - pos - 1); + } + + private static int estimateExecute(String portal, int rowCount) { + return 1 + 4 + (portal != null ? DataTypeEstimator.estimateUTF8(portal) : 0) + 1 + 4; + } + + private static void renderDescribe(Describe describe, ByteBuf out) { int pos = out.writerIndex(); out.writeByte(DESCRIBE); out.writeInt(0); @@ -327,8 +366,19 @@ void writeDescribe(Describe describe) { out.setInt(pos + 1, out.writerIndex() - pos- 1); } - void writeParse(String sql, byte[] statement, DataType[] parameterTypes) { - ensureBuffer(); + private static int estimateDescribe(Describe describe) { + int length = 1 + 4; + if (describe.statement.length > 1) { + length += 1 + DataTypeEstimator.estimateByteArray(describe.statement); + } else if (describe.portal != null) { + length += 1 + DataTypeEstimator.estimateCStringUTF8(describe.portal); + } else { + length += 1 + DataTypeEstimator.estimateCStringUTF8(""); + } + return length; + } + + private static void renderParse(String sql, byte[] statement, DataType[] parameterTypes, ByteBuf out) { int pos = out.writerIndex(); out.writeByte(PARSE); out.writeInt(0); @@ -346,47 +396,15 @@ void writeParse(String sql, byte[] statement, DataType[] parameterTypes) { out.setInt(pos + 1, out.writerIndex() - pos - 1); } - /** - * The message specifies the portal and a maximum row count (zero meaning "fetch all rows") of the result. - * - * The row count of the result is only meaningful for portals containing commands that return row sets; - * in other cases the command is always executed to completion, and the row count of the result is ignored. - * - * The possible responses to this message are the same as {@link Query} message, except that - * it doesn't cause {@link ReadyForQuery} or {@link RowDesc} to be issued. - * - * If Execute terminates before completing the execution of a portal, it will send a {@link PortalSuspended} message; - * the appearance of this message tells the frontend that another Execute should be issued against the same portal to - * complete the operation. The {@link CommandComplete} message indicating completion of the source SQL command - * is not sent until the portal's execution is completed. Therefore, This message is always terminated by - * the appearance of exactly one of these messages: {@link CommandComplete}, - * {@link EmptyQueryResponse}, {@link ErrorResponse} or {@link PortalSuspended}. - * - * @author Emad Alblueshi - */ - void writeExecute(String portal, int rowCount) { - ensureBuffer(); - int pos = out.writerIndex(); - out.writeByte(EXECUTE); - out.writeInt(0); - if (portal != null) { - out.writeCharSequence(portal, UTF_8); - } - out.writeByte(0); - out.writeInt(rowCount); // Zero denotes "no limit" maybe for ReadStream - out.setInt(pos + 1, out.writerIndex() - pos - 1); + private static int estimateParse(String sql, byte[] statement, DataType[] parameterTypes) { + return 1 + + 4 + + DataTypeEstimator.estimateByteArray(statement) + + DataTypeEstimator.estimateCStringUTF8(sql) + + 2 + (parameterTypes == null ? 0 : parameterTypes.length * 4); } - /** - * - * The message gives the name of the prepared statement, the name of portal, - * and the values to use for any parameter values present in the prepared statement. - * The supplied parameter set must match those needed by the prepared statement. - * - * The response is either {@link BindComplete} or {@link ErrorResponse}. - */ - void writeBind(Bind bind, String portal, Tuple paramValues) { - ensureBuffer(); + private static void renderBind(Bind bind, String portal, Tuple paramValues, ByteBuf out) { int pos = out.writerIndex(); out.writeByte(BIND); out.writeInt(0); @@ -436,12 +454,218 @@ void writeBind(Bind bind, String portal, Tuple paramValues) { out.setInt(pos + 1, out.writerIndex() - pos - 1); } - private void ensureBuffer() { - if (out == null) { - out = ctx.alloc().ioBuffer(); + private static int estimateBind(Bind bind, String portal, Tuple paramValues) { + + int paramLen = paramValues.size(); + + int length = 1 + + 4 + + (portal != null ? DataTypeEstimator.estimateUTF8(portal) : 0) + + 1 + + DataTypeEstimator.estimateByteArray(bind.statement) + + 2 + + paramLen * 2 + + 2; + + for (int c = 0;c < paramLen;c++) { + Object param = paramValues.getValue(c); + if (param == null) { + length += 4; + } else { + DataType dataType = bind.paramTypes[c]; + length += 4; + if (dataType.supportsBinary) { + int estimator = dataType.lengthEstimator; + if (dataType.array) { + length += 4 + 4 + 4 + 4 + 4; + Object[] array = (Object[]) param; + for (Object elt : array) { + length += 4; + if (elt != null) { + length += DataTypeEstimator.estimate(estimator, elt); + } + } + } else { + length += DataTypeEstimator.estimate(estimator, param); + } + } else { + length += DataTypeEstimator.estimate(dataType.lengthEstimator, param); + } + } + } + + // Result columns are all in Binary format + if (bind.resultColumns.length > 0) { + length += 2 + bind.resultColumns.length * 2; + } else { + length += 2 + 2; + } + return length; + } + + private static void renderClosePortal(String portal, ByteBuf out) { + int pos = out.writerIndex(); + out.writeByte(CLOSE); + out.writeInt(0); + out.writeByte('P'); // 'S' to close a prepared statement or 'P' to close a portal + Util.writeCStringUTF8(out, portal); + out.setInt(pos + 1, out.writerIndex() - pos - 1); + } + + private static int estimateClosePortal(String portal) { + return 1 + 4 + 1 + DataTypeEstimator.estimateCStringUTF8(portal); + } + + void close() { + ByteBuf buff = renderPendingMessages(); + ctx.writeAndFlush(buff).addListener(v -> { + Channel ch = channelHandlerContext().channel(); + if (ch instanceof SocketChannel) { + SocketChannel channel = (SocketChannel) ch; + channel.shutdownOutput(); + } + }); + } + + void flush() { + ByteBuf buff = renderPendingMessages(); + if (buff == Unpooled.EMPTY_BUFFER) { + ctx.flush(); + } else { + ctx.writeAndFlush(buff, ctx.voidPromise()); } } + /** + * This message immediately closes the connection. On receipt of this message, + * the backend closes the connection and terminates. + */ + void writeTerminate() { + enqueueMessage(TerminateMessage.INSTANCE, estimateTerminate()); + } + + /** + * + * The purpose of this message is to provide a resynchronization point for error recovery. + * When an error is detected while processing any extended-query message, the backend issues {@link ErrorResponse}, + * then reads and discards messages until this message is reached, then issues {@link ReadyForQuery} and returns to normal + * message processing. + * + * Note that no skipping occurs if an error is detected while processing this message which ensures that there is one + * and only one {@link ReadyForQuery} sent for each of this message. + * + * Note this message does not cause a transaction block opened with BEGIN to be closed. It is possible to detect this + * situation in {@link ReadyForQuery#txStatus()} that includes {@link TxStatus} information. + */ + void writeSync() { + enqueueMessage(SyncMessage.INSTANCE, estimateSync()); + } + + /** + * + * The message closes an existing prepared statement or portal and releases resources. + * Note that closing a prepared statement implicitly closes any open portals that were constructed from that statement. + * + * The response is either {@link CloseComplete} or {@link ErrorResponse} + * + * @param portal + */ + void writeClosePortal(String portal) { + enqueueMessage(ClosePortalMessage.INSTANCE, portal, estimateClosePortal(portal)); + } + + void writeClosePreparedStatement(byte[] statementName) { + enqueueMessage(ClosePreparedStatementMessage.INSTANCE, statementName, estimateClosePreparedStatement(statementName)); + } + + void writeStartupMessage(StartupMessage msg) { + enqueueMessage(msg, estimateStartupMessage(msg)); + } + + void writePasswordMessage(PasswordMessage msg) { + enqueueMessage(msg, estimatePasswordMessage(msg)); + } + + void writeScramClientInitialMessage(ScramClientInitialMessage msg) { + enqueueMessage(msg, estimateScramInitialMessage(msg)); + } + + void writeScramClientFinalMessage(ScramClientFinalMessage msg) { + enqueueMessage(msg, estimateScramFinalMessage(msg)); + } + + /** + * + * This message includes an SQL command (or commands) expressed as a text string. + * + * The possible response messages from the backend are + * {@link CommandComplete}, {@link RowDesc}, {@link DataRow}, {@link EmptyQueryResponse}, {@link ErrorResponse}, + * {@link ReadyForQuery} and {@link NoticeResponse} + */ + void writeQuery(Query query) { + enqueueMessage(query, estimateQueryMessage(query)); + } + + /** + * + * The message that using "statement" variant specifies the name of an existing prepared statement. + * + * The response is a {@link ParamDesc} message describing the parameters needed by the statement, + * followed by a {@link RowDesc} message describing the rows that will be returned when the statement is eventually + * executed or a {@link NoData} message if the statement will not return rows. + * {@link ErrorResponse} is issued if there is no such prepared statement. + * + * Note that since {@link Bind} has not yet been issued, the formats to be used for returned columns are not yet known to + * the backend; the format code fields in the {@link RowDesc} message will be zeroes in this case. + * + * The message that using "portal" variant specifies the name of an existing portal. + * + * The response is a {@link RowDesc} message describing the rows that will be returned by executing the portal; + * or a {@link NoData} message if the portal does not contain a query that will return rows; or {@link ErrorResponse} + * if there is no such portal. + */ + void writeDescribe(Describe describe) { + enqueueMessage(describe, estimateDescribe(describe)); + } + + void writeParse(String sql, byte[] statement, DataType[] parameterTypes) { + enqueueMessage(ParseMessage.INSTANCE, sql, statement, parameterTypes, estimateParse(sql, statement, parameterTypes)); + } + + /** + * The message specifies the portal and a maximum row count (zero meaning "fetch all rows") of the result. + * + * The row count of the result is only meaningful for portals containing commands that return row sets; + * in other cases the command is always executed to completion, and the row count of the result is ignored. + * + * The possible responses to this message are the same as {@link Query} message, except that + * it doesn't cause {@link ReadyForQuery} or {@link RowDesc} to be issued. + * + * If Execute terminates before completing the execution of a portal, it will send a {@link PortalSuspended} message; + * the appearance of this message tells the frontend that another Execute should be issued against the same portal to + * complete the operation. The {@link CommandComplete} message indicating completion of the source SQL command + * is not sent until the portal's execution is completed. Therefore, This message is always terminated by + * the appearance of exactly one of these messages: {@link CommandComplete}, + * {@link EmptyQueryResponse}, {@link ErrorResponse} or {@link PortalSuspended}. + * + * @author Emad Alblueshi + */ + void writeExecute(String portal, int rowCount) { + enqueueMessage(ExecuteMessage.INSTANCE, portal, rowCount, estimateExecute(portal, rowCount)); + } + + /** + * + * The message gives the name of the prepared statement, the name of portal, + * and the values to use for any parameter values present in the prepared statement. + * The supplied parameter set must match those needed by the prepared statement. + * + * The response is either {@link BindComplete} or {@link ErrorResponse}. + */ + void writeBind(Bind bind, String portal, Tuple paramValues) { + enqueueMessage(bind, portal, paramValues, estimateBind(bind, portal, paramValues)); + } + byte[] nextStatementName() { return psSeq.next(); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java index 5c2d133db1..9f4fd9e98c 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java @@ -16,11 +16,14 @@ */ package io.vertx.pgclient.impl.codec; +import io.vertx.core.VertxException; import io.vertx.sqlclient.impl.ErrorMessageFactory; +import io.vertx.sqlclient.internal.ArrayTuple; import io.vertx.sqlclient.internal.ParamDesc; import io.vertx.sqlclient.internal.TupleInternal; import java.util.Arrays; +import java.util.function.Function; /** * @author Emad Alblueshi @@ -38,15 +41,16 @@ DataType[] paramDataTypes() { return paramDataTypes; } - public String prepare(TupleInternal values) { + public TupleInternal prepare(TupleInternal values) { int numberOfParams = values.size(); if (numberOfParams > 65535) { - return "The number of parameters (" + numberOfParams + ") exceeds the maximum of 65535. Use arrays or split the query."; + throw new VertxException("The number of parameters (" + numberOfParams + ") exceeds the maximum of 65535. Use arrays or split the query.", true); } int paramDescLength = paramDataTypes.length; if (numberOfParams != paramDescLength) { - return ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParams); + throw new VertxException(ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParams), true); } + TupleInternal prepared = values; for (int i = 0; i < paramDescLength; i++) { DataType paramDataType = paramDataTypes[i]; ParamExtractor extractor = paramDataType.paramExtractor; @@ -54,11 +58,29 @@ public String prepare(TupleInternal values) { try { val = extractor.get(values, i); } catch (Exception e) { - return ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(paramDataType.decodingType, i, values.getValue(i)); + throw new VertxException(ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(paramDataType.decodingType, i, values.getValue(i)), true); } - values.setValue(i, val); + if (val != null) { + Function preparator = paramDataType.preEncoder; + if (preparator != null) { + if (prepared == values) { + prepared = new ArrayTuple(prepared); + } + if (paramDataType.array) { + Object[] array = (Object[]) val; + Object[] tmp = new Object[array.length]; + for (int j = 0;j < array.length;j++) { + tmp[j] = preparator.apply(array[j]); + } + val = tmp; + } else { + val = preparator.apply(val); + } + } + } + prepared.setValue(i, val); } - return null; + return prepared; } @Override diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java index be9c5036eb..1f078cb558 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java @@ -53,7 +53,7 @@ public String sql() { } @Override - public String prepare(TupleInternal values) { + public TupleInternal prepare(TupleInternal values) { return paramDesc.prepare(values); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java index f6fe07a152..72710b9ba9 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java @@ -20,7 +20,7 @@ /** * @author Emad Alblueshi */ -class Query { +class Query extends OutboundMessage { final String sql; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java index 5861f1bd4f..adbdb5fe3b 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java @@ -19,7 +19,7 @@ /** */ -class ScramClientFinalMessage { +class ScramClientFinalMessage extends OutboundMessage { final String message; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java index 270991de88..d35930e10f 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java @@ -19,7 +19,7 @@ /** */ -public class ScramClientInitialMessage { +public class ScramClientInitialMessage extends OutboundMessage { final String mechanism; final String message; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java index a1d6ceff7e..31eb47144e 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java @@ -27,11 +27,14 @@ /** * @author Emad Alblueshi */ -class StartupMessage { +class StartupMessage extends OutboundMessage { static final ByteBuf BUFF_USER = Unpooled.copiedBuffer("user", UTF_8).asReadOnly(); static final ByteBuf BUFF_DATABASE = Unpooled.copiedBuffer("database", UTF_8).asReadOnly(); + static final int BUFF_USER_LENGTH = 4; + static final int BUFF_DATABASE_LENGTH = 8; + final String username; final String database; final Map properties; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java new file mode 100644 index 0000000000..a870d5e88d --- /dev/null +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java @@ -0,0 +1,7 @@ +package io.vertx.pgclient.impl.codec; + +class SyncMessage extends OutboundMessage { + + static final SyncMessage INSTANCE = new SyncMessage(); + +} diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java new file mode 100644 index 0000000000..abf0335e3b --- /dev/null +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java @@ -0,0 +1,7 @@ +package io.vertx.pgclient.impl.codec; + +class TerminateMessage extends OutboundMessage { + + static final TerminateMessage INSTANCE = new TerminateMessage(); + +} diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java index 188d0c0af6..87551f1b06 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java @@ -18,6 +18,7 @@ package io.vertx.pgclient.impl.util; import io.netty.buffer.ByteBuf; +import io.netty.util.CharsetUtil; import io.vertx.core.buffer.Buffer; import io.vertx.sqlclient.Tuple; diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java index cc9d6e456e..b60c1f2970 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java @@ -29,6 +29,7 @@ import io.vertx.sqlclient.internal.PreparedStatement; import io.vertx.sqlclient.internal.TupleInternal; +import java.util.ArrayList; import java.util.List; import java.util.function.Function; import java.util.stream.Collector; @@ -65,23 +66,24 @@ QueryResultBuilder executeExtendedQuery(CommandScheduler scheduler, PreparedStatement preparedStatement, PrepareOptions options, boolean autoCommit, - Tuple arguments, + Tuple values, int fetch, String cursorId, boolean suspended, PromiseInternal promise) { ContextInternal context = promise.context(); QueryResultBuilder handler = createHandler(promise); - String msg = preparedStatement.prepare((TupleInternal) arguments); - if (msg != null) { - handler.fail(msg); + try { + values = preparedStatement.prepare((TupleInternal) values); + } catch (Exception e) { + handler.fail(e); return null; } ExtendedQueryCommand cmd = ExtendedQueryCommand.createQuery( preparedStatement.sql(), options, preparedStatement, - arguments, + values, fetch, cursorId, suspended, @@ -122,12 +124,11 @@ void executeBatchQuery(CommandScheduler scheduler, PromiseInternal promise) { ContextInternal context = promise.context(); QueryResultBuilder handler = createHandler(promise); - for (Tuple args : batch) { - String msg = preparedStatement.prepare((TupleInternal)args); - if (msg != null) { - handler.fail(msg); - return; - } + try { + batch = preparedStatement.prepare((List) batch); + } catch (Exception e) { + handler.fail(e); + return; } ExtendedQueryCommand cmd = ExtendedQueryCommand.createBatch(preparedStatement.sql(), options, preparedStatement, batch, autoCommit, collector, handler); scheduler.schedule(cmd, handler); diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java index c223728359..55591917f6 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java @@ -17,6 +17,11 @@ package io.vertx.sqlclient.internal; +import io.vertx.sqlclient.Tuple; + +import java.util.ArrayList; +import java.util.List; + public interface PreparedStatement { ParamDesc paramDesc(); @@ -25,6 +30,23 @@ public interface PreparedStatement { String sql(); - String prepare(TupleInternal values); + default List prepare(List batch) throws Exception { + List actual = batch; + int size = actual.size(); + for (int idx = 0;idx < size;idx++) { + TupleInternal values = batch.get(idx); + TupleInternal prepared = prepare(values); + if (prepared != values) { + if (batch == actual) { + actual = new ArrayList<>(actual); + } + actual.set(idx, prepared); + } + } + return actual; + } + default TupleInternal prepare(TupleInternal values) throws Exception { + return values; + } } diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java index dd44fb5b02..ba3956f95a 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java @@ -72,10 +72,11 @@ public static ExtendedQueryCommand createBatch( protected final PrepareOptions options; public PreparedStatement ps; protected final boolean batch; - protected final Object tuples; + private Object tuples; protected final int fetch; protected final String cursorId; protected final boolean suspended; + private boolean prepared; private ExtendedQueryCommand(String sql, PrepareOptions options, @@ -97,6 +98,7 @@ private ExtendedQueryCommand(String sql, this.fetch = fetch; this.cursorId = cursorId; this.suspended = suspended; + this.prepared = ps != null; } public PrepareOptions options() { @@ -109,17 +111,17 @@ public PrepareOptions options() { * @return {@code null} if the tuple preparation was successfull otherwise the validation error */ public String prepare() { - if (ps != null) { - if (batch) { - for (TupleInternal tuple : (List) tuples) { - String msg = ps.prepare(tuple); - if (msg != null) { - return msg; - } + if (ps != null && !prepared) { + prepared = true; // TODO : fix this + try { + if (batch) { + tuples = ps.prepare((List) tuples); + return null; + } else { + tuples = ps.prepare((TupleInternal) tuples); } - return null; - } else { - return ps.prepare((TupleInternal) tuples); + } catch (Exception e) { + return e.getMessage(); } } return null;
- * The row count of the result is only meaningful for portals containing commands that return row sets; - * in other cases the command is always executed to completion, and the row count of the result is ignored. - *
- * The possible responses to this message are the same as {@link Query} message, except that - * it doesn't cause {@link ReadyForQuery} or {@link RowDesc} to be issued. - *
- * If Execute terminates before completing the execution of a portal, it will send a {@link PortalSuspended} message; - * the appearance of this message tells the frontend that another Execute should be issued against the same portal to - * complete the operation. The {@link CommandComplete} message indicating completion of the source SQL command - * is not sent until the portal's execution is completed. Therefore, This message is always terminated by - * the appearance of exactly one of these messages: {@link CommandComplete}, - * {@link EmptyQueryResponse}, {@link ErrorResponse} or {@link PortalSuspended}. - * - * @author Emad Alblueshi - */ - void writeExecute(String portal, int rowCount) { - ensureBuffer(); - int pos = out.writerIndex(); - out.writeByte(EXECUTE); - out.writeInt(0); - if (portal != null) { - out.writeCharSequence(portal, UTF_8); - } - out.writeByte(0); - out.writeInt(rowCount); // Zero denotes "no limit" maybe for ReadStream - out.setInt(pos + 1, out.writerIndex() - pos - 1); + private static int estimateParse(String sql, byte[] statement, DataType[] parameterTypes) { + return 1 + + 4 + + DataTypeEstimator.estimateByteArray(statement) + + DataTypeEstimator.estimateCStringUTF8(sql) + + 2 + (parameterTypes == null ? 0 : parameterTypes.length * 4); } - /** - * - * The message gives the name of the prepared statement, the name of portal, - * and the values to use for any parameter values present in the prepared statement. - * The supplied parameter set must match those needed by the prepared statement. - * - * The response is either {@link BindComplete} or {@link ErrorResponse}. - */ - void writeBind(Bind bind, String portal, Tuple paramValues) { - ensureBuffer(); + private static void renderBind(Bind bind, String portal, Tuple paramValues, ByteBuf out) { int pos = out.writerIndex(); out.writeByte(BIND); out.writeInt(0); @@ -436,12 +454,218 @@ void writeBind(Bind bind, String portal, Tuple paramValues) { out.setInt(pos + 1, out.writerIndex() - pos - 1); } - private void ensureBuffer() { - if (out == null) { - out = ctx.alloc().ioBuffer(); + private static int estimateBind(Bind bind, String portal, Tuple paramValues) { + + int paramLen = paramValues.size(); + + int length = 1 + + 4 + + (portal != null ? DataTypeEstimator.estimateUTF8(portal) : 0) + + 1 + + DataTypeEstimator.estimateByteArray(bind.statement) + + 2 + + paramLen * 2 + + 2; + + for (int c = 0;c < paramLen;c++) { + Object param = paramValues.getValue(c); + if (param == null) { + length += 4; + } else { + DataType dataType = bind.paramTypes[c]; + length += 4; + if (dataType.supportsBinary) { + int estimator = dataType.lengthEstimator; + if (dataType.array) { + length += 4 + 4 + 4 + 4 + 4; + Object[] array = (Object[]) param; + for (Object elt : array) { + length += 4; + if (elt != null) { + length += DataTypeEstimator.estimate(estimator, elt); + } + } + } else { + length += DataTypeEstimator.estimate(estimator, param); + } + } else { + length += DataTypeEstimator.estimate(dataType.lengthEstimator, param); + } + } + } + + // Result columns are all in Binary format + if (bind.resultColumns.length > 0) { + length += 2 + bind.resultColumns.length * 2; + } else { + length += 2 + 2; + } + return length; + } + + private static void renderClosePortal(String portal, ByteBuf out) { + int pos = out.writerIndex(); + out.writeByte(CLOSE); + out.writeInt(0); + out.writeByte('P'); // 'S' to close a prepared statement or 'P' to close a portal + Util.writeCStringUTF8(out, portal); + out.setInt(pos + 1, out.writerIndex() - pos - 1); + } + + private static int estimateClosePortal(String portal) { + return 1 + 4 + 1 + DataTypeEstimator.estimateCStringUTF8(portal); + } + + void close() { + ByteBuf buff = renderPendingMessages(); + ctx.writeAndFlush(buff).addListener(v -> { + Channel ch = channelHandlerContext().channel(); + if (ch instanceof SocketChannel) { + SocketChannel channel = (SocketChannel) ch; + channel.shutdownOutput(); + } + }); + } + + void flush() { + ByteBuf buff = renderPendingMessages(); + if (buff == Unpooled.EMPTY_BUFFER) { + ctx.flush(); + } else { + ctx.writeAndFlush(buff, ctx.voidPromise()); } } + /** + * This message immediately closes the connection. On receipt of this message, + * the backend closes the connection and terminates. + */ + void writeTerminate() { + enqueueMessage(TerminateMessage.INSTANCE, estimateTerminate()); + } + + /** + * + * The purpose of this message is to provide a resynchronization point for error recovery. + * When an error is detected while processing any extended-query message, the backend issues {@link ErrorResponse}, + * then reads and discards messages until this message is reached, then issues {@link ReadyForQuery} and returns to normal + * message processing. + * + * Note that no skipping occurs if an error is detected while processing this message which ensures that there is one + * and only one {@link ReadyForQuery} sent for each of this message. + * + * Note this message does not cause a transaction block opened with BEGIN to be closed. It is possible to detect this + * situation in {@link ReadyForQuery#txStatus()} that includes {@link TxStatus} information. + */ + void writeSync() { + enqueueMessage(SyncMessage.INSTANCE, estimateSync()); + } + + /** + * + * The message closes an existing prepared statement or portal and releases resources. + * Note that closing a prepared statement implicitly closes any open portals that were constructed from that statement. + * + * The response is either {@link CloseComplete} or {@link ErrorResponse} + * + * @param portal + */ + void writeClosePortal(String portal) { + enqueueMessage(ClosePortalMessage.INSTANCE, portal, estimateClosePortal(portal)); + } + + void writeClosePreparedStatement(byte[] statementName) { + enqueueMessage(ClosePreparedStatementMessage.INSTANCE, statementName, estimateClosePreparedStatement(statementName)); + } + + void writeStartupMessage(StartupMessage msg) { + enqueueMessage(msg, estimateStartupMessage(msg)); + } + + void writePasswordMessage(PasswordMessage msg) { + enqueueMessage(msg, estimatePasswordMessage(msg)); + } + + void writeScramClientInitialMessage(ScramClientInitialMessage msg) { + enqueueMessage(msg, estimateScramInitialMessage(msg)); + } + + void writeScramClientFinalMessage(ScramClientFinalMessage msg) { + enqueueMessage(msg, estimateScramFinalMessage(msg)); + } + + /** + * + * This message includes an SQL command (or commands) expressed as a text string. + * + * The possible response messages from the backend are + * {@link CommandComplete}, {@link RowDesc}, {@link DataRow}, {@link EmptyQueryResponse}, {@link ErrorResponse}, + * {@link ReadyForQuery} and {@link NoticeResponse} + */ + void writeQuery(Query query) { + enqueueMessage(query, estimateQueryMessage(query)); + } + + /** + * + * The message that using "statement" variant specifies the name of an existing prepared statement. + * + * The response is a {@link ParamDesc} message describing the parameters needed by the statement, + * followed by a {@link RowDesc} message describing the rows that will be returned when the statement is eventually + * executed or a {@link NoData} message if the statement will not return rows. + * {@link ErrorResponse} is issued if there is no such prepared statement. + * + * Note that since {@link Bind} has not yet been issued, the formats to be used for returned columns are not yet known to + * the backend; the format code fields in the {@link RowDesc} message will be zeroes in this case. + * + * The message that using "portal" variant specifies the name of an existing portal. + * + * The response is a {@link RowDesc} message describing the rows that will be returned by executing the portal; + * or a {@link NoData} message if the portal does not contain a query that will return rows; or {@link ErrorResponse} + * if there is no such portal. + */ + void writeDescribe(Describe describe) { + enqueueMessage(describe, estimateDescribe(describe)); + } + + void writeParse(String sql, byte[] statement, DataType[] parameterTypes) { + enqueueMessage(ParseMessage.INSTANCE, sql, statement, parameterTypes, estimateParse(sql, statement, parameterTypes)); + } + + /** + * The message specifies the portal and a maximum row count (zero meaning "fetch all rows") of the result. + * + * The row count of the result is only meaningful for portals containing commands that return row sets; + * in other cases the command is always executed to completion, and the row count of the result is ignored. + * + * The possible responses to this message are the same as {@link Query} message, except that + * it doesn't cause {@link ReadyForQuery} or {@link RowDesc} to be issued. + * + * If Execute terminates before completing the execution of a portal, it will send a {@link PortalSuspended} message; + * the appearance of this message tells the frontend that another Execute should be issued against the same portal to + * complete the operation. The {@link CommandComplete} message indicating completion of the source SQL command + * is not sent until the portal's execution is completed. Therefore, This message is always terminated by + * the appearance of exactly one of these messages: {@link CommandComplete}, + * {@link EmptyQueryResponse}, {@link ErrorResponse} or {@link PortalSuspended}. + * + * @author Emad Alblueshi + */ + void writeExecute(String portal, int rowCount) { + enqueueMessage(ExecuteMessage.INSTANCE, portal, rowCount, estimateExecute(portal, rowCount)); + } + + /** + * + * The message gives the name of the prepared statement, the name of portal, + * and the values to use for any parameter values present in the prepared statement. + * The supplied parameter set must match those needed by the prepared statement. + * + * The response is either {@link BindComplete} or {@link ErrorResponse}. + */ + void writeBind(Bind bind, String portal, Tuple paramValues) { + enqueueMessage(bind, portal, paramValues, estimateBind(bind, portal, paramValues)); + } + byte[] nextStatementName() { return psSeq.next(); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java index 5c2d133db1..9f4fd9e98c 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java @@ -16,11 +16,14 @@ */ package io.vertx.pgclient.impl.codec; +import io.vertx.core.VertxException; import io.vertx.sqlclient.impl.ErrorMessageFactory; +import io.vertx.sqlclient.internal.ArrayTuple; import io.vertx.sqlclient.internal.ParamDesc; import io.vertx.sqlclient.internal.TupleInternal; import java.util.Arrays; +import java.util.function.Function; /** * @author Emad Alblueshi @@ -38,15 +41,16 @@ DataType[] paramDataTypes() { return paramDataTypes; } - public String prepare(TupleInternal values) { + public TupleInternal prepare(TupleInternal values) { int numberOfParams = values.size(); if (numberOfParams > 65535) { - return "The number of parameters (" + numberOfParams + ") exceeds the maximum of 65535. Use arrays or split the query."; + throw new VertxException("The number of parameters (" + numberOfParams + ") exceeds the maximum of 65535. Use arrays or split the query.", true); } int paramDescLength = paramDataTypes.length; if (numberOfParams != paramDescLength) { - return ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParams); + throw new VertxException(ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParams), true); } + TupleInternal prepared = values; for (int i = 0; i < paramDescLength; i++) { DataType paramDataType = paramDataTypes[i]; ParamExtractor extractor = paramDataType.paramExtractor; @@ -54,11 +58,29 @@ public String prepare(TupleInternal values) { try { val = extractor.get(values, i); } catch (Exception e) { - return ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(paramDataType.decodingType, i, values.getValue(i)); + throw new VertxException(ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(paramDataType.decodingType, i, values.getValue(i)), true); } - values.setValue(i, val); + if (val != null) { + Function preparator = paramDataType.preEncoder; + if (preparator != null) { + if (prepared == values) { + prepared = new ArrayTuple(prepared); + } + if (paramDataType.array) { + Object[] array = (Object[]) val; + Object[] tmp = new Object[array.length]; + for (int j = 0;j < array.length;j++) { + tmp[j] = preparator.apply(array[j]); + } + val = tmp; + } else { + val = preparator.apply(val); + } + } + } + prepared.setValue(i, val); } - return null; + return prepared; } @Override diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java index be9c5036eb..1f078cb558 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java @@ -53,7 +53,7 @@ public String sql() { } @Override - public String prepare(TupleInternal values) { + public TupleInternal prepare(TupleInternal values) { return paramDesc.prepare(values); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java index f6fe07a152..72710b9ba9 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java @@ -20,7 +20,7 @@ /** * @author Emad Alblueshi */ -class Query { +class Query extends OutboundMessage { final String sql; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java index 5861f1bd4f..adbdb5fe3b 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java @@ -19,7 +19,7 @@ /** */ -class ScramClientFinalMessage { +class ScramClientFinalMessage extends OutboundMessage { final String message; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java index 270991de88..d35930e10f 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java @@ -19,7 +19,7 @@ /** */ -public class ScramClientInitialMessage { +public class ScramClientInitialMessage extends OutboundMessage { final String mechanism; final String message; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java index a1d6ceff7e..31eb47144e 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java @@ -27,11 +27,14 @@ /** * @author Emad Alblueshi */ -class StartupMessage { +class StartupMessage extends OutboundMessage { static final ByteBuf BUFF_USER = Unpooled.copiedBuffer("user", UTF_8).asReadOnly(); static final ByteBuf BUFF_DATABASE = Unpooled.copiedBuffer("database", UTF_8).asReadOnly(); + static final int BUFF_USER_LENGTH = 4; + static final int BUFF_DATABASE_LENGTH = 8; + final String username; final String database; final Map properties; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java new file mode 100644 index 0000000000..a870d5e88d --- /dev/null +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java @@ -0,0 +1,7 @@ +package io.vertx.pgclient.impl.codec; + +class SyncMessage extends OutboundMessage { + + static final SyncMessage INSTANCE = new SyncMessage(); + +} diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java new file mode 100644 index 0000000000..abf0335e3b --- /dev/null +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java @@ -0,0 +1,7 @@ +package io.vertx.pgclient.impl.codec; + +class TerminateMessage extends OutboundMessage { + + static final TerminateMessage INSTANCE = new TerminateMessage(); + +} diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java index 188d0c0af6..87551f1b06 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java @@ -18,6 +18,7 @@ package io.vertx.pgclient.impl.util; import io.netty.buffer.ByteBuf; +import io.netty.util.CharsetUtil; import io.vertx.core.buffer.Buffer; import io.vertx.sqlclient.Tuple; diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java index cc9d6e456e..b60c1f2970 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java @@ -29,6 +29,7 @@ import io.vertx.sqlclient.internal.PreparedStatement; import io.vertx.sqlclient.internal.TupleInternal; +import java.util.ArrayList; import java.util.List; import java.util.function.Function; import java.util.stream.Collector; @@ -65,23 +66,24 @@ QueryResultBuilder executeExtendedQuery(CommandScheduler scheduler, PreparedStatement preparedStatement, PrepareOptions options, boolean autoCommit, - Tuple arguments, + Tuple values, int fetch, String cursorId, boolean suspended, PromiseInternal promise) { ContextInternal context = promise.context(); QueryResultBuilder handler = createHandler(promise); - String msg = preparedStatement.prepare((TupleInternal) arguments); - if (msg != null) { - handler.fail(msg); + try { + values = preparedStatement.prepare((TupleInternal) values); + } catch (Exception e) { + handler.fail(e); return null; } ExtendedQueryCommand cmd = ExtendedQueryCommand.createQuery( preparedStatement.sql(), options, preparedStatement, - arguments, + values, fetch, cursorId, suspended, @@ -122,12 +124,11 @@ void executeBatchQuery(CommandScheduler scheduler, PromiseInternal promise) { ContextInternal context = promise.context(); QueryResultBuilder handler = createHandler(promise); - for (Tuple args : batch) { - String msg = preparedStatement.prepare((TupleInternal)args); - if (msg != null) { - handler.fail(msg); - return; - } + try { + batch = preparedStatement.prepare((List) batch); + } catch (Exception e) { + handler.fail(e); + return; } ExtendedQueryCommand cmd = ExtendedQueryCommand.createBatch(preparedStatement.sql(), options, preparedStatement, batch, autoCommit, collector, handler); scheduler.schedule(cmd, handler); diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java index c223728359..55591917f6 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java @@ -17,6 +17,11 @@ package io.vertx.sqlclient.internal; +import io.vertx.sqlclient.Tuple; + +import java.util.ArrayList; +import java.util.List; + public interface PreparedStatement { ParamDesc paramDesc(); @@ -25,6 +30,23 @@ public interface PreparedStatement { String sql(); - String prepare(TupleInternal values); + default List prepare(List batch) throws Exception { + List actual = batch; + int size = actual.size(); + for (int idx = 0;idx < size;idx++) { + TupleInternal values = batch.get(idx); + TupleInternal prepared = prepare(values); + if (prepared != values) { + if (batch == actual) { + actual = new ArrayList<>(actual); + } + actual.set(idx, prepared); + } + } + return actual; + } + default TupleInternal prepare(TupleInternal values) throws Exception { + return values; + } } diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java index dd44fb5b02..ba3956f95a 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java @@ -72,10 +72,11 @@ public static ExtendedQueryCommand createBatch( protected final PrepareOptions options; public PreparedStatement ps; protected final boolean batch; - protected final Object tuples; + private Object tuples; protected final int fetch; protected final String cursorId; protected final boolean suspended; + private boolean prepared; private ExtendedQueryCommand(String sql, PrepareOptions options, @@ -97,6 +98,7 @@ private ExtendedQueryCommand(String sql, this.fetch = fetch; this.cursorId = cursorId; this.suspended = suspended; + this.prepared = ps != null; } public PrepareOptions options() { @@ -109,17 +111,17 @@ public PrepareOptions options() { * @return {@code null} if the tuple preparation was successfull otherwise the validation error */ public String prepare() { - if (ps != null) { - if (batch) { - for (TupleInternal tuple : (List) tuples) { - String msg = ps.prepare(tuple); - if (msg != null) { - return msg; - } + if (ps != null && !prepared) { + prepared = true; // TODO : fix this + try { + if (batch) { + tuples = ps.prepare((List) tuples); + return null; + } else { + tuples = ps.prepare((TupleInternal) tuples); } - return null; - } else { - return ps.prepare((TupleInternal) tuples); + } catch (Exception e) { + return e.getMessage(); } } return null;
- * The message gives the name of the prepared statement, the name of portal, - * and the values to use for any parameter values present in the prepared statement. - * The supplied parameter set must match those needed by the prepared statement. - *
- * The response is either {@link BindComplete} or {@link ErrorResponse}. - */ - void writeBind(Bind bind, String portal, Tuple paramValues) { - ensureBuffer(); + private static void renderBind(Bind bind, String portal, Tuple paramValues, ByteBuf out) { int pos = out.writerIndex(); out.writeByte(BIND); out.writeInt(0); @@ -436,12 +454,218 @@ void writeBind(Bind bind, String portal, Tuple paramValues) { out.setInt(pos + 1, out.writerIndex() - pos - 1); } - private void ensureBuffer() { - if (out == null) { - out = ctx.alloc().ioBuffer(); + private static int estimateBind(Bind bind, String portal, Tuple paramValues) { + + int paramLen = paramValues.size(); + + int length = 1 + + 4 + + (portal != null ? DataTypeEstimator.estimateUTF8(portal) : 0) + + 1 + + DataTypeEstimator.estimateByteArray(bind.statement) + + 2 + + paramLen * 2 + + 2; + + for (int c = 0;c < paramLen;c++) { + Object param = paramValues.getValue(c); + if (param == null) { + length += 4; + } else { + DataType dataType = bind.paramTypes[c]; + length += 4; + if (dataType.supportsBinary) { + int estimator = dataType.lengthEstimator; + if (dataType.array) { + length += 4 + 4 + 4 + 4 + 4; + Object[] array = (Object[]) param; + for (Object elt : array) { + length += 4; + if (elt != null) { + length += DataTypeEstimator.estimate(estimator, elt); + } + } + } else { + length += DataTypeEstimator.estimate(estimator, param); + } + } else { + length += DataTypeEstimator.estimate(dataType.lengthEstimator, param); + } + } + } + + // Result columns are all in Binary format + if (bind.resultColumns.length > 0) { + length += 2 + bind.resultColumns.length * 2; + } else { + length += 2 + 2; + } + return length; + } + + private static void renderClosePortal(String portal, ByteBuf out) { + int pos = out.writerIndex(); + out.writeByte(CLOSE); + out.writeInt(0); + out.writeByte('P'); // 'S' to close a prepared statement or 'P' to close a portal + Util.writeCStringUTF8(out, portal); + out.setInt(pos + 1, out.writerIndex() - pos - 1); + } + + private static int estimateClosePortal(String portal) { + return 1 + 4 + 1 + DataTypeEstimator.estimateCStringUTF8(portal); + } + + void close() { + ByteBuf buff = renderPendingMessages(); + ctx.writeAndFlush(buff).addListener(v -> { + Channel ch = channelHandlerContext().channel(); + if (ch instanceof SocketChannel) { + SocketChannel channel = (SocketChannel) ch; + channel.shutdownOutput(); + } + }); + } + + void flush() { + ByteBuf buff = renderPendingMessages(); + if (buff == Unpooled.EMPTY_BUFFER) { + ctx.flush(); + } else { + ctx.writeAndFlush(buff, ctx.voidPromise()); } } + /** + * This message immediately closes the connection. On receipt of this message, + * the backend closes the connection and terminates. + */ + void writeTerminate() { + enqueueMessage(TerminateMessage.INSTANCE, estimateTerminate()); + } + + /** + *
+ * The purpose of this message is to provide a resynchronization point for error recovery. + * When an error is detected while processing any extended-query message, the backend issues {@link ErrorResponse}, + * then reads and discards messages until this message is reached, then issues {@link ReadyForQuery} and returns to normal + * message processing. + *
+ * Note that no skipping occurs if an error is detected while processing this message which ensures that there is one + * and only one {@link ReadyForQuery} sent for each of this message. + *
+ * Note this message does not cause a transaction block opened with BEGIN to be closed. It is possible to detect this + * situation in {@link ReadyForQuery#txStatus()} that includes {@link TxStatus} information. + */ + void writeSync() { + enqueueMessage(SyncMessage.INSTANCE, estimateSync()); + } + + /** + *
+ * The message closes an existing prepared statement or portal and releases resources. + * Note that closing a prepared statement implicitly closes any open portals that were constructed from that statement. + *
+ * The response is either {@link CloseComplete} or {@link ErrorResponse} + * + * @param portal + */ + void writeClosePortal(String portal) { + enqueueMessage(ClosePortalMessage.INSTANCE, portal, estimateClosePortal(portal)); + } + + void writeClosePreparedStatement(byte[] statementName) { + enqueueMessage(ClosePreparedStatementMessage.INSTANCE, statementName, estimateClosePreparedStatement(statementName)); + } + + void writeStartupMessage(StartupMessage msg) { + enqueueMessage(msg, estimateStartupMessage(msg)); + } + + void writePasswordMessage(PasswordMessage msg) { + enqueueMessage(msg, estimatePasswordMessage(msg)); + } + + void writeScramClientInitialMessage(ScramClientInitialMessage msg) { + enqueueMessage(msg, estimateScramInitialMessage(msg)); + } + + void writeScramClientFinalMessage(ScramClientFinalMessage msg) { + enqueueMessage(msg, estimateScramFinalMessage(msg)); + } + + /** + *
+ * This message includes an SQL command (or commands) expressed as a text string. + *
+ * The possible response messages from the backend are + * {@link CommandComplete}, {@link RowDesc}, {@link DataRow}, {@link EmptyQueryResponse}, {@link ErrorResponse}, + * {@link ReadyForQuery} and {@link NoticeResponse} + */ + void writeQuery(Query query) { + enqueueMessage(query, estimateQueryMessage(query)); + } + + /** + *
+ * The message that using "statement" variant specifies the name of an existing prepared statement. + *
+ * The response is a {@link ParamDesc} message describing the parameters needed by the statement, + * followed by a {@link RowDesc} message describing the rows that will be returned when the statement is eventually + * executed or a {@link NoData} message if the statement will not return rows. + * {@link ErrorResponse} is issued if there is no such prepared statement. + *
+ * Note that since {@link Bind} has not yet been issued, the formats to be used for returned columns are not yet known to + * the backend; the format code fields in the {@link RowDesc} message will be zeroes in this case. + *
+ * The message that using "portal" variant specifies the name of an existing portal. + *
+ * The response is a {@link RowDesc} message describing the rows that will be returned by executing the portal; + * or a {@link NoData} message if the portal does not contain a query that will return rows; or {@link ErrorResponse} + * if there is no such portal. + */ + void writeDescribe(Describe describe) { + enqueueMessage(describe, estimateDescribe(describe)); + } + + void writeParse(String sql, byte[] statement, DataType[] parameterTypes) { + enqueueMessage(ParseMessage.INSTANCE, sql, statement, parameterTypes, estimateParse(sql, statement, parameterTypes)); + } + + /** + * The message specifies the portal and a maximum row count (zero meaning "fetch all rows") of the result. + *
+ * The row count of the result is only meaningful for portals containing commands that return row sets; + * in other cases the command is always executed to completion, and the row count of the result is ignored. + *
+ * The possible responses to this message are the same as {@link Query} message, except that + * it doesn't cause {@link ReadyForQuery} or {@link RowDesc} to be issued. + *
+ * If Execute terminates before completing the execution of a portal, it will send a {@link PortalSuspended} message; + * the appearance of this message tells the frontend that another Execute should be issued against the same portal to + * complete the operation. The {@link CommandComplete} message indicating completion of the source SQL command + * is not sent until the portal's execution is completed. Therefore, This message is always terminated by + * the appearance of exactly one of these messages: {@link CommandComplete}, + * {@link EmptyQueryResponse}, {@link ErrorResponse} or {@link PortalSuspended}. + * + * @author Emad Alblueshi + */ + void writeExecute(String portal, int rowCount) { + enqueueMessage(ExecuteMessage.INSTANCE, portal, rowCount, estimateExecute(portal, rowCount)); + } + + /** + *
+ * The message gives the name of the prepared statement, the name of portal, + * and the values to use for any parameter values present in the prepared statement. + * The supplied parameter set must match those needed by the prepared statement. + *
+ * The response is either {@link BindComplete} or {@link ErrorResponse}. + */ + void writeBind(Bind bind, String portal, Tuple paramValues) { + enqueueMessage(bind, portal, paramValues, estimateBind(bind, portal, paramValues)); + } + byte[] nextStatementName() { return psSeq.next(); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java index 5c2d133db1..9f4fd9e98c 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgParamDesc.java @@ -16,11 +16,14 @@ */ package io.vertx.pgclient.impl.codec; +import io.vertx.core.VertxException; import io.vertx.sqlclient.impl.ErrorMessageFactory; +import io.vertx.sqlclient.internal.ArrayTuple; import io.vertx.sqlclient.internal.ParamDesc; import io.vertx.sqlclient.internal.TupleInternal; import java.util.Arrays; +import java.util.function.Function; /** * @author Emad Alblueshi @@ -38,15 +41,16 @@ DataType[] paramDataTypes() { return paramDataTypes; } - public String prepare(TupleInternal values) { + public TupleInternal prepare(TupleInternal values) { int numberOfParams = values.size(); if (numberOfParams > 65535) { - return "The number of parameters (" + numberOfParams + ") exceeds the maximum of 65535. Use arrays or split the query."; + throw new VertxException("The number of parameters (" + numberOfParams + ") exceeds the maximum of 65535. Use arrays or split the query.", true); } int paramDescLength = paramDataTypes.length; if (numberOfParams != paramDescLength) { - return ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParams); + throw new VertxException(ErrorMessageFactory.buildWhenArgumentsLengthNotMatched(paramDescLength, numberOfParams), true); } + TupleInternal prepared = values; for (int i = 0; i < paramDescLength; i++) { DataType paramDataType = paramDataTypes[i]; ParamExtractor extractor = paramDataType.paramExtractor; @@ -54,11 +58,29 @@ public String prepare(TupleInternal values) { try { val = extractor.get(values, i); } catch (Exception e) { - return ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(paramDataType.decodingType, i, values.getValue(i)); + throw new VertxException(ErrorMessageFactory.buildWhenArgumentsTypeNotMatched(paramDataType.decodingType, i, values.getValue(i)), true); } - values.setValue(i, val); + if (val != null) { + Function preparator = paramDataType.preEncoder; + if (preparator != null) { + if (prepared == values) { + prepared = new ArrayTuple(prepared); + } + if (paramDataType.array) { + Object[] array = (Object[]) val; + Object[] tmp = new Object[array.length]; + for (int j = 0;j < array.length;j++) { + tmp[j] = preparator.apply(array[j]); + } + val = tmp; + } else { + val = preparator.apply(val); + } + } + } + prepared.setValue(i, val); } - return null; + return prepared; } @Override diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java index be9c5036eb..1f078cb558 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/PgPreparedStatement.java @@ -53,7 +53,7 @@ public String sql() { } @Override - public String prepare(TupleInternal values) { + public TupleInternal prepare(TupleInternal values) { return paramDesc.prepare(values); } diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java index f6fe07a152..72710b9ba9 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/Query.java @@ -20,7 +20,7 @@ /** * @author Emad Alblueshi */ -class Query { +class Query extends OutboundMessage { final String sql; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java index 5861f1bd4f..adbdb5fe3b 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientFinalMessage.java @@ -19,7 +19,7 @@ /** */ -class ScramClientFinalMessage { +class ScramClientFinalMessage extends OutboundMessage { final String message; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java index 270991de88..d35930e10f 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/ScramClientInitialMessage.java @@ -19,7 +19,7 @@ /** */ -public class ScramClientInitialMessage { +public class ScramClientInitialMessage extends OutboundMessage { final String mechanism; final String message; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java index a1d6ceff7e..31eb47144e 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/StartupMessage.java @@ -27,11 +27,14 @@ /** * @author Emad Alblueshi */ -class StartupMessage { +class StartupMessage extends OutboundMessage { static final ByteBuf BUFF_USER = Unpooled.copiedBuffer("user", UTF_8).asReadOnly(); static final ByteBuf BUFF_DATABASE = Unpooled.copiedBuffer("database", UTF_8).asReadOnly(); + static final int BUFF_USER_LENGTH = 4; + static final int BUFF_DATABASE_LENGTH = 8; + final String username; final String database; final Map properties; diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java new file mode 100644 index 0000000000..a870d5e88d --- /dev/null +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/SyncMessage.java @@ -0,0 +1,7 @@ +package io.vertx.pgclient.impl.codec; + +class SyncMessage extends OutboundMessage { + + static final SyncMessage INSTANCE = new SyncMessage(); + +} diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java new file mode 100644 index 0000000000..abf0335e3b --- /dev/null +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/codec/TerminateMessage.java @@ -0,0 +1,7 @@ +package io.vertx.pgclient.impl.codec; + +class TerminateMessage extends OutboundMessage { + + static final TerminateMessage INSTANCE = new TerminateMessage(); + +} diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java index 188d0c0af6..87551f1b06 100644 --- a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java +++ b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/util/Util.java @@ -18,6 +18,7 @@ package io.vertx.pgclient.impl.util; import io.netty.buffer.ByteBuf; +import io.netty.util.CharsetUtil; import io.vertx.core.buffer.Buffer; import io.vertx.sqlclient.Tuple; diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java index cc9d6e456e..b60c1f2970 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/QueryExecutor.java @@ -29,6 +29,7 @@ import io.vertx.sqlclient.internal.PreparedStatement; import io.vertx.sqlclient.internal.TupleInternal; +import java.util.ArrayList; import java.util.List; import java.util.function.Function; import java.util.stream.Collector; @@ -65,23 +66,24 @@ QueryResultBuilder executeExtendedQuery(CommandScheduler scheduler, PreparedStatement preparedStatement, PrepareOptions options, boolean autoCommit, - Tuple arguments, + Tuple values, int fetch, String cursorId, boolean suspended, PromiseInternal promise) { ContextInternal context = promise.context(); QueryResultBuilder handler = createHandler(promise); - String msg = preparedStatement.prepare((TupleInternal) arguments); - if (msg != null) { - handler.fail(msg); + try { + values = preparedStatement.prepare((TupleInternal) values); + } catch (Exception e) { + handler.fail(e); return null; } ExtendedQueryCommand cmd = ExtendedQueryCommand.createQuery( preparedStatement.sql(), options, preparedStatement, - arguments, + values, fetch, cursorId, suspended, @@ -122,12 +124,11 @@ void executeBatchQuery(CommandScheduler scheduler, PromiseInternal promise) { ContextInternal context = promise.context(); QueryResultBuilder handler = createHandler(promise); - for (Tuple args : batch) { - String msg = preparedStatement.prepare((TupleInternal)args); - if (msg != null) { - handler.fail(msg); - return; - } + try { + batch = preparedStatement.prepare((List) batch); + } catch (Exception e) { + handler.fail(e); + return; } ExtendedQueryCommand cmd = ExtendedQueryCommand.createBatch(preparedStatement.sql(), options, preparedStatement, batch, autoCommit, collector, handler); scheduler.schedule(cmd, handler); diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java index c223728359..55591917f6 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/PreparedStatement.java @@ -17,6 +17,11 @@ package io.vertx.sqlclient.internal; +import io.vertx.sqlclient.Tuple; + +import java.util.ArrayList; +import java.util.List; + public interface PreparedStatement { ParamDesc paramDesc(); @@ -25,6 +30,23 @@ public interface PreparedStatement { String sql(); - String prepare(TupleInternal values); + default List prepare(List batch) throws Exception { + List actual = batch; + int size = actual.size(); + for (int idx = 0;idx < size;idx++) { + TupleInternal values = batch.get(idx); + TupleInternal prepared = prepare(values); + if (prepared != values) { + if (batch == actual) { + actual = new ArrayList<>(actual); + } + actual.set(idx, prepared); + } + } + return actual; + } + default TupleInternal prepare(TupleInternal values) throws Exception { + return values; + } } diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java index dd44fb5b02..ba3956f95a 100644 --- a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java +++ b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/command/ExtendedQueryCommand.java @@ -72,10 +72,11 @@ public static ExtendedQueryCommand createBatch( protected final PrepareOptions options; public PreparedStatement ps; protected final boolean batch; - protected final Object tuples; + private Object tuples; protected final int fetch; protected final String cursorId; protected final boolean suspended; + private boolean prepared; private ExtendedQueryCommand(String sql, PrepareOptions options, @@ -97,6 +98,7 @@ private ExtendedQueryCommand(String sql, this.fetch = fetch; this.cursorId = cursorId; this.suspended = suspended; + this.prepared = ps != null; } public PrepareOptions options() { @@ -109,17 +111,17 @@ public PrepareOptions options() { * @return {@code null} if the tuple preparation was successfull otherwise the validation error */ public String prepare() { - if (ps != null) { - if (batch) { - for (TupleInternal tuple : (List) tuples) { - String msg = ps.prepare(tuple); - if (msg != null) { - return msg; - } + if (ps != null && !prepared) { + prepared = true; // TODO : fix this + try { + if (batch) { + tuples = ps.prepare((List) tuples); + return null; + } else { + tuples = ps.prepare((TupleInternal) tuples); } - return null; - } else { - return ps.prepare((TupleInternal) tuples); + } catch (Exception e) { + return e.getMessage(); } } return null;