/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package org.apache.directory.server.ldap.replication.consumer;


import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections.map.LRUMap;
import org.apache.directory.api.ldap.codec.controls.manageDsaIT.ManageDsaITDecorator;
import org.apache.directory.api.ldap.extras.controls.SyncDoneValue;
import org.apache.directory.api.ldap.extras.controls.SyncInfoValue;
import org.apache.directory.api.ldap.extras.controls.SyncModifyDnType;
import org.apache.directory.api.ldap.extras.controls.SyncRequestValue;
import org.apache.directory.api.ldap.extras.controls.SyncStateTypeEnum;
import org.apache.directory.api.ldap.extras.controls.SyncStateValue;
import org.apache.directory.api.ldap.extras.controls.SynchronizationModeEnum;
import org.apache.directory.api.ldap.extras.controls.syncrepl_impl.SyncInfoValueDecorator;
import org.apache.directory.api.ldap.extras.controls.syncrepl_impl.SyncRequestValueDecorator;
import org.apache.directory.api.ldap.model.constants.Loggers;
import org.apache.directory.api.ldap.model.constants.SchemaConstants;
import org.apache.directory.api.ldap.model.csn.Csn;
import org.apache.directory.api.ldap.model.entry.Attribute;
import org.apache.directory.api.ldap.model.entry.DefaultAttribute;
import org.apache.directory.api.ldap.model.entry.DefaultEntry;
import org.apache.directory.api.ldap.model.entry.DefaultModification;
import org.apache.directory.api.ldap.model.entry.Entry;
import org.apache.directory.api.ldap.model.entry.Modification;
import org.apache.directory.api.ldap.model.entry.ModificationOperation;
import org.apache.directory.api.ldap.model.exception.LdapException;
import org.apache.directory.api.ldap.model.exception.LdapNoSuchObjectException;
import org.apache.directory.api.ldap.model.filter.AndNode;
import org.apache.directory.api.ldap.model.filter.EqualityNode;
import org.apache.directory.api.ldap.model.filter.ExprNode;
import org.apache.directory.api.ldap.model.filter.NotNode;
import org.apache.directory.api.ldap.model.filter.OrNode;
import org.apache.directory.api.ldap.model.filter.PresenceNode;
import org.apache.directory.api.ldap.model.message.AliasDerefMode;
import org.apache.directory.api.ldap.model.message.IntermediateResponse;
import org.apache.directory.api.ldap.model.message.Response;
import org.apache.directory.api.ldap.model.message.ResultCodeEnum;
import org.apache.directory.api.ldap.model.message.SearchRequest;
import org.apache.directory.api.ldap.model.message.SearchRequestImpl;
import org.apache.directory.api.ldap.model.message.SearchResultDone;
import org.apache.directory.api.ldap.model.message.SearchResultEntry;
import org.apache.directory.api.ldap.model.message.SearchResultReference;
import org.apache.directory.api.ldap.model.message.SearchScope;
import org.apache.directory.api.ldap.model.message.controls.ManageDsaITImpl;
import org.apache.directory.api.ldap.model.name.Dn;
import org.apache.directory.api.ldap.model.name.Rdn;
import org.apache.directory.api.ldap.model.schema.AttributeType;
import org.apache.directory.api.ldap.model.schema.SchemaManager;
import org.apache.directory.api.util.StringConstants;
import org.apache.directory.api.util.Strings;
import org.apache.directory.ldap.client.api.ConnectionClosedEventListener;
import org.apache.directory.ldap.client.api.LdapNetworkConnection;
import org.apache.directory.ldap.client.api.future.SearchFuture;
import org.apache.directory.server.core.api.CoreSession;
import org.apache.directory.server.core.api.DirectoryService;
import org.apache.directory.server.core.api.OperationManager;
import org.apache.directory.server.core.api.filtering.EntryFilteringCursor;
import org.apache.directory.server.core.api.interceptor.context.AddOperationContext;
import org.apache.directory.server.core.api.interceptor.context.LookupOperationContext;
import org.apache.directory.server.core.api.interceptor.context.ModifyOperationContext;
import org.apache.directory.server.core.api.interceptor.context.MoveAndRenameOperationContext;
import org.apache.directory.server.core.api.interceptor.context.MoveOperationContext;
import org.apache.directory.server.core.api.interceptor.context.RenameOperationContext;
import org.apache.directory.server.ldap.LdapProtocolUtils;
import org.apache.directory.server.ldap.replication.ReplicationConsumerConfig;
import org.apache.directory.server.ldap.replication.SyncReplConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;


/**
 * Implementation of syncrepl slave a.k.a consumer.
 *
 * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
 */
public class ReplicationConsumerImpl implements ConnectionClosedEventListener, ReplicationConsumer
{
    /** the logger */
    private static final Logger LOG = LoggerFactory.getLogger( ReplicationConsumerImpl.class );

    /** A dedicated logger for the consumer */
    private static final Logger CONSUMER_LOG = LoggerFactory.getLogger( Loggers.CONSUMER_LOG.getName() );

    /** the syncrepl configuration */
    private SyncReplConfiguration config;

    /** the sync cookie sent by the server */
    private byte[] syncCookie;

    /** connection to the syncrepl provider */
    private LdapNetworkConnection connection;

    /** the search request with control */
    private SearchRequest searchRequest;

    /** a reference to the directoryService */
    private DirectoryService directoryService;

    /** the schema manager */
    private SchemaManager schemaManager;

    /** flag to indicate whether the consumer was disconnected */
    private volatile boolean disconnected;

    /** the core session */
    private CoreSession session;

    /** attributes on which modification should be ignored */
    private static final String[] MOD_IGNORE_AT = new String[]
        {
            SchemaConstants.ENTRY_UUID_AT,
            SchemaConstants.ENTRY_DN_AT,
            SchemaConstants.CREATE_TIMESTAMP_AT,
            SchemaConstants.CREATORS_NAME_AT,
            SchemaConstants.ENTRY_PARENT_ID_AT,
            SchemaConstants.COLLECTIVE_ATTRIBUTE_SUBENTRIES_AT
    };

