modules/andes-core/broker/src/main/java/org/wso2/andes/store/rdbms/RDBMSMessageStoreImpl.java
private RetainedItemData createRetainedEntry(Connection connection, AndesMessage message) throws SQLException {
        PreparedStatement preparedStatementForContent = null;
        PreparedStatement preparedStatementForMetadata = null;

        AndesMessageMetadata metadata = message.getMetadata();
        String destination = metadata.getDestination();
        Integer topicID = destination.hashCode();
        long messageID = metadata.getMessageID();
        Context contextWrite = MetricManager.timer(MetricsConstants.DB_WRITE, Level.INFO).start();
        try {
            // create metadata entry
            preparedStatementForMetadata = connection.prepareStatement(RDBMSConstants.PS_INSERT_RETAINED_METADATA);
            preparedStatementForMetadata.setInt(1, topicID);
            preparedStatementForMetadata.setString(2, destination);
            preparedStatementForMetadata.setLong(3, messageID);
            preparedStatementForMetadata.setBytes(4, metadata.getMetadata());
            preparedStatementForMetadata.addBatch();

            // create content
            preparedStatementForContent = connection.prepareStatement(RDBMSConstants.PS_INSERT_RETAIN_MESSAGE_PART);
            for (AndesMessagePart messagePart : message.getContentChunkList()) {
                preparedStatementForContent.setLong(1, messageID);
                preparedStatementForContent.setInt(2, messagePart.getOffset());
                preparedStatementForContent.setBytes(3, messagePart.getData());
                preparedStatementForContent.addBatch();
            }

            preparedStatementForMetadata.executeBatch();
            preparedStatementForContent.executeBatch();
            connection.commit();

            return new RetainedItemData(topicID, messageID);
        } finally {
            contextWrite.stop();
            close(preparedStatementForContent, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(preparedStatementForMetadata, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
        }
    }
/**
     * Create a new entry for retained message.
     *
     * @param connection database connection
     * @param message    retained message
     * @return Retained item data
     * @throws SQLException
     */
Create a new entry for retained message.
@param connection database connection
@param message    retained message
@return Retained item data
@throws SQLException
Create a new entry for retained message . @param connection database connection @param message retained message @return Retained item data @throws SQLException
private RetainedItemData createRetainedEntry(Connection connection, AndesMessage message) throws SQLException {
        PreparedStatement preparedStatementForContent = null;
        PreparedStatement preparedStatementForMetadata = null;
        AndesMessageMetadata metadata = message.getMetadata();
        String destination = metadata.getDestination();
        Integer topicID = destination.hashCode();
        long messageID = metadata.getMessageID();
        Context contextWrite = MetricManager.timer(MetricsConstants.DB_WRITE, Level.INFO).start();
        try {
            preparedStatementForMetadata = connection.prepareStatement(RDBMSConstants.PS_INSERT_RETAINED_METADATA);
            preparedStatementForMetadata.setInt(1, topicID);
            preparedStatementForMetadata.setString(2, destination);
            preparedStatementForMetadata.setLong(3, messageID);
            preparedStatementForMetadata.setBytes(4, metadata.getMetadata());
            preparedStatementForMetadata.addBatch();
            preparedStatementForContent = connection.prepareStatement(RDBMSConstants.PS_INSERT_RETAIN_MESSAGE_PART);
            for (AndesMessagePart messagePart : message.getContentChunkList()) {
                preparedStatementForContent.setLong(1, messageID);
                preparedStatementForContent.setInt(2, messagePart.getOffset());
                preparedStatementForContent.setBytes(3, messagePart.getData());
                preparedStatementForContent.addBatch();
            }
            preparedStatementForMetadata.executeBatch();
            preparedStatementForContent.executeBatch();
            connection.commit();
            return new RetainedItemData(topicID, messageID);
        } finally {
            contextWrite.stop();
            close(preparedStatementForContent, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
            close(preparedStatementForMetadata, RDBMSConstants.TASK_STORING_RETAINED_MESSAGE);
        }
    }
private RetainedItemData createRetainedEntry ( Connection connection , AndesMessage message ) throws SQLException { PreparedStatement preparedStatementForContent = null ; PreparedStatement preparedStatementForMetadata = null ; AndesMessageMetadata metadata = message . getMetadata ( ) ; String destination = metadata . getDestination ( ) ; Integer topicID = destination . hashCode ( ) ; long messageID = metadata . getMessageID ( ) ; Context contextWrite = MetricManager . timer ( MetricsConstants . DB_WRITE , Level . INFO ) . start ( ) ; try { preparedStatementForMetadata = connection . prepareStatement ( RDBMSConstants . PS_INSERT_RETAINED_METADATA ) ; preparedStatementForMetadata . setInt ( 1 , topicID ) ; preparedStatementForMetadata . setString ( 2 , destination ) ; preparedStatementForMetadata . setLong ( 3 , messageID ) ; preparedStatementForMetadata . setBytes ( 4 , metadata . getMetadata ( ) ) ; preparedStatementForMetadata . addBatch ( ) ; preparedStatementForContent = connection . prepareStatement ( RDBMSConstants . PS_INSERT_RETAIN_MESSAGE_PART ) ; for ( AndesMessagePart messagePart : message . getContentChunkList ( ) ) { preparedStatementForContent . setLong ( 1 , messageID ) ; preparedStatementForContent . setInt ( 2 , messagePart . getOffset ( ) ) ; preparedStatementForContent . setBytes ( 3 , messagePart . getData ( ) ) ; preparedStatementForContent . addBatch ( ) ; } preparedStatementForMetadata . executeBatch ( ) ; preparedStatementForContent . executeBatch ( ) ; connection . commit ( ) ; return new RetainedItemData ( topicID , messageID ) ; } finally { contextWrite . stop ( ) ; close ( preparedStatementForContent , RDBMSConstants . TASK_STORING_RETAINED_MESSAGE ) ; close ( preparedStatementForMetadata , RDBMSConstants . TASK_STORING_RETAINED_MESSAGE ) ; } }
Create a new entry for retained message.
Create a new entry for retained message .
// create metadata entry // create content