diff options
Diffstat (limited to 'cpp/src/tests/ExchangeTest.cpp')
| -rw-r--r-- | cpp/src/tests/ExchangeTest.cpp | 159 |
1 files changed, 140 insertions, 19 deletions
diff --git a/cpp/src/tests/ExchangeTest.cpp b/cpp/src/tests/ExchangeTest.cpp index e496f0c478..88a1cd99c2 100644 --- a/cpp/src/tests/ExchangeTest.cpp +++ b/cpp/src/tests/ExchangeTest.cpp @@ -7,9 +7,9 @@ * 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 @@ -39,9 +39,12 @@ using namespace qpid::framing; using namespace qpid::sys; using namespace qpid; +namespace qpid { +namespace tests { + QPID_AUTO_TEST_SUITE(ExchangeTestSuite) -QPID_AUTO_TEST_CASE(testMe) +QPID_AUTO_TEST_CASE(testMe) { Queue::shared_ptr queue(new Queue("queue", true)); Queue::shared_ptr queue2(new Queue("queue2", true)); @@ -57,7 +60,7 @@ QPID_AUTO_TEST_CASE(testMe) queue.reset(); queue2.reset(); - intrusive_ptr<Message> msgPtr(MessageUtils::createMessage("exchange", "key", "id")); + intrusive_ptr<Message> msgPtr(MessageUtils::createMessage("exchange", "key", false, "id")); DeliverableMessage msg(msgPtr); topic.route(msg, "abc", 0); direct.route(msg, "abc", 0); @@ -70,15 +73,15 @@ QPID_AUTO_TEST_CASE(testIsBound) Queue::shared_ptr b(new Queue("b", true)); Queue::shared_ptr c(new Queue("c", true)); Queue::shared_ptr d(new Queue("d", true)); - + string k1("abc"); string k2("def"); string k3("xyz"); FanOutExchange fanout("fanout"); - fanout.bind(a, "", 0); - fanout.bind(b, "", 0); - fanout.bind(c, "", 0); + BOOST_CHECK(fanout.bind(a, "", 0)); + BOOST_CHECK(fanout.bind(b, "", 0)); + BOOST_CHECK(fanout.bind(c, "", 0)); BOOST_CHECK(fanout.isBound(a, 0, 0)); BOOST_CHECK(fanout.isBound(b, 0, 0)); @@ -86,10 +89,10 @@ QPID_AUTO_TEST_CASE(testIsBound) BOOST_CHECK(!fanout.isBound(d, 0, 0)); DirectExchange direct("direct"); - direct.bind(a, k1, 0); - direct.bind(a, k3, 0); - direct.bind(b, k2, 0); - direct.bind(c, k1, 0); + BOOST_CHECK(direct.bind(a, k1, 0)); + BOOST_CHECK(direct.bind(a, k3, 0)); + BOOST_CHECK(direct.bind(b, k2, 0)); + BOOST_CHECK(direct.bind(c, k1, 0)); BOOST_CHECK(direct.isBound(a, 0, 0)); BOOST_CHECK(direct.isBound(a, &k1, 0)); @@ -104,10 +107,10 @@ QPID_AUTO_TEST_CASE(testIsBound) BOOST_CHECK(!direct.isBound(d, &k3, 0)); TopicExchange topic("topic"); - topic.bind(a, k1, 0); - topic.bind(a, k3, 0); - topic.bind(b, k2, 0); - topic.bind(c, k1, 0); + BOOST_CHECK(topic.bind(a, k1, 0)); + BOOST_CHECK(topic.bind(a, k3, 0)); + BOOST_CHECK(topic.bind(b, k2, 0)); + BOOST_CHECK(topic.bind(c, k1, 0)); BOOST_CHECK(topic.isBound(a, 0, 0)); BOOST_CHECK(topic.isBound(a, &k1, 0)); @@ -139,7 +142,7 @@ QPID_AUTO_TEST_CASE(testIsBound) headers.bind(a, "", &args3); headers.bind(b, "", &args2); headers.bind(c, "", &args1); - + BOOST_CHECK(headers.isBound(a, 0, 0)); BOOST_CHECK(headers.isBound(a, 0, &args1)); BOOST_CHECK(headers.isBound(a, 0, &args3)); @@ -153,7 +156,7 @@ QPID_AUTO_TEST_CASE(testIsBound) BOOST_CHECK(!headers.isBound(d, 0, &args3)); } -QPID_AUTO_TEST_CASE(testDeleteGetAndRedeclare) +QPID_AUTO_TEST_CASE(testDeleteGetAndRedeclare) { ExchangeRegistry exchanges; exchanges.declare("my-exchange", "direct", false, FieldTable()); @@ -162,7 +165,125 @@ QPID_AUTO_TEST_CASE(testDeleteGetAndRedeclare) exchanges.get("my-exchange"); } catch (const NotFoundException&) {} std::pair<Exchange::shared_ptr, bool> response = exchanges.declare("my-exchange", "direct", false, FieldTable()); - BOOST_CHECK_EQUAL(string("direct"), response.first->getType()); + BOOST_CHECK_EQUAL(string("direct"), response.first->getType()); +} + +intrusive_ptr<Message> cmessage(std::string exchange, std::string routingKey) { + intrusive_ptr<Message> msg(new Message()); + AMQFrame method((MessageTransferBody(ProtocolVersion(), exchange, 0, 0))); + AMQFrame header((AMQHeaderBody())); + msg->getFrames().append(method); + msg->getFrames().append(header); + msg->getFrames().getHeaders()->get<DeliveryProperties>(true)->setRoutingKey(routingKey); + return msg; +} + +QPID_AUTO_TEST_CASE(testSequenceOptions) +{ + FieldTable args; + args.setInt("qpid.msg_sequence",1); + char* buff = new char[10000]; + framing::Buffer buffer(buff,10000); + { + DirectExchange direct("direct1", false, args); + + intrusive_ptr<Message> msg1 = cmessage("e", "A"); + intrusive_ptr<Message> msg2 = cmessage("e", "B"); + intrusive_ptr<Message> msg3 = cmessage("e", "C"); + + DeliverableMessage dmsg1(msg1); + DeliverableMessage dmsg2(msg2); + DeliverableMessage dmsg3(msg3); + + direct.route(dmsg1, "abc", 0); + direct.route(dmsg2, "abc", 0); + direct.route(dmsg3, "abc", 0); + + BOOST_CHECK_EQUAL(1, msg1->getApplicationHeaders()->getAsInt64("qpid.msg_sequence")); + BOOST_CHECK_EQUAL(2, msg2->getApplicationHeaders()->getAsInt64("qpid.msg_sequence")); + BOOST_CHECK_EQUAL(3, msg3->getApplicationHeaders()->getAsInt64("qpid.msg_sequence")); + + FanOutExchange fanout("fanout1", false, args); + HeadersExchange header("headers1", false, args); + TopicExchange topic ("topic1", false, args); + + // check other exchanges, that they preroute + intrusive_ptr<Message> msg4 = cmessage("e", "A"); + intrusive_ptr<Message> msg5 = cmessage("e", "B"); + intrusive_ptr<Message> msg6 = cmessage("e", "C"); + + DeliverableMessage dmsg4(msg4); + DeliverableMessage dmsg5(msg5); + DeliverableMessage dmsg6(msg6); + + fanout.route(dmsg4, "abc", 0); + BOOST_CHECK_EQUAL(1, msg4->getApplicationHeaders()->getAsInt64("qpid.msg_sequence")); + + FieldTable headers; + header.route(dmsg5, "abc", &headers); + BOOST_CHECK_EQUAL(1, msg5->getApplicationHeaders()->getAsInt64("qpid.msg_sequence")); + + topic.route(dmsg6, "abc", 0); + BOOST_CHECK_EQUAL(1, msg6->getApplicationHeaders()->getAsInt64("qpid.msg_sequence")); + direct.encode(buffer); + } + { + + ExchangeRegistry exchanges; + buffer.reset(); + DirectExchange::shared_ptr exch_dec = Exchange::decode(exchanges, buffer); + + intrusive_ptr<Message> msg1 = cmessage("e", "A"); + DeliverableMessage dmsg1(msg1); + exch_dec->route(dmsg1, "abc", 0); + + BOOST_CHECK_EQUAL(4, msg1->getApplicationHeaders()->getAsInt64("qpid.msg_sequence")); + + } + delete [] buff; +} + +QPID_AUTO_TEST_CASE(testIVEOption) +{ + FieldTable args; + args.setInt("qpid.ive",1); + DirectExchange direct("direct1", false, args); + FanOutExchange fanout("fanout1", false, args); + HeadersExchange header("headers1", false, args); + TopicExchange topic ("topic1", false, args); + + intrusive_ptr<Message> msg1 = cmessage("direct1", "abc"); + msg1->getProperties<MessageProperties>()->getApplicationHeaders().setString("a", "abc"); + DeliverableMessage dmsg1(msg1); + + FieldTable args2; + args2.setString("x-match", "any"); + args2.setString("a", "abc"); + + direct.route(dmsg1, "abc", 0); + fanout.route(dmsg1, "abc", 0); + header.route(dmsg1, "abc", &args2); + topic.route(dmsg1, "abc", 0); + Queue::shared_ptr queue(new Queue("queue", true)); + Queue::shared_ptr queue1(new Queue("queue1", true)); + Queue::shared_ptr queue2(new Queue("queue2", true)); + Queue::shared_ptr queue3(new Queue("queue3", true)); + + BOOST_CHECK(HeadersExchange::match(args2, msg1->getProperties<MessageProperties>()->getApplicationHeaders())); + + BOOST_CHECK(direct.bind(queue, "abc", 0)); + BOOST_CHECK(fanout.bind(queue1, "abc", 0)); + BOOST_CHECK(header.bind(queue2, "", &args2)); + BOOST_CHECK(topic.bind(queue3, "abc", 0)); + + BOOST_CHECK_EQUAL(1u,queue->getMessageCount()); + BOOST_CHECK_EQUAL(1u,queue1->getMessageCount()); + BOOST_CHECK_EQUAL(1u,queue2->getMessageCount()); + BOOST_CHECK_EQUAL(1u,queue3->getMessageCount()); + } + QPID_AUTO_TEST_SUITE_END() + +}} // namespace qpid::tests |