    /** the cookie that was saved last time */
    private byte[] lastSavedCookie;

    /** The (entrtyUuid=*) filter */
    private static final PresenceNode ENTRY_UUID_PRESENCE_FILTER = new PresenceNode( SchemaConstants.ENTRY_UUID_AT );

    private Modification cookieMod;

    private Modification ridMod;

    /** AttributeTypes used for replication */
    private static AttributeType REPL_COOKIE_AT;
    private static AttributeType ENTRY_UUID_AT;
    private static AttributeType RID_AT_TYPE;

    private static final Map<String, Object> uuidLockMap = new LRUMap( 1000 );


    /**
     * @return the config
     */
    public SyncReplConfiguration getConfig()
    {
        return config;
    }


    /**
     * Init the replication service
     * @param directoryservice The directory service
     */
    public void init( DirectoryService directoryservice ) throws Exception
    {
        this.directoryService = directoryservice;

        session = directoryService.getAdminSession();

        schemaManager = directoryservice.getSchemaManager();

        ENTRY_UUID_AT = schemaManager.lookupAttributeTypeRegistry( SchemaConstants.ENTRY_UUID_AT );
        REPL_COOKIE_AT = schemaManager.lookupAttributeTypeRegistry( SchemaConstants.ADS_REPL_COOKIE );
        RID_AT_TYPE = schemaManager.lookupAttributeTypeRegistry( SchemaConstants.ADS_DS_REPLICA_ID );

        Attribute cookieAttr = new DefaultAttribute( REPL_COOKIE_AT );
        cookieMod = new DefaultModification( ModificationOperation.REPLACE_ATTRIBUTE, cookieAttr );

        Attribute ridAttr = new DefaultAttribute( RID_AT_TYPE );
        ridMod = new DefaultModification( ModificationOperation.REPLACE_ATTRIBUTE, ridAttr );

        prepareSyncSearchRequest();
    }


    /**
     * Connect to the remote server. Note that a SyncRepl consumer will be connected to only
     * one remote server
     * 
     * @return true if the connections have been successful. 
     */
    public boolean connect()
    {
        String providerHost = config.getRemoteHost();
        int port = config.getRemotePort();

        try
        {
            // Create a connection
            if ( connection == null )
            {
                connection = new LdapNetworkConnection( providerHost, port );
                connection.setTimeOut( -1L );

                if ( config.isUseTls() )
                {
                    connection.getConfig().setTrustManagers( config.getTrustManager() );
                    connection.startTls();
                }

                connection.addConnectionClosedEventListener( this );
            }

            // Try to connect
            if ( connection.connect() )
            {
                CONSUMER_LOG.info( "Consumer {} connected to producer {}", config.getReplicaId(), config.getProducer() );

                // Do a bind
                try
                {
                    connection.bind( config.getReplUserDn(), Strings.utf8ToString( config.getReplUserPassword() ) );
                    disconnected = false;

                    return true;
                }
                catch ( LdapException le )
                {
                    CONSUMER_LOG.warn( "Failed to bind to the producer {} with the given bind Dn {}",
                        config.getProducer(), config.getReplUserDn() );
                    LOG.warn( "Failed to bind to the server with the given bind Dn {}", config.getReplUserDn() );
                    LOG.warn( "", le );
                    disconnected = true;
                }
            }
            else
            {
                CONSUMER_LOG.warn( "Consumer {} cannot connect to producer {}", config.getReplicaId(),
                    config.getProducer() );
                disconnected = true;

                return false;
            }
        }
        catch ( Exception e )
        {
            CONSUMER_LOG.error( "Failed to connect to the server {}, cause : {}", config.getProducer(), e.getMessage() );
            LOG.error( "Failed to connect to the server {}, cause : {}", config.getProducer(), e.getMessage() );
            disconnected = true;
        }

        return false;
    }


    /**
     *  prepares a SearchRequest for syncing DIT content.
     *
     */
    private void prepareSyncSearchRequest() throws LdapException
    {
        String baseDn = config.getBaseDn();

        searchRequest = new SearchRequestImpl();

        searchRequest.setBase( new Dn( baseDn ) );
        searchRequest.setFilter( config.getFilter() );
        searchRequest.setSizeLimit( config.getSearchSizeLimit() );
        searchRequest.setTimeLimit( config.getSearchTimeout() );

        searchRequest.setDerefAliases( config.getAliasDerefMode() );
        searchRequest.setScope( config.getSearchScope() );
        searchRequest.setTypesOnly( false );

        searchRequest.addAttributes( config.getAttributes() );

        if ( !config.isChaseReferrals() )
        {
            searchRequest.addControl( new ManageDsaITDecorator( directoryService.getLdapCodecService(),
                new ManageDsaITImpl() ) );
        }

        if ( CONSUMER_LOG.isDebugEnabled() )
        {
            MDC.put( "Replica", Integer.toString( config.getReplicaId() ) );
            CONSUMER_LOG.debug( "Configuring consumer {}", config );
        }
    }


    private ResultCodeEnum handleSearchResultDone( SearchResultDone searchDone )
    {
        LOG.debug( "///////////////// handleSearchDone //////////////////" );

        SyncDoneValue ctrl = ( SyncDoneValue ) searchDone.getControls().get( SyncDoneValue.OID );

        if ( ( ctrl != null ) && ( ctrl.getCookie() != null ) )
        {
            syncCookie = ctrl.getCookie();
            LOG.debug( "assigning cookie from sync done value control: " + Strings.utf8ToString( syncCookie ) );
            storeCookie();
        }

        LOG.debug( "//////////////// END handleSearchDone//////////////////////" );

        return searchDone.getLdapResult().getResultCode();
    }


