modules/andes-core/broker/src/main/java/org/wso2/andes/store/rdbms/RDBMSMessageStoreImpl.java
@Override
    public void moveMetadataToDLC(long messageId, String dlcQueueName) throws AndesException {
        Connection connection = null;
        PreparedStatement metadataPS = null;
        PreparedStatement expiryDataPS = null;
        Context moveMetadataToDLCContext = MetricManager.timer(MetricsConstants.MOVE_METADATA_TO_DLC, Level.INFO)
                .start();
        //Remove the message from cache
        removeFromCache(messageId);

        Context contextWrite = MetricManager.timer(MetricsConstants.DB_WRITE, Level.INFO).start();
        try {
            connection = getConnection();
            //update the DLC queue ID in metadata table
            metadataPS = connection.prepareStatement(RDBMSConstants.PS_MOVE_METADATA_TO_DLC);
            int cachedQueueId = getCachedQueueID(dlcQueueName);
            metadataPS.setInt(1, cachedQueueId);
            metadataPS.setLong(2, messageId);
            metadataPS.execute();
            //update the DLC queue ID in the expiration table
            expiryDataPS = connection.prepareStatement(RDBMSConstants.PS_UPDATE_DLC_STATUS_IN_EXPIRY_TABLE);
            expiryDataPS.setInt(1, cachedQueueId);
            expiryDataPS.setLong(2, messageId);
            expiryDataPS.execute();
            connection.commit();
        } catch (SQLException e) {
            rollback(connection, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            throw rdbmsStoreUtils.convertSQLException("Error occurred while moving message metadata to dead letter "
                    + "channel.", e);
        } finally {
            contextWrite.stop();
            moveMetadataToDLCContext.stop();
            close(connection, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            close(metadataPS, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            close(expiryDataPS, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
        }
    }
/**
     * Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages.
     * {@inheritDoc}
     */
Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages.
Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages .
@Override
    public void moveMetadataToDLC(long messageId, String dlcQueueName) throws AndesException {
        Connection connection = null;
        PreparedStatement metadataPS = null;
        PreparedStatement expiryDataPS = null;
        Context moveMetadataToDLCContext = MetricManager.timer(MetricsConstants.MOVE_METADATA_TO_DLC, Level.INFO)
                .start();
        removeFromCache(messageId);
        Context contextWrite = MetricManager.timer(MetricsConstants.DB_WRITE, Level.INFO).start();
        try {
            connection = getConnection();
            metadataPS = connection.prepareStatement(RDBMSConstants.PS_MOVE_METADATA_TO_DLC);
            int cachedQueueId = getCachedQueueID(dlcQueueName);
            metadataPS.setInt(1, cachedQueueId);
            metadataPS.setLong(2, messageId);
            metadataPS.execute();
            expiryDataPS = connection.prepareStatement(RDBMSConstants.PS_UPDATE_DLC_STATUS_IN_EXPIRY_TABLE);
            expiryDataPS.setInt(1, cachedQueueId);
            expiryDataPS.setLong(2, messageId);
            expiryDataPS.execute();
            connection.commit();
        } catch (SQLException e) {
            rollback(connection, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            throw rdbmsStoreUtils.convertSQLException("Error occurred while moving message metadata to dead letter "
                    + "channel.", e);
        } finally {
            contextWrite.stop();
            moveMetadataToDLCContext.stop();
            close(connection, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            close(metadataPS, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
            close(expiryDataPS, RDBMSConstants.TASK_MOVING_METADATA_TO_DLC);
        }
    }
@ Override public void moveMetadataToDLC ( long messageId , String dlcQueueName ) throws AndesException { Connection connection = null ; PreparedStatement metadataPS = null ; PreparedStatement expiryDataPS = null ; Context moveMetadataToDLCContext = MetricManager . timer ( MetricsConstants . MOVE_METADATA_TO_DLC , Level . INFO ) . start ( ) ; removeFromCache ( messageId ) ; Context contextWrite = MetricManager . timer ( MetricsConstants . DB_WRITE , Level . INFO ) . start ( ) ; try { connection = getConnection ( ) ; metadataPS = connection . prepareStatement ( RDBMSConstants . PS_MOVE_METADATA_TO_DLC ) ; int cachedQueueId = getCachedQueueID ( dlcQueueName ) ; metadataPS . setInt ( 1 , cachedQueueId ) ; metadataPS . setLong ( 2 , messageId ) ; metadataPS . execute ( ) ; expiryDataPS = connection . prepareStatement ( RDBMSConstants . PS_UPDATE_DLC_STATUS_IN_EXPIRY_TABLE ) ; expiryDataPS . setInt ( 1 , cachedQueueId ) ; expiryDataPS . setLong ( 2 , messageId ) ; expiryDataPS . execute ( ) ; connection . commit ( ) ; } catch ( SQLException e ) { rollback ( connection , RDBMSConstants . TASK_MOVING_METADATA_TO_DLC ) ; throw rdbmsStoreUtils . convertSQLException ( "Error occurred while moving message metadata to dead letter " + "channel." , e ) ; } finally { contextWrite . stop ( ) ; moveMetadataToDLCContext . stop ( ) ; close ( connection , RDBMSConstants . TASK_MOVING_METADATA_TO_DLC ) ; close ( metadataPS , RDBMSConstants . TASK_MOVING_METADATA_TO_DLC ) ; close ( expiryDataPS , RDBMSConstants . TASK_MOVING_METADATA_TO_DLC ) ; } }
Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages.
Update the DLC queue id in metadata table and expiration table since need to check for expiry in DLC messages .
//Remove the message from cache //update the DLC queue ID in metadata table //update the DLC queue ID in the expiration table