<commit_before>package enmasse.systemtest;

import io.vertx.proton.ProtonClient;
import io.vertx.proton.ProtonConnection;
import io.vertx.proton.ProtonReceiver;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.AmqpSequence;
import org.apache.qpid.proton.amqp.messaging.Source;
import org.junit.Test;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;

public class ConfigServTest extends VertxTestBase {
    @SuppressWarnings("unchecked")
    @Test
    public void testPodSense() throws Exception {
        Endpoint configserv = getConfigServEndpoint();
        BlockingQueue<List<String>> latestPods = new LinkedBlockingDeque<>();
        ProtonClient.create(vertx).connect(configserv.getHost(), configserv.getPort(), event -> {
            if (event.succeeded()) {
                ProtonConnection connection = event.result().open();
                Source source = new Source();
                source.setAddress("podsense");
                source.setFilter(Collections.singletonMap(Symbol.getSymbol("role"), "broker"));
                ProtonReceiver receiver = connection.createReceiver("podsense").setSource(source);
                receiver.handler((protonDelivery, message) -> {
                    List<String> pods = new ArrayList<>();
                    AmqpSequence seq = (AmqpSequence) message.getBody();
                    for (Object obj : seq.getValue()) {
                        Map<String, Object> pod = (Map<String, Object>) obj;
                        pods.add((String) pod.get("host"));
                    }
                    try {
                        latestPods.put(pods);
                    } catch (InterruptedException e) {
                        fail();
                    }
                });
                receiver.open();
            }
        });
        deploy(Destination.queue("myqueue"));
        assertPods(latestPods, 1, new TimeoutBudget(2, TimeUnit.MINUTES));
        deploy(Destination.queue("myqueue"), Destination.queue("anotherqueue"));
        assertPods(latestPods, 2, new TimeoutBudget(2, TimeUnit.MINUTES));
        deploy(Destination.queue("anotherqueue"));
        assertPods(latestPods, 1, new TimeoutBudget(2, TimeUnit.MINUTES));
    }

    private void assertPods(BlockingQueue<List<String>> latestPods, int numPods, TimeoutBudget timeoutBudget) throws InterruptedException {
        List<String> pods = null;
        do {
            pods = latestPods.poll(timeoutBudget.timeLeft(), TimeUnit.MILLISECONDS);
        } while ((pods == null || pods.size() != numPods) && timeoutBudget.timeLeft() >= 0);
        assertNotNull(pods);<commit_msg>Use different queue names to make logs less confusing<commit_after>package enmasse.systemtest;

import io.vertx.proton.ProtonClient;
import io.vertx.proton.ProtonConnection;
import io.vertx.proton.ProtonReceiver;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.AmqpSequence;
import org.apache.qpid.proton.amqp.messaging.Source;
import org.junit.Test;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;

public class ConfigServTest extends VertxTestBase {
    @SuppressWarnings("unchecked")
    @Test
    public void testPodSense() throws Exception {
        Endpoint configserv = getConfigServEndpoint();
        BlockingQueue<List<String>> latestPods = new LinkedBlockingDeque<>();
        ProtonClient.create(vertx).connect(configserv.getHost(), configserv.getPort(), event -> {
            if (event.succeeded()) {
                ProtonConnection connection = event.result().open();
                Source source = new Source();
                source.setAddress("podsense");
                source.setFilter(Collections.singletonMap(Symbol.getSymbol("role"), "broker"));
                ProtonReceiver receiver = connection.createReceiver("podsense").setSource(source);
                receiver.handler((protonDelivery, message) -> {
                    List<String> pods = new ArrayList<>();
                    AmqpSequence seq = (AmqpSequence) message.getBody();
                    for (Object obj : seq.getValue()) {
                        Map<String, Object> pod = (Map<String, Object>) obj;
                        pods.add((String) pod.get("host"));
                    }
                    try {
                        latestPods.put(pods);
                    } catch (InterruptedException e) {
                        fail();
                    }
                });
                receiver.open();
            }
        });
        deploy(Destination.queue("testqueue"));
        assertPods(latestPods, 1, new TimeoutBudget(2, TimeUnit.MINUTES));
        deploy(Destination.queue("testqueue"), Destination.queue("anotherqueue"));
        assertPods(latestPods, 2, new TimeoutBudget(2, TimeUnit.MINUTES));
        deploy(Destination.queue("anotherqueue"));
        assertPods(latestPods, 1, new TimeoutBudget(2, TimeUnit.MINUTES));
    }

    private void assertPods(BlockingQueue<List<String>> latestPods, int numPods, TimeoutBudget timeoutBudget) throws InterruptedException {
        List<String> pods = null;
        do {
            pods = latestPods.poll(timeoutBudget.timeLeft(), TimeUnit.MILLISECONDS);
        } while ((pods == null || pods.size() != numPods) && timeoutBudget.timeLeft() >= 0);
        assertNotNull(pods);
{'fraction_non_alphanumeric': 0.11692726036709722, 'fraction_numerical': 0.002379333786539769, 'mean_word_length': 2.639455782312925, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5895728', 'n_tokens_mistral': 1673, 'n_tokens_neox': 1602, 'n_words': 299}