    private void handleSearchReference( SearchResultReference searchRef )
    {
        // this method won't be called cause the provider will serve the referrals as
        // normal entry objects due to the usage of ManageDsaITControl in the search request
    }


    /**
     * Process a SearchResultEntry received from a consumer. We have to handle all the 
     * cases :
     * - Add
     * - Modify
     * - Moddn
     * - Delete
     * - Present
     * @param syncResult
     */
    private void handleSearchResultEntry( SearchResultEntry syncResult )
    {
        LOG.debug( "------------- starting handleSearchResult ------------" );

        SyncStateValue syncStateCtrl = ( SyncStateValue ) syncResult.getControl( SyncStateValue.OID );

        try
        {
            Entry remoteEntry = new DefaultEntry( schemaManager, syncResult.getEntry() );
            String uuid = remoteEntry.get( ENTRY_UUID_AT ).getString();
            // lock on UUID to serialize the updates when there are multiple consumers
            // connected to several producers and to the *same* base/partition
            Object lock = getLockFor( uuid );

            synchronized ( lock )
            {
                int rid = -1;

                if ( syncStateCtrl.getCookie() != null )
                {
                    syncCookie = syncStateCtrl.getCookie();
                    rid = LdapProtocolUtils.getReplicaId( Strings.utf8ToString( syncCookie ) );
                    LOG.debug( "assigning the cookie from sync state value control: {}",
                        Strings.utf8ToString( syncCookie ) );
                }

                SyncStateTypeEnum state = syncStateCtrl.getSyncStateType();

                LOG.debug( "state name {}", state.name() );

                // check to avoid conversion of UUID from byte[] to String
                if ( LOG.isDebugEnabled() )
                {
                    LOG.debug( "entryUUID = {}", Strings.uuidToString( syncStateCtrl.getEntryUUID() ) );
                }

                Dn remoteDn = remoteEntry.getDn();

                switch ( state )
                {
                    case ADD:
                        boolean remoteDnExist = false;

                        try
                        {
                            remoteDnExist = session.exists( remoteDn );
                        }
                        catch ( LdapNoSuchObjectException lnsoe )
                        {
                            CONSUMER_LOG.error( lnsoe.getMessage() );
                        }

                        if ( !remoteDnExist )
                        {
                            LOG.debug( "adding entry with dn {}", remoteDn );
                            LOG.debug( remoteEntry.toString() );
                            AddOperationContext addContext = new AddOperationContext( session, remoteEntry );
                            addContext.setReplEvent( true );
                            addContext.setRid( rid );

                            OperationManager operationManager = directoryService.getOperationManager();
                            operationManager.add( addContext );
                        }
                        else
                        {
                            LOG.debug( "updating entry in refreshOnly mode {}", remoteDn );
                            modify( remoteEntry, rid );
                        }

                        break;

                    case MODIFY:
                        LOG.debug( "modifying entry with dn {}", remoteEntry.getDn().getName() );
                        modify( remoteEntry, rid );

                        break;

                    case MODDN:
                        String entryUuid = Strings.uuidToString( syncStateCtrl.getEntryUUID() ).toString();
                        applyModDnOperation( remoteEntry, entryUuid, rid );

                        break;

                    case DELETE:
                        LOG.debug( "deleting entry with dn {}", remoteEntry.getDn().getName() );

                        if ( !session.exists( remoteDn ) )
                        {
                            LOG.debug(
                                "looks like entry {} was already deleted in a prior update (possibly from another provider), skipping delete",
                                remoteDn );
                        }
                        else
                        {
                            // incase of a MODDN operation resulting in a branch to be moved out of scope
                            // ApacheDS replication provider sends a single delete event on the Dn of the moved branch
                            // so the branch needs to be recursively deleted here
                            deleteRecursive( remoteEntry.getDn(), null );
                        }

                        break;

                    case PRESENT:
                        LOG.debug( "entry present {}", remoteEntry );
                        break;
                }

                // store the cookie only if the above operation was successful
                if ( syncStateCtrl.getCookie() != null )
                {
                    storeCookie();
                }
            }
        }
        catch ( Exception e )
        {
            LOG.error( e.getMessage(), e );
        }

        LOG.debug( "------------- Ending handleSearchResult ------------" );
    }


    /**
     * {@inheritDoc}
     */
    private void handleSyncInfo( IntermediateResponse syncInfoResp )
    {
        try
        {
            LOG.debug( "............... inside handleSyncInfo ..............." );

            byte[] syncInfoBytes = syncInfoResp.getResponseValue();

            if ( syncInfoBytes == null )
            {
                return;
            }

            SyncInfoValueDecorator decorator = new SyncInfoValueDecorator( directoryService.getLdapCodecService() );
            SyncInfoValue syncInfoValue = ( SyncInfoValue ) decorator.decode( syncInfoBytes );

            byte[] cookie = syncInfoValue.getCookie();

            if ( CONSUMER_LOG.isDebugEnabled() )
            {
                CONSUMER_LOG.debug( "Received a SyncInfoValue from producer {} : {}", config.getProducer(),
                    syncInfoValue );
            }

            int replicaId = -1;

            if ( cookie != null )
            {
                LOG.debug( "setting the cookie from the sync info: " + Strings.utf8ToString( cookie ) );
                CONSUMER_LOG.debug( "setting the cookie from the sync info: " + Strings.utf8ToString( cookie ) );
                syncCookie = cookie;

                String cookieString = Strings.utf8ToString( syncCookie );
                replicaId = LdapProtocolUtils.getReplicaId( cookieString );
            }

            LOG.info( "refreshDeletes: " + syncInfoValue.isRefreshDeletes() );

            List<byte[]> uuidList = syncInfoValue.getSyncUUIDs();

            // if refreshDeletes set to true then delete all the entries with entryUUID
            // present in the syncIdSet
            if ( syncInfoValue.isRefreshDeletes() )
            {
                deleteEntries( uuidList, false, replicaId );
            }
            else
            {
                deleteEntries( uuidList, true, replicaId );
            }

            LOG.info( "refreshDone: " + syncInfoValue.isRefreshDone() );

            storeCookie();
        }
        catch ( Exception de )
        {
            LOG.error( "Failed to handle syncinfo message", de );
            CONSUMER_LOG.error( "Failed to handle syncinfo message", de );
        }

        LOG.debug( ".................... END handleSyncInfo ..............." );
    }


