diff options
Diffstat (limited to 'cpp/tests')
| -rw-r--r-- | cpp/tests/BasicP2PTest.h | 2 | ||||
| -rw-r--r-- | cpp/tests/ChannelTest.cpp | 32 |
2 files changed, 33 insertions, 1 deletions
diff --git a/cpp/tests/BasicP2PTest.h b/cpp/tests/BasicP2PTest.h index 989344bb47..8b5d0e7a8c 100644 --- a/cpp/tests/BasicP2PTest.h +++ b/cpp/tests/BasicP2PTest.h @@ -71,7 +71,7 @@ public: std::string queue = params.getString("P2P_QUEUE_AND_KEY_NAME"); int messages = params.getInt("P2P_NUM_MESSAGES"); if (role == "SENDER") { - worker = std::auto_ptr<Worker>(new Sender(options, Exchange::DEFAULT_EXCHANGE, queue, messages)); + worker = std::auto_ptr<Worker>(new Sender(options, Exchange::STANDARD_DIRECT_EXCHANGE, queue, messages)); } else if(role == "RECEIVER"){ worker = std::auto_ptr<Worker>(new Receiver(options, queue, messages)); } else { diff --git a/cpp/tests/ChannelTest.cpp b/cpp/tests/ChannelTest.cpp index f0860b8a28..cc0a90bad9 100644 --- a/cpp/tests/ChannelTest.cpp +++ b/cpp/tests/ChannelTest.cpp @@ -53,6 +53,7 @@ class ChannelTest : public CppUnit::TestCase CPPUNIT_TEST(testDeliveryAndRecovery); CPPUNIT_TEST(testStaging); CPPUNIT_TEST(testQueuePolicy); + CPPUNIT_TEST(testFlow); CPPUNIT_TEST_SUITE_END(); class MockMessageStore : public NullMessageStore @@ -303,6 +304,37 @@ class ChannelTest : public CppUnit::TestCase store.check(); } + + void testFlow(){ + DummyHandler handler; + Channel channel(qpid::framing::highestProtocolVersion, &handler, 7, 10000); + + const string data("abcdefghijklmn"); + + Message::shared_ptr msg(createMessage("test", "my_routing_key", "my_message_id", 14)); + addContent(msg, data); + Queue::shared_ptr queue(new Queue("my_queue")); + ConnectionToken* owner(0); + string tag("no_ack"); + channel.consume(tag, queue, false, false, owner); + channel.flow(false); + queue->deliver(msg); + CPPUNIT_ASSERT_EQUAL((size_t) 0, handler.frames.size()); + CPPUNIT_ASSERT_EQUAL((u_int32_t) 1, queue->getMessageCount()); + channel.flow(true); + CPPUNIT_ASSERT_EQUAL((size_t) 3, handler.frames.size()); + CPPUNIT_ASSERT_EQUAL((u_int16_t) 7, handler.frames[0]->getChannel()); + CPPUNIT_ASSERT_EQUAL((u_int16_t) 7, handler.frames[1]->getChannel()); + CPPUNIT_ASSERT_EQUAL((u_int16_t) 7, handler.frames[2]->getChannel()); + BasicDeliverBody::shared_ptr deliver(dynamic_pointer_cast<BasicDeliverBody, AMQBody>(handler.frames[0]->getBody())); + AMQHeaderBody::shared_ptr contentHeader(dynamic_pointer_cast<AMQHeaderBody, AMQBody>(handler.frames[1]->getBody())); + AMQContentBody::shared_ptr contentBody(dynamic_pointer_cast<AMQContentBody, AMQBody>(handler.frames[2]->getBody())); + CPPUNIT_ASSERT(deliver); + CPPUNIT_ASSERT(contentHeader); + CPPUNIT_ASSERT(contentBody); + CPPUNIT_ASSERT_EQUAL(data, contentBody->getData()); + } + Message* createMessage(const string& exchange, const string& routingKey, const string& messageId, u_int64_t contentSize) { Message* msg = new Message(0, exchange, routingKey, false, false); |