// This Source Code is in the Public Domain per: http://litesoft.org/License.txt
package org.litesoft.GWT.eventbus.client.nonpublic;

import java.util.*;

import org.litesoft.GWT.eventbus.client.*;
import org.litesoft.GWT.eventbus.client.rpc.*;
import org.litesoft.core.typeutils.*;
import org.litesoft.core.util.*;
import org.litesoft.logger.*;

public class StatePeerSupportingEventBusRegular extends StatePeerSupportingEventBus
{
    private final PeerInterestManager mPeerInterestManager;
    private final SubscriptionManager mSubscriptionManager;
    private final EventBus mEventBus;
    private EventBusMessageCollector mCollector;
    private RemotePeerService mRemotePeerService;
    private EventConstrictor mEventConstrictor;
    private IProcessingManager mProcessingManager;

    /* package friendly */
    StatePeerSupportingEventBusRegular( Logger pLogger, PeerSupportingEventBus pPeerSupportingEventBus,
                                        EventConstrictor pEventConstrictor, EventBus pEventBus,
                                        IProcessingManager pProcessingManager,
                                        RemotePeerService pRemotePeerService,
                                        EventBusMessageCollector pCollector,
                                        SavedEventsExpectingResponseTracker pSavedEventsExpectingResponseTracker )
    {
        super( pLogger, pPeerSupportingEventBus, pSavedEventsExpectingResponseTracker );

        mProcessingManager = pProcessingManager;

        mPeerInterestManager = new PeerInterestManager( pPeerSupportingEventBus.getForm() );
        mSubscriptionManager = new SubscriptionManager( pPeerSupportingEventBus.getForm() );

        mRemotePeerService = (pRemotePeerService != null) ? //
                             pRemotePeerService : //
                             RemotePeerService.Null.INSTANCE;
        Objects.assertNotNull( "EventConstrictor", mEventConstrictor = pEventConstrictor );
        Objects.assertNotNull( "EventBus", mEventBus = pEventBus );
        Objects.assertNotNull( "Collector", mCollector = pCollector );

        new LoggingStringSet( pLogger, "AckPeerInterests" );
    }

    public StatePeerSupportingEventBusRegular( Logger pLogger, PeerSupportingEventBus pPeerSupportingEventBus,
                                               EventConstrictor pEventConstrictor, EventBus pEventBus,
                                               IProcessingManager pProcessingManager,
                                               RemotePeerService pRemotePeerService )
    {
        this( pLogger, pPeerSupportingEventBus, //
              pEventConstrictor, pEventBus, //
              pProcessingManager, //
              pRemotePeerService, //
              new EventBusMessageCollector( pLogger ), //
              new SavedEventsExpectingResponseTracker( pLogger ) );
    }

    public void disposing()
    {
        mRemotePeerService.stopUnrequestedMessages();
        mProcessingManager.waitTillNoOthers();
    }

    public synchronized void dispose()
    {
        mPeerInterestManager.clear();
        mSubscriptionManager.clear();
        clearPeerService();
        super.dispose();
    }

    public IStatePeerSupportingEventBus suspendedToRegular( EventConstrictor pEventConstrictor,
                                                            EventBus pEventBus )
    {
        throw new IllegalStateException( "Not Suspended" );
    }

    public void publishExpectingResponse( EventPackage pPackage )
    {
        String sourceName = pPackage.getSourceName();
        synchronized ( this )
        {
            if ( mSubscriptionManager.hasPeerAcknowledged( sourceName ) )
            {
                getEventBusMessageCollector().addPublishedEvent( pPackage );
            }
            else
            {
                mSavedEventsExpectingResponseTracker.add( pPackage );
            }
        }
        releaseCollected();
    }

    public boolean releaseCollected()
    {
        boolean rv = true;
        if ( mProcessingManager.incThreadsAndIndicateIfWeAreFirst() )
        {
            for ( EventBusMessageCollector collector; null != (collector = grabCollector()); )
            {
                if ( !process( collector ) )
                {
                    rv = false;
                }
            }
        }
        mProcessingManager.decThreads();
        return rv;
    }

    protected synchronized EventBusMessageCollector getEventBusMessageCollector()
    {
        return mCollector;
    }

    private synchronized EventBusMessageCollector grabCollector()
    {
        EventBusMessageCollector rv = null;
        if ( !mCollector.isEmpty() )
        {
            rv = mCollector;
            mCollector = new EventBusMessageCollector( mLogger );
        }
        return rv;
    }

    private synchronized void clearPeerService()
    {
        mRemotePeerService = mRemotePeerService.dispose();
    }

    private synchronized RemotePeerService getRemotePeerService()
    {
        return mRemotePeerService;
    }