    /**
     * {@inheritDoc}
     */
    public void connectionClosed()
    {
        if ( CONSUMER_LOG.isDebugEnabled() )
        {
            MDC.put( "Replica", Integer.toString( config.getReplicaId() ) );
            CONSUMER_LOG.debug( "Consumer {} session with {} has been closed ", config.getReplicaId(),
                config.getProducer() );
        }

        disconnect();
    }


    /**
     * Starts the synchronization operation
     */
    public ReplicationStatusEnum startSync()
    {
        CONSUMER_LOG.debug( "Starting the SyncRepl process for consumer {}", config.getReplicaId() );

        // read the cookie if persisted
        readCookie();

        if ( config.isRefreshNPersist() )
        {
            try
            {
                LOG.debug( "==================== Refresh And Persist ==========" );

                return doSyncSearch( SynchronizationModeEnum.REFRESH_AND_PERSIST, false );
            }
            catch ( Exception e )
            {
                LOG.error( "Failed to sync with refreshAndPersist mode", e );
                return ReplicationStatusEnum.DISCONNECTED;
            }
        }
        else
        {
            return doRefreshOnly();
        }
    }


    private ReplicationStatusEnum doRefreshOnly()
    {
        while ( !disconnected )
        {
            LOG.debug( "==================== Refresh Only ==========" );

            try
            {
                doSyncSearch( SynchronizationModeEnum.REFRESH_ONLY, false );

                LOG.info( "--------------------- Sleep for a little while ------------------" );
                Thread.sleep( config.getRefreshInterval() );
                LOG.debug( "--------------------- syncing again ------------------" );

            }
            catch ( InterruptedException ie )
            {
                LOG.warn( "refresher thread interrupted" );

                return ReplicationStatusEnum.DISCONNECTED;
            }
            catch ( Exception e )
            {
                LOG.error( "Failed to sync with refresh only mode", e );
                return ReplicationStatusEnum.DISCONNECTED;
            }
        }

        return ReplicationStatusEnum.STOPPED;
    }


    /** 
     * {@inheritDoc}
     */
    public void setConfig( ReplicationConsumerConfig config )
    {
        this.config = ( SyncReplConfiguration ) config;
    }


    /**
     * {@inheritDoc}
     */
    public boolean connect( boolean now )
    {
        boolean connected = false;

        if ( now )
        {
            connected = connect();
        }

        while ( !connected )
        {
            try
            {
                CONSUMER_LOG.debug( "Consumer {} cannot connect to {}, wait 5 seconds.", config.getReplicaId(),
                    config.getProducer() );

                // try to establish a connection for every 5 seconds
                Thread.sleep( 5000 );
            }
            catch ( InterruptedException e )
            {
                LOG.warn( "Consumer {} Interrupted while trying to reconnect to the provider {}",
                    config.getReplicaId(), config.getProducer() );
            }

            connected = connect();
        }

        // TODO : we may have cases were we get here with the connected flag to false. With the above
        // code, thi sis not possible

        return connected;
    }


    /**
     * {@inheritDoc}
     */
    public void ping()
    {
        boolean connected = !disconnected;

        if ( disconnected )
        {
            connected = connect();
        }

        if ( connected )
        {
            CONSUMER_LOG.debug( "PING : The consumer {} is alive", config.getReplicaId() );

            try
            {
                Entry baseDn = connection.lookup( config.getBaseDn(), "1.1" );

                if ( baseDn == null )
                {
                    // Cannot get the entry : this is bad, but possible
                    CONSUMER_LOG.debug( "Cannot fetch '{}' from provider for consumer {}", config.getBaseDn(),
                        config.getReplicaId() );
                }
                else
                {
                    CONSUMER_LOG.debug( "Fetched '{}' from provider for consumer {}", config.getBaseDn(),
                        config.getReplicaId() );
                }
            }
            catch ( LdapException le )
            {
                // Error : we must disconnect
                disconnect();
            }
        }
        else
        {
            CONSUMER_LOG.debug( "PING : The consumer {} cannot be connected", config.getReplicaId() );
        }
    }


    /**
     * {@inheritDoc}
     */
    public void stop()
    {
        if ( !disconnected )
        {
            disconnect();
        }
    }


    /**
     * {@inheritDoc}
     */
    public String getId()
    {
        return String.valueOf( getConfig().getReplicaId() );
    }


