/*
 *  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 org.apache.directory.server.core.event.DirectoryListener;
import org.apache.directory.server.core.event.EventType;
import org.apache.directory.server.core.interceptor.context.AddOperationContext;
import org.apache.directory.server.core.interceptor.context.DeleteOperationContext;
import org.apache.directory.server.core.interceptor.context.ModifyOperationContext;
import org.apache.directory.server.core.interceptor.context.MoveAndRenameOperationContext;
import org.apache.directory.server.core.interceptor.context.MoveOperationContext;
import org.apache.directory.server.core.interceptor.context.RenameOperationContext;
import org.apache.directory.server.i18n.I18n;
import org.apache.directory.server.ldap.LdapSession;
import org.apache.directory.shared.ldap.codec.controls.replication.syncStateValue.SyncStateValueDecorator;
import org.apache.directory.shared.ldap.codec.controls.replication.syncmodifydn.SyncModifyDnDecorator;
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.model.constants.SchemaConstants;
import org.apache.directory.shared.ldap.model.entry.Entry;
import org.apache.directory.shared.ldap.model.exception.LdapInvalidAttributeValueException;
import org.apache.directory.shared.ldap.model.message.AbandonListener;
import org.apache.directory.shared.ldap.model.message.AbandonableRequest;
import org.apache.directory.shared.ldap.model.message.SearchRequest;
import org.apache.directory.shared.ldap.model.message.SearchResultEntry;
import org.apache.directory.shared.ldap.model.message.SearchResultEntryImpl;
import org.apache.directory.shared.util.Strings;
import org.apache.mina.core.future.WriteFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 
 * modeled after PersistentSearchListener
 *  NOTE: doco is missing at many parts. Will be added once the functionality is satisfactory
 * 
 * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
 */
public class SyncReplSearchListener implements DirectoryListener, AbandonListener
{
    private static final Logger LOG = LoggerFactory.getLogger( SyncReplSearchListener.class );

    private LdapSession session;
    private SearchRequest req;

    private volatile boolean pushInRealTime;

    private final ReplicaEventLog clientMsgLog;


    SyncReplSearchListener( LdapSession session, SearchRequest req, ReplicaEventLog clientMsgLog,
        boolean pushInRealTime )
    {
        this.pushInRealTime = pushInRealTime;
        setSession( session );
        setReq( req );

        this.clientMsgLog = clientMsgLog;
    }


    public void setSession( LdapSession session )
    {
        this.session = session;
    }


    public void setReq( SearchRequest req )
    {
        this.req = req;
        if ( req != null )
        {
            req.addAbandonListener( this );
        }
    }


    public void abandon() throws Exception
    {
        if ( session != null )
        {
            // must abandon the operation 
            session.getCoreSession().getDirectoryService().getEventService().removeListener( this );
        }

        /*
         * From RFC 2251 Section 4.11:
         * 
         * In the event that a server receives an Abandon Request on a Search  
         * operation in the midst of transmitting responses to the Search, that
         * server MUST cease transmitting entry responses to the abandoned
         * request immediately, and MUST NOT send the SearchResultDone. Of
         * course, the server MUST ensure that only properly encoded LDAPMessage
         * PDUs are transmitted. 
         * 
         * SO DON'T SEND BACK ANYTHING!!!!!
         */
    }


    public void requestAbandoned( AbandonableRequest req )
    {
        try
        {
            abandon();
        }
        catch ( Exception e )
        {
            LOG.error( I18n.err( I18n.ERR_164 ), e );
        }
    }


    public void entryAdded( AddOperationContext addContext )
    {
        Entry entry = addContext.getEntry();

        LOG.debug( "sending added entry {}", entry.getDn() );

        try
        {
            if ( pushInRealTime )
            {

                SearchResultEntry respEntry = new SearchResultEntryImpl( req.getMessageId() );
                respEntry.setObjectName( entry.getDn() );
                respEntry.setEntry( entry );

                SyncStateValueDecorator syncAdd = new SyncStateValueDecorator( 
                    session.getLdapServer().getDirectoryService().getLdapCodecService() );
                syncAdd.setSyncStateType( SyncStateTypeEnum.ADD );
                syncAdd
                    .setEntryUUID( Strings.uuidToBytes(entry.get(SchemaConstants.ENTRY_UUID_AT).getString()) );
                syncAdd.setCookie( getCookie( entry ) );
                respEntry.addControl( syncAdd );

                WriteFuture future = session.getIoSession().write( respEntry );
                handleWriteFuture( future, entry, EventType.ADD, null );
            }
            else
            {
                clientMsgLog.log( EventType.ADD, addContext.getEntry() );
            }
        }
        catch ( LdapInvalidAttributeValueException e )
        {
            // shouldn't happen
            LOG.error( e.getMessage(), e );
        }
    }


    public void entryDeleted( DeleteOperationContext deleteContext )
    {
        sendDeletedEntry( deleteContext.getEntry() );
    }


