modules/andes-core/broker/src/main/java/org/wso2/andes/store/rdbms/RDBMSMessageStoreImpl.java
private void storeMessage(AndesMessage message) throws AndesException {
        Connection connection = null;
        PreparedStatement storeMetadataPS = null;
        PreparedStatement storeContentPS = null;
        PreparedStatement storeExpiryMetadataPS = null;
        AndesMessageMetadata metadata = null;

        try {
            connection = getConnection();
            storeMetadataPS = connection.prepareStatement(PS_INSERT_METADATA);
            storeExpiryMetadataPS = connection.prepareStatement(PS_INSERT_EXPIRY_DATA);
            storeContentPS = connection.prepareStatement(PS_INSERT_MESSAGE_PART);

            metadata = message.getMetadata();
            storeMetadataPS.setLong(1, metadata.getMessageID());
            storeMetadataPS.setInt(2, getCachedQueueID(metadata.getStorageQueueName()));
            storeMetadataPS.setBytes(3, metadata.getMetadata());
            storeMetadataPS.execute();

            for (AndesMessagePart messagePart : message.getContentChunkList()) {
                addContentToBatch(storeContentPS, messagePart);
            }
            storeContentPS.executeBatch();
            if (metadata.isExpirationDefined()) {
                storeExpiryMetadataPS.setLong(1, metadata.getMessageID());
                storeExpiryMetadataPS.setLong(2, metadata.getExpirationTime());
                storeExpiryMetadataPS.setString(3, metadata.getStorageQueueName());
                storeExpiryMetadataPS.execute();
            }
            connection.commit();
            addToCache(message);
        } catch (AndesException e) {
            rollback(connection, RDBMSConstants.TASK_ADDING_MESSAGE);
            throw e;
        } catch (SQLException e) {
            if (metadata != null) {
                log.warn("Dropped message with metadata :: " + metadata.toString());
            }
            AndesException andesException = rdbmsStoreUtils
                    .convertSQLException("Error occurred while inserting message to queue ", e);
            if (AndesDataIntegrityViolationException.class.isInstance(andesException)) {
                log.warn("Dropped message with ID: " + message.getMetadata().getMessageID() + " since queue: " + message
                        .getMetadata().getStorageQueueName() + " does not exist", e);
            } else {
                rollback(connection, RDBMSConstants.TASK_ADDING_MESSAGE);
                throw andesException;
            }
        } finally {
            close(storeExpiryMetadataPS,RDBMSConstants.TASK_ADDING_MESSAGE);
            close(storeMetadataPS, RDBMSConstants.TASK_ADDING_MESSAGE);
            close(storeContentPS, RDBMSConstants.TASK_ADDING_MESSAGE);
            close(connection, RDBMSConstants.TASK_ADDING_MESSAGE);
        }
    }
/**
     * Store a given Andes message to the database and the cache
     *
     * @param message
     * @throws AndesException
     */
Store a given Andes message to the database and the cache
@param message
@throws AndesException
Store a given Andes message to the database and the cache @param message @throws AndesException
private void storeMessage(AndesMessage message) throws AndesException {
        Connection connection = null;
        PreparedStatement storeMetadataPS = null;
        PreparedStatement storeContentPS = null;
        PreparedStatement storeExpiryMetadataPS = null;
        AndesMessageMetadata metadata = null;
        try {
            connection = getConnection();
            storeMetadataPS = connection.prepareStatement(PS_INSERT_METADATA);
            storeExpiryMetadataPS = connection.prepareStatement(PS_INSERT_EXPIRY_DATA);
            storeContentPS = connection.prepareStatement(PS_INSERT_MESSAGE_PART);
            metadata = message.getMetadata();
            storeMetadataPS.setLong(1, metadata.getMessageID());
            storeMetadataPS.setInt(2, getCachedQueueID(metadata.getStorageQueueName()));
            storeMetadataPS.setBytes(3, metadata.getMetadata());
            storeMetadataPS.execute();
            for (AndesMessagePart messagePart : message.getContentChunkList()) {
                addContentToBatch(storeContentPS, messagePart);
            }
            storeContentPS.executeBatch();
            if (metadata.isExpirationDefined()) {
                storeExpiryMetadataPS.setLong(1, metadata.getMessageID());
                storeExpiryMetadataPS.setLong(2, metadata.getExpirationTime());
                storeExpiryMetadataPS.setString(3, metadata.getStorageQueueName());
                storeExpiryMetadataPS.execute();
            }
            connection.commit();
            addToCache(message);
        } catch (AndesException e) {
            rollback(connection, RDBMSConstants.TASK_ADDING_MESSAGE);
            throw e;
        } catch (SQLException e) {
            if (metadata != null) {
                log.warn("Dropped message with metadata :: " + metadata.toString());
            }
            AndesException andesException = rdbmsStoreUtils
                    .convertSQLException("Error occurred while inserting message to queue ", e);
            if (AndesDataIntegrityViolationException.class.isInstance(andesException)) {
                log.warn("Dropped message with ID: " + message.getMetadata().getMessageID() + " since queue: " + message
                        .getMetadata().getStorageQueueName() + " does not exist", e);
            } else {
                rollback(connection, RDBMSConstants.TASK_ADDING_MESSAGE);
                throw andesException;
            }
        } finally {
            close(storeExpiryMetadataPS,RDBMSConstants.TASK_ADDING_MESSAGE);
            close(storeMetadataPS, RDBMSConstants.TASK_ADDING_MESSAGE);
            close(storeContentPS, RDBMSConstants.TASK_ADDING_MESSAGE);
            close(connection, RDBMSConstants.TASK_ADDING_MESSAGE);
        }
    }
