/*
 *  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;


import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

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.CoreSession;
import org.apache.directory.server.core.DirectoryService;
import org.apache.directory.server.core.entry.ClonedServerEntry;
import org.apache.directory.server.core.filtering.EntryFilteringCursor;
import org.apache.directory.shared.ldap.codec.controls.ManageDsaITControl;
import org.apache.directory.shared.ldap.codec.controls.replication.syncDoneValue.SyncDoneValueControl;
import org.apache.directory.shared.ldap.codec.controls.replication.syncDoneValue.SyncDoneValueControlDecoder;
import org.apache.directory.shared.ldap.codec.controls.replication.syncInfoValue.SyncInfoValueControl;
import org.apache.directory.shared.ldap.codec.controls.replication.syncInfoValue.SyncInfoValueControlDecoder;
import org.apache.directory.shared.ldap.codec.controls.replication.syncRequestValue.SyncRequestValueControl;
import org.apache.directory.shared.ldap.codec.controls.replication.syncStateValue.SyncStateValueControl;
import org.apache.directory.shared.ldap.codec.controls.replication.syncStateValue.SyncStateValueControlDecoder;
import org.apache.directory.shared.ldap.codec.controls.replication.syncmodifydn.SyncModifyDnControl;
import org.apache.directory.shared.ldap.codec.controls.replication.syncmodifydn.SyncModifyDnControlDecoder;
import org.apache.directory.shared.ldap.constants.SchemaConstants;
import org.apache.directory.shared.ldap.entry.DefaultEntry;
import org.apache.directory.shared.ldap.entry.DefaultEntryAttribute;
import org.apache.directory.shared.ldap.entry.DefaultModification;
import org.apache.directory.shared.ldap.entry.Entry;
import org.apache.directory.shared.ldap.entry.EntryAttribute;
import org.apache.directory.shared.ldap.entry.Modification;
import org.apache.directory.shared.ldap.entry.ModificationOperation;
import org.apache.directory.shared.ldap.exception.LdapException;
import org.apache.directory.shared.ldap.filter.AndNode;
import org.apache.directory.shared.ldap.filter.EqualityNode;
import org.apache.directory.shared.ldap.filter.ExprNode;
import org.apache.directory.shared.ldap.filter.NotNode;
import org.apache.directory.shared.ldap.filter.OrNode;
import org.apache.directory.shared.ldap.filter.PresenceNode;
import org.apache.directory.shared.ldap.filter.SearchScope;
import org.apache.directory.shared.ldap.message.AliasDerefMode;
import org.apache.directory.shared.ldap.message.BindResponse;
import org.apache.directory.shared.ldap.message.IntermediateResponse;
import org.apache.directory.shared.ldap.message.LdapResult;
import org.apache.directory.shared.ldap.message.Response;
import org.apache.directory.shared.ldap.message.ResultCodeEnum;
import org.apache.directory.shared.ldap.message.SearchRequest;
import org.apache.directory.shared.ldap.message.SearchResultDone;
import org.apache.directory.shared.ldap.message.SearchResultEntry;
import org.apache.directory.shared.ldap.message.SearchResultReference;
import org.apache.directory.shared.ldap.message.control.Control;
import org.apache.directory.shared.ldap.message.control.replication.SyncModifyDnType;
import org.apache.directory.shared.ldap.message.control.replication.SyncStateTypeEnum;
import org.apache.directory.shared.ldap.message.control.replication.SynchronizationModeEnum;
import org.apache.directory.shared.ldap.name.DN;
import org.apache.directory.shared.ldap.name.RDN;
import org.apache.directory.shared.ldap.schema.AttributeType;
import org.apache.directory.shared.ldap.schema.AttributeTypeOptions;
import org.apache.directory.shared.ldap.schema.SchemaManager;
import org.apache.directory.shared.ldap.util.StringTools;
import org.apache.directory.shared.util.Strings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 *
 * Implementation of syncrepl slave a.k.a consumer.
 *
 * TODO write test cases
 *
 * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
 */
public class SyncReplConsumer implements ConnectionClosedEventListener
{

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

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

    /** the logger */
    private static final Logger LOG = LoggerFactory.getLogger( SyncReplConsumer.class );

    /** 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;

    /** the decoder for syncinfovalue control */
    private SyncInfoValueControlDecoder decoder = new SyncInfoValueControlDecoder();