    /**
     * Performs a search on connection with updated syncRequest control. The provider
     * will initiate an UpdateContant or an initContent depending on the current consumer
     * status, accordingly to the cookie's content.
     * If the mode is refreshOnly, the server will send a SearchResultDone when all the modified
     * entries have been sent.
     * If the mode is refreshAndPersist, the provider never send a SearchResultDone, so we keep
     * receiving modifications' notifications on the consumer, and never exit the loop, unless
     * some communication error occurs.
     *
     * @param syncType The synchornization type, either REFRESH_ONLY or REFRESH_AND_PERSIST
     * @param reloadHint A flag used to tell the server that we want a reload
     * @return The replication status
     * @throws Exception in case of any problems encountered while searching
     */
    private ReplicationStatusEnum doSyncSearch( SynchronizationModeEnum syncType, boolean reloadHint ) throws Exception
    {
        CONSUMER_LOG.debug( "Starting synchronization mode {}, reloadHint {}", syncType, reloadHint );
        // Prepare the Syncrepl Request
        SyncRequestValue syncReq = new SyncRequestValueDecorator( directoryService.getLdapCodecService() );

        syncReq.setMode( syncType );
        syncReq.setReloadHint( reloadHint );

        // If we have a persisted cookie, send it.
        if ( syncCookie != null )
        {
            CONSUMER_LOG.debug( "searching on {} with searchRequest, cookie '{}'", config.getProducer(),
                Strings.utf8ToString( syncCookie ) );
            syncReq.setCookie( syncCookie );
        }
        else
        {
            CONSUMER_LOG.debug( "searching on {} with searchRequest, no cookie", config.getProducer() );
        }

        searchRequest.addControl( syncReq );

        // Do the search. We use a searchAsync because we want to get SearchResultDone responses
        SearchFuture sf = connection.searchAsync( searchRequest );

        Response resp = sf.get();

        CONSUMER_LOG.debug( "Response from {} : {}", config.getProducer(), resp );

        // Now, process the responses. We loop until we have a connection termination or
        // a SearchResultDone (RefreshOnly mode)
        while ( !( resp instanceof SearchResultDone ) && !sf.isCancelled() && !disconnected )
        {
            if ( resp instanceof SearchResultEntry )
            {
                SearchResultEntry result = ( SearchResultEntry ) resp;

                handleSearchResultEntry( result );
            }
            else if ( resp instanceof SearchResultReference )
            {
                handleSearchReference( ( SearchResultReference ) resp );
            }
            else if ( resp instanceof IntermediateResponse )
            {
                handleSyncInfo( ( IntermediateResponse ) resp );
            }

            // Next entry
            resp = sf.get();
            CONSUMER_LOG.debug( "Response from {} : {}", config.getProducer(), resp );
        }

        if ( sf.isCancelled() )
        {

            CONSUMER_LOG.debug( "Search sync on {} has been canceled ", config.getProducer(), sf.getCause() );

            return ReplicationStatusEnum.DISCONNECTED;
        }
        else if ( disconnected )
        {
            CONSUMER_LOG.debug( "Disconnected from {}", config.getProducer() );

            return ReplicationStatusEnum.DISCONNECTED;
        }
        else
        {
            ResultCodeEnum resultCode = handleSearchResultDone( ( SearchResultDone ) resp );

            CONSUMER_LOG.debug( "Response from {} : {}", config.getProducer(), resultCode );
            LOG.debug( "sync operation returned result code {}", resultCode );

            if ( resultCode == ResultCodeEnum.NO_SUCH_OBJECT )
            {
                // log the error and handle it appropriately
                CONSUMER_LOG.warn( "The base Dn {} is not found on provider {}", config.getBaseDn(),
                    config.getProducer() );
                LOG.warn( "The base Dn {} is not found on provider {}", config.getBaseDn(), config.getProducer() );

                CONSUMER_LOG.warn( "Disconnecting the Refresh&Persist consumer from provider {}", config.getProducer() );
                LOG.warn( "Disconnecting the Refresh&Persist consumer from provider {}", config.getProducer() );
                disconnect();

                return ReplicationStatusEnum.DISCONNECTED;
            }
            else if ( resultCode == ResultCodeEnum.E_SYNC_REFRESH_REQUIRED )
            {
                CONSUMER_LOG.info( "Full SYNC_REFRESH required from {}", config.getProducer() );
                LOG.info( "Full SYNC_REFRESH required from {}", config.getProducer() );

                try
                {
                    CONSUMER_LOG.debug( "Deleting baseDN {}", config.getBaseDn() );
                    deleteRecursive( new Dn( config.getBaseDn() ), null );
                }
                catch ( Exception e )
                {
                    LOG
                        .error(
                            "Failed to delete the replica base as part of handling E_SYNC_REFRESH_REQUIRED, disconnecting the consumer",
                            e );
                }

                // Do a full update.
                removeCookie();

                CONSUMER_LOG.debug( "Re-doing a syncRefresh from producer {}", config.getProducer() );

                return ReplicationStatusEnum.REFRESH_REQUIRED;
            }
            else
            {
                CONSUMER_LOG.debug( "Got result code {} from producer {}. Replication stopped", resultCode,
                    config.getProducer() );
                return ReplicationStatusEnum.DISCONNECTED;
            }
        }
    }


    /**
     * Disconnect from the producer
     */
    private void disconnect()
    {
        disconnected = true;

        try
        {
            if ( ( connection != null ) && connection.isConnected() )
            {
                connection.unBind();
                LOG.info( "Unbound from the server {}", config.getProducer() );
                
                if ( CONSUMER_LOG.isDebugEnabled() )
                {
                    MDC.put( "Replica", Integer.toString( config.getReplicaId() ) );
                    CONSUMER_LOG.info( "Unbound from the server {}", config.getProducer() );
                }
                
                connection.close();
                LOG.info( "Connection closed for the server {}", config.getProducer() );
                CONSUMER_LOG.info( "Connection closed for the server {}", config.getProducer() );
                
                connection = null;
            }
        }
        catch ( Exception e )
        {
            LOG.error( "Failed to close the connection", e );
        }
        finally
        {
            // persist the cookie
            storeCookie();
            
            // reset the cookie
            syncCookie = null;
        }
    }


    /**
     * stores the cookie.
     */
    private void storeCookie()
    {
        CONSUMER_LOG.debug( "Storing the cookie '{}'", Strings.utf8ToString( syncCookie ) );

        if ( syncCookie == null )
        {
            return;
        }

        if ( ( lastSavedCookie != null ) && Arrays.equals( syncCookie, lastSavedCookie ) )
        {
            return;
        }

        try
        {
            Attribute attr = cookieMod.getAttribute();
            attr.clear();
            attr.add( syncCookie );

            String cookieString = Strings.utf8ToString( syncCookie );
            int replicaId = LdapProtocolUtils.getReplicaId( cookieString );

            Attribute ridAt = ridMod.getAttribute();
            ridAt.clear();
            ridAt.add( String.valueOf( replicaId ) );

            CONSUMER_LOG.debug( "Storing the cookie in the DIT : {}", config.getConfigEntryDn() );

            //session.modify( config.getConfigEntryDn(), cookieMod, ridMod );
            session.modify( config.getConfigEntryDn(), cookieMod );
            CONSUMER_LOG.debug( "stored the cookie in entry {}", config.getConfigEntryDn() );

            lastSavedCookie = new byte[syncCookie.length];
            System.arraycopy( syncCookie, 0, lastSavedCookie, 0, syncCookie.length );

            LOG.debug( "stored the cookie" );
        }
        catch ( Exception e )
        {
            LOG.error( "Failed to store the cookie in consumer entry {}", config.getConfigEntryDn(), e );
        }
    }


