/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.master;

import alluxio.grpc.NodeState;
import alluxio.resource.LockResource;
import alluxio.util.interfaces.Scoped;

import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Base class for implementing primary selectors.
 *
 * This class handles the synchronization logic of getting the current state, waiting for a certain
 * state, or registering a state change listener. Subclasses just need to call
 * {@link #setState(NodeState)} when they detect a state change. The selector starts off in STANDBY
 * state.
 */
@ThreadSafe
public abstract class AbstractPrimarySelector implements PrimarySelector {
  private static final Logger LOG = LoggerFactory.getLogger(AbstractPrimarySelector.class);

  @GuardedBy("mListeners")
  private final Set<AtomicReference<Consumer<NodeState>>> mListeners = new HashSet<>();

  private final Lock mStateLock = new ReentrantLock();
  private final Condition mStateCond = mStateLock.newCondition();
  @GuardedBy("mStateLock")
  private volatile NodeState mState = NodeState.STANDBY;

  protected final void setState(NodeState state) {
    try (LockResource lr = new LockResource(mStateLock)) {
      mState = state;
      mStateCond.signalAll();
      synchronized (mListeners) {
        mListeners.forEach(listener -> listener.get().accept(state));
      }
      LOG.info("Primary selector transitioning to {}", state);
    }
  }

  @Override
  public final NodeState getState() {
    try (LockResource lr = new LockResource(mStateLock)) {
      return mState;
    }
  }

  @Override
  public final NodeState getStateUnsafe() {
    return mState;
  }

  @Override
  public final Scoped onStateChange(Consumer<NodeState> listener) {
    // Wrap listeners in a reference of our own to guarantee uniqueness for listener references.
    AtomicReference<Consumer<NodeState>> listenerRef = new AtomicReference<>(listener);
    synchronized (mListeners) {
      Preconditions.checkState(mListeners.add(listenerRef), "listener already exists");
    }
    return () -> {
      synchronized (mListeners) {
        mListeners.remove(listenerRef);
      }
    };
  }

  @Override
  public final void waitForState(NodeState state) throws InterruptedException {
    try (LockResource lr = new LockResource(mStateLock)) {
      while (mState != state) {
        mStateCond.await();
      }
    }
  }
}