    /** the cookie file */
    private File cookieFile;

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

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

    private SyncDoneValueControlDecoder syncDoneControlDecoder = new SyncDoneValueControlDecoder();

    private SyncStateValueControlDecoder syncStateControlDecoder = new SyncStateValueControlDecoder();

    private SyncModifyDnControlDecoder syncModifyDnControlDecoder = new SyncModifyDnControlDecoder();

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

    private RefresherThread refreshThread;

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

    private static AttributeType ENTRY_UUID_AT;

    private static final PresenceNode ENTRY_UUID_PRESENCE_FILTER = new PresenceNode( SchemaConstants.ENTRY_UUID_AT );

    private static final Set<AttributeTypeOptions> ENTRY_UUID_ATOP_SET = new HashSet<AttributeTypeOptions>();

    private List<Modification> cookieModLst;

    private DN configEntryDn;

    private static AttributeType COOKIE_AT_TYPE;


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


    public void init( DirectoryService directoryservice, SyncreplConfiguration config ) throws Exception
    {
        this.config = config;
        this.directoryService = directoryservice;

        if ( config.isStoreCookieInFile() )
        {
            File cookieDir = new File( directoryservice.getInstanceLayout().getRunDirectory(), "cookies" );
            cookieDir.mkdir();

            cookieFile = new File( cookieDir, String.valueOf( config.getReplicaId() ) );
        }

        session = directoryService.getAdminSession();

        schemaManager = directoryservice.getSchemaManager();

        ENTRY_UUID_AT = schemaManager.lookupAttributeTypeRegistry( SchemaConstants.ENTRY_UUID_AT );

        ENTRY_UUID_ATOP_SET.add( new AttributeTypeOptions( ENTRY_UUID_AT ) );

        COOKIE_AT_TYPE = schemaManager.lookupAttributeTypeRegistry( "ads-replCookie" );
        EntryAttribute cookieAttr = new DefaultEntryAttribute( COOKIE_AT_TYPE );

        Modification cookieMod = new DefaultModification( ModificationOperation.REPLACE_ATTRIBUTE, cookieAttr );
        cookieModLst = new ArrayList<Modification>( 1 );
        cookieModLst.add( cookieMod );

        configEntryDn = new DN( config.getConfigEntryDn(), schemaManager );

        prepareSyncSearchRequest();
    }


    public boolean connect()
    {
        try
        {
            String providerHost = config.getProviderHost();
            int port = config.getPort();
            
            // Create a connection
            if ( connection == null )
            {
                connection = new LdapNetworkConnection( providerHost, port );
                
                if( config.isUseTls() )
                {
                    connection.getConfig().setTrustManagers( config.getTrustManager() );
                    connection.startTls();
                }
                
                connection.addConnectionClosedEventListener( this );
            }

            // Do a bind
            BindResponse bindResponse = connection.bind( config.getReplUserDn(), config.getReplUserPassword() );

            // Now get the result
            LdapResult ldapResult = bindResponse.getLdapResult();

            if ( ldapResult.getResultCode() != ResultCodeEnum.SUCCESS )
            {
                LOG.warn( "Failed to bind to the server with the given bind DN {} and credentials: {}", config.getReplUserDn(), ldapResult );
            }
            else
            {
                return true;
            }
        }
        catch ( Exception e )
        {
            LOG.error( "Failed to bind with the given bindDN and credentials", e );
        }

        return false;
    }


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