    /**
     * Read the cookie for a consumer
     */
    private void readCookie()
    {
        try
        {
            Entry entry = session.lookup( config.getConfigEntryDn(), SchemaConstants.ADS_REPL_COOKIE );

            if ( entry != null )
            {
                Attribute attr = entry.get( REPL_COOKIE_AT );

                if ( attr != null )
                {
                    syncCookie = attr.getBytes();
                    lastSavedCookie = syncCookie;
                    String syncCookieString = Strings.utf8ToString( syncCookie );
                    CONSUMER_LOG.debug( "Cookie for consumer {} : {}", config.getReplicaId(), syncCookieString );
                    LOG.debug( "loaded cookie from DIT" );
                }
                else
                {
                    CONSUMER_LOG.debug( "No cookie found for consumer {}", config.getReplicaId() );
                }
            }
            else
            {
                CONSUMER_LOG.debug( "Cannot find the configuration '{}' in the DIT for consumer {}",
                    config.getConfigEntryDn(), config.getReplicaId() );
            }
        }
        catch ( Exception e )
        {
            // can be ignored, most likely happens if there is no entry with the given Dn
            // log in debug mode
            CONSUMER_LOG.debug( "Cannot find the  '{}' in the DIT for consumer {}", config.getConfigEntryDn(),
                config.getReplicaId() );
            LOG.debug( "Failed to read the cookie from the entry {}", config.getConfigEntryDn(), e );
        }
    }


    /**
     * deletes the cookie and resets the syncCookie to null
     */
    private void removeCookie()
    {
        try
        {
            Attribute cookieAttr = new DefaultAttribute( REPL_COOKIE_AT );
            Modification deleteCookieMod = new DefaultModification( ModificationOperation.REMOVE_ATTRIBUTE,
                cookieAttr );
            session.modify( config.getConfigEntryDn(), deleteCookieMod );
        }
        catch ( Exception e )
        {
            LOG.warn( "Failed to delete the cookie from the entry with Dn {}", config.getConfigEntryDn() );
            LOG.warn( "{}", e );
        }

        LOG.info( "resetting sync cookie" );

        syncCookie = null;
        lastSavedCookie = null;
    }


    private void applyModDnOperation( Entry remoteEntry, String entryUuid, int rid ) throws Exception
    {
        LOG.debug( "MODDN for entry {}, new entry : {}", entryUuid, remoteEntry );

        // First, compute the MODDN type
        SyncModifyDnType modDnType = null;

        try
        {
            // Retrieve locally the moved or renamed entry
            String filter = "(entryUuid=" + entryUuid + ")";
            SearchRequest searchRequest = new SearchRequestImpl();
            searchRequest.setBase( new Dn( schemaManager, config.getBaseDn() ) );
            searchRequest.setFilter( filter );
            searchRequest.setScope( SearchScope.SUBTREE );
            searchRequest.addAttributes( "entryUuid", "entryCsn", "*" );

            EntryFilteringCursor cursor = session.search( searchRequest );
            cursor.beforeFirst();

            Entry localEntry = null;

            if ( cursor.next() )
            {
                localEntry = cursor.get();
            }

            cursor.close();

            // can happen in MMR scenario
            if ( localEntry == null )
            {
                return;
            }

            if ( config.isMmrMode() )
            {
                Csn localCsn = new Csn( localEntry.get( SchemaConstants.ENTRY_CSN_AT ).getString() );
                Csn remoteCsn = new Csn( remoteEntry.get( SchemaConstants.ENTRY_CSN_AT ).getString() );

                if ( localCsn.compareTo( remoteCsn ) >= 0 )
                {
                    // just discard the received modified entry, that is old
                    LOG.debug( "local modification is latest, discarding the modDn operation dn {}",
                        remoteEntry.getDn() );
                    return;
                }
            }

            // Compute the DN, parentDn and Rdn for both entries
            Dn localDn = localEntry.getDn();
            Dn remoteDn = directoryService.getDnFactory().create( remoteEntry.getDn().getName() );

            Dn localParentDn = localDn.getParent();
            Dn remoteParentDn = directoryService.getDnFactory().create( remoteDn.getParent().getName() );

            Rdn localRdn = localDn.getRdn();
            Rdn remoteRdn = directoryService.getDnFactory().create( remoteDn.getRdn().getName() ).getRdn();

            if ( localRdn.equals( remoteRdn ) )
            {
                // If the RDN are equals, it's a MOVE
                modDnType = SyncModifyDnType.MOVE;
            }
            else if ( localParentDn.equals( remoteParentDn ) )
            {
                // If the parentDn are equals, it's a RENAME
                modDnType = SyncModifyDnType.RENAME;
            }
            else
            {
                // Otherwise, it's a MOVE and RENAME
                modDnType = SyncModifyDnType.MOVE_AND_RENAME;
            }

            // Check if the OldRdn has been deleted
            boolean deleteOldRdn = !remoteEntry.contains( localRdn.getNormType(), localRdn.getNormValue() );

            switch ( modDnType )
            {
                case MOVE:
                    LOG.debug( "moving {} to the new parent {}", localDn, remoteParentDn );
                    MoveOperationContext movCtx = new MoveOperationContext( session, localDn, remoteParentDn );
                    movCtx.setReplEvent( true );
                    movCtx.setRid( rid );
                    directoryService.getOperationManager().move( movCtx );

                    break;

                case RENAME:
                    LOG.debug( "renaming the Dn {} with new Rdn {} and deleteOldRdn flag set to {}", new String[]
                        { localDn.getName(), remoteRdn.getName(), String.valueOf( deleteOldRdn ) } );

                    RenameOperationContext renCtx = new RenameOperationContext( session, localDn, remoteRdn,
                        deleteOldRdn );
                    renCtx.setReplEvent( true );
                    renCtx.setRid( rid );
                    directoryService.getOperationManager().rename( renCtx );

                    break;

                case MOVE_AND_RENAME:
                    LOG.debug(
                        "moveAndRename on the Dn {} with new newParent Dn {}, new Rdn {} and deleteOldRdn flag set to {}",
                        new String[]
                            {
                                localDn.getName(),
                                remoteParentDn.getName(),
                                remoteRdn.getName(),
                                String.valueOf( deleteOldRdn ) } );

                    MoveAndRenameOperationContext movRenCtx = new MoveAndRenameOperationContext( session, localDn,
                        remoteParentDn, remoteRdn, deleteOldRdn );
                    movRenCtx.setReplEvent( true );
                    movRenCtx.setRid( rid );
                    directoryService.getOperationManager().moveAndRename( movRenCtx );

                    break;
            }
        }
        catch ( Exception e )
        {
            throw e;
        }
    }