    protected boolean process( EventBusMessageCollector pCollector )
    {
        List zAddPeerInterests4Remote = new ArrayList();
        List zRemovePeerInterests4Remote = new ArrayList();
        List zEvents4Remote = new ArrayList();

        for ( Iterator it = pCollector.getSubscribes(); it.hasNext(); )
        {
            EventSubscriptionStructure member = (EventSubscriptionStructure) it.next();
            if ( mSubscriptionManager.subscribe( member.getSubscribeWith(), member.getEventListener() ) )
            {
                zAddPeerInterests4Remote.add( member.getSubscribeWith() );
            }
        }
        for ( Iterator it = pCollector.getUnsubscribes(); it.hasNext(); )
        {
            EventSubscriptionStructure member = (EventSubscriptionStructure) it.next();
            if ( mSubscriptionManager.unsubscribe( member.getSubscribeWith(), member.getEventListener() ) )
            {
                zRemovePeerInterests4Remote.add( member.getSubscribeWith() );
            }
        }

        Set zAddedRemotePeerInterests = pCollector.getAddRemotePeerInterests();
        mPeerInterestManager.add( zAddedRemotePeerInterests );

        mPeerInterestManager.remove( pCollector.getRemoveRemotePeerInterests() );

        for ( Iterator it = pCollector.getFromRemotePeerEvents(); it.hasNext(); )
        {
            EventPackage member = (EventPackage) it.next();
            String error = mEventConstrictor.isLocalDistributionFromPeerOK( member );
            if ( error == null )
            {
                distributeToLocals( member );
            }
            else
            {
                mEventBus.handleUnexpectedProblem( "EventConstrictor:" + error,
                                                   new Problem( "DistributionNotAllowed", //
                                                                error, //
                                                                member.getSourceName(), //
                                                                member.getTargetName(), //
                                                                member.getType() ) );
            }
        }

        for ( Iterator it = pCollector.getPublishedEvents(); it.hasNext(); )
        {
            EventPackage member = (EventPackage) it.next();
            publishFromLocal( zEvents4Remote, member );
        }

        for ( Iterator it = pCollector.getFromRemoteAckPeerInterests(); it.hasNext(); )
        {
            peerHasAcked( it.next().toString() );
        }

        if ( zAddPeerInterests4Remote.isEmpty() && zRemovePeerInterests4Remote.isEmpty() &&
             zEvents4Remote.isEmpty() && zAddedRemotePeerInterests.isEmpty() )
        {
            return true;
        }

        ChannelServicePackage csp = new ChannelServicePackage( toStringArray( zAddPeerInterests4Remote ),
                                                               toStringArray( zRemovePeerInterests4Remote ),
                                                               toChannelEventArray( zEvents4Remote ),
                                                               toStringArray( zAddedRemotePeerInterests ) );
        return getRemotePeerService().forwardTo( csp );
    }

    private ChannelEventPackage[] toChannelEventArray( List pList )
    {
        ChannelEventPackage[] rv = new ChannelEventPackage[pList.size()];
        for ( int i = 0; i < pList.size(); i++ )
        {
            rv[i] = (ChannelEventPackage) pList.get( i );
        }
        return rv;
    }

    private String[] toStringArray( Collection pCollection )
    {
        return Strings.toArray( pCollection );
    }

    private void publishFromLocal( List pPackages4Remote, EventPackage pPackage )
    {
        distributeToLocals( pPackage );
        if ( mPeerInterestManager.isPeerInterested( pPackage ) )
        {
            pPackages4Remote.add( pPackage );
        }
    }

    private void distributeToLocals( EventPackage pPackage )
    {
        EventPackageListener[] listeners = mSubscriptionManager.getListenersFor( pPackage.getTargetName() );
        if ( listeners != null )
        {
            mLogger.trace.log( "(distributeToLocals):", pPackage );
            //noinspection ForLoopReplaceableByForEach
            for ( int i = 0; i < listeners.length; i++ )
            {
                EventPackageListener zListener = listeners[i];
                try
                {
                    zListener.packageReceivedVia( pPackage, mEventBus );
                }
                catch ( RuntimeException e )
                {
                    String zListenerType = Objects.classNameOf( zListener );
                    mLogger.error.log( e, zListenerType, pPackage );
                    // Note: GWT does not provide stack traces in client mode
                    mEventBus.handleUnexpectedProblem( Objects.justClassName( zListenerType ), //
                                                       new Problem( e, "UnableDistributeToLocals",
                                                                    pPackage.toString() ) );
                }
            }
        }
    }

    public String toString()
    {
        return super.toString() + "Regular";
    }

    private void peerHasAcked( String pTargetName )
    {
        EventPackage[] packages;
        synchronized ( this )
        {
            mSubscriptionManager.setPeerAcknowledged( pTargetName );
            packages = mSavedEventsExpectingResponseTracker.getAndReleaseEventsFor( pTargetName );
        }
        mEventBus.publish( packages );
    }
}