private void storeMessage ( AndesMessage message ) throws AndesException { Connection connection = null ; PreparedStatement storeMetadataPS = null ; PreparedStatement storeContentPS = null ; PreparedStatement storeExpiryMetadataPS = null ; AndesMessageMetadata metadata = null ; try { connection = getConnection ( ) ; storeMetadataPS = connection . prepareStatement ( PS_INSERT_METADATA ) ; storeExpiryMetadataPS = connection . prepareStatement ( PS_INSERT_EXPIRY_DATA ) ; storeContentPS = connection . prepareStatement ( PS_INSERT_MESSAGE_PART ) ; metadata = message . getMetadata ( ) ; storeMetadataPS . setLong ( 1 , metadata . getMessageID ( ) ) ; storeMetadataPS . setInt ( 2 , getCachedQueueID ( metadata . getStorageQueueName ( ) ) ) ; storeMetadataPS . setBytes ( 3 , metadata . getMetadata ( ) ) ; storeMetadataPS . execute ( ) ; for ( AndesMessagePart messagePart : message . getContentChunkList ( ) ) { addContentToBatch ( storeContentPS , messagePart ) ; } storeContentPS . executeBatch ( ) ; if ( metadata . isExpirationDefined ( ) ) { storeExpiryMetadataPS . setLong ( 1 , metadata . getMessageID ( ) ) ; storeExpiryMetadataPS . setLong ( 2 , metadata . getExpirationTime ( ) ) ; storeExpiryMetadataPS . setString ( 3 , metadata . getStorageQueueName ( ) ) ; storeExpiryMetadataPS . execute ( ) ; } connection . commit ( ) ; addToCache ( message ) ; } catch ( AndesException e ) { rollback ( connection , RDBMSConstants . TASK_ADDING_MESSAGE ) ; throw e ; } catch ( SQLException e ) { if ( metadata != null ) { log . warn ( "Dropped message with metadata :: " + metadata . toString ( ) ) ; } AndesException andesException = rdbmsStoreUtils . convertSQLException ( "Error occurred while inserting message to queue " , e ) ; if ( AndesDataIntegrityViolationException . class . isInstance ( andesException ) ) { log . warn ( "Dropped message with ID: " + message . getMetadata ( ) . getMessageID ( ) + " since queue: " + message . getMetadata ( ) . getStorageQueueName ( ) + " does not exist" , e ) ; } else { rollback ( connection , RDBMSConstants . TASK_ADDING_MESSAGE ) ; throw andesException ; } } finally { close ( storeExpiryMetadataPS , RDBMSConstants . TASK_ADDING_MESSAGE ) ; close ( storeMetadataPS , RDBMSConstants . TASK_ADDING_MESSAGE ) ; close ( storeContentPS , RDBMSConstants . TASK_ADDING_MESSAGE ) ; close ( connection , RDBMSConstants . TASK_ADDING_MESSAGE ) ; } }
Store a given Andes message to the database and the cache
 @param message
 @throws AndesException
Store a given Andes message to the database and the cache @param message @throws AndesException