/*
 * Copyright 2015 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.atomix.catalyst.util;

import io.atomix.catalyst.util.concurrent.ThreadContext;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;

/**
 * Utility for managing a set of listeners.
 *
 * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
 */
public class Listeners<T> implements Iterable<Listener<T>> {
  private final List<Listener<T>> listeners = new CopyOnWriteArrayList<>();

  /**
   * Returns the number of registered listeners.
   *
   * @return The number of registered listeners.
   */
  public int size() {
    return listeners.size();
  }

  /**
   * Adds a listener to the set of listeners.
   *
   * @param listener The listener to add.
   * @return The listener context.
   * @throws NullPointerException if {@code listener} is null
   */
  public Listener<T> add(Consumer<T> listener) {
    Assert.notNull(listener, "listener");
    
    ThreadContext context = ThreadContext.currentContext();

    Listener<T> wrapper = new Listener<T>() {
      @Override
      public void accept(T event) {
        if (context != null) {
          context.executor().execute(() -> listener.accept(event));
        } else {
          listener.accept(event);
        }
      }

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

    listeners.add(wrapper);
    return wrapper;
  }

  @Override
  public Iterator<Listener<T>> iterator() {
    return listeners.iterator();
  }

}

@@ -696,16 +696,44 @@
 ntext;%0A%0A
+import java.util.ArrayList;%0A
 import j
@@ -770,24 +770,71 @@
 .util.List;%0A
+import java.util.concurrent.CompletableFuture;%0A
 import java.
@@ -1119,19 +1119,22 @@
 Listener
-%3CT%3E
+Holder
 %3E listen
@@ -1642,36 +1642,60 @@
     
-%0A    ThreadContext context =
+ListenerHolder holder = new ListenerHolder(listener,
  Thr
@@ -1725,77 +1725,247 @@
 xt()
+)
 ;
-%0A
 %0A    
-L
+l
 istener
-%3CT%3E wrapper = new Listener%3CT%3E() %7B
+s.add(holder);%0A    return holder;%0A  %7D%0A%0A  /**%0A   * Applies an event to all listeners.
 %0A   
+*%0A
    
-@Override
+* @param event The event to apply.%0A   * @return A completable future to be completed once all listeners have been completed.
 %0A   
- 
+*/%0A
   pu
@@ -1973,19 +1973,37 @@
 lic 
-v
+CompletableFuture%3CV
 oid
+%3E
  a
-ccept
+pply
 (T e
@@ -2018,63 +2018,205 @@
     
-    if (context != null) %7B%0A          context.executor()
+List%3CCompletableFuture%3CVoid%3E%3E futures = new ArrayList%3C%3E(listeners.size());%0A    for (ListenerHolder listener : listeners) %7B%0A      if (listener.context != null) %7B%0A        futures.add(listener.context
 .exe
@@ -2226,16 +2226,25 @@
 e(() -%3E 
+listener.
 listener
@@ -2258,20 +2258,19 @@
 (event))
+)
 ;%0A
-  
       %7D 
@@ -2280,26 +2280,33 @@
 e %7B%0A        
-  
+listener.
 listener.acc
@@ -2317,24 +2317,28 @@
 event);%0A    
+  %7D%0A
     %7D%0A      
@@ -2339,17 +2339,101 @@
     
+return CompletableFuture.allOf(futures.toArray(new CompletableFuture%5Bfutures.size()%5D));%0A
   %7D%0A%0A
-    
   @O
@@ -2446,135 +2446,449 @@
 e%0A  
-    public void close() %7B%0A        listeners.remove(this);%0A      %7D%0A    %7D;%0A%0A    listeners.add(wrapper);%0A    return wrapper;%0A
+@SuppressWarnings(%22unchecked%22)%0A  public Iterator%3CListener%3CT%3E%3E iterator() %7B%0A    return (Iterator) listeners.iterator();%0A  %7D%0A%0A  /**%0A   * Listener holder.%0A   */%0A  private class ListenerHolder implements Listener%3CT%3E %7B%0A    private final Consumer%3CT%3E listener;%0A    private final ThreadContext context;%0A%0A    private ListenerHolder(Consumer%3CT%3E listener, ThreadContext context) %7B%0A      this.listener = listener;%0A      this.context = context;%0A  
   %7D%0A%0A
+  
   @O
@@ -2895,16 +2895,18 @@
 verride%0A
+  
   public
@@ -2910,75 +2910,263 @@
 lic 
-Iterator%3CListener%3CT%3E%3E iterator() %7B%0A    return listeners.iterator();
+void accept(T event) %7B%0A      if (context != null) %7B%0A        context.executor().execute(() -%3E listener.accept(event));%0A      %7D else %7B%0A        listener.accept(event);%0A      %7D%0A    %7D%0A%0A    @Override%0A    public void close() %7B%0A      listeners.remove(this);%0A    %7D
 %0A  %7D

Allow listeners to be executed as a single unit.