/*
 * Copyright 2013 by Cloudsoft Corp.
 *
 * 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 brooklyn.entity.messaging.kafka;

import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;

import java.util.concurrent.Callable;

import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import brooklyn.entity.basic.ApplicationBuilder;
import brooklyn.entity.basic.Entities;
import brooklyn.entity.messaging.activemq.ActiveMQBroker;
import brooklyn.entity.proxying.EntitySpec;
import brooklyn.entity.trait.Startable;
import brooklyn.location.Location;
import brooklyn.location.LocationSpec;
import brooklyn.location.basic.LocalhostMachineProvisioningLocation;
import brooklyn.test.Asserts;
import brooklyn.test.entity.TestApplication;
import brooklyn.util.collections.MutableMap;
import brooklyn.util.time.Duration;

import com.google.common.collect.ImmutableList;

/**
 * Test the operation of the {@link ActiveMQBroker} class.
 *
 * TODO test that sensors update.
 */
public class KafkaIntegrationTest {

    private TestApplication app;
    private Location testLocation;

    @BeforeMethod(alwaysRun = true)
    public void setup() {
        app = ApplicationBuilder.newManagedApp(TestApplication.class);
        LocationSpec<LocalhostMachineProvisioningLocation> locationSpec = LocationSpec.create(LocalhostMachineProvisioningLocation.class);
        testLocation = app.getManagementContext().getLocationManager().createLocation(locationSpec);
    }

    @AfterMethod(alwaysRun = true)
    public void shutdown() {
        if (app != null) Entities.destroyAll(app.getManagementContext());
    }

    /**
     * Test that we can start a zookeeper.
     */
    @Test(groups = "Integration")
    public void testZookeeper() {
        final KafkaZooKeeper zookeeper = app.createAndManageChild(EntitySpec.create(KafkaZooKeeper.class));

        zookeeper.start(ImmutableList.of(testLocation));
        Asserts.succeedsEventually(MutableMap.of("timeout", 60000l), new Callable<Void>() {
            @Override
            public Void call() {
                assertTrue(zookeeper.getAttribute(Startable.SERVICE_UP));
                return null;
            }
        });

        zookeeper.stop();
        assertFalse(zookeeper.getAttribute(Startable.SERVICE_UP));
    }

    /**
     * Test that we can start a  broker and zookeeper together.
     */
    @Test(groups = "Integration")
    public void testBrokerPlusZookeeper() {
        final KafkaZooKeeper zookeeper = app.createAndManageChild(EntitySpec.create(KafkaZooKeeper.class));
        final KafkaBroker broker = app.createAndManageChild(EntitySpec.create(KafkaBroker.class).configure(KafkaBroker.ZOOKEEPER, zookeeper));

        zookeeper.start(ImmutableList.of(testLocation));
        Asserts.succeedsEventually(MutableMap.of("timeout", 60000l), new Callable<Void>() {
            @Override
            public Void call() {
                assertTrue(zookeeper.getAttribute(Startable.SERVICE_UP));
                return null;
            }
        });

        broker.start(ImmutableList.of(testLocation));
        Asserts.succeedsEventually(MutableMap.of("timeout", 60000l), new Callable<Void>() {
            @Override
            public Void call() {
                assertTrue(broker.getAttribute(Startable.SERVICE_UP));
                return null;
            }
        });

        zookeeper.stop();
        assertFalse(zookeeper.getAttribute(Startable.SERVICE_UP));

        broker.stop();
        assertFalse(broker.getAttribute(Startable.SERVICE_UP));
    }

    /**
     * Test that we can start a cluster with zookeeper and one broker.
     *
     * Connects to the zookeeper controller and tests sending and receiving messages on a topic.
     */
    @Test(groups = "Integration")
    public void testTwoBrokerCluster() {
        final KafkaCluster cluster = app.createAndManageChild(EntitySpec.create(KafkaCluster.class)
                .configure(KafkaCluster.INITIAL_SIZE, 2));

        cluster.start(ImmutableList.of(testLocation));
        Asserts.succeedsEventually(MutableMap.of("timeout", Duration.FIVE_MINUTES), new Callable<Void>() {
            @Override
            public Void call() {
                assertTrue(cluster.getAttribute(Startable.SERVICE_UP));
                assertTrue(cluster.getZooKeeper().getAttribute(Startable.SERVICE_UP));
                assertEquals(cluster.getCurrentSize().intValue(), 2);
                return null;
            }
        });

        Entities.dumpInfo(cluster);

        KafkaSupport support = new KafkaSupport(cluster);

        support.sendMessage("brooklyn", "TEST_MESSAGE");
        String message = support.getMessage("brooklyn");
        assertEquals(message, "TEST_MESSAGE");
    }

}

@@ -1335,16 +1335,54 @@
 sserts;%0A
+import brooklyn.test.EntityTestUtils;%0A
 import b
@@ -1547,16 +1547,63 @@
 bleList;
+%0Aimport com.google.common.collect.ImmutableMap;
 %0A%0A/**%0A *
@@ -2628,39 +2628,60 @@
 ));%0A        
-Asserts.succeed
+EntityTestUtils.assertAttributeEqual
 sEventually(
@@ -2672,33 +2672,35 @@
 qualsEventually(
-M
+Imm
 utableMap.of(%22ti
@@ -2713,219 +2713,53 @@
 , 60
+*1
 000
-l
 ), 
-new Callable%3CVoid%3E() %7B%0A            @Override%0A            public Void call() %7B%0A                assertTrue(zookeeper.getAttribute(Startable.SERVICE_UP));%0A                return null;%0A            %7D%0A        %7D
+zookeeper, Startable.SERVICE_UP, true
 );%0A%0A
@@ -3329,39 +3329,60 @@
 ));%0A        
-Asserts.succeed
+EntityTestUtils.assertAttributeEqual
 sEventually(
@@ -3373,33 +3373,35 @@
 qualsEventually(
-M
+Imm
 utableMap.of(%22ti
@@ -3414,219 +3414,53 @@
 , 60
+*1
 000
-l
 ), 
-new Callable%3CVoid%3E() %7B%0A            @Override%0A            public Void call() %7B%0A                assertTrue(zookeeper.getAttribute(Startable.SERVICE_UP));%0A                return null;%0A            %7D%0A        %7D
+zookeeper, Startable.SERVICE_UP, true
 );%0A%0A
@@ -3517,39 +3517,60 @@
 ));%0A        
-Asserts.succeed
+EntityTestUtils.assertAttributeEqual
 sEventually(
@@ -3561,33 +3561,35 @@
 qualsEventually(
-M
+Imm
 utableMap.of(%22ti
@@ -3602,216 +3602,50 @@
 , 60
+*1
 000
-l
 ), 
-new Callable%3CVoid%3E() %7B%0A            @Override%0A            public Void call() %7B%0A                assertTrue(broker.getAttribute(Startable.SERVICE_UP));%0A                return null;%0A            %7D%0A        %7D
+broker, Startable.SERVICE_UP, true
 );%0A%0A

Tidy KafkaIntegrationTest