<commit_before>/*
 * Copyright 2016 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 io.enmasse.systemtest.standard;

import io.enmasse.systemtest.*;
import io.enmasse.systemtest.amqp.AmqpClient;
import io.enmasse.systemtest.bases.StandardTestBase;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.message.Message;
import org.junit.Test;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.*;

public class QueueTest extends StandardTestBase {

    @Test
    public void testColocatedQueues() throws Exception {
        Destination q1 = Destination.queue("queue1", "pooled-queue");
        Destination q2 = Destination.queue("queue2", "pooled-queue");
        Destination q3 = Destination.queue("queue3", "pooled-queue");
        setAddresses(q1, q2, q3);

        AmqpClient client = amqpClientFactory.createQueueClient();
        runQueueTest(client, q1);
        runQueueTest(client, q2);
        runQueueTest(client, q3);
    }

    public void testShardedQueues() throws Exception {
        Destination q1 = Destination.queue("persistedQueue1", "sharded-queue");
        Destination q2 = Destination.queue("persistedQueue2", "sharded-queue");

        setAddresses(q1, q2);

        AmqpClient client = amqpClientFactory.createQueueClient();
        runQueueTest(client, q1);
        runQueueTest(client, q2);
    }

    @Test
    public void testRestApi() throws Exception {
        List<String> queues = Arrays.asList("queue1", "queue2");
        Destination q1 = Destination.queue(queues.get(0), "pooled-queue");
        Destination q2 = Destination.queue(queues.get(1), "pooled-queue");

        runRestApiTest(queues, q1, q2);
    }

    @Test
    public void testCreateDeleteQueue() throws Exception {
        List<String> queues = IntStream.range(0, 16).mapToObj(i -> "queue-create-delete-" + i).collect(Collectors.toList());
        Destination destExtra = Destination.queue("ext-queue", "pooled-queue");

        List<Destination> addresses = new ArrayList<>();
        queues.forEach(queue -> addresses.add(Destination.queue(queue, "pooled-queue")));

        AmqpClient client = amqpClientFactory.createQueueClient();
        for (Destination address : addresses) {
            setAddresses(address, destExtra);
            Thread.sleep(20_000);

            //runQueueTest(client, address, 1); //TODO! commented due to issue #429

            deleteAddresses(address);
            Future<List<String>> response = getAddresses(Optional.empty());
            assertThat("Extra destination was not created ",
                    response.get(20, TimeUnit.SECONDS), is(Arrays.asList(destExtra.getAddress())));
            deleteAddresses(destExtra);
            response = getAddresses(Optional.empty());
            assertThat("No destinations are expected",
                    response.get(20, TimeUnit.SECONDS), is(java.util.Collections.emptyList()));
            Thread.sleep(20_000);
        }
    }

    @Test
    public void testMessagePriorities() throws Exception {
        Destination dest = Destination.queue("messagePrioritiesQueue", getDefaultPlan(AddressType.QUEUE));
        setAddresses(dest);

        AmqpClient client = amqpClientFactory.createQueueClient();
        Thread.sleep(30_000);

        int msgsCount = 1024;
        List<Message> listOfMessages = new ArrayList<>();
        for (int i = 0; i < msgsCount; i++) {
            Message msg = Message.Factory.create();
            msg.setAddress(dest.getAddress());
            msg.setBody(new AmqpValue(dest.getAddress()));
            msg.setSubject("subject");
            msg.setPriority((short) (i % 10));
            listOfMessages.add(msg);
        }

        Future<Integer> sent = client.sendMessages(dest.getAddress(),
                listOfMessages.toArray(new Message[listOfMessages.size()]));
        assertThat("Wrong count of messages sent", sent.get(1, TimeUnit.MINUTES), is(msgsCount));

        Future<List<Message>> received = client.recvMessages(dest.getAddress(), msgsCount);
        assertThat("Wrong count of messages received", received.get(1, TimeUnit.MINUTES).size(), is(msgsCount));

        int sub = 1;
        for (Message m : received.get()) {
            for (Message mSub : received.get().subList(sub, received.get().size())) {
                assertTrue("Wrong order of messages", m.getPriority() >= mSub.getPriority());
            }
            sub++;
        }
    }

    @Test
    public void testScaledown() throws Exception {
        Destination dest = Destination.queue("scalequeue", "sharded-queue");
        setAddresses(dest);
        scale(dest, 4);
        AmqpClient client = amqpClientFactory.createQueueClient();
        List<Future<Integer>> sent = Arrays.asList(
                client.sendMessages(dest.getAddress(), TestUtils.generateMessages("foo", 1000)),
                client.sendMessages(dest.getAddress(), TestUtils.generateMessages("bar", 1000)),
                client.sendMessages(dest.getAddress(), TestUtils.generateMessages("baz", 1000)),
                client.sendMessages(dest.getAddress(), TestUtils.generateMessages("quux", 1000)));

        assertThat("Wrong count of messages sent: sender0",
                sent.get(0).get(1, TimeUnit.MINUTES), is(1000));
        assertThat("Wrong count of messages sent: sender1",
                sent.get(1).get(1, TimeUnit.MINUTES), is(1000));
        assertThat("Wrong count of messages sent: sender2",
                sent.get(2).get(1, TimeUnit.MINUTES), is(1000));
        assertThat("Wrong count of messages sent: sender3",
                sent.get(3).get(1, TimeUnit.MINUTES), is(1000));

        Future<List<Message>> received = client.recvMessages(dest.getAddress(), 500);
        assertThat("Wrong count of messages received",
                received.get(1, TimeUnit.MINUTES).size(), is(500));

        scale(dest, 1);

        received = client.recvMessages(dest.getAddress(), 3500);

        assertThat("Wrong count of messages received",
                received.get(1, TimeUnit.MINUTES).size(), is(3500));
    }

    public static void runQueueTest(AmqpClient client, Destination dest) throws InterruptedException, ExecutionException, TimeoutException, IOException {
        runQueueTest(client, dest, 1024);
    }

    public static void runQueueTest(AmqpClient client, Destination dest, int countMessages) throws InterruptedException, TimeoutException, ExecutionException, IOException {
        List<String> msgs = TestUtils.generateMessages(countMessages);
        Count<Message> predicate = new Count<>(msgs.size());
        Future<Integer> numSent = client.sendMessages(dest.getAddress(), msgs, predicate);

        assertNotNull("Sending messages didn't start", numSent);
        int actual = 0;
        try {
            actual = numSent.get(1, TimeUnit.MINUTES);
        } catch (TimeoutException t) {
            fail("Sending messages timed out after sending " + predicate.actual());
        }
        assertThat("Wrong count of messages sent", actual, is(msgs.size()));

        predicate = new Count<>(msgs.size());
        Future<List<Message>> received = client.recvMessages(dest.getAddress(), predicate);
        actual = 0;
        try {
            actual = received.get(1, TimeUnit.MINUTES).size();
        } catch (TimeoutException t) {
            fail("Receiving messages timed out after " + predicate.actual() + " msgs received");
        }

        assertThat("Wrong count of messages received", actual, is(msgs.size()));
    }
}

<commit_msg>UPD standard/QueueTest - testScaledown disabled due to issue #851<commit_after>/*
 * Copyright 2016 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 io.enmasse.systemtest.standard;

import io.enmasse.systemtest.*;
import io.enmasse.systemtest.amqp.AmqpClient;
import io.enmasse.systemtest.bases.StandardTestBase;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.message.Message;
import org.junit.Test;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.*;

public class QueueTest extends StandardTestBase {

    @Test
    public void testColocatedQueues() throws Exception {
        Destination q1 = Destination.queue("queue1", "pooled-queue");
        Destination q2 = Destination.queue("queue2", "pooled-queue");
        Destination q3 = Destination.queue("queue3", "pooled-queue");
        setAddresses(q1, q2, q3);

        AmqpClient client = amqpClientFactory.createQueueClient();
        runQueueTest(client, q1);
        runQueueTest(client, q2);
        runQueueTest(client, q3);
    }

    public void testShardedQueues() throws Exception {
        Destination q1 = Destination.queue("persistedQueue1", "sharded-queue");
        Destination q2 = Destination.queue("persistedQueue2", "sharded-queue");

        setAddresses(q1, q2);

        AmqpClient client = amqpClientFactory.createQueueClient();
        runQueueTest(client, q1);
        runQueueTest(client, q2);
    }

    @Test
    public void testRestApi() throws Exception {
        List<String> queues = Arrays.asList("queue1", "queue2");
        Destination q1 = Destination.queue(queues.get(0), "pooled-queue");
        Destination q2 = Destination.queue(queues.get(1), "pooled-queue");

        runRestApiTest(queues, q1, q2);
    }

    @Test
    public void testCreateDeleteQueue() throws Exception {
        List<String> queues = IntStream.range(0, 16).mapToObj(i -> "queue-create-delete-" + i).collect(Collectors.toList());
        Destination destExtra = Destination.queue("ext-queue", "pooled-queue");

        List<Destination> addresses = new ArrayList<>();
        queues.forEach(queue -> addresses.add(Destination.queue(queue, "pooled-queue")));

        AmqpClient client = amqpClientFactory.createQueueClient();
        for (Destination address : addresses) {
            setAddresses(address, destExtra);
            Thread.sleep(20_000);

            //runQueueTest(client, address, 1); //TODO! commented due to issue #429

            deleteAddresses(address);
            Future<List<String>> response = getAddresses(Optional.empty());
            assertThat("Extra destination was not created ",
                    response.get(20, TimeUnit.SECONDS), is(Arrays.asList(destExtra.getAddress())));
            deleteAddresses(destExtra);
            response = getAddresses(Optional.empty());
            assertThat("No destinations are expected",
                    response.get(20, TimeUnit.SECONDS), is(java.util.Collections.emptyList()));
            Thread.sleep(20_000);
        }
    }

    @Test
    public void testMessagePriorities() throws Exception {
        Destination dest = Destination.queue("messagePrioritiesQueue", getDefaultPlan(AddressType.QUEUE));
        setAddresses(dest);

        AmqpClient client = amqpClientFactory.createQueueClient();
        Thread.sleep(30_000);

        int msgsCount = 1024;
        List<Message> listOfMessages = new ArrayList<>();
        for (int i = 0; i < msgsCount; i++) {
            Message msg = Message.Factory.create();
            msg.setAddress(dest.getAddress());
            msg.setBody(new AmqpValue(dest.getAddress()));
            msg.setSubject("subject");
            msg.setPriority((short) (i % 10));
            listOfMessages.add(msg);
        }

        Future<Integer> sent = client.sendMessages(dest.getAddress(),
                listOfMessages.toArray(new Message[listOfMessages.size()]));
        assertThat("Wrong count of messages sent", sent.get(1, TimeUnit.MINUTES), is(msgsCount));

        Future<List<Message>> received = client.recvMessages(dest.getAddress(), msgsCount);
        assertThat("Wrong count of messages received", received.get(1, TimeUnit.MINUTES).size(), is(msgsCount));

        int sub = 1;
        for (Message m : received.get()) {
            for (Message mSub : received.get().subList(sub, received.get().size())) {
                assertTrue("Wrong order of messages", m.getPriority() >= mSub.getPriority());
            }
            sub++;
        }
    }

    //@Test test disabled due to issue: #851
    public void testScaledown() throws Exception {
        Destination dest = Destination.queue("scalequeue", "sharded-queue");
        setAddresses(dest);
        scale(dest, 4);

        Thread.sleep(30000);
        AmqpClient client = amqpClientFactory.createQueueClient();
        List<Future<Integer>> sent = Arrays.asList(
                client.sendMessages(dest.getAddress(), TestUtils.generateMessages("foo", 1000)),
                client.sendMessages(dest.getAddress(), TestUtils.generateMessages("bar", 1000)),
                client.sendMessages(dest.getAddress(), TestUtils.generateMessages("baz", 1000)),
                client.sendMessages(dest.getAddress(), TestUtils.generateMessages("quux", 1000)));

        assertThat("Wrong count of messages sent: sender0",
                sent.get(0).get(1, TimeUnit.MINUTES), is(1000));
        assertThat("Wrong count of messages sent: sender1",
                sent.get(1).get(1, TimeUnit.MINUTES), is(1000));
        assertThat("Wrong count of messages sent: sender2",
                sent.get(2).get(1, TimeUnit.MINUTES), is(1000));
        assertThat("Wrong count of messages sent: sender3",
                sent.get(3).get(1, TimeUnit.MINUTES), is(1000));

        Future<List<Message>> received = client.recvMessages(dest.getAddress(), 500);
        assertThat("Wrong count of messages received",
                received.get(1, TimeUnit.MINUTES).size(), is(500));

        scale(dest, 1);

        Thread.sleep(30000);

        received = client.recvMessages(dest.getAddress(), 3500);

        assertThat("Wrong count of messages received",
                received.get(1, TimeUnit.MINUTES).size(), is(3500));
    }

    public static void runQueueTest(AmqpClient client, Destination dest) throws InterruptedException, ExecutionException, TimeoutException, IOException {
        runQueueTest(client, dest, 1024);
    }

    public static void runQueueTest(AmqpClient client, Destination dest, int countMessages) throws InterruptedException, TimeoutException, ExecutionException, IOException {
        List<String> msgs = TestUtils.generateMessages(countMessages);
        Count<Message> predicate = new Count<>(msgs.size());
        Future<Integer> numSent = client.sendMessages(dest.getAddress(), msgs, predicate);

        assertNotNull("Sending messages didn't start", numSent);
        int actual = 0;
        try {
            actual = numSent.get(1, TimeUnit.MINUTES);
        } catch (TimeoutException t) {
            fail("Sending messages timed out after sending " + predicate.actual());
        }
        assertThat("Wrong count of messages sent", actual, is(msgs.size()));

        predicate = new Count<>(msgs.size());
        Future<List<Message>> received = client.recvMessages(dest.getAddress(), predicate);
        actual = 0;
        try {
            actual = received.get(1, TimeUnit.MINUTES).size();
        } catch (TimeoutException t) {
            fail("Receiving messages timed out after " + predicate.actual() + " msgs received");
        }

        assertThat("Wrong count of messages received", actual, is(msgs.size()));
    }
}


{'fraction_non_alphanumeric': 0.12160452930193338, 'fraction_numerical': 0.018069023670421007, 'mean_word_length': 3.0115970041072724, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1695768', 'n_tokens_mistral': 4764, 'n_tokens_neox': 4522, 'n_words': 1083}