You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by gt...@apache.org on 2011/09/21 14:45:49 UTC
svn commit: r1173605 - in
/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter:
BlobJDBCAdapter.java OracleJDBCAdapter.java
Author: gtully
Date: Wed Sep 21 12:45:48 2011
New Revision: 1173605
URL: http://svn.apache.org/viewvc?rev=1173605&view=rev
Log:
https://issues.apache.org/jira/browse/AMQ-3289 - updates to Blob/Oracle JDBCAdapter - applied patch from William McDonald with thanks
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/BlobJDBCAdapter.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/OracleJDBCAdapter.java
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/BlobJDBCAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/BlobJDBCAdapter.java?rev=1173605&r1=1173604&r2=1173605&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/BlobJDBCAdapter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/BlobJDBCAdapter.java Wed Sep 21 12:45:48 2011
@@ -26,6 +26,7 @@ import java.sql.ResultSet;
import java.sql.SQLException;
import javax.jms.JMSException;
+import javax.sql.rowset.serial.SerialBlob;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.MessageId;
@@ -58,7 +59,6 @@ public class BlobJDBCAdapter extends Def
ResultSet rs = null;
cleanupExclusiveLock.readLock().lock();
try {
-
// Add the Blob record.
s = c.getConnection().prepareStatement(statements.getAddMessageStatement());
s.setLong(1, sequence);
@@ -67,7 +67,6 @@ public class BlobJDBCAdapter extends Def
s.setString(4, destination.getQualifiedName());
s.setLong(5, expiration);
s.setLong(6, priority);
- s.setString(7, " ");
if (s.executeUpdate() != 1) {
throw new IOException("Failed to add broker message: " + messageID + " in container.");
@@ -75,7 +74,8 @@ public class BlobJDBCAdapter extends Def
s.close();
// Select the blob record so that we can update it.
- s = c.getConnection().prepareStatement(statements.getFindMessageByIdStatement());
+ s = c.getConnection().prepareStatement(statements.getFindMessageByIdStatement(),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_UPDATABLE);
s.setLong(1, sequence);
rs = s.executeQuery();
if (!rs.next()) {
@@ -84,15 +84,10 @@ public class BlobJDBCAdapter extends Def
// Update the blob
Blob blob = rs.getBlob(1);
- OutputStream stream = blob.setBinaryStream(data.length);
- stream.write(data);
- stream.close();
- s.close();
-
- // Update the row with the updated blob
- s = c.getConnection().prepareStatement(statements.getUpdateMessageStatement());
- s.setBlob(1, blob);
- s.setLong(2, sequence);
+ blob.truncate(0);
+ blob.setBytes(1, data);
+ rs.updateBlob(1, blob);
+ rs.updateRow(); // Update the row with the updated blob
} finally {
cleanupExclusiveLock.readLock().unlock();
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/OracleJDBCAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/OracleJDBCAdapter.java?rev=1173605&r1=1173604&r2=1173605&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/OracleJDBCAdapter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/OracleJDBCAdapter.java Wed Sep 21 12:45:48 2011
@@ -48,6 +48,16 @@ public class OracleJDBCAdapter extends B
public void setStatements(Statements statements) {
statements.setLongDataType("NUMBER");
statements.setSequenceDataType("NUMBER");
+
+ String addMessageStatement = "INSERT INTO "
+ + statements.getFullMessageTableName()
+ + "(ID, MSGID_PROD, MSGID_SEQ, CONTAINER, EXPIRATION, PRIORITY, MSG) VALUES (?, ?, ?, ?, ?, ?, empty_blob())";
+ statements.setAddMessageStatement(addMessageStatement);
+
+ String findMessageByIdStatement = "SELECT MSG FROM " +
+ statements.getFullMessageTableName() + " WHERE ID=? FOR UPDATE";
+ statements.setFindMessageByIdStatement(findMessageByIdStatement);
+
super.setStatements(statements);
}