Return-Path: Delivered-To: apmail-db-derby-commits-archive@www.apache.org Received: (qmail 11551 invoked from network); 16 Dec 2006 14:03:54 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (140.211.11.2) by minotaur.apache.org with SMTP; 16 Dec 2006 14:03:54 -0000 Received: (qmail 97667 invoked by uid 500); 16 Dec 2006 11:17:04 -0000 Delivered-To: apmail-db-derby-commits-archive@db.apache.org Received: (qmail 97601 invoked by uid 500); 16 Dec 2006 11:17:04 -0000 Mailing-List: contact derby-commits-help@db.apache.org; run by ezmlm Precedence: bulk list-help: list-unsubscribe: List-Post: Reply-To: "Derby Development" List-Id: Delivered-To: mailing list derby-commits@db.apache.org Received: (qmail 97588 invoked by uid 99); 16 Dec 2006 11:17:04 -0000 Received: from herse.apache.org (HELO herse.apache.org) (140.211.11.133) by apache.org (qpsmtpd/0.29) with ESMTP; Sat, 16 Dec 2006 03:17:04 -0800 X-ASF-Spam-Status: No, hits=-9.4 required=10.0 tests=ALL_TRUSTED,NO_REAL_NAME X-Spam-Check-By: apache.org Received: from [140.211.11.3] (HELO eris.apache.org) (140.211.11.3) by apache.org (qpsmtpd/0.29) with ESMTP; Sat, 16 Dec 2006 03:16:54 -0800 Received: by eris.apache.org (Postfix, from userid 65534) id 5D5481A981A; Sat, 16 Dec 2006 03:16:07 -0800 (PST) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r487788 [1/2] - in /db/derby/code/trunk/java: client/org/apache/derby/client/am/ client/org/apache/derby/client/net/ drda/org/apache/derby/impl/drda/ engine/org/apache/derby/iapi/jdbc/ engine/org/apache/derby/loc/ shared/org/apache/derby/sh... Date: Sat, 16 Dec 2006 11:16:06 -0000 To: derby-commits@db.apache.org From: tmnk@apache.org X-Mailer: svnmailer-1.1.0 Message-Id: <20061216111607.5D5481A981A@eris.apache.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: tmnk Date: Sat Dec 16 03:16:04 2006 New Revision: 487788 URL: http://svn.apache.org/viewvc?view=rev&rev=487788 Log: - DERBY-1471 Implement layer B streaming for new methods defined in JDBC4.0 - Patch by Tomohito Nakayama (tomonaka@basil.ocn.ne.jp) Added: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/LayerBStreamedEXTDTAReaderInputStream.java (with props) db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/StandardEXTDTAReaderInputStream.java (with props) Modified: db/derby/code/trunk/java/client/org/apache/derby/client/am/Blob.java db/derby/code/trunk/java/client/org/apache/derby/client/am/Clob.java db/derby/code/trunk/java/client/org/apache/derby/client/am/Lob.java db/derby/code/trunk/java/client/org/apache/derby/client/net/NetConnection.java db/derby/code/trunk/java/client/org/apache/derby/client/net/NetDatabaseMetaData.java db/derby/code/trunk/java/client/org/apache/derby/client/net/NetStatementRequest.java db/derby/code/trunk/java/client/org/apache/derby/client/net/Request.java db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DDMReader.java db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DRDAConnThread.java db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/EXTDTAReaderInputStream.java db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/ReEncodedInputStream.java db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/TestProto.java db/derby/code/trunk/java/engine/org/apache/derby/iapi/jdbc/BrokeredPreparedStatement.java db/derby/code/trunk/java/engine/org/apache/derby/iapi/jdbc/BrokeredPreparedStatement40.java db/derby/code/trunk/java/engine/org/apache/derby/iapi/jdbc/EnginePreparedStatement.java db/derby/code/trunk/java/engine/org/apache/derby/loc/messages.xml db/derby/code/trunk/java/shared/org/apache/derby/shared/common/reference/SQLState.java db/derby/code/trunk/java/testing/org/apache/derbyTesting/functionTests/tests/jdbc4/PreparedStatementTest.java Modified: db/derby/code/trunk/java/client/org/apache/derby/client/am/Blob.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/client/org/apache/derby/client/am/Blob.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/client/org/apache/derby/client/am/Blob.java (original) +++ db/derby/code/trunk/java/client/org/apache/derby/client/am/Blob.java Sat Dec 16 03:16:04 2006 @@ -42,13 +42,16 @@ // must generate an independent stream. java.io.InputStream binaryStream_ = null; int dataOffset_; - + //---------------------constructors/finalizer--------------------------------- public Blob(byte[] binaryString, Agent agent, int dataOffset) { - super(agent); + + super(agent, + false); + binaryString_ = binaryString; dataType_ |= BINARY_STRING; sqlLength_ = binaryString.length - dataOffset; @@ -60,7 +63,10 @@ public Blob(Agent agent, java.io.InputStream binaryStream, int length) { - super(agent); + + super(agent, + false); + binaryStream_ = binaryStream; dataType_ |= BINARY_STREAM; sqlLength_ = length; @@ -88,11 +94,15 @@ * @param binaryStream the stream to get data from */ public Blob(Agent agent, java.io.InputStream binaryStream) { - super(agent); + + super(agent, + isLayerBStreamingPossible(agent)); + binaryStream_ = binaryStream; dataType_ |= BINARY_STREAM; sqlLength_ = -1; lengthObtained_ = false; + } // ---------------------------jdbc 2------------------------------------------ @@ -109,6 +119,11 @@ } // Code to handle the lengthless constructor. if (!lengthObtained_) { + + if( willBeLayerBStreamed() ) + throw new SqlException(agent_.logWriter_, + LOB_OBJECT_LENGTH_UNKNOWN_YET); + binaryStream_ = super.materializeStream(binaryStream_, "java.sql.Blob"); } @@ -562,4 +577,5 @@ throw new SqlException(null,new ClientMessageId(SQLState.LOB_OBJECT_INVALID)) .getSQLException(); } + } Modified: db/derby/code/trunk/java/client/org/apache/derby/client/am/Clob.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/client/org/apache/derby/client/am/Clob.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/client/org/apache/derby/client/am/Clob.java (original) +++ db/derby/code/trunk/java/client/org/apache/derby/client/am/Clob.java Sat Dec 16 03:16:04 2006 @@ -54,10 +54,13 @@ //This boolean variable indicates whether the Clob object has //been invalidated by calling free() on it private boolean isValid = true; - + //---------------------constructors/finalizer--------------------------------- public Clob(Agent agent, String string) { - this(agent); + + this(agent, + false); + string_ = string; sqlLength_ = string_.length(); lengthObtained_ = true; @@ -69,7 +72,10 @@ byte[] unconvertedBytes, String charsetName, int dataOffset) throws SqlException { - this(agent); + + this(agent, + false); + try { // check for null encoding is needed because the net layer // will no longer throw an exception if the server didn't specify @@ -102,7 +108,9 @@ java.io.InputStream inputStream, String encoding, int length) throws SqlException { - this(agent); + + this(agent, + false); sqlLength_ = length; lengthObtained_ = true; @@ -142,10 +150,12 @@ */ public Clob(Agent agent, java.io.InputStream inputStream, String encoding) throws SqlException { - this(agent); - + + this(agent, + isLayerBStreamingPossible( agent )); + lengthObtained_ = false; - + if (encoding.equals("US-ASCII")) { asciiStream_ = inputStream; dataType_ |= ASCII_STREAM; @@ -159,7 +169,10 @@ // CTOR for character stream input // THE ENCODING IS ASSUMED TO BE "UTF-16BE" public Clob(Agent agent, java.io.Reader reader, int length) { - this(agent); + + this(agent, + false); + sqlLength_ = length; lengthObtained_ = true; characterStream_ = reader; @@ -177,7 +190,10 @@ * @param reader the data to insert */ public Clob(Agent agent, Reader reader) { - this(agent); + + this(agent, + isLayerBStreamingPossible( agent ) ); + lengthObtained_ = false; // Wrap reader in stream to share code. unicodeStream_ = EncodedInputStream.createUTF8Stream(reader); @@ -185,8 +201,10 @@ dataType_ |= UNICODE_STREAM; } - private Clob(Agent agent) { - super(agent); + private Clob(Agent agent, + boolean willBeLayerBStreamed) { + super(agent, + willBeLayerBStreamed); } protected void finalize() throws java.lang.Throwable { @@ -214,6 +232,11 @@ if (lengthObtained_) { return sqlLength_; } + + if( willBeLayerBStreamed() ) + throw new SqlException(agent_.logWriter_, + LOB_OBJECT_LENGTH_UNKNOWN_YET); + materializeStream(); lengthInBytes_ = super.sqlLength(); @@ -875,4 +898,5 @@ "java.sql.Clob"); dataType_ = UNICODE_STREAM; } + } Modified: db/derby/code/trunk/java/client/org/apache/derby/client/am/Lob.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/client/org/apache/derby/client/am/Lob.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/client/org/apache/derby/client/am/Lob.java (original) +++ db/derby/code/trunk/java/client/org/apache/derby/client/am/Lob.java Sat Dec 16 03:16:04 2006 @@ -25,6 +25,11 @@ import java.io.IOException; import java.util.ArrayList; +import java.sql.SQLException; + +import org.apache.derby.client.net.NetConfiguration; +import org.apache.derby.client.net.NetConnection; + import org.apache.derby.shared.common.reference.SQLState; public abstract class Lob implements UnitOfWorkListener { @@ -44,11 +49,20 @@ protected long sqlLength_; // length of the LOB value, as defined by the server protected boolean lengthObtained_; + + final private boolean willBeLayerBStreamed_; + //-----------------------------messageId------------------------------------------ + final static protected ClientMessageId LOB_OBJECT_LENGTH_UNKNOWN_YET = + new ClientMessageId( SQLState.LOB_OBJECT_LENGTH_UNKNOWN_YET ); + + //---------------------constructors/finalizer--------------------------------- - protected Lob(Agent agent) { + protected Lob(Agent agent, + boolean willBeLayerBStreamed) { agent_ = agent; lengthObtained_ = false; + willBeLayerBStreamed_ = willBeLayerBStreamed; } protected void finalize() throws java.lang.Throwable { @@ -172,4 +186,30 @@ ); } } + + public static boolean isLengthObtained(Lob l){ + return l.lengthObtained_; + } + + public abstract long length() throws SQLException; + + protected static boolean isLayerBStreamingPossible( Agent agent ){ + + final NetConnection netConn = + ( NetConnection ) agent.connection_ ; + + final int securityMechanism = + netConn.getSecurityMechanism(); + + return + netConn.serverSupportsLayerBStreaming() && + securityMechanism != NetConfiguration.SECMEC_EUSRIDDTA && + securityMechanism != NetConfiguration.SECMEC_EUSRPWDDTA; + + } + + public boolean willBeLayerBStreamed() { + return willBeLayerBStreamed_; + } + } Modified: db/derby/code/trunk/java/client/org/apache/derby/client/net/NetConnection.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/client/org/apache/derby/client/net/NetConnection.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/client/org/apache/derby/client/net/NetConnection.java (original) +++ db/derby/code/trunk/java/client/org/apache/derby/client/net/NetConnection.java Sat Dec 16 03:16:04 2006 @@ -1768,6 +1768,17 @@ (NetDatabaseMetaData) databaseMetaData_; return metadata.serverSupportsQryclsimp(); } + + + public final boolean serverSupportsLayerBStreaming() { + + NetDatabaseMetaData metadata = + (NetDatabaseMetaData) databaseMetaData_; + + return metadata.serverSupportsLayerBStreaming(); + + } + /** * Returns if a transaction is in process Modified: db/derby/code/trunk/java/client/org/apache/derby/client/net/NetDatabaseMetaData.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/client/org/apache/derby/client/net/NetDatabaseMetaData.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/client/org/apache/derby/client/net/NetDatabaseMetaData.java (original) +++ db/derby/code/trunk/java/client/org/apache/derby/client/net/NetDatabaseMetaData.java Sat Dec 16 03:16:04 2006 @@ -32,6 +32,8 @@ /** True if the server supports QRYCLSIMP. */ private boolean supportsQryclsimp_; + + private boolean supportsLayerBStreaming_; public NetDatabaseMetaData(NetAgent netAgent, NetConnection netConnection) { // Consider setting product level during parse @@ -87,6 +89,9 @@ } else { supportsQryclsimp_ = false; } + + supportsLayerBStreaming_ = + productLevel_.greaterThanOrEqualTo(10, 3, 0); } /** @@ -97,6 +102,10 @@ */ final boolean serverSupportsQryclsimp() { return supportsQryclsimp_; + } + + final boolean serverSupportsLayerBStreaming() { + return supportsLayerBStreaming_; } } Modified: db/derby/code/trunk/java/client/org/apache/derby/client/net/NetStatementRequest.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/client/org/apache/derby/client/net/NetStatementRequest.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/client/org/apache/derby/client/net/NetStatementRequest.java (original) +++ db/derby/code/trunk/java/client/org/apache/derby/client/net/NetStatementRequest.java Sat Dec 16 03:16:04 2006 @@ -21,6 +21,7 @@ package org.apache.derby.client.net; import org.apache.derby.iapi.reference.DRDAConstants; +import org.apache.derby.client.am.Lob; import org.apache.derby.client.am.Blob; import org.apache.derby.client.am.Clob; import org.apache.derby.client.am.ColumnMetaData; @@ -646,7 +647,7 @@ Object[] inputs) throws SqlException { try { - long dataLength = 0; + Object o = null; markLengthBytes(CodePoint.FDODTA); @@ -674,9 +675,10 @@ if (o == null) { writeSingleorMixedCcsidLDString((String) inputs[i], netAgent_.typdef_.getCcsidMbcEncoding()); } else { // use the promototed object instead - Clob c = (Clob) o; - dataLength = c.length(); - setFDODTALobLength(protocolTypesAndLengths, i, dataLength); + setFDODTALob(netAgent_.netConnection_.getSecurityMechanism(), + (Clob) o, + protocolTypesAndLengths, + i); } break; @@ -686,8 +688,12 @@ if (o == null) { } else { // use the promototed object instead - dataLength = ((Clob) o).length(); - setFDODTALobLength(protocolTypesAndLengths, i, dataLength); + + setFDODTALob(netAgent_.netConnection_.getSecurityMechanism(), + (Clob) o, + protocolTypesAndLengths, + i); + } break; @@ -726,9 +732,11 @@ if (o == null) { writeLDBytes((byte[]) inputs[i]); } else { // use the promototed object instead - Blob b = (Blob) o; - dataLength = b.length(); - setFDODTALobLength(protocolTypesAndLengths, i, dataLength); + + setFDODTALob(netAgent_.netConnection_.getSecurityMechanism(), + (Clob) o, + protocolTypesAndLengths, + i); } break; case DRDAConstants.DRDA_TYPE_NLOBCSBCS: @@ -737,47 +745,93 @@ o = retrievePromotedParameterIfExists(i); if (o == null) { try { - dataLength = ((java.sql.Clob) inputs[i]).length(); + + java.sql.Clob c = (java.sql.Clob) inputs[i]; + + if( c instanceof Clob && + ( (Clob) c ).willBeLayerBStreamed() ) { + setFDODTALobLengthUnknown( i ); + + }else{ + long dataLength = c.length(); + setFDODTALobLength(protocolTypesAndLengths, i, dataLength); + + } + } catch (java.sql.SQLException e) { throw new SqlException(netAgent_.logWriter_, new ClientMessageId(SQLState.NET_ERROR_GETTING_BLOB_LENGTH), e); } } else { - dataLength = ((Clob) o).length(); + setFDODTALob(netAgent_.netConnection_.getSecurityMechanism(), + (Clob) o, + protocolTypesAndLengths, + i); } - setFDODTALobLength(protocolTypesAndLengths, i, dataLength); + break; case DRDAConstants.DRDA_TYPE_NLOBBYTES: // check for a promoted Clob o = retrievePromotedParameterIfExists(i); if (o == null) { try { - dataLength = ((java.sql.Blob) inputs[i]).length(); + + java.sql.Blob b = (java.sql.Blob) inputs[i]; + + if( b instanceof Blob && + ( (Blob) b ).willBeLayerBStreamed() ) { + + setFDODTALobLengthUnknown( i ); + + }else{ + long dataLength = b.length(); + setFDODTALobLength(protocolTypesAndLengths, i, dataLength); + + } + } catch (java.sql.SQLException e) { throw new SqlException(netAgent_.logWriter_, new ClientMessageId(SQLState.NET_ERROR_GETTING_BLOB_LENGTH), e); } } else { // use promoted Blob - dataLength = ((Blob) o).length(); + setFDODTALob(netAgent_.netConnection_.getSecurityMechanism(), + (Blob) o, + protocolTypesAndLengths, + i); } - setFDODTALobLength(protocolTypesAndLengths, i, dataLength); break; case DRDAConstants.DRDA_TYPE_NLOBCMIXED: // check for a promoted Clob o = retrievePromotedParameterIfExists(i); if (o == null) { - if (((Clob) inputs[i]).isString()) { - dataLength = ((Clob) inputs[i]).getUTF8Length(); - } else // must be a Unicode stream - { - dataLength = ((Clob) inputs[i]).length(); + + final Clob c = (Clob) inputs[i]; + + if (c.isString()) { + setFDODTALobLength(protocolTypesAndLengths, + i, + c.getUTF8Length() ); + + } else if ( ! c.willBeLayerBStreamed() ){ + // must be a Unicode stream + setFDODTALobLength(protocolTypesAndLengths, + i, + c.length() ); + + } else { + setFDODTALobLengthUnknown( i ); + } + } else { // use promoted Clob - dataLength = ((Clob) o).length(); + setFDODTALob(netAgent_.netConnection_.getSecurityMechanism(), + (Clob) o, + protocolTypesAndLengths, + i); } - setFDODTALobLength(protocolTypesAndLengths, i, dataLength); + break; default: throw new SqlException(netAgent_.logWriter_, @@ -851,14 +905,24 @@ new ClientMessageId(SQLState.NET_ERROR_GETTING_BLOB_LENGTH), e); } - } else if (((Blob) b).isBinaryStream()) { - writeScalarStream(chainFlag, - chainedWithSameCorrelator, - CodePoint.EXTDTA, - (int) ((Blob) b).length(), - ((Blob) b).getBinaryStream(), - writeNullByte, - index + 1); + } else if ( ( (Blob) b).isBinaryStream()) { + + if( ( (Blob) b).willBeLayerBStreamed() ){ + writeScalarStream(chainFlag, + chainedWithSameCorrelator, + CodePoint.EXTDTA, + ((Blob) b).getBinaryStream(), + writeNullByte, + index + 1); + }else{ + writeScalarStream(chainFlag, + chainedWithSameCorrelator, + CodePoint.EXTDTA, + (int) ((Blob) b).length(), + ((Blob) b).getBinaryStream(), + writeNullByte, + index + 1); + } } else { // must be a binary string // note: a possible optimization is to use writeScalarLobBytes // when the input is small @@ -898,30 +962,62 @@ new ClientMessageId(SQLState.NET_ERROR_GETTING_BLOB_LENGTH), e); } - } else if (((Clob) c).isCharacterStream()) { - writeScalarStream(chainFlag, - chainedWithSameCorrelator, - CodePoint.EXTDTA, - (int) ((Clob) c).length(), - ((Clob) c).getCharacterStream(), - writeNullByte, - index + 1); + } else if ( ( (Clob) c).isCharacterStream()) { + + if( ( (Clob) c).willBeLayerBStreamed() ) { + writeScalarStream(chainFlag, + chainedWithSameCorrelator, + CodePoint.EXTDTA, + ((Clob) c).getCharacterStream(), + writeNullByte, + index + 1); + }else{ + writeScalarStream(chainFlag, + chainedWithSameCorrelator, + CodePoint.EXTDTA, + (int) ((Clob) c).length(), + ((Clob) c).getCharacterStream(), + writeNullByte, + index + 1); + } + } else if (((Clob) c).isAsciiStream()) { - writeScalarStream(chainFlag, - chainedWithSameCorrelator, - CodePoint.EXTDTA, - (int) ((Clob) c).length(), - ((Clob) c).getAsciiStream(), - writeNullByte, - index + 1); + + if( ( (Clob) c).willBeLayerBStreamed() ){ + writeScalarStream(chainFlag, + chainedWithSameCorrelator, + CodePoint.EXTDTA, + ((Clob) c).getAsciiStream(), + writeNullByte, + index + 1); + }else { + writeScalarStream(chainFlag, + chainedWithSameCorrelator, + CodePoint.EXTDTA, + (int) ((Clob) c).length(), + ((Clob) c).getAsciiStream(), + writeNullByte, + index + 1); + } + } else if (((Clob) c).isUnicodeStream()) { - writeScalarStream(chainFlag, - chainedWithSameCorrelator, - CodePoint.EXTDTA, - (int) ((Clob) c).length(), - ((Clob) c).getUnicodeStream(), - writeNullByte, - index + 1); + + if( ( (Clob) c).willBeLayerBStreamed() ){ + writeScalarStream(chainFlag, + chainedWithSameCorrelator, + CodePoint.EXTDTA, + ((Clob) c).getUnicodeStream(), + writeNullByte, + index + 1); + }else{ + writeScalarStream(chainFlag, + chainedWithSameCorrelator, + CodePoint.EXTDTA, + (int) ((Clob) c).length(), + ((Clob) c).getUnicodeStream(), + writeNullByte, + index + 1); + } } else { // must be a String // note: a possible optimization is to use writeScalarLobBytes // when the input is small. @@ -1028,9 +1124,19 @@ // Place the new Lob in the promototedParameter_ collection for // NetStatementRequest use promototedParameters_.put(new Integer(i), c); - + lidAndLengths[i][0] = DRDAConstants.DRDA_TYPE_NLOBCMIXED; - lidAndLengths[i][1] = buildPlaceholderLength(c.length()); + + if( c.willBeLayerBStreamed() ){ + + //Correspond to totalLength 0 as default length for unknown + lidAndLengths[i][1] = 0x8002; + + }else { + lidAndLengths[i][1] = buildPlaceholderLength(c.length()); + + } + } catch (java.io.UnsupportedEncodingException e) { throw new SqlException(netAgent_.logWriter_, new ClientMessageId(SQLState.UNSUPPORTED_ENCODING), @@ -1206,7 +1312,17 @@ } else { lidAndLengths[i][0] = DRDAConstants.DRDA_TYPE_NLOBBYTES; try { - lidAndLengths[i][1] = buildPlaceholderLength(b.length()); + + if( b instanceof Blob && + ( (Blob) b).willBeLayerBStreamed() ){ + + //Correspond to totalLength 0 as default length for unknown + lidAndLengths[i][1] = 0x8002; + + }else { + lidAndLengths[i][1] = buildPlaceholderLength(b.length()); + + } } catch (java.sql.SQLException e) { throw new SqlException(netAgent_.logWriter_, new ClientMessageId(SQLState.NET_ERROR_GETTING_BLOB_LENGTH), e); @@ -1219,6 +1335,9 @@ java.sql.Clob c = (java.sql.Clob) inputRow[i]; boolean isExternalClob = !(c instanceof org.apache.derby.client.am.Clob); long lobLength = 0; + + boolean doesLayerBStreaming = false; + if (c == null) { lobLength = parameterMetaData.sqlLength_[i]; } else if (isExternalClob) { @@ -1230,8 +1349,16 @@ e); } } else { - lobLength = ((Clob) c).length(); + if( ( (Clob) c ).willBeLayerBStreamed() ){ + doesLayerBStreaming = true; + + }else{ + lobLength = ((Clob) c).length(); + + } + } + if (c == null) { lidAndLengths[i][0] = DRDAConstants.DRDA_TYPE_NLOBCMIXED; lidAndLengths[i][1] = buildPlaceholderLength(lobLength); @@ -1240,16 +1367,52 @@ lidAndLengths[i][1] = buildPlaceholderLength(lobLength); } else if (((Clob) c).isCharacterStream()) { lidAndLengths[i][0] = DRDAConstants.DRDA_TYPE_NLOBCDBCS; - lidAndLengths[i][1] = buildPlaceholderLength(lobLength); + + if( doesLayerBStreaming ) { + + //Correspond to totalLength 0 as default length for unknown + lidAndLengths[i][1] = 0x8002; + + }else { + lidAndLengths[i][1] = buildPlaceholderLength(lobLength); + } + } else if (((Clob) c).isUnicodeStream()) { lidAndLengths[i][0] = DRDAConstants.DRDA_TYPE_NLOBCMIXED; - lidAndLengths[i][1] = buildPlaceholderLength(lobLength); + + if( doesLayerBStreaming ) { + + //Correspond to totalLength 0 as default length for unknown + lidAndLengths[i][1] = 0x8002; + + }else { + lidAndLengths[i][1] = buildPlaceholderLength(lobLength); + } + } else if (((Clob) c).isAsciiStream()) { lidAndLengths[i][0] = DRDAConstants.DRDA_TYPE_NLOBCSBCS; - lidAndLengths[i][1] = buildPlaceholderLength(lobLength); + + if( doesLayerBStreaming ) { + + //Correspond to totalLength 0 as default length for unknown + lidAndLengths[i][1] = 0x8002; + + }else { + lidAndLengths[i][1] = buildPlaceholderLength(lobLength); + } + } else if (((Clob) c).isString()) { lidAndLengths[i][0] = DRDAConstants.DRDA_TYPE_NLOBCMIXED; - lidAndLengths[i][1] = buildPlaceholderLength(((Clob) c).getUTF8Length()); + + if( doesLayerBStreaming ) { + + //Correspond to totalLength 0 as default length for unknown + lidAndLengths[i][1] = 0x8002; + + }else { + lidAndLengths[i][1] = buildPlaceholderLength(lobLength); + } + } } break; @@ -1509,6 +1672,16 @@ extdtaPositions_.add(new Integer(i)); } } + + private void setFDODTALobLengthUnknown(int i) throws SqlException { + short v = 1; + writeShort( v <<= 15 ); + if (extdtaPositions_ == null) { + extdtaPositions_ = new java.util.ArrayList(); + } + + extdtaPositions_.add(new Integer(i)); + } private boolean checkSendQryrowset(int fetchSize, int resultSetType) { @@ -1567,6 +1740,26 @@ private int getNextOverrideLid() { return overrideLid_++; } + + private void setFDODTALob(int securityMechanism, + Lob lob, + int[][] protocolTypesAndLengths, + int i) + throws SqlException, java.sql.SQLException{ + + if( lob.willBeLayerBStreamed() ) { + + setFDODTALobLengthUnknown( i ); + + } else { + setFDODTALobLength(protocolTypesAndLengths, + i, + lob.length() ); + } + + } + + } Modified: db/derby/code/trunk/java/client/org/apache/derby/client/net/Request.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/client/org/apache/derby/client/net/Request.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/client/org/apache/derby/client/net/Request.java (original) +++ db/derby/code/trunk/java/client/org/apache/derby/client/net/Request.java Sat Dec 16 03:16:04 2006 @@ -27,8 +27,11 @@ import org.apache.derby.client.am.Utils; import org.apache.derby.shared.common.reference.SQLState; +import java.io.BufferedInputStream; import java.io.UnsupportedEncodingException; +import java.io.IOException; + public class Request { // byte array buffer used for constructing requests. @@ -217,6 +220,23 @@ simpleDssFinalize = simpleFinalizeBuildingNextDss; } + + final void writeScalarStream(boolean chained, + boolean chainedWithSameCorrelator, + int codePoint, + java.io.InputStream in, + boolean writeNullByte, + int parameterIndex) throws DisconnectException, SqlException { + + writePlainScalarStream(chained, + chainedWithSameCorrelator, + codePoint, + in, + writeNullByte, + parameterIndex); + + } + final void writeScalarStream(boolean chained, boolean chainedWithSameCorrelator, @@ -250,7 +270,8 @@ } } - + + // We need to reuse the agent's sql exception accumulation mechanism // for this write exception, pad if the length is too big, and truncation if the length is too small final private void writeEncryptedScalarStream(boolean chained, @@ -498,6 +519,107 @@ } + // We need to reuse the agent's sql exception accumulation mechanism + // for this write exception, pad if the length is too big, and truncation if the length is too small + final private void writePlainScalarStream(boolean chained, + boolean chainedWithSameCorrelator, + int codePoint, + java.io.InputStream in, + boolean writeNullByte, + int parameterIndex) throws DisconnectException, SqlException { + + in = new BufferedInputStream( in ); + + flushExistingDSS(); + + ensureLength( DssConstants.MAX_DSS_LEN ); + + buildDss(true, + chained, + chainedWithSameCorrelator, + DssConstants.GDSFMT_OBJDSS, + correlationID_, + true); + + int spareInDss; + + if (writeNullByte) { + spareInDss = DssConstants.MAX_DSS_LEN - 6 - 4 - 1; + } else { + spareInDss = DssConstants.MAX_DSS_LEN - 6 - 4; + } + + buildLengthAndCodePointForLob(codePoint, + writeNullByte); + + try{ + + int bytesRead = 0; + + while( ( bytesRead = + in.read(bytes_, offset_, spareInDss ) + ) > -1 ) { + + spareInDss -= bytesRead; + offset_ += bytesRead; + + if( spareInDss <= 0 ){ + + if( ! peekStream( ( BufferedInputStream ) in ) ) + break; + + flushScalarStreamSegment(); + + bytes_[offset_++] = (byte) (0xff); + bytes_[offset_++] = (byte) (0xff); + + spareInDss = DssConstants.MAX_DSS_LEN - 2; + + } + + } + + + } catch (java.io.IOException e) { + + final SqlException sqlex = + new SqlException(netAgent_.logWriter_, + new ClientMessageId(SQLState.NET_IOEXCEPTION_ON_READ), + new Integer(parameterIndex), + e.getMessage(), + e); + + netAgent_.accumulateReadException(sqlex); + + return; + } + + + + // check to make sure that the specified length wasn't too small + try { + if (in.read() != -1) { + // set with SQLSTATE 01004: The value of a string was truncated when assigned to a host variable. + + final SqlException sqlex = + new SqlException(netAgent_.logWriter_, + new ClientMessageId(SQLState.NET_INPUTSTREAM_LENGTH_TOO_SMALL), + new Integer(parameterIndex)); + + netAgent_.accumulateReadException(sqlex); + } + } catch (java.io.IOException e) { + netAgent_.accumulateReadException(new SqlException( + netAgent_.logWriter_, + new ClientMessageId( + SQLState.NET_IOEXCEPTION_ON_STREAMLEN_VERIFICATION), + new Integer(parameterIndex), + e.getMessage(), + e)); + } + } + + // Throw DataTruncation, instead of closing connection if input size mismatches // An implication of this, is that we need to extend the chaining model // for writes to accomodate chained write exceptoins @@ -518,8 +640,24 @@ writeNullByte, parameterIndex); } - - + + + final void writeScalarStream(boolean chained, + boolean chainedWithSameCorrelator, + int codePoint, + java.io.Reader r, + boolean writeNullByte, + int parameterIndex) throws DisconnectException, + SqlException{ + writeScalarStream(chained, + chainedWithSameCorrelator, + codePoint, + EncodedInputStream.createUTF16BEStream(r), + writeNullByte, + parameterIndex); + } + + // prepScalarStream does the following prep for writing stream data: // 1. Flushes an existing DSS segment, if necessary // 2. Determines if extended length bytes are needed @@ -576,6 +714,22 @@ return extendedLengthByteCount; } + + protected final void flushExistingDSS() throws DisconnectException { + + try { + if (simpleDssFinalize) { + finalizeDssLength(); + } else { + finalizePreviousChainedDss(true); + } + sendBytes(netAgent_.getOutputStream()); + } catch (java.io.IOException e) { + netAgent_.throwCommunicationsFailure(e); + } + + } + // Writes out a scalar stream DSS segment, along with DSS continuation headers, // if necessary. @@ -601,6 +755,19 @@ return newBytesToRead; } + + protected final int flushScalarStreamSegment() throws DisconnectException { + + try { + sendBytes(netAgent_.getOutputStream()); + } catch (java.io.IOException ioe) { + netAgent_.throwCommunicationsFailure(ioe); + } + + dssLengthLocation_ = offset_; + return DssConstants.MAX_DSS_LEN; + } + // the offset_ must not be updated when an error is encountered // note valid data may be overwritten @@ -1494,12 +1661,41 @@ } } + + + private void buildLengthAndCodePointForLob(int codePoint, + boolean writeNullByte) throws DisconnectException { + + //0x8004 is for Layer B Streaming. + //See DRDA, Version 3, Volume 3: Distributed Data Management (DDM) Architecture page 315. + writeLengthCodePoint(0x8004, codePoint); + + // write the null byte, if necessary + if (writeNullByte) { + write1Byte(0x0); + } + + } + public void setDssLengthLocation(int location) { dssLengthLocation_ = location; } - + + public void setCorrelationID(int id) { correlationID_ = id; } + + + private static boolean peekStream( BufferedInputStream in ) + throws IOException { + + in.mark( 1 ); + boolean notYet = in.read() > -1; + in.reset(); + return notYet; + + } + } Modified: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DDMReader.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DDMReader.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DDMReader.java (original) +++ db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DDMReader.java Sat Dec 16 03:16:04 2006 @@ -99,6 +99,7 @@ private final static boolean NO_ADJUST_LENGTHS = false; private final static long MAX_EXTDTA_SIZE= Long.MAX_VALUE; private static boolean internalTrace = true; + // magnitude represented in an int array, used in BigDecimal conversion private static final int[][] tenRadixMagnitude = { @@ -152,6 +153,9 @@ // input stream private InputStream inputStream; + + // State whether doing layer B Streaming or not. + private boolean doingLayerBStreaming = false;; // constructor DDMReader (DRDAConnThread agent, DssTrace dssTrace) @@ -496,11 +500,13 @@ /** * Read the DDM Length and CodePoint * + * @param isLayerBStreamingPossible true only when layer B streaming is possible * @return - returns codepoint * * @exception DRDProtocolException */ - protected int readLengthAndCodePoint() throws DRDAProtocolException + protected int readLengthAndCodePoint( boolean isLayerBStreamingPossible ) + throws DRDAProtocolException { ensureBLayerDataInBuffer (4, NO_ADJUST_LENGTHS); @@ -552,6 +558,23 @@ ((buffer[pos++] & 0xff) << 0); adjustSize = 8; break; + + case 0: + + if( isLayerBStreamingPossible && + ( codePoint == CodePoint.EXTDTA || + codePoint == CodePoint.QRYDTA ) ){ + + startLayerBStreaming(); + adjustSize = 4; + + }else { + agent.throwSyntaxrm(CodePoint.SYNERRCD_INCORRECT_EXTENDED_LEN, + DRDAProtocolException.NO_CODPNT_ARG); + } + + break; + default: agent.throwSyntaxrm(CodePoint.SYNERRCD_INCORRECT_EXTENDED_LEN, DRDAProtocolException.NO_CODPNT_ARG); @@ -610,13 +633,13 @@ { // if the collecion is exhausted then return NO_CODEPOINT if (ddmCollectionLenStack[topDdmCollectionStack] == 0) - { + { // done with this collection so remove it's length from the stack ddmCollectionLenStack[topDdmCollectionStack--] = 0; return NO_CODEPOINT; } else { - return readLengthAndCodePoint(); + return readLengthAndCodePoint( false ); } } } @@ -963,15 +986,40 @@ { if (checkNullability && isEXTDTANull()) { return null; - } else { - return new EXTDTAReaderInputStream(this); - } + + } else if ( doingLayerBStreaming ){ + return new LayerBStreamedEXTDTAReaderInputStream(this); + + } else { + return new StandardEXTDTAReaderInputStream(this); + + } + } + + /** + * This method is used by EXTDTAReaderInputStream to read the first chunk + * of data. + * This lengthless method must be called only when layer B streaming. + * + * @exception DRDAProtocolException standard DRDA protocol exception + */ + ByteArrayInputStream readLOBInitStream() + throws DRDAProtocolException + { + if ( SanityManager.DEBUG ) { + SanityManager.ASSERT( doingLayerBStreaming ); + } + + return readLOBInitStream( 0 ); + + } + /** * This method is used by EXTDTAReaderInputStream to read the first chunk * of data. - * @param desiredLength the desired length of chunk + * @param desiredLength the desired length of chunk. This parameter is ignored when layerB Streaming is doing. * @exception DRDAProtocolException standard DRDA protocol exception */ ByteArrayInputStream readLOBInitStream(final long desiredLength) @@ -980,10 +1028,37 @@ return readLOBChunk(false, desiredLength); } + + /** + * This method is used by EXTDTAReaderInputStream to read the next chunk + * of data. + * + * Calling this method finishes layer B streaming + * if continuation of DSS segment was finished. + * This lengthless method must be called only when layer B streaming. + * + * @param desiredLength the desired length of chunk. This parameter is ignored when layerB Streaming is doing. + * @exception IOException IOException + */ + ByteArrayInputStream readLOBContinuationStream () + throws IOException + { + if ( SanityManager.DEBUG ) { + SanityManager.ASSERT( doingLayerBStreaming ); + } + return readLOBContinuationStream( 0 ); + } + + /** * This method is used by EXTDTAReaderInputStream to read the next chunk * of data. - * @param desiredLength the desired length of chunk + * + * Furthermore, when Layer B streaming is carried out, + * calling this method finishes layer B streaming + * if continuation of DSS segment was finished. + * + * @param desiredLength the desired length of chunk. This parameter is ignored when layerB Streaming is doing. * @exception IOException IOException */ ByteArrayInputStream readLOBContinuationStream (final long desiredLength) @@ -1000,8 +1075,12 @@ /** * This method is used by EXTDTAReaderInputStream to read the next chunk * of data. + * + * Furthermore, when Layer B streaming is carried out, + * calling this method may finish layer B streaming. + * * @param readHeader set to true if the dss continuation should be read - * @param desiredLength the desired length of chunk + * @param desiredLength the desired length of chunk. This parameter is ignored when layerB Streaming is doing. * @exception DRDAProtocolException standard DRDA protocol exception */ private ByteArrayInputStream readLOBChunk @@ -1011,11 +1090,21 @@ if (readHeader) { readDSSContinuationHeader(); } - int copySize = (int) Math.min(dssLength, desiredLength); + + int copySize = doingLayerBStreaming ? + dssLength : + (int) Math.min(dssLength, desiredLength); // read the segment ensureALayerDataInBuffer (copySize); - adjustLengths (copySize); + + if( ! doingLayerBStreaming ){ + adjustLengths (copySize); + + }else{ + dssLength -= copySize; + + } // Create ByteArrayInputStream on top of buffer. // This will not make a copy of the buffer. @@ -1023,11 +1112,20 @@ new ByteArrayInputStream(buffer, pos, copySize); pos += copySize; + if( doingLayerBStreaming && + ! dssIsContinued ) + finishLayerBStreaming(); + return bais; } byte[] getExtData (long desiredLength, boolean checkNullability) throws DRDAProtocolException { + + if ( SanityManager.DEBUG ) { + SanityManager.ASSERT( ! doingLayerBStreaming ); + } + boolean readHeader; int copySize; ByteArrayOutputStream baos; @@ -1865,5 +1963,21 @@ return DssConstants.DSSCHAIN; } - + + + private void startLayerBStreaming() { + doingLayerBStreaming = true; + } + + + private void finishLayerBStreaming() { + doingLayerBStreaming = false; + } + + + boolean doingLayerBStreaming() { + return doingLayerBStreaming; + } + + } Modified: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DRDAConnThread.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DRDAConnThread.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DRDAConnThread.java (original) +++ db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/DRDAConnThread.java Sat Dec 16 03:16:04 2006 @@ -26,9 +26,10 @@ package org.apache.derby.impl.drda; import java.io.ByteArrayInputStream; -import java.io.IOException; +import java.io.ByteArrayOutputStream; import java.io.InputStream; import java.io.OutputStream; +import java.io.InputStreamReader; import java.io.IOException; import java.io.UnsupportedEncodingException; import java.math.BigDecimal; @@ -64,6 +65,7 @@ import org.apache.derby.impl.jdbc.EmbedSQLException; import org.apache.derby.impl.jdbc.Util; import org.apache.derby.jdbc.InternalDriver; +import org.apache.derby.iapi.jdbc.EnginePreparedStatement; class DRDAConnThread extends Thread { @@ -661,7 +663,7 @@ do { correlationID = reader.readDssHeader(); - int codePoint = reader.readLengthAndCodePoint(); + int codePoint = reader.readLengthAndCodePoint( false ); int writerMark = writer.markDSSClearPoint(); if (checkSecurityCodepoint) @@ -1022,7 +1024,7 @@ } } - codePoint = reader.readLengthAndCodePoint(); + codePoint = reader.readLengthAndCodePoint( false ); // The first code point in the exchange of attributes must be EXCSAT if (codePoint != CodePoint.EXCSAT) @@ -1058,7 +1060,7 @@ } correlationID = reader.readDssHeader(); - codePoint = reader.readLengthAndCodePoint(); + codePoint = reader.readLengthAndCodePoint( false ); verifyRequiredObject(codePoint,CodePoint.ACCRDB); int svrcod = parseACCRDB(); @@ -1965,7 +1967,7 @@ correlationID = reader.readDssHeader(); while (reader.moreDssData()) { - codePoint = reader.readLengthAndCodePoint(); + codePoint = reader.readLengthAndCodePoint( false ); switch(codePoint) { // optional @@ -2411,7 +2413,7 @@ correlationID = reader.readDssHeader(); while (reader.moreDssData()) { - codePoint = reader.readLengthAndCodePoint(); + codePoint = reader.readLengthAndCodePoint( false ); switch(codePoint) { // optional @@ -3446,7 +3448,7 @@ correlationID = reader.readDssHeader(); while (reader.moreDssData()) { - codePoint = reader.readLengthAndCodePoint(); + codePoint = reader.readLengthAndCodePoint( false ); switch(codePoint) { // required @@ -3945,7 +3947,7 @@ correlationID = reader.readDssHeader(); while (reader.moreDssData()) { - codePoint = reader.readLengthAndCodePoint(); + codePoint = reader.readLengthAndCodePoint( true ); switch(codePoint) { // optional @@ -4481,7 +4483,7 @@ if (i < numExt -1) { correlationID = reader.readDssHeader(); - int codePoint = reader.readLengthAndCodePoint(); + int codePoint = reader.readLengthAndCodePoint( true ); } } @@ -4524,11 +4526,27 @@ paramBytes = null; final EXTDTAReaderInputStream stream = reader.getEXTDTAReaderInputStream(checkNullability); - if (stream==null) { - ps.setBytes(i+1, null); - } else { - ps.setBinaryStream(i+1, stream, (int) stream.getLength()); - } + + if( stream instanceof StandardEXTDTAReaderInputStream ){ + + final StandardEXTDTAReaderInputStream stdeis = + (StandardEXTDTAReaderInputStream) stream ; + + ps.setBinaryStream( i + 1, + stdeis, + (int) stdeis.getLength() ); + + } else if( stream instanceof LayerBStreamedEXTDTAReaderInputStream ) { + + ( ( EnginePreparedStatement ) ps).setBinaryStream( i + 1, + stream); + + } else if( stream == null ){ + ps.setBytes(i+1, null); + + } else { + throw new IllegalStateException(); + } if (SanityManager.DEBUG) { if (stream==null) { @@ -4538,50 +4556,68 @@ } } } else { - paramBytes = reader.getExtData(checkNullability); + final EXTDTAReaderInputStream stream = + reader.getEXTDTAReaderInputStream(checkNullability); - if ( paramBytes==null ) { - ps.setBytes(i+1, + if ( stream == null ) { + + ps.setBytes(i+1, null ); - } else { - ps.setBinaryStream(i+1, - new ByteArrayInputStream(paramBytes), - paramBytes.length); - } - if (SanityManager.DEBUG) { - if (paramBytes==null) { + if (SanityManager.DEBUG) { trace("parameter value : NULL"); - } else { + } + + } else { + + ByteArrayInputStream bais = + convertAsByteArrayInputStream( stream ); + + if (SanityManager.DEBUG) { trace("parameter value is a LOB with length:" + - paramBytes.length); + bais.available() ); } + + ps.setBinaryStream(i+1, + bais, + bais.available() ); + } + } break; case DRDAConstants.DRDA_TYPE_LOBCSBCS: case DRDAConstants.DRDA_TYPE_NLOBCSBCS: - paramBytes = reader.getExtData(checkNullability); - paramString = new String(paramBytes, stmt.ccsidSBCEncoding); - if (SanityManager.DEBUG) - trace("parameter value is: "+ paramString); - ps.setString(i+1,paramString); + + setAsCharacterStream(ps, + i, + checkNullability, + reader, + streamLOB, + stmt.ccsidSBCEncoding ); + break; case DRDAConstants.DRDA_TYPE_LOBCDBCS: case DRDAConstants.DRDA_TYPE_NLOBCDBCS: - paramBytes = reader.getExtData(checkNullability); - paramString = new String(paramBytes, stmt.ccsidDBCEncoding ); - if (SanityManager.DEBUG) - trace("parameter value is: "+ paramString); - ps.setString(i+1,paramString); + + setAsCharacterStream(ps, + i, + checkNullability, + reader, + streamLOB, + stmt.ccsidDBCEncoding); + break; case DRDAConstants.DRDA_TYPE_LOBCMIXED: case DRDAConstants.DRDA_TYPE_NLOBCMIXED: - paramBytes = reader.getExtData(checkNullability); - paramString = new String(paramBytes, stmt.ccsidMBCEncoding); - if (SanityManager.DEBUG) - trace("parameter value is: "+ paramString); - ps.setString(i+1,paramString); + + setAsCharacterStream(ps, + i, + checkNullability, + reader, + streamLOB, + stmt.ccsidMBCEncoding); + break; default: paramBytes = null; @@ -4593,7 +4629,11 @@ } catch (java.io.UnsupportedEncodingException e) { throw new SQLException (e.getMessage()); - } + + } catch( IOException e ){ + throw new SQLException ( e.getMessage() ); + + } } /** @@ -4746,7 +4786,7 @@ correlationID = reader.readDssHeader(); while (reader.moreDssData()) { - codePoint = reader.readLengthAndCodePoint(); + codePoint = reader.readLengthAndCodePoint( false ); switch(codePoint) { // optional @@ -4805,7 +4845,7 @@ while (reader.moreDssData()) { - codePoint = reader.readLengthAndCodePoint(); + codePoint = reader.readLengthAndCodePoint( false ); switch(codePoint) { @@ -5063,7 +5103,7 @@ private String parseSQLSTTDss() throws DRDAProtocolException { correlationID = reader.readDssHeader(); - int codePoint = reader.readLengthAndCodePoint(); + int codePoint = reader.readLengthAndCodePoint( false ); String strVal = parseEncodedString(); if (SanityManager.DEBUG) trace("SQL Statement = " + strVal); @@ -8089,4 +8129,80 @@ } } + + + private static ByteArrayInputStream + convertAsByteArrayInputStream( EXTDTAReaderInputStream stream ) + throws IOException { + + final int byteArrayLength = + stream instanceof StandardEXTDTAReaderInputStream ? + (int) ( ( StandardEXTDTAReaderInputStream ) stream ).getLength() : + 32;// default length + + PublicBufferOutputStream pbos = + new PublicBufferOutputStream( byteArrayLength ); + + byte[] buffer = new byte[32 * 1024]; + + int c = 0; + + while( ( c = stream.read( buffer, + 0, + buffer.length ) ) > -1 ) { + pbos.write( buffer, 0, c ); + } + + return new ByteArrayInputStream( pbos.getBuffer(), + 0, + pbos.getCount() ); + + } + + + private static class PublicBufferOutputStream extends ByteArrayOutputStream{ + + PublicBufferOutputStream(int size){ + super(size); + } + + public byte[] getBuffer(){ + return buf; + } + + public int getCount(){ + return count; + } + + } + + private static void setAsCharacterStream(PreparedStatement ps, + int i, + boolean checkNullability, + DDMReader reader, + boolean streamLOB, + String encoding) + throws DRDAProtocolException , + SQLException , + IOException { + + EnginePreparedStatement engnps = + ( EnginePreparedStatement ) ps; + + final EXTDTAReaderInputStream extdtastream = + reader.getEXTDTAReaderInputStream(checkNullability); + + final InputStream is = + streamLOB ? + (InputStream) extdtastream : + convertAsByteArrayInputStream( extdtastream ); + + final InputStreamReader streamReader = + new InputStreamReader( is, + encoding ) ; + + engnps.setCharacterStream( i + 1, + streamReader ); + } + } Modified: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/EXTDTAReaderInputStream.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/EXTDTAReaderInputStream.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/EXTDTAReaderInputStream.java (original) +++ db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/EXTDTAReaderInputStream.java Sat Dec 16 03:16:04 2006 @@ -28,149 +28,6 @@ * This class can be used to stream LOBs from Network client to the * Network server. */ -final class EXTDTAReaderInputStream extends InputStream +abstract class EXTDTAReaderInputStream extends InputStream { - /** - * Constructor - * @param reader The reader to get data from - * @exception DRDAProtocolException if thrown while initializing current - * buffer. - */ - EXTDTAReaderInputStream(final DDMReader reader) - throws DRDAProtocolException - { - super(); - this.reader = reader; - this.length = reader.getDdmLength(); - this.remainingBytes = length; - this.currentBuffer = - reader.readLOBInitStream(remainingBytes); - } - - /** - * Reads the next byte of data from the input stream. - * - *

