package org.apache.camel.component.infinispan;

import java.io.IOException;

import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.infinispan.remote.InfinispanRemoteCustomListener;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.impl.JndiRegistry;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.infinispan.client.hotrod.RemoteCache;
import org.infinispan.client.hotrod.RemoteCacheManager;
import org.infinispan.client.hotrod.annotation.ClientListener;
import org.infinispan.client.hotrod.configuration.ConfigurationBuilder;
import org.junit.Test;

public class InfinispanRemoteAsyncConsumerCustomListenerIT extends CamelTestSupport {
    @EndpointInject(uri = "mock:result")
    private MockEndpoint mockResult;

    private RemoteCacheManager manager;

    @ClientListener(includeCurrentState=true)
    private static class MyCustomListener extends InfinispanRemoteCustomListener {
    }

    @Override
    protected void doPreSetup() throws IOException {
        ConfigurationBuilder builder = new ConfigurationBuilder();
        builder.addServer().host("localhost").port(11222);

        manager = new RemoteCacheManager(builder.build());

        // Pre-existing cache entry.
        getCache().put("existingKey", "existingValue");
    }

    @Override
    protected JndiRegistry createRegistry() throws Exception {
        JndiRegistry registry = super.createRegistry();
        registry.bind("myCustomListener", new MyCustomListener());
        registry.bind("myCustomContainer", manager);
        return registry;
    }

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() {
                from("infinispan://?cacheContainer=#myCustomContainer&cacheName=include_current_state&customListener=#myCustomListener&sync=false")
                .to("mock:result");
            }
        };
    }

    @Test
    public void createEventConsumed() throws InterruptedException {
        //One for the existing entry and one for the new entry.
        mockResult.expectedMessageCount(2);

        getCache().put("newKey", "newValue");

        mockResult.assertIsSatisfied();
    }

    private RemoteCache<Object, Object> getCache() {
        return manager.getCache("include_current_state");
    }
}
