You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ma...@apache.org on 2017/02/10 15:31:17 UTC
[1/3] activemq-artemis git commit: ARTEMIS-938 JDBC persistence-store
should use BIGINT type for IDs in database tables
Repository: activemq-artemis
Updated Branches:
refs/heads/1.x 7de8dd77e -> 3d765ae4c
ARTEMIS-938 JDBC persistence-store should use BIGINT type for IDs in database tables
(cherry picked from commit 807dbf90510c7dce35fd141f0cd49932d772e115)
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/d0b568c8
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/d0b568c8
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/d0b568c8
Branch: refs/heads/1.x
Commit: d0b568c895532099f19724f104dc3c543f2c3404
Parents: 7de8dd7
Author: Francesco Nigro <ni...@gmail.com>
Authored: Wed Jan 18 17:08:11 2017 +0100
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Fri Feb 10 14:37:16 2017 +0000
----------------------------------------------------------------------
.../store/drivers/derby/DerbySQLProvider.java | 2 +-
.../store/drivers/mysql/MySQLSQLProvider.java | 2 +-
.../drivers/postgres/PostgresSQLProvider.java | 2 +-
.../artemis/jdbc/store/file/JDBCFileUtils.java | 7 +++-
.../jdbc/store/file/JDBCSequentialFile.java | 6 ++--
.../file/JDBCSequentialFileFactoryDriver.java | 38 ++++++++++----------
.../PostgresSequentialSequentialFileDriver.java | 21 ++++++++---
.../jdbc/store/sql/GenericSQLProvider.java | 2 +-
8 files changed, 49 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d0b568c8/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java
index 281ea88..ab61cc2 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java
@@ -32,7 +32,7 @@ public class DerbySQLProvider extends GenericSQLProvider {
super(tableName.toUpperCase());
createFileTableSQL = "CREATE TABLE " + tableName +
- "(ID INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1)," +
+ "(ID BIGINT NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1)," +
"FILENAME VARCHAR(255), EXTENSION VARCHAR(10), DATA BLOB, PRIMARY KEY(ID))";
appendToFileSQL = "UPDATE " + tableName + " SET DATA = DATA || ? WHERE ID=?";
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d0b568c8/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/mysql/MySQLSQLProvider.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/mysql/MySQLSQLProvider.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/mysql/MySQLSQLProvider.java
index 9b0c4e6..a538ebd 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/mysql/MySQLSQLProvider.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/mysql/MySQLSQLProvider.java
@@ -33,7 +33,7 @@ public class MySQLSQLProvider extends GenericSQLProvider {
super(tName.toLowerCase());
createFileTableSQL = "CREATE TABLE " + tableName +
- "(ID INTEGER NOT NULL AUTO_INCREMENT," +
+ "(ID BIGINT NOT NULL AUTO_INCREMENT," +
"FILENAME VARCHAR(255), EXTENSION VARCHAR(10), DATA LONGBLOB, PRIMARY KEY(ID)) ENGINE=InnoDB;";
createJournalTableSQL = new String[] {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d0b568c8/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/postgres/PostgresSQLProvider.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/postgres/PostgresSQLProvider.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/postgres/PostgresSQLProvider.java
index fb0d34a..1baccae 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/postgres/PostgresSQLProvider.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/postgres/PostgresSQLProvider.java
@@ -31,7 +31,7 @@ public class PostgresSQLProvider extends GenericSQLProvider {
private PostgresSQLProvider(String tName) {
super(tName.toLowerCase());
createFileTableSQL = "CREATE TABLE " + tableName +
- "(ID SERIAL, FILENAME VARCHAR(255), EXTENSION VARCHAR(10), DATA OID, PRIMARY KEY(ID))";
+ "(ID BIGSERIAL, FILENAME VARCHAR(255), EXTENSION VARCHAR(10), DATA OID, PRIMARY KEY(ID))";
createJournalTableSQL = new String[] {
"CREATE TABLE " + tableName + "(id BIGINT,recordType SMALLINT,compactCount SMALLINT,txId BIGINT,userRecordType SMALLINT,variableSize INTEGER,record BYTEA,txDataSize INTEGER,txData BYTEA,txCheckNoRecords INTEGER,seq BIGINT)",
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d0b568c8/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCFileUtils.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCFileUtils.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCFileUtils.java
index 58494b0..b5f1aa6 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCFileUtils.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCFileUtils.java
@@ -29,7 +29,12 @@ class JDBCFileUtils {
static JDBCSequentialFileFactoryDriver getDBFileDriver(String driverClass,
String jdbcConnectionUrl,
SQLProvider provider) throws SQLException {
- JDBCSequentialFileFactoryDriver dbDriver = new JDBCSequentialFileFactoryDriver();
+ final JDBCSequentialFileFactoryDriver dbDriver;
+ if (provider instanceof PostgresSQLProvider) {
+ dbDriver = new PostgresSequentialSequentialFileDriver();
+ } else {
+ dbDriver = new JDBCSequentialFileFactoryDriver();
+ }
dbDriver.setSqlProvider(provider);
dbDriver.setJdbcConnectionUrl(jdbcConnectionUrl);
dbDriver.setJdbcDriverClass(driverClass);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d0b568c8/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFile.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFile.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFile.java
index 85c4877..7e72785 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFile.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFile.java
@@ -47,7 +47,7 @@ public class JDBCSequentialFile implements SequentialFile {
private boolean isCreated = false;
- private int id = -1;
+ private long id = -1;
private long readPosition = 0;
@@ -328,11 +328,11 @@ public class JDBCSequentialFile implements SequentialFile {
}
}
- public int getId() {
+ public long getId() {
return id;
}
- public void setId(int id) {
+ public void setId(long id) {
this.id = id;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d0b568c8/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java
index f9f206a..41ad105 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java
@@ -35,19 +35,19 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
protected PreparedStatement deleteFile;
- PreparedStatement createFile;
+ protected PreparedStatement createFile;
- private PreparedStatement selectFileByFileName;
+ protected PreparedStatement selectFileByFileName;
- private PreparedStatement copyFileRecord;
+ protected PreparedStatement copyFileRecord;
- private PreparedStatement renameFile;
+ protected PreparedStatement renameFile;
- PreparedStatement readLargeObject;
+ protected PreparedStatement readLargeObject;
- private PreparedStatement appendToLargeObject;
+ protected PreparedStatement appendToLargeObject;
- private PreparedStatement selectFileNamesByExtension;
+ protected PreparedStatement selectFileNamesByExtension;
JDBCSequentialFileFactoryDriver() {
super();
@@ -105,7 +105,7 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
* @throws SQLException
*/
public void openFile(JDBCSequentialFile file) throws SQLException {
- int fileId = fileExists(file);
+ final long fileId = fileExists(file);
if (fileId < 0) {
createFile(file);
} else {
@@ -121,13 +121,13 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
* @return
* @throws SQLException
*/
- public int fileExists(JDBCSequentialFile file) throws SQLException {
+ public long fileExists(JDBCSequentialFile file) throws SQLException {
try {
synchronized (connection) {
connection.setAutoCommit(false);
selectFileByFileName.setString(1, file.getFileName());
try (ResultSet rs = selectFileByFileName.executeQuery()) {
- int id = rs.next() ? rs.getInt(1) : -1;
+ final long id = rs.next() ? rs.getLong(1) : -1;
connection.commit();
return id;
} catch (Exception e) {
@@ -150,7 +150,7 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
public void loadFile(JDBCSequentialFile file) throws SQLException {
synchronized (connection) {
connection.setAutoCommit(false);
- readLargeObject.setInt(1, file.getId());
+ readLargeObject.setLong(1, file.getId());
try (ResultSet rs = readLargeObject.executeQuery()) {
if (rs.next()) {
@@ -180,7 +180,7 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
createFile.executeUpdate();
try (ResultSet keys = createFile.getGeneratedKeys()) {
keys.next();
- file.setId(keys.getInt(1));
+ file.setId(keys.getLong(1));
}
connection.commit();
} catch (SQLException e) {
@@ -202,7 +202,7 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
try {
connection.setAutoCommit(false);
renameFile.setString(1, newFileName);
- renameFile.setInt(2, file.getId());
+ renameFile.setLong(2, file.getId());
renameFile.executeUpdate();
connection.commit();
} catch (SQLException e) {
@@ -222,7 +222,7 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
synchronized (connection) {
try {
connection.setAutoCommit(false);
- deleteFile.setInt(1, file.getId());
+ deleteFile.setLong(1, file.getId());
deleteFile.executeUpdate();
connection.commit();
} catch (SQLException e) {
@@ -245,7 +245,7 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
try {
connection.setAutoCommit(false);
appendToLargeObject.setBytes(1, data);
- appendToLargeObject.setInt(2, file.getId());
+ appendToLargeObject.setLong(2, file.getId());
appendToLargeObject.executeUpdate();
connection.commit();
return data.length;
@@ -267,11 +267,11 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
public int readFromFile(JDBCSequentialFile file, ByteBuffer bytes) throws SQLException {
synchronized (connection) {
connection.setAutoCommit(false);
- readLargeObject.setInt(1, file.getId());
+ readLargeObject.setLong(1, file.getId());
int readLength = 0;
try (ResultSet rs = readLargeObject.executeQuery()) {
if (rs.next()) {
- Blob blob = rs.getBlob(1);
+ final Blob blob = rs.getBlob(1);
readLength = (int) calculateReadLength(blob.length(), bytes.remaining(), file.position());
byte[] data = blob.getBytes(file.position() + 1, readLength);
bytes.put(data);
@@ -296,8 +296,8 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
synchronized (connection) {
try {
connection.setAutoCommit(false);
- copyFileRecord.setInt(1, fileFrom.getId());
- copyFileRecord.setInt(2, fileTo.getId());
+ copyFileRecord.setLong(1, fileFrom.getId());
+ copyFileRecord.setLong(2, fileTo.getId());
copyFileRecord.executeUpdate();
connection.commit();
} catch (SQLException e) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d0b568c8/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/PostgresSequentialSequentialFileDriver.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/PostgresSequentialSequentialFileDriver.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/PostgresSequentialSequentialFileDriver.java
index 8c0f975..cdc167f 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/PostgresSequentialSequentialFileDriver.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/PostgresSequentialSequentialFileDriver.java
@@ -19,13 +19,14 @@ package org.apache.activemq.artemis.jdbc.store.file;
import java.nio.ByteBuffer;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
import org.postgresql.PGConnection;
import org.postgresql.largeobject.LargeObject;
import org.postgresql.largeobject.LargeObjectManager;
@SuppressWarnings("SynchronizeOnNonFinalField")
-public class PostgresSequentialSequentialFileDriver extends JDBCSequentialFileFactoryDriver {
+public final class PostgresSequentialSequentialFileDriver extends JDBCSequentialFileFactoryDriver {
private static final String POSTGRES_OID_KEY = "POSTGRES_OID_KEY";
@@ -34,6 +35,18 @@ public class PostgresSequentialSequentialFileDriver extends JDBCSequentialFileFa
}
@Override
+ protected void prepareStatements() throws SQLException {
+ this.deleteFile = connection.prepareStatement(sqlProvider.getDeleteFileSQL());
+ this.createFile = connection.prepareStatement(sqlProvider.getInsertFileSQL(), Statement.RETURN_GENERATED_KEYS);
+ this.selectFileByFileName = connection.prepareStatement(sqlProvider.getSelectFileByFileName());
+ this.copyFileRecord = connection.prepareStatement(sqlProvider.getCopyFileRecordByIdSQL());
+ this.renameFile = connection.prepareStatement(sqlProvider.getUpdateFileNameByIdSQL());
+ this.readLargeObject = connection.prepareStatement(sqlProvider.getReadLargeObjectSQL());
+ this.appendToLargeObject = connection.prepareStatement(sqlProvider.getAppendToLargeObjectSQL());
+ this.selectFileNamesByExtension = connection.prepareStatement(sqlProvider.getSelectFileNamesByExtensionSQL());
+ }
+
+ @Override
public void createFile(JDBCSequentialFile file) throws SQLException {
synchronized (connection) {
try {
@@ -49,7 +62,7 @@ public class PostgresSequentialSequentialFileDriver extends JDBCSequentialFileFa
try (ResultSet keys = createFile.getGeneratedKeys()) {
keys.next();
- file.setId(keys.getInt(1));
+ file.setId(keys.getLong(1));
}
connection.commit();
} catch (SQLException e) {
@@ -63,7 +76,7 @@ public class PostgresSequentialSequentialFileDriver extends JDBCSequentialFileFa
public void loadFile(JDBCSequentialFile file) throws SQLException {
synchronized (connection) {
connection.setAutoCommit(false);
- readLargeObject.setInt(1, file.getId());
+ readLargeObject.setLong(1, file.getId());
try (ResultSet rs = readLargeObject.executeQuery()) {
if (rs.next()) {
@@ -133,7 +146,7 @@ public class PostgresSequentialSequentialFileDriver extends JDBCSequentialFileFa
if (oid == null) {
synchronized (connection) {
connection.setAutoCommit(false);
- readLargeObject.setInt(1, file.getId());
+ readLargeObject.setLong(1, file.getId());
try (ResultSet rs = readLargeObject.executeQuery()) {
if (rs.next()) {
file.addMetaData(POSTGRES_OID_KEY, rs.getLong(1));
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/d0b568c8/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java
index a7a4b47..0527c54 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java
@@ -61,7 +61,7 @@ public class GenericSQLProvider implements SQLProvider {
this.tableName = tableName;
createFileTableSQL = "CREATE TABLE " + tableName +
- "(ID INT AUTO_INCREMENT, FILENAME VARCHAR(255), EXTENSION VARCHAR(10), DATA BLOB, PRIMARY KEY(ID))";
+ "(ID BIGINT AUTO_INCREMENT, FILENAME VARCHAR(255), EXTENSION VARCHAR(10), DATA BLOB, PRIMARY KEY(ID))";
insertFileSQL = "INSERT INTO " + tableName + " (FILENAME, EXTENSION, DATA) VALUES (?,?,?)";
[2/3] activemq-artemis git commit: ARTEMIS-917 Only return body of
retained message after reboot
Posted by ma...@apache.org.
ARTEMIS-917 Only return body of retained message after reboot
(cherry picked from commit 3900cb0ec7305ee1f341687ace6a4fab7469a817)
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/f231fe4e
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/f231fe4e
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/f231fe4e
Branch: refs/heads/1.x
Commit: f231fe4e9b4d0720c3bac19d097c1bde17badd46
Parents: d0b568c
Author: Martyn Taylor <mt...@redhat.com>
Authored: Sun Feb 5 19:20:32 2017 +0000
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Fri Feb 10 14:55:24 2017 +0000
----------------------------------------------------------------------
.../artemis/core/message/impl/MessageImpl.java | 2 ++
.../core/protocol/mqtt/MQTTPublishManager.java | 4 ++-
.../integration/mqtt/imported/MQTTTest.java | 30 ++++++++++++++++++++
.../mqtt/imported/MQTTTestSupport.java | 4 +++
4 files changed, 39 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f231fe4e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java
index 1e19817..69e6e65 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java
@@ -162,8 +162,10 @@ public abstract class MessageImpl implements MessageInternal {
buffer.setIndex(other.buffer.readerIndex(), buffer.capacity());
bodyBuffer = new ResetLimitWrappedActiveMQBuffer(BODY_OFFSET, buffer, this);
+
bodyBuffer.readerIndex(BODY_OFFSET);
bodyBuffer.writerIndex(other.getBodyBuffer().writerIndex());
+ endOfBodyPosition = other.endOfBodyPosition;
}
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f231fe4e/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
index 73a7c8e..ca926b9 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
@@ -22,6 +22,7 @@ import java.io.UnsupportedEncodingException;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.EmptyByteBuf;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.Pair;
@@ -230,7 +231,8 @@ public class MQTTPublishManager {
log.warn("Unable to send message: " + message.getMessageID() + " Cause: " + e.getMessage());
}
default:
- payload = message.getBodyBufferDuplicate().byteBuf();
+ ActiveMQBuffer bufferDup = message.getBodyBufferDuplicate();
+ payload = bufferDup.readBytes(message.getEndOfBodyPosition() - bufferDup.readerIndex()).byteBuf();
break;
}
session.getProtocolHandler().send(messageId, address, qos, payload, deliveryCount);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f231fe4e/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java
index 1d6b98d..886edd3 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTest.java
@@ -1,3 +1,4 @@
+
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
@@ -1613,4 +1614,33 @@ public class MQTTTest extends MQTTTestSupport {
connection.disconnect();
}
+
+ @Test
+ public void testRetainedMessagesAreCorrectlyFormedAfterRestart() throws Exception {
+ String clientId = "testMqtt";
+ String address = "testAddress";
+ String payload = "This is a test message";
+
+ // Send MQTT Retain Message
+ Topic[] mqttTopic = new Topic[]{new Topic(address, QoS.AT_LEAST_ONCE)};
+
+ MQTT mqtt = createMQTTConnection();
+ mqtt.setClientId(clientId);
+ BlockingConnection connection1 = mqtt.blockingConnection();
+ connection1.connect();
+ connection1.publish(address, payload.getBytes(), QoS.AT_LEAST_ONCE, true);
+
+ getServer().stop(false);
+ getServer().start();
+ waitForServerToStart(getServer());
+
+ MQTT mqtt2 = createMQTTConnection();
+ mqtt2.setClientId(clientId + "2");
+ BlockingConnection connection2 = mqtt2.blockingConnection();
+ connection2.connect();
+ connection2.subscribe(mqttTopic);
+
+ Message message = connection2.receive();
+ assertEquals(payload, new String(message.getPayload()));
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/f231fe4e/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTestSupport.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTestSupport.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTestSupport.java
index 8b85f83..476a562 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTestSupport.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTTestSupport.java
@@ -93,6 +93,10 @@ public class MQTTTestSupport extends ActiveMQTestBase {
return name.getMethodName();
}
+ public ActiveMQServer getServer() {
+ return server;
+ }
+
@Override
@Before
public void setUp() throws Exception {
[3/3] activemq-artemis git commit: ARTEMIS-957 Use setBytes JDBC API
vs Concat BLOB
Posted by ma...@apache.org.
ARTEMIS-957 Use setBytes JDBC API vs Concat BLOB
(cherry picked from commit b8595d610df77271029669a3d8b7ed1b6ef9478f)
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/3d765ae4
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/3d765ae4
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/3d765ae4
Branch: refs/heads/1.x
Commit: 3d765ae4cf1b122328ce7bfa8f8db671ad758b92
Parents: f231fe4
Author: Martyn Taylor <mt...@redhat.com>
Authored: Fri Feb 10 14:27:26 2017 +0000
Committer: Martyn Taylor <mt...@redhat.com>
Committed: Fri Feb 10 15:29:27 2017 +0000
----------------------------------------------------------------------
.../store/drivers/derby/DerbySQLProvider.java | 9 --------
.../file/JDBCSequentialFileFactoryDriver.java | 23 +++++++++++++-------
.../jdbc/store/sql/GenericSQLProvider.java | 2 +-
3 files changed, 16 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/3d765ae4/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java
index ab61cc2..7e0ce58 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/drivers/derby/DerbySQLProvider.java
@@ -26,16 +26,12 @@ public class DerbySQLProvider extends GenericSQLProvider {
private final String createFileTableSQL;
- private final String appendToFileSQL;
-
private DerbySQLProvider(String tableName) {
super(tableName.toUpperCase());
createFileTableSQL = "CREATE TABLE " + tableName +
"(ID BIGINT NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1)," +
"FILENAME VARCHAR(255), EXTENSION VARCHAR(10), DATA BLOB, PRIMARY KEY(ID))";
-
- appendToFileSQL = "UPDATE " + tableName + " SET DATA = DATA || ? WHERE ID=?";
}
@Override
@@ -49,11 +45,6 @@ public class DerbySQLProvider extends GenericSQLProvider {
}
@Override
- public String getAppendToLargeObjectSQL() {
- return appendToFileSQL;
- }
-
- @Override
public boolean closeConnectionOnShutdown() {
return false;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/3d765ae4/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java
index 41ad105..b773cd9 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFileFactoryDriver.java
@@ -74,7 +74,7 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
this.copyFileRecord = connection.prepareStatement(sqlProvider.getCopyFileRecordByIdSQL());
this.renameFile = connection.prepareStatement(sqlProvider.getUpdateFileNameByIdSQL());
this.readLargeObject = connection.prepareStatement(sqlProvider.getReadLargeObjectSQL());
- this.appendToLargeObject = connection.prepareStatement(sqlProvider.getAppendToLargeObjectSQL());
+ this.appendToLargeObject = connection.prepareStatement(sqlProvider.getAppendToLargeObjectSQL(), ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_UPDATABLE);
this.selectFileNamesByExtension = connection.prepareStatement(sqlProvider.getSelectFileNamesByExtensionSQL());
}
@@ -154,7 +154,8 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
try (ResultSet rs = readLargeObject.executeQuery()) {
if (rs.next()) {
- file.setWritePosition((int) rs.getBlob(1).length());
+ Blob blob = rs.getBlob(1);
+ file.setWritePosition((int) blob.length());
}
connection.commit();
} catch (SQLException e) {
@@ -242,13 +243,19 @@ public class JDBCSequentialFileFactoryDriver extends AbstractJDBCDriver {
*/
public int writeToFile(JDBCSequentialFile file, byte[] data) throws SQLException {
synchronized (connection) {
- try {
- connection.setAutoCommit(false);
- appendToLargeObject.setBytes(1, data);
- appendToLargeObject.setLong(2, file.getId());
- appendToLargeObject.executeUpdate();
+ connection.setAutoCommit(false);
+ appendToLargeObject.setLong(1, file.getId());
+
+ int bytesWritten = 0;
+ try (ResultSet rs = appendToLargeObject.executeQuery()) {
+ if (rs.next()) {
+ Blob blob = rs.getBlob(1);
+ bytesWritten = blob.setBytes(blob.length() + 1, data);
+ rs.updateBlob(1, blob);
+ rs.updateRow();
+ }
connection.commit();
- return data.length;
+ return bytesWritten;
} catch (SQLException e) {
connection.rollback();
throw e;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/3d765ae4/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java
----------------------------------------------------------------------
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java
index 0527c54..1d9b829 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/sql/GenericSQLProvider.java
@@ -69,7 +69,7 @@ public class GenericSQLProvider implements SQLProvider {
selectIdByFileNameSQL = "SELECT ID, FILENAME, EXTENSION, DATA FROM " + tableName + " WHERE fileName=?";
- appendToFileSQL = "UPDATE " + tableName + " SET DATA = CONCAT(DATA, ?) WHERE ID=?";
+ appendToFileSQL = "SELECT DATA FROM " + tableName + " WHERE ID=? FOR UPDATE";
readLargeObjectSQL = "SELECT DATA FROM " + tableName + " WHERE ID=?";