modules/andes-core/broker/src/main/java/org/wso2/andes/store/rdbms/RDBMSMessageStoreImpl.java
private void addRetainedMessageToUpdateBatch(PreparedStatement updateMetadataPreparedStatement,
            PreparedStatement deleteContentPreparedStatement, PreparedStatement deleteMetadataPreparedStatement,
            PreparedStatement insertContentPreparedStatement, AndesMessage message, AndesMessageMetadata metadata,
            RetainedItemData retainedItemData) throws SQLException {

        // Will set to true if payload of retained message is empty. Therefore, instead of update
        // retain topic entry will be deleted from database.
        boolean isRetainTopicSetToDelete = false;

        // If retain topic message received with empty payload that particular retain topic will be
        // deleted from broker instead of updating.
        if (!message.getContentChunkList().isEmpty()) {
            // if first chunk doesn't contain any data that payload is empty and therefore
            // retain message(metadata + content) should remove.
            if (message.getContentChunkList().get(0).getDataLength() == 0) {

                isRetainTopicSetToDelete = true;

                deleteMetadataPreparedStatement.setLong(1, retainedItemData.messageID);
                deleteMetadataPreparedStatement.addBatch();

                deleteContentPreparedStatement.setLong(1, retainedItemData.messageID);
                deleteContentPreparedStatement.addBatch();

            }
        }

        // If retain topic is not set to delete (payload is not empty) following 3 operations will occur.
        // 1. metadata will be updated with new parameters.
        // 2. old content corresponding to previous retain message will be deleted.
        // 3. new content will be created for newly arrived retained message.
        if (!isRetainTopicSetToDelete) {

            // update metadata
            updateMetadataPreparedStatement.setLong(1, metadata.getMessageID());
            updateMetadataPreparedStatement.setBytes(2, metadata.getMetadata());
            updateMetadataPreparedStatement.setInt(3, retainedItemData.topicID);
            updateMetadataPreparedStatement.addBatch();

            // update content
            deleteContentPreparedStatement.setLong(1, retainedItemData.messageID);
            deleteContentPreparedStatement.addBatch();
            for (AndesMessagePart messagePart : message.getContentChunkList()) {
                insertContentPreparedStatement.setLong(1, metadata.getMessageID());
                insertContentPreparedStatement.setInt(2, messagePart.getOffset());
                insertContentPreparedStatement.setBytes(3, messagePart.getData());
                insertContentPreparedStatement.addBatch();
            }
        }
    }
/**
     * Update batching prepared statements with current message.
     *
     * @param updateMetadataPreparedStatement update prepared statement
     * @param deleteContentPreparedStatement  delete content prepared statement
     * @param deleteMetadataPreparedStatement delete metadata prepared statement
     * @param insertContentPreparedStatement  insert prepared statement
     * @param message                         current message
     * @param metadata                        current message metadata
     * @param retainedItemData                retained item data
     * @throws SQLException
     */
Update batching prepared statements with current message.
Update batching prepared statements with current message .
private void addRetainedMessageToUpdateBatch(PreparedStatement updateMetadataPreparedStatement,
            PreparedStatement deleteContentPreparedStatement, PreparedStatement deleteMetadataPreparedStatement,
            PreparedStatement insertContentPreparedStatement, AndesMessage message, AndesMessageMetadata metadata,
            RetainedItemData retainedItemData) throws SQLException {
        boolean isRetainTopicSetToDelete = false;
        if (!message.getContentChunkList().isEmpty()) {
            if (message.getContentChunkList().get(0).getDataLength() == 0) {
                isRetainTopicSetToDelete = true;
                deleteMetadataPreparedStatement.setLong(1, retainedItemData.messageID);
                deleteMetadataPreparedStatement.addBatch();
                deleteContentPreparedStatement.setLong(1, retainedItemData.messageID);
                deleteContentPreparedStatement.addBatch();
            }
        }
        if (!isRetainTopicSetToDelete) {
            updateMetadataPreparedStatement.setLong(1, metadata.getMessageID());
            updateMetadataPreparedStatement.setBytes(2, metadata.getMetadata());
            updateMetadataPreparedStatement.setInt(3, retainedItemData.topicID);
            updateMetadataPreparedStatement.addBatch();
            deleteContentPreparedStatement.setLong(1, retainedItemData.messageID);
            deleteContentPreparedStatement.addBatch();
            for (AndesMessagePart messagePart : message.getContentChunkList()) {
                insertContentPreparedStatement.setLong(1, metadata.getMessageID());
                insertContentPreparedStatement.setInt(2, messagePart.getOffset());
                insertContentPreparedStatement.setBytes(3, messagePart.getData());
                insertContentPreparedStatement.addBatch();
            }
        }
    }
private void addRetainedMessageToUpdateBatch ( PreparedStatement updateMetadataPreparedStatement , PreparedStatement deleteContentPreparedStatement , PreparedStatement deleteMetadataPreparedStatement , PreparedStatement insertContentPreparedStatement , AndesMessage message , AndesMessageMetadata metadata , RetainedItemData retainedItemData ) throws SQLException { boolean isRetainTopicSetToDelete = false ; if ( ! message . getContentChunkList ( ) . isEmpty ( ) ) { if ( message . getContentChunkList ( ) . get ( 0 ) . getDataLength ( ) == 0 ) { isRetainTopicSetToDelete = true ; deleteMetadataPreparedStatement . setLong ( 1 , retainedItemData . messageID ) ; deleteMetadataPreparedStatement . addBatch ( ) ; deleteContentPreparedStatement . setLong ( 1 , retainedItemData . messageID ) ; deleteContentPreparedStatement . addBatch ( ) ; } } if ( ! isRetainTopicSetToDelete ) { updateMetadataPreparedStatement . setLong ( 1 , metadata . getMessageID ( ) ) ; updateMetadataPreparedStatement . setBytes ( 2 , metadata . getMetadata ( ) ) ; updateMetadataPreparedStatement . setInt ( 3 , retainedItemData . topicID ) ; updateMetadataPreparedStatement . addBatch ( ) ; deleteContentPreparedStatement . setLong ( 1 , retainedItemData . messageID ) ; deleteContentPreparedStatement . addBatch ( ) ; for ( AndesMessagePart messagePart : message . getContentChunkList ( ) ) { insertContentPreparedStatement . setLong ( 1 , metadata . getMessageID ( ) ) ; insertContentPreparedStatement . setInt ( 2 , messagePart . getOffset ( ) ) ; insertContentPreparedStatement . setBytes ( 3 , messagePart . getData ( ) ) ; insertContentPreparedStatement . addBatch ( ) ; } } }
Update batching prepared statements with current message.
Update batching prepared statements with current message .
// Will set to true if payload of retained message is empty. Therefore, instead of update // retain topic entry will be deleted from database. // If retain topic message received with empty payload that particular retain topic will be // deleted from broker instead of updating. // if first chunk doesn't contain any data that payload is empty and therefore // retain message(metadata + content) should remove. // If retain topic is not set to delete (payload is not empty) following 3 operations will occur. // 1. metadata will be updated with new parameters. // 2. old content corresponding to previous retain message will be deleted. // 3. new content will be created for newly arrived retained message. // update metadata // update content