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

import org.litesoft.GWT.eventbus.client.nonpublic.*;
import org.litesoft.GWT.eventbus.client.rpc.*;
import org.litesoft.commonfoundation.base.*;
import org.litesoft.logger.*;

import java.util.*;

public class ServerSideRemotePeerService implements RemotePeerService {
    private final MillisecTimeSource mMillisecTimeSource;
    private final Set<String> mAddPeerInterestsForClient = new HashSet<String>();
    private final Set<String> mRemovePeerInterestsForClient = new HashSet<String>();
    private final List<ChannelEventPackage> mChannelEventsForClient = new ArrayList<ChannelEventPackage>();
    private final Set<String> mAckPeerInterestsForClient = new HashSet<String>();

    private final Logger mLogger;

    private Long mGiveUpAfter = null;
    private PeerSupportingEventBus mServerEventBus;

    public ServerSideRemotePeerService( MillisecTimeSource pMillisecTimeSource, Logger pLogger, PeerSupportingEventBus pServerEventBus ) {
        mMillisecTimeSource = MillisecTimeSource.deNull( pMillisecTimeSource );
        mLogger = pLogger;
        mServerEventBus = (pServerEventBus != null) ? //
                          pServerEventBus : //
                          PeerSupportingEventBus.Null.INSTANCE;
    }

    public ServerSideRemotePeerService( Logger pLogger, PeerSupportingEventBus pServerEventBus ) {
        this( null, pLogger, pServerEventBus );
    }

    private synchronized void setGiveUpAfter( int pPollMillisecs ) {
        if ( pPollMillisecs > 0 ) {
            mGiveUpAfter = mMillisecTimeSource.now() + pPollMillisecs + 60000L * 5; // 5 minutes
        }
    }

    private boolean okToCollect() // synchronized by forwardTo()
    {
        return (mGiveUpAfter == null) || (mMillisecTimeSource.now() < mGiveUpAfter);
    }

    private synchronized PeerSupportingEventBus getServerEventBus() {
        return mServerEventBus;
    }

    private synchronized void clrServerEventBus() {
        mServerEventBus = PeerSupportingEventBus.Null.INSTANCE;
    }

    public void stopUnrequestedMessages() {
    }

    public synchronized RemotePeerService dispose() {
        mLogger.trace.log( "Dispose: ServerSideRemotePeerService" );
        mAddPeerInterestsForClient.clear();
        mRemovePeerInterestsForClient.clear();
        mChannelEventsForClient.clear();
        mServerEventBus = PeerSupportingEventBus.Null.INSTANCE;
        return RemotePeerService.Null.INSTANCE;
    }

    public synchronized boolean forwardTo( ChannelServicePackage pToClient ) {
        if ( pToClient != null ) {
            if ( pToClient.anyAddPeerInterests() ) {
                for ( String peerInterest : pToClient.getAddPeerInterests() ) {
                    mAddPeerInterestsForClient.add( peerInterest );
                }
            }
            if ( pToClient.anyRemovePeerInterests() ) {
                for ( String peerInterest : pToClient.getRemovePeerInterests() ) {
                    mRemovePeerInterestsForClient.add( peerInterest );
                }
            }
            if ( pToClient.anyEvents() ) {
                for ( ChannelEventPackage event : pToClient.getEvents() ) {
                    mChannelEventsForClient.add( event );
                }
            }
            if ( pToClient.anyAckPeerInterests() ) {
                for ( String ackInterest : pToClient.getAckPeerInterests() ) {
                    mAckPeerInterestsForClient.add( ackInterest );
                }
            }
        }
        return okToCollect();
    }

    public ChannelServicePackage processClientRequest( int pPollMillisecs, ChannelServicePackage pFromClient ) {
        setGiveUpAfter( pPollMillisecs );
        PeerSupportingEventBus eventBus = getServerEventBus();
        boolean peerOK;
        if ( (pFromClient != null) && pFromClient.isNotEmpty() ) {
            peerOK = eventBus.propagateFromRemotePeerService( pFromClient );
        } else {
            peerOK = eventBus.ping();
        }
        if ( !peerOK ) {
            clrServerEventBus();
        }
        synchronized ( this ) {
            return new ChannelServicePackage( extractAddPeerInterestsForClient(), //
                                              extractRemovePeerInterestsForClient(), //
                                              extractChannelEventsForClient(), //
                                              extractAckPeerInterestsForClient() );
        }
    }

    public String toString() {
        return "ServerSideRemotePeerService";
    }

    private String[] extractAddPeerInterestsForClient() // synchronized in processClientRequest()
    {
        String[] rv = mAddPeerInterestsForClient.toArray( new String[mAddPeerInterestsForClient.size()] );
        mAddPeerInterestsForClient.clear();
        return rv;
    }

    private String[] extractRemovePeerInterestsForClient() // synchronized in processClientRequest()
    {
        String[] rv =
                mRemovePeerInterestsForClient.toArray( new String[mRemovePeerInterestsForClient.size()] );
        mRemovePeerInterestsForClient.clear();
        return rv;
    }

    private ChannelEventPackage[] extractChannelEventsForClient() // synchronized in processClientRequest()
    {
        ChannelEventPackage[] rv =
                mChannelEventsForClient.toArray( new ChannelEventPackage[mChannelEventsForClient.size()] );
        mChannelEventsForClient.clear();
        return rv;
    }

    private String[] extractAckPeerInterestsForClient() // synchronized in processClientRequest()
    {
        String[] rv = mAckPeerInterestsForClient.toArray( new String[mAckPeerInterestsForClient.size()] );
        mAckPeerInterestsForClient.clear();
        return rv;
    }
}