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

import io.atomix.resource.ResourceType;
import io.atomix.testing.AbstractCopycatTest;
import org.testng.annotations.Test;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Distributed task queue test.
 *
 * @author <a href="http://github.com/kuujo>Jordan Halterman</a>
 */
@Test
@SuppressWarnings("unchecked")
public class DistributedTaskQueueTest extends AbstractCopycatTest<DistributedTaskQueue> {

  @Override
  protected ResourceType<DistributedTaskQueue> type() {
    return DistributedTaskQueue.TYPE;
  }

  /**
   * Tests submitting and processing tasks.
   */
  public void testSubmitAsync() throws Throwable {
    createServers(3);

    DistributedTaskQueue<String> worker1 = createResource().async();
    DistributedTaskQueue<String> worker2 = createResource().async();
    DistributedTaskQueue<String> queue = createResource().async();

    Set<String> tasks = new HashSet<>(Arrays.asList("foo", "bar", "baz"));
    Set<String> completed = new ConcurrentSkipListSet<>();
    worker1.consumer(task -> {
      threadAssertTrue(tasks.contains(task));
      threadAssertFalse(completed.contains(task));
      completed.add(task);
      resume();
    }).thenRun(this::resume);
    worker2.consumer(task -> {
      threadAssertTrue(tasks.contains(task));
      threadAssertFalse(completed.contains(task));
      completed.add(task);
      resume();
    }).thenRun(this::resume);

    await(5000, 2);

    queue.submit("foo");
    queue.submit("bar");
    queue.submit("baz");
    await(5000, 3);
  }

  public void testSubmitRecursive() throws Throwable {
    createServers(3);

    DistributedTaskQueue<String> worker = createResource().async();
    DistributedTaskQueue<String> queue = createResource().async();

    AtomicInteger counter = new AtomicInteger();
    worker.consumer(task -> {
      if (counter.incrementAndGet() % 1000 == 0) {
        System.out.println("1000 messages received");
      }
      if (counter.get() == 20000) {
        resume();
      }
    });
    submitTasks(queue);
    await(10000);
  }

  private void submitTasks(DistributedTaskQueue<String> queue) {
    for (int i = 0; i < 100; i++) {
      submitTask(queue);
    }
  }

  private void submitTask(DistributedTaskQueue<String> queue) {
    queue.submit(UUID.randomUUID().toString()).whenComplete((result, error) -> submitTask(queue));
  }

  /**
   * Tests submitting and processing tasks and awaiting the acknowledgement.
   */
  public void testSubmitSync() throws Throwable {
    createServers(3);

    DistributedTaskQueue<String> worker1 = createResource().sync();
    DistributedTaskQueue<String> worker2 = createResource().sync();
    DistributedTaskQueue<String> queue = createResource().sync();

    Set<String> tasks = new HashSet<>(Arrays.asList("foo", "bar", "baz"));
    Set<String> completed = new ConcurrentSkipListSet<>();
    worker1.consumer(task -> {
      threadAssertTrue(tasks.contains(task));
      threadAssertFalse(completed.contains(task));
      completed.add(task);
      resume();
    }).thenRun(this::resume);
    worker2.consumer(task -> {
      threadAssertTrue(tasks.contains(task));
      threadAssertFalse(completed.contains(task));
      completed.add(task);
      resume();
    }).thenRun(this::resume);

    await(5000, 2);

    queue.submit("foo").thenRun(this::resume);
    queue.submit("bar").thenRun(this::resume);
    queue.submit("baz").thenRun(this::resume);
    await(5000, 6);
  }

}

@@ -833,31 +833,8 @@
 et;%0A
-import java.util.UUID;%0A
 impo
@@ -883,58 +883,8 @@
 Set;
-%0Aimport java.util.concurrent.atomic.AtomicInteger;
 %0A%0A/*
@@ -2226,836 +2226,8 @@
  %7D%0A%0A
-  public void testSubmitRecursive() throws Throwable %7B%0A    createServers(3);%0A%0A    DistributedTaskQueue%3CString%3E worker = createResource().async();%0A    DistributedTaskQueue%3CString%3E queue = createResource().async();%0A%0A    AtomicInteger counter = new AtomicInteger();%0A    worker.consumer(task -%3E %7B%0A      if (counter.incrementAndGet() %25 1000 == 0) %7B%0A        System.out.println(%221000 messages received%22);%0A      %7D%0A      if (counter.get() == 20000) %7B%0A        resume();%0A      %7D%0A    %7D);%0A    submitTasks(queue);%0A    await(10000);%0A  %7D%0A%0A  private void submitTasks(DistributedTaskQueue%3CString%3E queue) %7B%0A    for (int i = 0; i %3C 100; i++) %7B%0A      submitTask(queue);%0A    %7D%0A  %7D%0A%0A  private void submitTask(DistributedTaskQueue%3CString%3E queue) %7B%0A    queue.submit(UUID.randomUUID().toString()).whenComplete((result, error) -%3E submitTask(queue));%0A  %7D%0A%0A
   /*

Remove rogue task tests.