@Override
public void commit(final Transaction tx, boolean syncCommit)
{
    try
    {
        // Using commit() instead of commitNoSync() for the HA store to allow
        // the HA durability configuration to influence resulting behaviour.
        tx.commit(_realMessageStoreDurability);
    }
    catch (DatabaseException de)
    {
        throw handleDatabaseException("Got DatabaseException on commit, closing environment", de);
    }

    if (_coalescingCommiter != null && _realMessageStoreDurability.getLocalSync() == SyncPolicy.NO_SYNC
            && _messageStoreDurability.getLocalSync() == SyncPolicy.SYNC)
    {
        _coalescingCommiter.commit(tx, syncCommit);
    }

}
 

@Override
public <X> ListenableFuture<X> commitAsync(final Transaction tx, final X val)
{
    try
    {
        // Using commit() instead of commitNoSync() for the HA store to allow
        // the HA durability configuration to influence resulting behaviour.
        tx.commit(_realMessageStoreDurability);
    }
    catch (DatabaseException de)
    {
        throw handleDatabaseException("Got DatabaseException on commit, closing environment", de);
    }

    if (_coalescingCommiter != null && _realMessageStoreDurability.getLocalSync() == SyncPolicy.NO_SYNC
        && _messageStoreDurability.getLocalSync() == SyncPolicy.SYNC)
    {
        return _coalescingCommiter.commitAsync(tx, val);
    }
    return Futures.immediateFuture(val);
}
 

/**
 * This method should only be invoked from configuration thread on virtual host activation.
 * Otherwise, invocation of this method whilst coalescing committer is committing transactions might result in transaction aborts.
 */
public void setMessageStoreDurability(SyncPolicy localTransactionSynchronizationPolicy, SyncPolicy remoteTransactionSynchronizationPolicy, ReplicaAckPolicy replicaAcknowledgmentPolicy)
{
    if (_messageStoreDurability == null || localTransactionSynchronizationPolicy != _messageStoreDurability.getLocalSync()
            || remoteTransactionSynchronizationPolicy != _messageStoreDurability.getReplicaSync()
            || replicaAcknowledgmentPolicy != _messageStoreDurability.getReplicaAck())
    {
        _messageStoreDurability = new Durability(localTransactionSynchronizationPolicy, remoteTransactionSynchronizationPolicy, replicaAcknowledgmentPolicy);

        if (_coalescingCommiter != null)
        {
            _coalescingCommiter.stop();
            _coalescingCommiter = null;
        }

        if (localTransactionSynchronizationPolicy == LOCAL_TRANSACTION_SYNCHRONIZATION_POLICY)
        {
            localTransactionSynchronizationPolicy = SyncPolicy.NO_SYNC;
            _coalescingCommiter = new CoalescingCommiter(_configuration.getGroupName(), this);
            _coalescingCommiter.start();
        }
        _realMessageStoreDurability = new Durability(localTransactionSynchronizationPolicy, remoteTransactionSynchronizationPolicy, replicaAcknowledgmentPolicy);
    }
}
 

@Override
public void onOpen()
{
    ReplicatedEnvironmentFacade facade = getReplicatedEnvironmentFacade();
    if (facade != null)
    {
        facade.setMessageStoreDurability(
                SyncPolicy.valueOf(getLocalTransactionSynchronizationPolicy()),
                SyncPolicy.valueOf(getRemoteTransactionSynchronizationPolicy()),
                ReplicatedEnvironmentFacade.REPLICA_REPLICA_ACKNOWLEDGMENT_POLICY);
    }
    super.onOpen();
}
 

private void validateTransactionSynchronizationPolicy(String policy)
{
    try
    {
        SyncPolicy.valueOf(policy);
    }
    catch(Exception e)
    {
        throw new IllegalArgumentException("Invalid transaction synchronization policy '" + policy + "'. " + e.getMessage());
    }
}
 

@Override
public boolean isCoalescingSync()
{
    return SyncPolicy.SYNC.name().equals(_localTransactionSynchronizationPolicy);
}