/*
 * Copyright 2018 Red Hat, Inc.
 *
 * 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 org.jboss.shamrock.vertx;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import javax.inject.Inject;

import org.jboss.protean.arc.Arc;
import org.jboss.shamrock.test.ShamrockUnitTest;
import org.jboss.shamrock.vertx.runtime.ConsumeEvent;
import org.jboss.shrinkwrap.api.ShrinkWrap;
import org.jboss.shrinkwrap.api.spec.JavaArchive;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;

import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.Message;

public class MessageConsumerMethodTest {

    @RegisterExtension
    static final ShamrockUnitTest config = new ShamrockUnitTest().setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class).addClasses(SimpleBean.class));

    @Inject
    SimpleBean simpleBean;

    @Test
    public void testSend() throws InterruptedException {
        EventBus eventBus = Arc.container().instance(EventBus.class).get();
        BlockingQueue<Object> synchronizer = new LinkedBlockingQueue<>();
        eventBus.send("foo", "hello", ar -> {
            if (ar.succeeded()) {
                try {
                    synchronizer.put(ar.result().body());
                } catch (InterruptedException e) {
                    fail(e);
                }
            } else {
                fail(ar.cause());
            }
        });
        assertEquals("HELLO", synchronizer.poll(2, TimeUnit.SECONDS));
    }
    
    @Test
    public void testSendAsync() throws InterruptedException {
        EventBus eventBus = Arc.container().instance(EventBus.class).get();
        BlockingQueue<Object> synchronizer = new LinkedBlockingQueue<>();
        eventBus.send("foo-async", "hello", ar -> {
            if (ar.succeeded()) {
                try {
                    synchronizer.put(ar.result().body());
                } catch (InterruptedException e) {
                    fail(e);
                }
            } else {
                fail(ar.cause());
            }
        });
        assertEquals("olleh", synchronizer.poll(2, TimeUnit.SECONDS));
    }
    
    @Test
    public void testSendDefaultAddress() throws InterruptedException {
        EventBus eventBus = Arc.container().instance(EventBus.class).get();
        BlockingQueue<Object> synchronizer = new LinkedBlockingQueue<>();
        eventBus.send("org.jboss.shamrock.vertx.MessageConsumerMethodTest$SimpleBean", "Hello", ar -> {
            if (ar.succeeded()) {
                try {
                    synchronizer.put(ar.result().body());
                } catch (InterruptedException e) {
                    fail(e);
                }
            } else {
                fail(ar.cause());
            }
        });
        assertEquals("hello", synchronizer.poll(2, TimeUnit.SECONDS));
    }

    @Test
    public void testPublish() throws InterruptedException {
        EventBus eventBus = Arc.container().instance(EventBus.class).get();
        SimpleBean.latch = new CountDownLatch(2);
        eventBus.publish("pub", "Hello");
        SimpleBean.latch.await(2, TimeUnit.SECONDS);
        assertTrue(SimpleBean.MESSAGES.contains("hello"));
        assertTrue(SimpleBean.MESSAGES.contains("HELLO"));
    }

    static class SimpleBean {

        static volatile CountDownLatch latch;

        static final List<String> MESSAGES = new CopyOnWriteArrayList<>();

        @ConsumeEvent // org.jboss.shamrock.vertx.MessageConsumerMethodTest$SimpleBean
        String sendDefaultAddress(String message) {
            return message.toLowerCase();
        }
        
        @ConsumeEvent("foo")
        String send(String message) {
            return message.toUpperCase();
        }

        @ConsumeEvent("pub")
        void pub(String message) {
            MESSAGES.add(message.toLowerCase());
            latch.countDown();
        }

        @ConsumeEvent("pub")
        void pub(Message<String> message) {
            MESSAGES.add(message.body().toUpperCase());
            latch.countDown();
        }
        
        @ConsumeEvent("foo-async")
        CompletionStage<String> sendAsync(String message) {
            return CompletableFuture.completedFuture(new StringBuilder(message).reverse().toString());
        }
    }

}

@@ -4646,20 +4646,21 @@
  String 
-send
+reply
 (String 
@@ -4765,19 +4765,23 @@
    void 
-pub
+consume
 (String 
@@ -4924,19 +4924,23 @@
    void 
-pub
+consume
 (Message
@@ -5132,20 +5132,21 @@
 String%3E 
-send
+reply
 Async(St

Use more meaningful test method names