This subclass of InputStream implements this method by reading - * the next byte from the current buffer. If there is more data, - * it will be requested a new buffer from the DDMReader. - * - * @return the next byte of data, or -1 if the end of the - * stream is reached. - * @exception IOException if an I/O error occurs. - * @see java.io.InputStream#read() - */ - public final int read() - throws IOException - { - if (remainingBytes <= 0) { - return -1; - } - int val = (currentBuffer == null) ? -1 : currentBuffer.read(); - if (val < 0) { - val = refreshCurrentBuffer(); - } - remainingBytes--; - return val; - } - - /** - * Reads up to len bytes of data from the input stream into - * an array of bytes. An attempt is made to read as many as - * len bytes, but a smaller number may be read, possibly - * zero. The number of bytes actually read is returned as an integer. - * - * This subclass implements this method by calling this method on the - * current buffer, which is an instance of ByteArrayInputStream. If the - * current buffer does not have any data, it will be requested a new - * buffer from the DDMReader. - * - * @param b the buffer into which the data is read. - * @param off the start offset in array b - * at which the data is written. - * @param len the maximum number of bytes to read. - * @return the total number of bytes read into the buffer, or - * -1 if there is no more data because the end of - * the stream has been reached. - * @exception IOException if an I/O error occurs. - * @exception NullPointerException if b is null. - * @see java.io.InputStream#read(byte[], int, int) - */ - public final int read(final byte[] b, - final int off, - final int len) - throws IOException - { - if (remainingBytes <= 0) { - return -1; - } - int val = currentBuffer.read(b, off, len); - if (val < 0) { - currentBuffer = - reader.readLOBContinuationStream(remainingBytes); - val = currentBuffer.read(b, off, len); - } - remainingBytes -= val; - return val; - } - - /** - * Returns the number of bytes that can be read (or skipped over) from - * this input stream without blocking by the next caller of a method for - * this input stream. - * - *