    private void modify( Entry remoteEntry, int rid ) throws Exception
    {
        String[] attributes = computeAttributes( config.getAttributes(), SchemaConstants.ALL_OPERATIONAL_ATTRIBUTES );

        LookupOperationContext lookupCtx =
            new LookupOperationContext( session, remoteEntry.getDn(), attributes );

        lookupCtx.setSyncreplLookup( true );

        Entry localEntry = session.getDirectoryService().getOperationManager().lookup( lookupCtx );

        if ( config.isMmrMode() )
        {
            Csn localCsn = new Csn( localEntry.get( SchemaConstants.ENTRY_CSN_AT ).getString() );
            Csn remoteCsn = new Csn( remoteEntry.get( SchemaConstants.ENTRY_CSN_AT ).getString() );

            if ( localCsn.compareTo( remoteCsn ) >= 0 )
            {
                // just discard the received modified entry, that is old
                LOG.debug( "local modification is latest, discarding the modification of dn {}", remoteEntry.getDn() );
                return;
            }
        }

        remoteEntry.removeAttributes( MOD_IGNORE_AT );
        localEntry.removeAttributes( MOD_IGNORE_AT );

        List<Modification> mods = new ArrayList<Modification>();
        Iterator<Attribute> itr = localEntry.iterator();

        while ( itr.hasNext() )
        {
            Attribute localAttr = itr.next();
            String attrId = localAttr.getId();
            Modification mod;
            Attribute remoteAttr = remoteEntry.get( attrId );

            if ( remoteAttr != null ) // would be better if we compare the values also? or will it consume more time?
            {
                mod = new DefaultModification( ModificationOperation.REPLACE_ATTRIBUTE, remoteAttr );
                remoteEntry.remove( remoteAttr );
            }
            else
            {
                mod = new DefaultModification( ModificationOperation.REMOVE_ATTRIBUTE, localAttr );
            }

            mods.add( mod );
        }

        if ( remoteEntry.size() > 0 )
        {
            itr = remoteEntry.iterator();

            while ( itr.hasNext() )
            {
                mods.add( new DefaultModification( ModificationOperation.ADD_ATTRIBUTE, itr.next() ) );
            }
        }

        List<Modification> serverModifications = new ArrayList<Modification>( mods.size() );

        for ( Modification mod : mods )
        {
            serverModifications.add( new DefaultModification( directoryService.getSchemaManager(), mod ) );
        }

        ModifyOperationContext modifyContext = new ModifyOperationContext( session, remoteEntry.getDn(),
            serverModifications );
        modifyContext.setReplEvent( true );
        modifyContext.setRid( rid );

        OperationManager operationManager = directoryService.getOperationManager();
        operationManager.modify( modifyContext );
    }


    /**
     * Create a new list combining a list and a newly added attribute
     */
    private String[] computeAttributes( String[] attributes, String addedAttribute )
    {
        if ( attributes != null )
        {
            if ( addedAttribute != null )
            {
                String[] combinedAttributes = new String[attributes.length + 1];

                System.arraycopy( attributes, 0, combinedAttributes, 0, attributes.length );
                combinedAttributes[attributes.length] = addedAttribute;

                return combinedAttributes;
            }
            else
            {
                return attributes;
            }
        }
        else
        {
            if ( addedAttribute != null )
            {
                return new String[]
                    { addedAttribute };
            }
            else
            {
                return StringConstants.EMPTY_STRINGS;
            }
        }
    }


    /**
     * deletes the entries having the UUID given in the list
     *
     * @param uuidList the list of UUIDs
     * @param replicaId TODO
     * @throws Exception in case of any problems while deleting the entries
     */
    private void deleteEntries( List<byte[]> uuidList, boolean isRefreshPresent, int replicaId ) throws Exception
    {
        if ( uuidList == null || uuidList.isEmpty() )
        {
            return;
        }

        for ( byte[] uuid : uuidList )
        {
            LOG.info( "uuid: {}", Strings.uuidToString( uuid ) );
        }

        // if it is refreshPresent list then send all the UUIDs for
        // filtering, otherwise breaking the list will cause the
        // other present entries to be deleted from DIT
        if ( isRefreshPresent )
        {
            LOG.debug( "refresh present syncinfo list has {} UUIDs", uuidList.size() );
            processDelete( uuidList, isRefreshPresent, replicaId );
            return;
        }

        int NODE_LIMIT = 10;

        int count = uuidList.size() / NODE_LIMIT;

        int startIndex = 0;
        int i = 0;
        for ( ; i < count; i++ )
        {
            startIndex = i * NODE_LIMIT;
            processDelete( uuidList.subList( startIndex, startIndex + NODE_LIMIT ), isRefreshPresent, replicaId );
        }

        if ( ( uuidList.size() % NODE_LIMIT ) != 0 )
        {
            // remove the remaining entries
            if ( count > 0 )
            {
                startIndex = i * NODE_LIMIT;
            }

            processDelete( uuidList.subList( startIndex, uuidList.size() ), isRefreshPresent, replicaId );
        }
    }


