/*
 * 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.transport;

import com.google.common.base.Preconditions;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.RejectedExecutionException;
import java.util.function.Consumer;

/**
 * A class to hold all listeners belong to given type T.
 *
 * @param <T> the listener type
 */
public class Listeners<T> implements Iterable<Listener<T>> {
  private final List<ListenerHolder> mListeners = new CopyOnWriteArrayList();

  /**
   * Empty constructor to construct {@link Listeners}.
   */
  public Listeners() {}

  /**
   * @return the current holding listeners
   */
  public int size() {
    return mListeners.size();
  }

  /**
   * Adds a new listener.
   *
   * @param listener the listener to add
   * @return listener holder of this listener
   */
  public Listener<T> add(Consumer<T> listener) {
    Preconditions.checkNotNull(listener, "listener should not be null");
    ListenerHolder holder
        = new ListenerHolder(listener, GrpcMessagingContext.currentContext());
    mListeners.add(holder);
    return holder;
  }

  /**
   * Accepts an event.
   *
   * @param event the event to accept
   * @return the listener accepted future
   */
  public CompletableFuture<Void> accept(T event) {
    List<CompletableFuture<Void>> futures = new ArrayList(mListeners.size());
    for (ListenerHolder listener : mListeners) {
      if (listener.getContext() != null) {
        futures.add(listener.getContext().execute(() -> listener.getListener().accept(event)));
      } else {
        listener.getListener().accept(event);
      }
    }
    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
  }

  @Override
  @SuppressWarnings("unchecked")
  public Iterator<Listener<T>> iterator() {
    return (Iterator) mListeners.iterator();
  }

  /**
   * The listener holder with messaging context.
   */
  public class ListenerHolder implements Listener<T> {
    private final Consumer<T> mListener;
    private final GrpcMessagingContext mContext;

    ListenerHolder(Consumer<T> listener, GrpcMessagingContext context) {
      mListener = listener;
      mContext = context;
    }

    @Override
    public void accept(T event) {
      if (mContext != null) {
        try {
          mContext.executor().execute(() -> {
            mListener.accept(event);
          });
        } catch (RejectedExecutionException var3) {
          // Ignore the rejected exception
        }
      } else {
        mListener.accept(event);
      }
    }

    @Override
    public void close() {
      mListeners.remove(this);
    }

    private GrpcMessagingContext getContext() {
      return mContext;
    }

    private Consumer<T> getListener() {
      return mListener;
    }
  }
}
