/*
 * Copyright 2014 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 net.kuujo.copycat.internal;

import net.kuujo.copycat.Coordinator;
import net.kuujo.copycat.CopycatContext;
import net.kuujo.copycat.EventLog;
import net.kuujo.copycat.SubmitOptions;
import net.kuujo.copycat.cluster.Cluster;
import net.kuujo.copycat.internal.util.FluentList;

import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;

/**
 * Event log implementation.
 *
 * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
 */
public class DefaultEventLog<T> extends AbstractResource implements EventLog<T> {
  private Consumer<T> consumer;

  public DefaultEventLog(String name, Coordinator coordinator, Cluster cluster, CopycatContext context) {
    super(name, coordinator, cluster, context);
    context.handler(this::handle);
  }

  @SuppressWarnings("unchecked")
  private Object handle(Object entry) {
    if (consumer != null) {
      consumer.accept((T) entry);
    }
    return null;
  }

  @Override
  public EventLog<T> consumer(Consumer<T> consumer) {
    this.consumer = consumer;
    return this;
  }

  @Override
  @SuppressWarnings("unchecked")
  public CompletableFuture<T> get(long index) {
    CompletableFuture<T> future = new CompletableFuture<>();
    context.executor().execute(() -> {
      try {
        future.complete(context.log().getEntry(index));
      } catch (Exception e) {
        future.completeExceptionally(e);
      }
    });
    return future;
  }

  @Override
  public CompletableFuture<Long> commit(T entry) {
    return context.submit(new FluentList().add("add").add(entry), new SubmitOptions().withConsistent(true)
      .withPersistent(true));
  }

  @Override
  @SuppressWarnings("unchecked")
  public CompletableFuture<Void> replay() {
    CompletableFuture<Void> future = new CompletableFuture<>();
    context.executor().execute(() -> {
      if (consumer != null) {
        for (long i = context.log().firstIndex(); i <= context.log().lastIndex(); i++) {
          consumer.accept(context.log().getEntry(i));
        }
        future.complete(null);
      } else {
        future.completeExceptionally(new IllegalStateException("No consumer registered"));
      }
    });
    return future;
  }

  @Override
  @SuppressWarnings("unchecked")
  public CompletableFuture<Void> replay(long index) {
    CompletableFuture<Void> future = new CompletableFuture<>();
    context.executor().execute(() -> {
      if (consumer != null) {
        for (long i = index; i <= context.log().lastIndex(); i++) {
          consumer.accept(context.log().getEntry(i));
        }
        future.complete(null);
      } else {
        future.completeExceptionally(new IllegalStateException("No consumer registered"));
      }
    });
    return future;
  }

}

@@ -638,16 +638,56 @@
 ernal;%0A%0A
+import net.kuujo.copycat.ActionOptions;%0A
 import n
@@ -716,16 +716,16 @@
 inator;%0A
-
 import n
@@ -796,48 +796,8 @@
 og;%0A
-import net.kuujo.copycat.SubmitOptions;%0A
 impo
@@ -837,59 +837,8 @@
 ter;
-%0Aimport net.kuujo.copycat.internal.util.FluentList;
 %0A%0Aim
@@ -1298,206 +1298,8 @@
 t);%0A
-    context.handler(this::handle);%0A  %7D%0A%0A  @SuppressWarnings(%22unchecked%22)%0A  private Object handle(Object entry) %7B%0A    if (consumer != null) %7B%0A      consumer.accept((T) entry);%0A    %7D%0A    return null;%0A
   %7D%0A
@@ -1882,116 +1882,23 @@
 mit(
-new FluentList().add(%22add%22).add(entry), new SubmitOptions().withConsistent(true)%0A      .withPersistent(true)
+%22commit%22, entry
 );%0A 
@@ -2962,32 +2962,32 @@
       %7D%0A    %7D);%0A
-
     return futur
@@ -2994,10 +2994,488 @@
 e;%0A  %7D%0A%0A
+  /**%0A   * Handles a commit.%0A   */%0A  private long handleCommit(Long index, T entry) %7B%0A    return index;%0A  %7D%0A%0A  @Override%0A  public CompletableFuture%3CVoid%3E open() %7B%0A    return super.open().thenRunAsync(() -%3E %7B%0A      context.register(%22commit%22, this::handleCommit, new ActionOptions().withConsistent(true).withPersistent(true));%0A    %7D, context.executor());%0A  %7D%0A%0A  @Override%0A  public CompletableFuture%3CVoid%3E close() %7B%0A    context.unregister(%22commit%22);%0A    return super.close();%0A  %7D%0A%0A
 %7D%0A

Refactor event log for new core API.