        searchRequest = new org.apache.directory.shared.ldap.codec.message.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 ManageDsaITControl() );
        }
    }


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

        Control ctrl = searchDone.getControls().get( SyncDoneValueControl.CONTROL_OID );
        SyncDoneValueControl syncDoneCtrl = new SyncDoneValueControl();
        try
        {
            if ( ctrl != null )
            {
                syncDoneCtrl = ( SyncDoneValueControl ) syncDoneControlDecoder.decode( ctrl.getValue(), syncDoneCtrl );
            }
        }
        catch ( Exception e )
        {
            LOG.error( "Failed to decode the syncDoneControlCodec", e );
        }

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

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

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


    public 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
    }


    public void handleSearchResult( SearchResultEntry syncResult )
    {

        LOG.debug( "------------- starting handleSearchResult ------------" );

        SyncStateValueControl syncStateCtrl = new SyncStateValueControl();

        try
        {
            Entry remoteEntry = syncResult.getEntry();

            Control ctrl = syncResult.getControls().get( SyncStateValueControl.CONTROL_OID );

            try
            {
                syncStateCtrl = ( SyncStateValueControl ) syncStateControlDecoder.decode( ctrl.getValue(),
                    syncStateCtrl );
            }
            catch ( Exception e )
            {
                LOG.error( "Failed to decode syncStateControl", e );
            }

            if ( syncStateCtrl.getCookie() != null )
            {
                syncCookie = syncStateCtrl.getCookie();
                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 = {}", StringTools.uuidToString( syncStateCtrl.getEntryUUID() ) );
            }

            switch ( state )
            {
                case ADD:
                    if ( !session.exists( remoteEntry.getDn() ) )
                    {
                        LOG.debug( "adding entry with dn {}", remoteEntry.getDn().getName() );
                        LOG.debug( remoteEntry.toString() );
                        session.add( new DefaultEntry( schemaManager, remoteEntry ) );
                    }
                    else
                    {
                        LOG.debug( "updating entry in refreshOnly mode {}", remoteEntry.getDn().getName() );
                        modify( remoteEntry );
                    }

                    break;

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

                case MODDN:
                    Control adsModDnControl = syncResult.getControls().get( SyncModifyDnControl.CONTROL_OID );
                    //Apache Directory Server's special control
                    SyncModifyDnControl syncModDnControl = new SyncModifyDnControl();

                    LOG.debug( "decoding the SyncModifyDnControl.." );
                    syncModDnControl = ( SyncModifyDnControl ) syncModifyDnControlDecoder.decode( adsModDnControl
                        .getValue(), syncModDnControl );

                    applyModDnOperation( syncModDnControl );
                    break;

                case DELETE:
                    LOG.debug( "deleting entry with dn {}", remoteEntry.getDn().getName() );
                    // 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}
     */
    public void handleSyncInfo( IntermediateResponse syncInfoResp )
    {
        try
        {
            LOG.debug( "............... inside handleSyncInfo ..............." );

            byte[] syncinfo = syncInfoResp.getResponseValue();
            SyncInfoValueControl syncInfoValue = ( SyncInfoValueControl ) decoder.decode( syncinfo, null );

            byte[] cookie = syncInfoValue.getCookie();

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

            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 );
            }
            else
            {
                deleteEntries( uuidList, true );
            }

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

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

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


    /**
     * {@inheritDoc}
     */
    public void connectionClosed()
    {
        if ( disconnected )
        {
            return;
        }

        boolean connected = false;

        while ( !connected )
        {
            try
            {
                Thread.sleep( config.getRefreshInterval() );
            }
            catch ( InterruptedException e )
            {
                LOG.error( "Interrupted while sleeping before trying to reconnect", e );
            }

            LOG.debug( "Trying to reconnect" );
            connected = connect();
        }

        startSync();
    }


    /**
     * starts the synchronization operation
     */
    public void startSync()
    {
        // read the cookie if persisted
        readCookie();

        if ( config.isRefreshNPersist() )
        {
            try
            {
                LOG.debug( "==================== Refresh And Persist ==========" );
                doSyncSearch( SynchronizationModeEnum.REFRESH_AND_PERSIST, false );
            }
            catch ( Exception e )
            {
                LOG.error( "Failed to sync with refreshAndPersist mode", e );
            }
        }
        else
        {
            refreshThread = new RefresherThread();
            refreshThread.start();
        }
    }


    /**
     * performs a search on connection with updated syncRequest control.
     *
     * @throws Exception in case of any problems encountered while searching
     */
    private void doSyncSearch( SynchronizationModeEnum syncType, boolean reloadHint ) throws Exception
    {
        SyncRequestValueControl syncReq = new SyncRequestValueControl();

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

        if ( syncCookie != null )
        {
            LOG.debug( "searching with searchRequest, cookie '{}'", Strings.utf8ToString(syncCookie) );
            syncReq.setCookie( syncCookie );
        }

        searchRequest.addControl( syncReq );

        // Do the search
        SearchFuture sf = connection.searchAsync( searchRequest );

        Response resp = sf.get();

        while ( !( resp instanceof SearchResultDone ) && !sf.isCancelled() )
        {
            if ( resp instanceof SearchResultEntry )
            {
                handleSearchResult( ( SearchResultEntry ) resp );
            }
            else if ( resp instanceof SearchResultReference )
            {
                handleSearchReference( ( SearchResultReference ) resp );
            }
            else if ( resp instanceof IntermediateResponse )
            {
                handleSyncInfo( ( IntermediateResponse ) resp );
            }

            resp = sf.get();
        }

        ResultCodeEnum resultCode = handleSearchDone( ( SearchResultDone ) resp );

        LOG.debug( "sync operation returned result code {}", resultCode );
        if ( resultCode == ResultCodeEnum.NO_SUCH_OBJECT )
        {
            // log the error and handle it appropriately
            LOG.warn( "given replication base DN {} is not found on provider", config.getBaseDn() );
            if ( syncType == SynchronizationModeEnum.REFRESH_AND_PERSIST )
            {
                LOG.warn( "disconnecting the consumer running in refreshAndPersist mode from the provider" );
                disconnet();
            }
        }
        else if ( resultCode == ResultCodeEnum.E_SYNC_REFRESH_REQUIRED )
        {
            LOG.info( "unable to perform the content synchronization cause E_SYNC_REFRESH_REQUIRED" );
            try
            {
                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 );
                disconnet();
            }

            removeCookie();
            doSyncSearch( syncType, true );
        }
    }


    public void disconnet()
    {
        disconnected = true;

        try
        {
            if ( refreshThread != null )
            {
                refreshThread.stopRefreshing();
            }

            connection.unBind();
            LOG.info( "Unbound from the server {}", config.getProviderHost() );

            connection.close();
            LOG.info( "Connection closed for the server {}", config.getProviderHost() );

            connection = null;

            // persist the cookie
            storeCookie();

            // reset the cookie
            syncCookie = null;
        }
        catch ( Exception e )
        {
            LOG.error( "Failed to close the connection", e );
        }

    }


    /**
     * stores the cookie.
     */
    private void storeCookie()
    {
        if ( syncCookie == null )
        {
            return;
        }

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

        try
        {
            if ( config.isStoreCookieInFile() )
            {
                FileOutputStream fout = new FileOutputStream( cookieFile );
                fout.write( syncCookie.length );
                fout.write( syncCookie );
                fout.close();
            }
            else
            {
                EntryAttribute attr = cookieModLst.get( 0 ).getAttribute();
                attr.clear();
                attr.add( syncCookie );

                session.modify( configEntryDn, cookieModLst );
            }

            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", e );
        }
    }


    /**
     * read the cookie
     */
    private void readCookie()
    {
        try
        {
            if ( config.isStoreCookieInFile() )
            {
                if ( cookieFile.exists() && ( cookieFile.length() > 0 ) )
                {
                    FileInputStream fin = new FileInputStream( cookieFile );
                    syncCookie = new byte[fin.read()];
                    fin.read( syncCookie );
                    fin.close();

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

                    LOG.debug( "read the cookie from file: " + Strings.utf8ToString(syncCookie) );
                }
            }
            else
            {
                try
                {
                    Entry entry = session.lookup( configEntryDn, new String[]
                        { COOKIE_AT_TYPE.getName() } );
                    if ( entry != null )
                    {
                        EntryAttribute attr = entry.get( COOKIE_AT_TYPE );
                        if ( attr != null )
                        {
                            syncCookie = attr.getBytes();
                            lastSavedCookie = syncCookie;
                            LOG.debug( "loaded cookie from DIT" );
                        }
                    }
                }
                catch ( Exception e )
                {
                    // can be ignored, most likely happens if there is no entry with the given DN
                    // log in debug mode
                    LOG.debug( "Failed to read the cookie from the entry", e );
                }
            }
        }
        catch ( Exception e )
        {
            LOG.error( "Failed to read the cookie", e );
        }
    }


    /**
     * deletes the cookie and resets the syncCookie to null
     */
    public void removeCookie()
    {
        if ( config.isStoreCookieInFile() )
        {
            if ( cookieFile.exists() && ( cookieFile.length() > 0 ) )
            {
                boolean deleted = cookieFile.delete();
                LOG.info( "deleted cookie file {}", deleted );
            }
        }
        else
        {
            try
            {
                EntryAttribute cookieAttr = new DefaultEntryAttribute( COOKIE_AT_TYPE );
                Modification deleteCookieMod = new DefaultModification( ModificationOperation.REMOVE_ATTRIBUTE,
                    cookieAttr );
                List<Modification> deleteModLst = new ArrayList<Modification>();
                deleteModLst.add( deleteCookieMod );
                session.modify( configEntryDn, deleteModLst );
            }
            catch ( Exception e )
            {
                LOG.warn( "Failed to delete the cookie from the entry with DN {}", configEntryDn );
                LOG.warn( "{}", e );
            }
        }

        LOG.info( "resetting sync cookie" );

        syncCookie = null;
        lastSavedCookie = null;
    }


    private void applyModDnOperation( SyncModifyDnControl modDnControl ) throws Exception
    {
        SyncModifyDnType modDnType = modDnControl.getModDnType();

        DN entryDn = new DN( modDnControl.getEntryDn() );
        switch ( modDnType )
        {
            case MOVE:

                LOG.debug( "moving {} to the new parent {}", entryDn, modDnControl.getNewSuperiorDn() );

                session.move( entryDn, new DN( modDnControl.getNewSuperiorDn() ) );
                break;

            case RENAME:

                RDN newRdn = new RDN( modDnControl.getNewRdn() );
                boolean deleteOldRdn = modDnControl.isDeleteOldRdn();
                LOG.debug( "renaming the DN {} with new RDN {} and deleteOldRdn flag set to {}", new String[]
                    { entryDn.getName(), newRdn.getName(), String.valueOf( deleteOldRdn ) } );

                session.rename( entryDn, newRdn, deleteOldRdn );
                break;

            case MOVEANDRENAME:

                DN newParentDn = new DN( modDnControl.getNewSuperiorDn() );
                newRdn = new RDN( modDnControl.getNewRdn() );
                deleteOldRdn = modDnControl.isDeleteOldRdn();

                LOG.debug(
                    "moveAndRename on the DN {} with new newParent DN {}, new RDN {} and deleteOldRdn flag set to {}",
                    new String[]
                        { entryDn.getName(), newParentDn.getName(), newRdn.getName(), String.valueOf( deleteOldRdn ) } );

                session.moveAndRename( entryDn, newParentDn, newRdn, deleteOldRdn );
        }
    }


    private void modify( Entry remoteEntry ) throws Exception
    {
        Entry localEntry = session.lookup( remoteEntry.getDn() );

        remoteEntry.removeAttributes( MOD_IGNORE_AT );

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

        while ( itr.hasNext() )
        {
            EntryAttribute localAttr = itr.next();
            String attrId = localAttr.getId();
            Modification mod;
            EntryAttribute 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() ) );
            }
        }

        session.modify( remoteEntry.getDn(), mods );
    }


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

        for ( byte[] uuid : uuidList )
        {
            LOG.info( "uuid: {}", StringTools.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() );
            _deleteEntries_( uuidList, isRefreshPresent );
            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;
            _deleteEntries_( uuidList.subList( startIndex, startIndex + NODE_LIMIT ), isRefreshPresent );
        }

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


    /**
     * 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
     */
    private void _deleteEntries_( List<byte[]> limitedUuidList, boolean isRefreshPresent ) throws Exception
    {
        ExprNode filter = null;
        int size = limitedUuidList.size();
        if ( size == 1 )
        {
            String uuid = StringTools.uuidToString( limitedUuidList.get( 0 ) );
            filter = new EqualityNode<String>( SchemaConstants.ENTRY_UUID_AT,
                new org.apache.directory.shared.ldap.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 = StringTools.uuidToString( limitedUuidList.get( i ) );
                ExprNode uuidEqNode = new EqualityNode<String>( SchemaConstants.ENTRY_UUID_AT,
                    new org.apache.directory.shared.ldap.entry.StringValue( uuid ) );

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

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

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

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

        cursor.close();
    }

    /**
     * A Thread implementation for synchronizing the DIT in refreshOnly mode
     */
    private class RefresherThread extends Thread
    {
        private volatile boolean stop;


        public RefresherThread()
        {
            setDaemon( true );
        }


        public void run()
        {
            while ( !stop )
            {
                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" );
                }
                catch ( Exception e )
                {
                    LOG.error( "Failed to sync with refresh only mode", e );
                }
            }
        }


        public void stopRefreshing()
        {
            stop = true;
            // just incase if it is sleeping
            this.interrupt();
        }
    }


    /**
     * 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, ENTRY_UUID_ATOP_SET );
                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
                {
                    ClonedServerEntry 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;
        }
    }

}