    /**
     * do not call this method directly, instead call deleteEntries()
     *
     * @param limitedUuidList a list of UUIDs whose size is less than or equal to #NODE_LIMIT (node limit applies only for refreshDeletes list)
     * @param isRefreshPresent a flag indicating the type of entries present in the UUID list
     * @param replicaId TODO
     */
    private void processDelete( List<byte[]> limitedUuidList, boolean isRefreshPresent, int replicaId )
        throws Exception
    {
        ExprNode filter = null;
        int size = limitedUuidList.size();
        if ( size == 1 )
        {
            String uuid = Strings.uuidToString( limitedUuidList.get( 0 ) );
            filter = new EqualityNode<String>( SchemaConstants.ENTRY_UUID_AT,
                new org.apache.directory.api.ldap.model.entry.StringValue( uuid ) );
            if ( isRefreshPresent )
            {
                filter = new NotNode( filter );
            }
        }
        else
        {
            if ( isRefreshPresent )
            {
                filter = new AndNode();
            }
            else
            {
                filter = new OrNode();
            }

            for ( int i = 0; i < size; i++ )
            {
                String uuid = Strings.uuidToString( limitedUuidList.get( i ) );
                ExprNode uuidEqNode = new EqualityNode<String>( SchemaConstants.ENTRY_UUID_AT,
                    new org.apache.directory.api.ldap.model.entry.StringValue( uuid ) );

                if ( isRefreshPresent )
                {
                    uuidEqNode = new NotNode( uuidEqNode );
                    ( ( AndNode ) filter ).addNode( uuidEqNode );
                }
                else
                {
                    ( ( OrNode ) filter ).addNode( uuidEqNode );
                }
            }
        }

        Dn dn = new Dn( schemaManager, config.getBaseDn() );

        LOG.debug( "selecting entries to be deleted using filter {}", filter.toString() );
        EntryFilteringCursor cursor = session.search( dn, SearchScope.SUBTREE, filter,
            AliasDerefMode.NEVER_DEREF_ALIASES, SchemaConstants.ENTRY_UUID_AT );
        cursor.beforeFirst();

        while ( cursor.next() )
        {
            Entry entry = cursor.get();
            deleteRecursive( entry.getDn(), null );
        }

        cursor.close();
    }


    private synchronized Object getLockFor( String uuid )
    {
        Object lock = uuidLockMap.get( uuid );

        if ( lock == null )
        {
            lock = new Object();
            uuidLockMap.put( uuid, lock );
        }

        return lock;
    }


    /**
     * removes all child entries present under the given Dn and finally the Dn itself
     *
     * Working:
     *          This is a recursive function which maintains a Map<Dn,Cursor>.
     *          The way the cascade delete works is by checking for children for a
     *          given Dn(i.e opening a search cursor) and if the cursor is empty
     *          then delete the Dn else for each entry's Dn present in cursor call
     *          deleteChildren() with the Dn and the reference to the map.
     *
     *          The reason for opening a search cursor is based on an assumption
     *          that an entry *might* contain children, consider the below DIT fragment
     *
     *          parent
     *          /     \
     *        child1   child2
     *                 /     \
     *               grand21  grand22
     *
     *           The below method works better in the case where the tree depth is >1
     *
     *   In the case of passing a non-null DeleteListener, the return value will always be null, cause the
     *   operation is treated as asynchronous and response result will be sent using the listener callback
     *
     * @param rootDn the Dn which will be removed after removing its children
     * @param map a map to hold the Cursor related to a Dn
     * @throws Exception If the Dn is not valid or if the deletion failed
     */
    private void deleteRecursive( Dn rootDn, Map<Dn, EntryFilteringCursor> cursorMap ) throws Exception
    {
        LOG.debug( "searching for {}", rootDn.getName() );
        EntryFilteringCursor cursor = null;

        try
        {
            if ( cursorMap == null )
            {
                cursorMap = new HashMap<Dn, EntryFilteringCursor>();
            }

            cursor = cursorMap.get( rootDn );

            if ( cursor == null )
            {
                cursor = session.search( rootDn, SearchScope.ONELEVEL, ENTRY_UUID_PRESENCE_FILTER,
                    AliasDerefMode.NEVER_DEREF_ALIASES, SchemaConstants.ENTRY_UUID_AT );
                cursor.beforeFirst();
                LOG.debug( "putting cursor for {}", rootDn.getName() );
                cursorMap.put( rootDn, cursor );
            }

            if ( !cursor.next() ) // if this is a leaf entry's Dn
            {
                LOG.debug( "deleting {}", rootDn.getName() );
                cursorMap.remove( rootDn );
                cursor.close();
                session.delete( rootDn );
            }
            else
            {
                do
                {
                    Entry entry = cursor.get();

                    deleteRecursive( entry.getDn(), cursorMap );
                }
                while ( cursor.next() );

                cursorMap.remove( rootDn );
                cursor.close();
                LOG.debug( "deleting {}", rootDn.getName() );
                session.delete( rootDn );
            }
        }
        catch ( Exception e )
        {
            String msg = "Failed to delete child entries under the Dn " + rootDn.getName();
            LOG.error( msg, e );
            throw e;
        }
    }


    /**
     * @see Object#toString()
     */
    public String toString()
    {
        StringBuilder sb = new StringBuilder();

        sb.append( "Consumer " ).append( config );

        return sb.toString();
    }
}