This subclass implements this method by calling available on - * the current buffer, which is a ByteInputStreamReader. - * - * @return the number of bytes that can be read from this input stream - * without blocking. - */ - public final int available() - { - if (remainingBytes <= 0) { - return 0; - } - return currentBuffer.available(); - } - - /** - * Return the length if this stream. The length includes data which has - * been read. - * @return length of this stream. - */ - final long getLength() - { - return length; - } - - /** - * Refresh the current buffer from the DDMReader - * @exception IOException if there is a IOException when - * refreshing the buffer from DDMReader - * @return the next byte of data, or -1 if the end of the - * stream is reached. - */ - private int refreshCurrentBuffer() - throws IOException - { - if (remainingBytes > 0) { - currentBuffer = - reader.readLOBContinuationStream(remainingBytes); - return currentBuffer.read(); - } else { - return -1; - } - } - - /** Length of stream */ - private final long length; - - /** DDMReader. Used to get more data. */ - private final DDMReader reader; - - /** Remaining bytes in stream */ - private long remainingBytes; - - /** Current data buffer */ - private ByteArrayInputStream currentBuffer; - } Added: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/LayerBStreamedEXTDTAReaderInputStream.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/LayerBStreamedEXTDTAReaderInputStream.java?view=auto&rev=487788 ============================================================================== --- db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/LayerBStreamedEXTDTAReaderInputStream.java (added) +++ db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/LayerBStreamedEXTDTAReaderInputStream.java Sat Dec 16 03:16:04 2006 @@ -0,0 +1,158 @@ +/* + Derby - Class org.apache.derby.impl.drda.LayerBStreamedEXTDTAReaderInputStream + + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. +*/ +package org.apache.derby.impl.drda; +import java.io.InputStream; +import java.io.ByteArrayInputStream; +import java.io.IOException; + +/** + * Implementation of InputStream which get EXTDTA from the DDMReader. + * This class can be used to stream LOBs from Network client to the + * Network server. + * + * Furthermore, this class is used when layer B streaming is carried out and + * expects corresponding DDMReader start layer B streaming + * when the object of this class is instantiated. + * + */ +final class LayerBStreamedEXTDTAReaderInputStream extends EXTDTAReaderInputStream +{ + /** + * Constructor + * @param reader The reader to get data from + * @exception DRDAProtocolException if thrown while initializing current + * buffer. + */ + LayerBStreamedEXTDTAReaderInputStream(final DDMReader reader) + throws DRDAProtocolException + { + super(); + this.reader = reader; + this.currentBuffer = + reader.readLOBInitStream(); + } + + /** + * Reads the next byte of data from the input stream. + * + *

This subclass of InputStream implements this method by reading + * the next byte from the current buffer. If there is more data, + * it will be requested a new buffer from the DDMReader. + * + * @return the next byte of data, or -1 if the end of the + * stream is reached. + * @exception IOException if an I/O error occurs. + * @see java.io.InputStream#read() + */ + public final int read() + throws IOException + { + int val = (currentBuffer == null) ? -1 : currentBuffer.read(); + if (val < 0) { + val = refreshCurrentBuffer(); + } + return val; + } + + /** + * Reads up to len bytes of data from the input stream into + * an array of bytes. An attempt is made to read as many as + * len bytes, but a smaller number may be read, possibly + * zero. The number of bytes actually read is returned as an integer. + * + * This subclass implements this method by calling this method on the + * current buffer, which is an instance of ByteArrayInputStream. If the + * current buffer does not have any data, it will be requested a new + * buffer from the DDMReader. + * + * @param b the buffer into which the data is read. + * @param off the start offset in array b + * at which the data is written. + * @param len the maximum number of bytes to read. + * @return the total number of bytes read into the buffer, or + * -1 if there is no more data because the end of + * the stream has been reached. + * @exception IOException if an I/O error occurs. + * @exception NullPointerException if b is null. + * @see java.io.InputStream#read(byte[], int, int) + */ + public final int read(final byte[] b, + final int off, + final int len) + throws IOException + { + int val = currentBuffer.read(b, off, len); + + if (val < 0 && + reader.doingLayerBStreaming() ) { + + currentBuffer = + reader.readLOBContinuationStream(); + val = currentBuffer.read(b, off, len); + + } + return val; + } + + /** + * Returns the number of bytes that can be read (or skipped over) from + * this input stream without blocking by the next caller of a method for + * this input stream. + * + *

This subclass implements this method by calling available on + * the current buffer, which is a ByteInputStreamReader. + * + * @return the number of bytes that can be read from this input stream + * without blocking. + */ + public final int available() + { + return currentBuffer.available(); + } + + + /** + * Refresh the current buffer from the DDMReader + * @exception IOException if there is a IOException when + * refreshing the buffer from DDMReader + * @return the next byte of data, or -1 if the end of the + * stream is reached and layer B streaming was finished. + */ + private int refreshCurrentBuffer() + throws IOException + { + + if( ! reader.doingLayerBStreaming() ) + return -1; + + currentBuffer = + reader.readLOBContinuationStream(); + return currentBuffer.read(); + } + + + /** DDMReader. Used to get more data. */ + private final DDMReader reader; + + /** Current data buffer */ + private ByteArrayInputStream currentBuffer; + +} Propchange: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/LayerBStreamedEXTDTAReaderInputStream.java ------------------------------------------------------------------------------ svn:eol-style = native Modified: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/ReEncodedInputStream.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/ReEncodedInputStream.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/ReEncodedInputStream.java (original) +++ db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/ReEncodedInputStream.java Sat Dec 16 03:16:04 2006 @@ -158,7 +158,7 @@ } - static class PublicBufferOutputStream extends ByteArrayOutputStream{ + private static class PublicBufferOutputStream extends ByteArrayOutputStream{ PublicBufferOutputStream(int size){ super(size); Added: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/StandardEXTDTAReaderInputStream.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/StandardEXTDTAReaderInputStream.java?view=auto&rev=487788 ============================================================================== --- db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/StandardEXTDTAReaderInputStream.java (added) +++ db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/StandardEXTDTAReaderInputStream.java Sat Dec 16 03:16:04 2006 @@ -0,0 +1,176 @@ +/* + Derby - Class org.apache.derby.impl.drda.StandardEXTDTAReaderInputStream + + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. +*/ +package org.apache.derby.impl.drda; +import java.io.InputStream; +import java.io.ByteArrayInputStream; +import java.io.IOException; + +/** + * Implementation of InputStream which get EXTDTA from the DDMReader. + * This class can be used to stream LOBs from Network client to the + * Network server. + */ +final class StandardEXTDTAReaderInputStream extends EXTDTAReaderInputStream +{ + /** + * Constructor + * @param reader The reader to get data from + * @exception DRDAProtocolException if thrown while initializing current + * buffer. + */ + StandardEXTDTAReaderInputStream(final DDMReader reader) + throws DRDAProtocolException + { + super(); + this.reader = reader; + this.length = reader.getDdmLength(); + this.remainingBytes = length; + this.currentBuffer = + reader.readLOBInitStream(remainingBytes); + } + + /** + * Reads the next byte of data from the input stream. + * + *

This subclass of InputStream implements this method by reading + * the next byte from the current buffer. If there is more data, + * it will be requested a new buffer from the DDMReader. + * + * @return the next byte of data, or -1 if the end of the + * stream is reached. + * @exception IOException if an I/O error occurs. + * @see java.io.InputStream#read() + */ + public final int read() + throws IOException + { + if (remainingBytes <= 0) { + return -1; + } + int val = (currentBuffer == null) ? -1 : currentBuffer.read(); + if (val < 0) { + val = refreshCurrentBuffer(); + } + remainingBytes--; + return val; + } + + /** + * Reads up to len bytes of data from the input stream into + * an array of bytes. An attempt is made to read as many as + * len bytes, but a smaller number may be read, possibly + * zero. The number of bytes actually read is returned as an integer. + * + * This subclass implements this method by calling this method on the + * current buffer, which is an instance of ByteArrayInputStream. If the + * current buffer does not have any data, it will be requested a new + * buffer from the DDMReader. + * + * @param b the buffer into which the data is read. + * @param off the start offset in array b + * at which the data is written. + * @param len the maximum number of bytes to read. + * @return the total number of bytes read into the buffer, or + * -1 if there is no more data because the end of + * the stream has been reached. + * @exception IOException if an I/O error occurs. + * @exception NullPointerException if b is null. + * @see java.io.InputStream#read(byte[], int, int) + */ + public final int read(final byte[] b, + final int off, + final int len) + throws IOException + { + if (remainingBytes <= 0) { + return -1; + } + int val = currentBuffer.read(b, off, len); + if (val < 0) { + currentBuffer = + reader.readLOBContinuationStream(remainingBytes); + val = currentBuffer.read(b, off, len); + } + remainingBytes -= val; + return val; + } + + /** + * Returns the number of bytes that can be read (or skipped over) from + * this input stream without blocking by the next caller of a method for + * this input stream. + * + *

This subclass implements this method by calling available on + * the current buffer, which is a ByteInputStreamReader. + * + * @return the number of bytes that can be read from this input stream + * without blocking. + */ + public final int available() + { + if (remainingBytes <= 0) { + return 0; + } + return currentBuffer.available(); + } + + /** + * Return the length if this stream. The length includes data which has + * been read. + * @return length of this stream. + */ + final long getLength() + { + return length; + } + + /** + * Refresh the current buffer from the DDMReader + * @exception IOException if there is a IOException when + * refreshing the buffer from DDMReader + * @return the next byte of data, or -1 if the end of the + * stream is reached. + */ + private int refreshCurrentBuffer() + throws IOException + { + if (remainingBytes > 0) { + currentBuffer = + reader.readLOBContinuationStream(remainingBytes); + return currentBuffer.read(); + } else { + return -1; + } + } + + /** Length of stream */ + private final long length; + + /** DDMReader. Used to get more data. */ + private final DDMReader reader; + + /** Remaining bytes in stream */ + private long remainingBytes; + + /** Current data buffer */ + private ByteArrayInputStream currentBuffer; + +} Propchange: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/StandardEXTDTAReaderInputStream.java ------------------------------------------------------------------------------ svn:eol-style = native Modified: db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/TestProto.java URL: http://svn.apache.org/viewvc/db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/TestProto.java?view=diff&rev=487788&r1=487787&r2=487788 ============================================================================== --- db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/TestProto.java (original) +++ db/derby/code/trunk/java/drda/org/apache/derby/impl/drda/TestProto.java Sat Dec 16 03:16:04 2006 @@ -536,7 +536,7 @@ */ private void skipDdm() throws DRDAProtocolException { - reader.readLengthAndCodePoint(); + reader.readLengthAndCodePoint( false ); reader.skipBytes(); } /** @@ -784,7 +784,7 @@ int reqVal; Vector manager = new Vector(), managerLevel = new Vector() ; reader.readReplyDss(); - int error = reader.readLengthAndCodePoint(); + int error = reader.readLengthAndCodePoint( false ); int reqCP = getCP(); if (error != reqCP) { @@ -793,7 +793,7 @@ } while (reader.moreDssData()) { - codepoint = reader.readLengthAndCodePoint(); + codepoint = reader.readLengthAndCodePoint( false ); switch (codepoint) { case CodePoint.SVRCOD: @@ -884,7 +884,7 @@ */ private void readLengthAndCodePoint() throws IOException, DRDAProtocolException { - int codepoint = reader.readLengthAndCodePoint(); + int codepoint = reader.readLengthAndCodePoint( false ); int reqCP = getCP(); if (codepoint != reqCP) cpError(codepoint, reqCP); @@ -905,7 +905,7 @@ int val = -1; do { - codepoint = reader.readLengthAndCodePoint(); + codepoint = reader.readLengthAndCodePoint( false ); switch(codepoint) { case CodePoint.SECMEC: @@ -1006,7 +1006,7 @@ throws IOException, DRDAProtocolException { reader.readReplyDss(); - int codepoint = reader.readLengthAndCodePoint(); + int codepoint = reader.readLengthAndCodePoint( false ); if (codepoint != CodePoint.SQLCARD) { fail("Expecting SQLCARD got "+ Integer.toHexString(codepoint));