    private void sendDeletedEntry( Entry entry )
    {
        LOG.debug( "sending deleted entry {}", entry.getDn() );

        try
        {
            if ( pushInRealTime )
            {
                SearchResultEntry respEntry = new SearchResultEntryImpl( req.getMessageId() );
                respEntry.setObjectName( entry.getDn() );
                respEntry.setEntry( entry );

                SyncStateValueDecorator syncDelete = new SyncStateValueDecorator( 
                    session.getLdapServer().getDirectoryService().getLdapCodecService() );
                syncDelete.setSyncStateType( SyncStateTypeEnum.DELETE );
                syncDelete.setEntryUUID( Strings.uuidToBytes(entry.get(SchemaConstants.ENTRY_UUID_AT)
                        .getString()) );
                syncDelete.setCookie( getCookie( entry ) );
                respEntry.addControl( syncDelete );

                WriteFuture future = session.getIoSession().write( respEntry );

                handleWriteFuture( future, entry, EventType.DELETE, null );
            }
            else
            {
                clientMsgLog.log( EventType.DELETE, entry );
            }
        }
        catch ( LdapInvalidAttributeValueException e )
        {
            // shouldn't happen
            LOG.error( e.getMessage(), e );
        }
    }


    public void entryModified( ModifyOperationContext modifyContext )
    {
        Entry alteredEntry = modifyContext.getAlteredEntry();

        LOG.debug( "sending modified entry {}", alteredEntry.getDn() );

        try
        {
            if ( pushInRealTime )
            {

                SearchResultEntry respEntry = new SearchResultEntryImpl( req.getMessageId() );
                respEntry.setObjectName( modifyContext.getDn() );
                respEntry.setEntry( alteredEntry );

                SyncStateValueDecorator syncModify = new SyncStateValueDecorator( 
                    session.getLdapServer().getDirectoryService().getLdapCodecService() );
                syncModify.setSyncStateType( SyncStateTypeEnum.MODIFY );
                syncModify.setEntryUUID( Strings.uuidToBytes(alteredEntry.get(SchemaConstants.ENTRY_UUID_AT)
                        .getString()) );
                syncModify.setCookie( getCookie( alteredEntry ) );
                respEntry.addControl( syncModify );

                WriteFuture future = session.getIoSession().write( respEntry );

                // store altered entry cause that holds the updated CSN
                handleWriteFuture( future, alteredEntry, EventType.MODIFY, null );
            }
            else
            {
                clientMsgLog.log( EventType.MODIFY, modifyContext.getAlteredEntry() );
            }
        }
        catch ( Exception e )
        {
            LOG.error( e.getMessage(), e );
        }
    }


    public void entryMoved( MoveOperationContext moveContext )
    {
        Entry entry = moveContext.getEntry();

        LOG.debug( "sending moved entry {}", entry.getDn() );

        try
        {
            if ( !moveContext.getNewSuperior().isChildOf( clientMsgLog.getSearchCriteria().getBase() ) )
            {
                sendDeletedEntry( moveContext.getEntry() );
                return;
            }

            SyncModifyDnDecorator modDnControl = new SyncModifyDnDecorator( 
                session.getLdapServer().getDirectoryService().getLdapCodecService(),
                SyncModifyDnType.MOVE );
            modDnControl.setEntryDn( moveContext.getDn().getNormName() );
            modDnControl.setNewSuperiorDn( moveContext.getNewSuperior().getNormName() );

            if ( pushInRealTime )
            {
                SearchResultEntry respEntry = new SearchResultEntryImpl( req.getMessageId() );
                respEntry.setObjectName( moveContext.getDn() );
                respEntry.setEntry( entry );

                SyncStateValueDecorator syncModify = new SyncStateValueDecorator(
                    session.getLdapServer().getDirectoryService().getLdapCodecService() );
                syncModify.setSyncStateType( SyncStateTypeEnum.MODDN );
                syncModify.setEntryUUID( Strings.uuidToBytes(entry.get(SchemaConstants.ENTRY_UUID_AT)
                        .getString()) );
                syncModify.setCookie( getCookie( entry ) );
                respEntry.addControl( syncModify );
                respEntry.addControl( modDnControl );

                WriteFuture future = session.getIoSession().write( respEntry );

                handleWriteFuture( future, entry, null, modDnControl );
            }
            else
            {
                clientMsgLog.log( new ReplicaEventMessage( modDnControl, moveContext.getEntry() ) );
            }
        }
        catch ( Exception e )
        {
            LOG.error( e.getMessage(), e );
        }
    }


