| /* |
| * |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you 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.apache.qpid.systests.jms_1_1.extensions.autocreation; |
| |
| import static org.hamcrest.CoreMatchers.anyOf; |
| import static org.hamcrest.CoreMatchers.equalTo; |
| import static org.hamcrest.CoreMatchers.is; |
| import static org.hamcrest.CoreMatchers.not; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| import static org.junit.Assume.assumeThat; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.TimeUnit; |
| import java.util.stream.Stream; |
| |
| import javax.jms.Connection; |
| import javax.jms.JMSException; |
| import javax.jms.Message; |
| import javax.jms.MessageConsumer; |
| import javax.jms.MessageProducer; |
| import javax.jms.Queue; |
| import javax.jms.Session; |
| import javax.jms.TextMessage; |
| import javax.jms.Topic; |
| |
| import com.fasterxml.jackson.databind.ObjectMapper; |
| import org.junit.Test; |
| |
| import org.apache.qpid.server.exchange.ExchangeDefaults; |
| import org.apache.qpid.server.model.AlternateBinding; |
| import org.apache.qpid.server.model.Exchange; |
| import org.apache.qpid.server.model.Protocol; |
| import org.apache.qpid.server.virtualhost.NodeAutoCreationPolicy; |
| import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost; |
| import org.apache.qpid.systests.JmsTestBase; |
| |
| public class NodeAutoCreationPolicyTest extends JmsTestBase |
| { |
| private static final String DEAD_LETTER_QUEUE_SUFFIX = "_DLQ"; |
| private static final String DEAD_LETTER_EXCHANGE_SUFFIX = "_DLE"; |
| private static final String AUTO_CREATION_POLICIES = createAutoCreationPolicies(); |
| private static final String TEST_MESSAGE = "Hello world!"; |
| private static final String VALID_QUEUE_NAME = "fooQueue"; |
| private static final String TYPE_QUEUE = "queue"; |
| private static final String TYPE_TOPIC = "topic"; |
| |
| |
| private static String createAutoCreationPolicies() |
| { |
| ObjectMapper mapper = new ObjectMapper(); |
| try |
| { |
| |
| NodeAutoCreationPolicy[] policies = new NodeAutoCreationPolicy[] { |
| new NodeAutoCreationPolicy() |
| { |
| @Override |
| public String getPattern() |
| { |
| return "fooQ.*"; |
| } |
| |
| @Override |
| public boolean isCreatedOnPublish() |
| { |
| return true; |
| } |
| |
| @Override |
| public boolean isCreatedOnConsume() |
| { |
| return true; |
| } |
| |
| @Override |
| public String getNodeType() |
| { |
| return "Queue"; |
| } |
| |
| @Override |
| public Map<String, Object> getAttributes() |
| { |
| return Collections.emptyMap(); |
| } |
| }, |
| new NodeAutoCreationPolicy() |
| { |
| @Override |
| public String getPattern() |
| { |
| return "barE.*"; |
| } |
| |
| @Override |
| public boolean isCreatedOnPublish() |
| { |
| return true; |
| } |
| |
| @Override |
| public boolean isCreatedOnConsume() |
| { |
| return false; |
| } |
| |
| @Override |
| public String getNodeType() |
| { |
| return "Exchange"; |
| } |
| |
| @Override |
| public Map<String, Object> getAttributes() |
| { |
| return Collections.singletonMap(Exchange.TYPE, "fanout"); |
| } |
| }, |
| |
| new NodeAutoCreationPolicy() |
| { |
| @Override |
| public String getPattern() |
| { |
| return ".*" + DEAD_LETTER_QUEUE_SUFFIX; |
| } |
| |
| @Override |
| public boolean isCreatedOnPublish() |
| { |
| return true; |
| } |
| |
| @Override |
| public boolean isCreatedOnConsume() |
| { |
| return true; |
| } |
| |
| @Override |
| public String getNodeType() |
| { |
| return "Queue"; |
| } |
| |
| @Override |
| public Map<String, Object> getAttributes() |
| { |
| return Collections.emptyMap(); |
| } |
| }, |
| |
| new NodeAutoCreationPolicy() |
| { |
| @Override |
| public String getPattern() |
| { |
| return ".*" + DEAD_LETTER_EXCHANGE_SUFFIX; |
| } |
| |
| @Override |
| public boolean isCreatedOnPublish() |
| { |
| return true; |
| } |
| |
| @Override |
| public boolean isCreatedOnConsume() |
| { |
| return false; |
| } |
| |
| @Override |
| public String getNodeType() |
| { |
| return "Exchange"; |
| } |
| |
| @Override |
| public Map<String, Object> getAttributes() |
| { |
| return Collections.singletonMap(Exchange.TYPE, ExchangeDefaults.FANOUT_EXCHANGE_CLASS); |
| } |
| } |
| }; |
| |
| return mapper.writeValueAsString(Arrays.asList(policies)); |
| } |
| catch (IOException e) |
| { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| private void updateAutoCreationPolicies() throws Exception |
| { |
| updateEntityUsingAmqpManagement(getVirtualHostName(), "org.apache.qpid.VirtualHost", Collections.singletonMap(QueueManagingVirtualHost.NODE_AUTO_CREATION_POLICIES, AUTO_CREATION_POLICIES)); |
| } |
| |
| @Test |
| public void testSendingToQueuePattern() throws Exception |
| { |
| updateAutoCreationPolicies(); |
| |
| Connection connection = getConnection(); |
| try |
| { |
| connection.start(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| final Queue queue = session.createQueue(getDestinationAddress(VALID_QUEUE_NAME, TYPE_QUEUE)); |
| final MessageProducer producer = session.createProducer(queue); |
| producer.send(session.createTextMessage(TEST_MESSAGE)); |
| |
| final MessageConsumer consumer = session.createConsumer(queue); |
| Message received = consumer.receive(getReceiveTimeout()); |
| assertNotNull(received); |
| assertTrue(received instanceof TextMessage); |
| assertEquals(TEST_MESSAGE, ((TextMessage) received).getText()); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| |
| @Test |
| public void testConcurrentQueueCreation() throws Exception |
| { |
| updateAutoCreationPolicies(); |
| |
| final String destination = getDestinationAddress(VALID_QUEUE_NAME, TYPE_QUEUE); |
| final int numberOfActors = 3; |
| final Connection[] connections = new Connection[numberOfActors]; |
| try |
| { |
| final Session[] sessions = new Session[numberOfActors]; |
| for (int i = 0; i < numberOfActors; i++) |
| { |
| final Connection connection = getConnection(); |
| connections[i] = connection; |
| sessions[i] = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| } |
| |
| final List<CompletableFuture<MessageProducer>> futures = new ArrayList<>(numberOfActors); |
| final ExecutorService executorService = Executors.newFixedThreadPool(numberOfActors); |
| try |
| { |
| Stream.of(sessions) |
| .forEach(session -> futures.add(CompletableFuture.supplyAsync(() -> publishMessage(session, |
| destination), |
| executorService))); |
| final CompletableFuture<Void> combinedFuture = |
| CompletableFuture.allOf(futures.toArray(new CompletableFuture[numberOfActors])); |
| combinedFuture.get(getReceiveTimeout(), TimeUnit.MILLISECONDS); |
| } |
| finally |
| { |
| executorService.shutdown(); |
| } |
| |
| final Connection connection = getConnection(); |
| try |
| { |
| connection.start(); |
| final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| final Queue queue = session.createQueue(destination); |
| final MessageConsumer consumer = session.createConsumer(queue); |
| |
| for (int i = 0; i < numberOfActors; i++) |
| { |
| Message received = consumer.receive(getReceiveTimeout()); |
| assertNotNull(received); |
| assertTrue(received instanceof TextMessage); |
| assertEquals(TEST_MESSAGE, ((TextMessage) received).getText()); |
| } |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| finally |
| { |
| for (Connection connection : connections) |
| { |
| if (connection != null) |
| { |
| connection.close(); |
| } |
| } |
| } |
| } |
| |
| private MessageProducer publishMessage(final Session session, final String destination) |
| { |
| try |
| { |
| final Queue queue = session.createQueue(destination); |
| final MessageProducer producer = session.createProducer(queue); |
| producer.send(session.createTextMessage(TEST_MESSAGE)); |
| return producer; |
| } |
| catch (JMSException e) |
| { |
| throw new IllegalStateException(e); |
| } |
| } |
| |
| @Test |
| public void testSendingToNonMatchingQueuePattern() throws Exception |
| { |
| updateAutoCreationPolicies(); |
| |
| Connection connection = getConnection(); |
| try |
| { |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| final Queue queue = session.createQueue(getDestinationAddress("foQueue", TYPE_QUEUE)); |
| try |
| { |
| session.createProducer(queue); |
| fail("Creating producer should fail"); |
| } |
| catch (JMSException e) |
| { |
| // pass |
| } |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| |
| @Test |
| public void testSendingToExchangePattern() throws Exception |
| { |
| updateAutoCreationPolicies(); |
| |
| Connection connection = getConnection(); |
| try |
| { |
| connection.start(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| final Topic topic = session.createTopic(getDestinationAddress("barExchange/foo", TYPE_TOPIC)); |
| final MessageProducer producer = session.createProducer(topic); |
| producer.send(session.createTextMessage(TEST_MESSAGE)); |
| |
| final MessageConsumer consumer = session.createConsumer(topic); |
| Message received = consumer.receive(getReceiveTimeout() / 4); |
| assertNull(received); |
| |
| producer.send(session.createTextMessage("Hello world2!")); |
| received = consumer.receive(getReceiveTimeout()); |
| |
| assertNotNull(received); |
| |
| assertTrue(received instanceof TextMessage); |
| assertEquals("Hello world2!", ((TextMessage) received).getText()); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| |
| @Test |
| public void testSendingToNonMatchingTopicPattern() throws Exception |
| { |
| updateAutoCreationPolicies(); |
| |
| Connection connection = getConnection(); |
| try |
| { |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| final Topic topic = session.createTopic(getDestinationAddress("baa", TYPE_TOPIC)); |
| try |
| { |
| session.createProducer(topic); |
| fail("Creating producer should fail"); |
| } |
| catch (JMSException e) |
| { |
| // pass |
| } |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| |
| @Test |
| public void testSendingToQueuePatternBURL() throws Exception |
| { |
| assumeThat("Qpid JMS Client does not support BURL syntax", |
| getProtocol(), |
| is(not(equalTo(Protocol.AMQP_1_0)))); |
| updateAutoCreationPolicies(); |
| |
| Connection connection = getConnection(); |
| try |
| { |
| connection.start(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| final Queue queue = session.createQueue("BURL:direct:///fooQ/fooQ"); |
| final MessageProducer producer = session.createProducer(queue); |
| producer.send(session.createTextMessage(TEST_MESSAGE)); |
| |
| final MessageConsumer consumer = session.createConsumer(queue); |
| Message received = consumer.receive(getReceiveTimeout()); |
| assertNotNull(received); |
| assertTrue(received instanceof TextMessage); |
| assertEquals(TEST_MESSAGE, ((TextMessage) received).getText()); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| |
| @Test |
| public void testSendingToNonMatchingQueuePatternBURL() throws Exception |
| { |
| assumeThat("Using AMQP 0-8..0-9-1 to test BURL syntax", |
| getProtocol(), |
| is(not(anyOf(equalTo(Protocol.AMQP_1_0), equalTo(Protocol.AMQP_0_10))))); |
| updateAutoCreationPolicies(); |
| |
| Connection connection = getConnectionBuilder().setSyncPublish(true).build(); |
| try |
| { |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| final Queue queue = session.createQueue("BURL:direct:///fo/fo"); |
| try |
| { |
| final MessageProducer producer = session.createProducer(queue); |
| producer.send(session.createTextMessage(TEST_MESSAGE)); |
| |
| fail("Sending a message should fail"); |
| } |
| catch (JMSException e) |
| { |
| // pass |
| } |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| |
| @Test |
| public void testQueueAlternateBindingCreation() throws Exception |
| { |
| updateAutoCreationPolicies(); |
| |
| String queueName = getTestName(); |
| String deadLetterQueueName = queueName + DEAD_LETTER_QUEUE_SUFFIX; |
| |
| final Map<String, Object> attributes = new HashMap<>(); |
| Map<String, Object> expectedAlternateBinding = |
| Collections.singletonMap(AlternateBinding.DESTINATION, deadLetterQueueName); |
| attributes.put(org.apache.qpid.server.model.Queue.ALTERNATE_BINDING, |
| new ObjectMapper().writeValueAsString(expectedAlternateBinding)); |
| createEntityUsingAmqpManagement(queueName, "org.apache.qpid.Queue", attributes); |
| |
| Map<String, Object> queueAttributes = readEntityUsingAmqpManagement(queueName, "org.apache.qpid.Queue", true); |
| |
| Object actualAlternateBinding = queueAttributes.get(org.apache.qpid.server.model.Queue.ALTERNATE_BINDING); |
| Map<String, Object> actualAlternateBindingMap = convertIfNecessary(actualAlternateBinding); |
| assertEquals("Unexpected alternate binding", |
| new HashMap<>(expectedAlternateBinding), |
| new HashMap<>(actualAlternateBindingMap)); |
| |
| Map<String, Object> dlqAttributes = |
| readEntityUsingAmqpManagement(deadLetterQueueName, "org.apache.qpid.Queue", true); |
| assertNotNull("Cannot get dead letter queue", dlqAttributes); |
| } |
| |
| @Test |
| public void testExchangeAlternateBindingCreation() throws Exception |
| { |
| updateAutoCreationPolicies(); |
| |
| String exchangeName = getTestName(); |
| String deadLetterExchangeName = exchangeName + DEAD_LETTER_EXCHANGE_SUFFIX; |
| |
| final Map<String, Object> attributes = new HashMap<>(); |
| Map<String, Object> expectedAlternateBinding = |
| Collections.singletonMap(AlternateBinding.DESTINATION, deadLetterExchangeName); |
| attributes.put(Exchange.ALTERNATE_BINDING, new ObjectMapper().writeValueAsString(expectedAlternateBinding)); |
| attributes.put(Exchange.TYPE, ExchangeDefaults.DIRECT_EXCHANGE_CLASS); |
| createEntityUsingAmqpManagement(exchangeName, "org.apache.qpid.DirectExchange", attributes); |
| |
| Map<String, Object> exchangeAttributes = readEntityUsingAmqpManagement(exchangeName, "org.apache.qpid.Exchange", true); |
| |
| Object actualAlternateBinding = exchangeAttributes.get(Exchange.ALTERNATE_BINDING); |
| Map<String, Object> actualAlternateBindingMap = convertIfNecessary(actualAlternateBinding); |
| assertEquals("Unexpected alternate binding", |
| new HashMap<>(expectedAlternateBinding), |
| new HashMap<>(actualAlternateBindingMap)); |
| |
| Map<String, Object> dlqExchangeAttributes = readEntityUsingAmqpManagement( |
| deadLetterExchangeName, |
| "org.apache.qpid.FanoutExchange", |
| true); |
| assertNotNull("Cannot get dead letter exchange", dlqExchangeAttributes); |
| } |
| |
| @SuppressWarnings("unchecked") |
| private Map<String, Object> convertIfNecessary(final Object actualAlternateBinding) throws IOException |
| { |
| Map<String, Object> actualAlternateBindingMap; |
| if (actualAlternateBinding instanceof String) |
| { |
| actualAlternateBindingMap = new ObjectMapper().readValue((String)actualAlternateBinding, Map.class); |
| } |
| else |
| { |
| actualAlternateBindingMap = (Map<String, Object>) actualAlternateBinding; |
| } |
| return actualAlternateBindingMap; |
| } |
| |
| private String getDestinationAddress(final String name, final String type) |
| { |
| return getProtocol() == Protocol.AMQP_1_0 |
| ? name |
| : String.format("ADDR: %s; { assert: never, node: { type: %s } }", name, type); |
| } |
| } |
