package org.realityforge.replicant.client.ee;

import java.util.HashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Supplier;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.enterprise.concurrent.ContextService;
import javax.enterprise.concurrent.ManagedScheduledExecutorService;
import org.realityforge.gwt.webpoller.client.WebPoller;
import org.realityforge.replicant.client.ChangeSet;
import org.realityforge.replicant.client.transport.CacheService;
import org.realityforge.replicant.client.transport.ClientSession;
import org.realityforge.replicant.client.transport.DataLoaderListener;
import org.realityforge.replicant.client.transport.DataLoaderServiceConfig;
import org.realityforge.replicant.client.transport.WebPollerDataLoaderService;

public abstract class EeDataLoaderService
  extends WebPollerDataLoaderService
{
  private static final int DEFAULT_EE_CHANGES_TO_PROCESS_PER_TICK = 10000;
  private static final int DEFAULT_EE_LINKS_TO_PROCESS_PER_TICK = 10000;

  @Nullable
  private ScheduledFuture _future;
  private final InMemoryCacheService _cacheService = new InMemoryCacheService();
  private final ReentrantReadWriteLock _lock = new ReentrantReadWriteLock( true );
  private DataLoaderServiceConfig _config;
  private final HashMap<DataLoaderListener, DataLoaderListener> _listenerMap = new HashMap<>();

  protected EeDataLoaderService()
  {
    setChangesToProcessPerTick( DEFAULT_EE_CHANGES_TO_PROCESS_PER_TICK );
    setLinksToProcessPerTick( DEFAULT_EE_LINKS_TO_PROCESS_PER_TICK );
  }

  protected void postConstruct()
  {
    _config = getContextService().
      createContextualProxy( new EeDataLoaderServiceConfigImpl( getJndiPrefix() ), DataLoaderServiceConfig.class );
  }

  @Nonnull
  @Override
  protected DataLoaderServiceConfig config()
  {
    return _config;
  }

  @Nonnull
  protected ReentrantReadWriteLock getLock()
  {
    return _lock;
  }

  @Nonnull
  protected abstract ContextService getContextService();

  @Nonnull
  @Override
  protected CacheService getCacheService()
  {
    return _cacheService;
  }

  @Nonnull
  protected abstract ManagedScheduledExecutorService getManagedScheduledExecutorService();

  @Override
  @Nonnull
  protected ChangeSet parseChangeSet( @Nonnull final String rawJsonData )
  {
    return ChangeSetDTO.asChangeSet( rawJsonData );
  }

  @Override
  protected void doScheduleDataLoad()
  {
    _future = getManagedScheduledExecutorService().
      scheduleAtFixedRate( this::scheduleTick, 0, 1, TimeUnit.MILLISECONDS );
  }

  private void scheduleTick()
  {
    if ( !stepDataLoad() && null != _future )
    {
      final ScheduledFuture future = _future;
      _future = null;
      future.cancel( false );
    }
  }

  /**
   * Return the JNDI prefix used to derive optional configuration.
   * Typically it returns a string such as: "myapp/replicant/client"
   */
  @Nonnull
  protected abstract String getJndiPrefix();


  @Override
  public void connect()
  {
    withLock( getLock().writeLock(), super::connect );
  }

  @Override
  public void disconnect()
  {
    withLock( getLock().writeLock(), super::disconnect );
  }

  @Override
  protected void startPolling()
  {
    withLock( getLock().writeLock(), () -> super.startPolling() );
  }

  @Override
  protected void stopPolling()
  {
    withLock( getLock().writeLock(), () -> super.stopPolling() );
  }

  @Nonnull
  @Override
  public State getState()
  {
    return withLock( getLock().readLock(), super::getState );
  }

  @Override
  protected void resumeWebPoller()
  {
    withLock( getLock().writeLock(), () -> super.resumeWebPoller() );
  }

  @Override
  protected void pauseWebPoller()
  {
    withLock( getLock().writeLock(), () -> super.pauseWebPoller() );
  }

  @Nonnull
  @Override
  protected WebPoller getWebPoller()
  {
    return withLock( getLock().readLock(), () -> super.getWebPoller() );
  }

  @Override
  protected boolean stepDataLoad()
  {
    return withLock( getLock().writeLock(), () -> super.stepDataLoad() );
  }

  @Override
  protected void setSession( @Nullable final ClientSession session, @Nullable final Runnable postAction )
  {
    withLock( getLock().writeLock(), () -> super.setSession( session, postAction ) );
  }

  @Override
  public void scheduleDataLoad()
  {
    withLock( getLock().writeLock(), super::scheduleDataLoad );
  }

  private <R> R withLock( final Lock lock, final Supplier<R> action )
  {
    lock.lock();
    try
    {
      return action.get();
    }
    finally
    {
      lock.unlock();
    }
  }

  private void withLock( final Lock lock, final Runnable action )
  {
    lock.lock();
    try
    {
      action.run();
    }
    finally
    {
      lock.unlock();
    }
  }

  @Override
  public boolean addDataLoaderListener( @Nonnull final DataLoaderListener listener )
  {
    synchronized ( _listenerMap )
    {
      if ( !_listenerMap.containsKey( listener ) )
      {
        final DataLoaderListener wrapped = wrap( listener );
        _listenerMap.put( listener, wrapped );
        return super.addDataLoaderListener( wrapped );
      }
      else
      {
        return false;
      }
    }
  }

  @Override
  public boolean removeDataLoaderListener( @Nonnull final DataLoaderListener listener )
  {
    synchronized ( _listenerMap )
    {
      final DataLoaderListener wrapped = _listenerMap.remove( listener );
      return null != wrapped && super.removeDataLoaderListener( wrapped );
    }
  }

  @Nonnull
  private DataLoaderListener wrap( @Nonnull final DataLoaderListener listener )
  {
    return getContextService().createContextualProxy( listener, DataLoaderListener.class );
  }
}

@@ -2755,18 +2755,103 @@
 aLoad() 
-&&
+)%0A    %7B%0A      cancelSchedule();%0A    %7D%0A  %7D%0A%0A  private void cancelSchedule()%0A  %7B%0A    if (
  null !=
@@ -3362,23 +3362,58 @@
 Lock(), 
-super::
+() -%3E%0A    %7B%0A      cancelSchedule();%0A      
 disconne
@@ -3410,24 +3410,33 @@
   disconnect
+();%0A    %7D
  );%0A  %7D%0A%0A  @

Make sure the dataloader schedule is disabled during disconnect