    public void entryMovedAndRenamed( MoveAndRenameOperationContext moveAndRenameContext )
    {

        LOG.debug( "sending moveAndRenamed entry {}", moveAndRenameContext.getDn() );

        try
        {
            if ( !moveAndRenameContext.getNewSuperiorDn().isChildOf( clientMsgLog.getSearchCriteria().getBase() ) )
            {
                sendDeletedEntry( moveAndRenameContext.getEntry() );
                return;
            }

            SyncModifyDnDecorator modDnControl = new SyncModifyDnDecorator( 
                session.getLdapServer().getDirectoryService().getLdapCodecService(),
                SyncModifyDnType.MOVEANDRENAME );
            modDnControl.setEntryDn( moveAndRenameContext.getDn().getNormName() );
            modDnControl.setNewSuperiorDn( moveAndRenameContext.getNewSuperiorDn().getNormName() );
            modDnControl.setNewRdn( moveAndRenameContext.getNewRdn().getNormName() );
            modDnControl.setDeleteOldRdn( moveAndRenameContext.getDeleteOldRdn() );

            if ( pushInRealTime )
            {
                Entry alteredEntry = moveAndRenameContext.getModifiedEntry();

                SearchResultEntry respEntry = new SearchResultEntryImpl( req.getMessageId() );
                respEntry.setObjectName( moveAndRenameContext.getModifiedEntry().getDn() );
                respEntry.setEntry( alteredEntry );

                SyncStateValueDecorator syncModify = new SyncStateValueDecorator(
                    session.getLdapServer().getDirectoryService().getLdapCodecService() );
                syncModify.setSyncStateType( SyncStateTypeEnum.MODDN );
                syncModify.setEntryUUID( Strings.uuidToBytes(alteredEntry.get(SchemaConstants.ENTRY_UUID_AT)
                        .getString()) );
                syncModify.setCookie( getCookie( alteredEntry ) );
                respEntry.addControl( syncModify );
                respEntry.addControl( modDnControl );

                WriteFuture future = session.getIoSession().write( respEntry );

                handleWriteFuture( future, alteredEntry, null, modDnControl );
            }
            else
            {
                clientMsgLog.log( new ReplicaEventMessage( modDnControl, moveAndRenameContext.getEntry() ) );
            }
        }
        catch ( Exception e )
        {
            LOG.error( e.getMessage(), e );
        }
    }


    public void entryRenamed( RenameOperationContext renameContext )
    {
        Entry entry = renameContext.getEntry();

        LOG.debug( "sending renamed entry {}", entry.getDn() );

        try
        {
            SyncModifyDnDecorator modDnControl = new SyncModifyDnDecorator(
                session.getLdapServer().getDirectoryService().getLdapCodecService() );
            modDnControl.setModDnType( SyncModifyDnType.RENAME );
            modDnControl.setEntryDn( renameContext.getDn().getName() );
            modDnControl.setNewRdn( renameContext.getNewRdn().getName() );
            modDnControl.setDeleteOldRdn( renameContext.getDeleteOldRdn() );

            if ( pushInRealTime )
            {
                SearchResultEntry respEntry = new SearchResultEntryImpl( req.getMessageId() );
                respEntry.setObjectName( entry.getDn() );
                respEntry.setEntry( entry );

                SyncStateValueDecorator syncModify = new SyncStateValueDecorator(
                    session.getLdapServer().getDirectoryService().getLdapCodecService() );
                syncModify.setSyncStateType( SyncStateTypeEnum.MODDN );
                syncModify.setEntryUUID( Strings.uuidToBytes(entry.get(SchemaConstants.ENTRY_UUID_AT)
                        .getString()) );
                syncModify.setCookie( getCookie( renameContext.getModifiedEntry() ) );
                respEntry.addControl( syncModify );
                respEntry.addControl( modDnControl );

                WriteFuture future = session.getIoSession().write( respEntry );

                handleWriteFuture( future, renameContext.getModifiedEntry(), null, modDnControl );
            }
            else
            {
                clientMsgLog.log( new ReplicaEventMessage( modDnControl, renameContext.getEntry() ) );
            }
        }
        catch ( Exception e )
        {
            LOG.error( e.getMessage(), e );
        }
    }


    public boolean isPushInRealTime()
    {
        return pushInRealTime;
    }


    public void setPushInRealTime( boolean pushInRealTime )
    {
        this.pushInRealTime = pushInRealTime;
    }


    private byte[] getCookie( Entry entry ) throws LdapInvalidAttributeValueException
    {
        String csn = entry.get( SchemaConstants.ENTRY_CSN_AT ).getString();
        return Strings.getBytesUtf8(clientMsgLog.getId() + SyncReplProvider.REPLICA_ID_DELIM + csn);
    }


    private void handleWriteFuture( WriteFuture future, Entry entry, EventType event, SyncModifyDnDecorator modDnControl )
    {
        future.awaitUninterruptibly();
        if ( !future.isWritten() )
        {
            LOG.error( "Failed to write to the consumer {}", clientMsgLog.getId() );
            LOG.error( "", future.getException() );

            // set realtime push to false, will be set back to true when the client
            // comes back and sends another request this flag will be set to true
            pushInRealTime = false;

            if ( modDnControl != null )
            {
                clientMsgLog.log( new ReplicaEventMessage( modDnControl, entry ) );
            }
            else
            {
                clientMsgLog.log( event, entry );
            }
        }
    }
}