cpp/src/credit_test.cpp - qpid-proton - Git at Google

 /*
  * 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.
  */

 #include "test_bits.hpp"
 #include "proton/connection.hpp"
 #include "proton/connection_options.hpp"
 #include "proton/container.hpp"
 #include "proton/delivery.hpp"
 #include "proton/error_condition.hpp"
 #include "proton/listen_handler.hpp"
 #include "proton/listener.hpp"
 #include "proton/message.hpp"
 #include "proton/messaging_handler.hpp"
 #include "proton/receiver_options.hpp"
 #include "proton/transport.hpp"
 #include "proton/work_queue.hpp"

 #include "proton/internal/pn_unique_ptr.hpp"

 #include <cstdlib>
 #include <ctime>
 #include <string>
 #include <cstdio>
 #include <sstream>

 namespace {

 // Wait for N things to be done.
 class waiter {
     size_t count;
   public:
     waiter(size_t n) : count(n) {}
     void done() { if (--count == 0) ready(); }
     virtual void ready() = 0;
 };

 class server_connection_handler : public proton::messaging_handler {

     struct listen_handler : public proton::listen_handler {
         proton::connection_options opts;
         std::string url;
         waiter& listen_waiter;

         listen_handler(proton::messaging_handler& h, waiter& w) : listen_waiter(w) {
             opts.handler(h);
         }

         void on_open(proton::listener& l) override {
             std::ostringstream o;
             o << "//:" << l.port(); // Connect to the actual listening port
             url = o.str();
             // Schedule rather than call done() direct to ensure serialization
             l.container().schedule(proton::duration::IMMEDIATE,
                                    proton::make_work(&waiter::done, &listen_waiter));
         }

         proton::connection_options on_accept(proton::listener&) override { return opts; }
     };

     proton::listener listener_;
     proton::sender sender_;
     int expect_;
     bool closing_;
     int available_;
     int acked_;
     listen_handler listen_handler_;
     proton::work_queue *notify_wq_;
     proton::work notify_work_;

     void close (proton::connection &c) {
         if (closing_) return;

         c.close(proton::error_condition("amqp:connection:forced", "Failover testing"));
         closing_ = true;
     }

   public:
     server_connection_handler(proton::container& c, int a, waiter& w)
         : expect_(0), closing_(false), available_(a), acked_(0), listen_handler_(*this, w), notify_wq_(0)
     {
         listener_ = c.listen("//:0", listen_handler_);
     }

     std::string url() const {
         if (listen_handler_.url.empty()) throw std::runtime_error("no url");
         return listen_handler_.url;
     }

     void on_connection_open(proton::connection &c) override {
         // Only listen for a single connection
         listener_.stop();
         c.open();
     }

     void on_sender_open(proton::sender &s) override {
         s.open();
         sender_ = s;
     }

     void on_sendable(proton::sender &s) override {
         send_available_messages(s);
     }

     void send_available_messages(proton::sender &s) {
         bool draining = s.draining();
         while (available_ && s.credit() > 0) {
             s.send(proton::message("hello"));
             available_--;
             expect_++;
         }
         if (draining && !available_ && s.credit()) {
             s.return_credit(); // return the rest
         }
     }

     void notify_idle() {
         notify_wq_->add(notify_work_);
     }

     void on_tracker_accept(proton::tracker& d) override {
         acked_++;
         if (acked_ == expect_ && (available_ == 0 || d.sender().credit() == 0))
             notify_idle();
     }

     void on_transport_error(proton::transport & ) override {
         // If we get an error then (try to) stop the listener
         // - this will stop the listener if we didn't already accept a connection
         listener_.stop();
     }

     void notify_on_idle(proton::work_queue &wq, proton::work &w) { notify_wq_ = &wq; notify_work_ = w;}

     void available(int i) {
         available_ = i;
     }
 };

 class tester : public proton::messaging_handler, public waiter {
   public:
     tester() : waiter(1), container_(*this, "credit_tester"),
                received_(0), initial_credit_(0) {}

     void on_container_start(proton::container &c) override {
         srv_.reset(new server_connection_handler(c, 100000, *this));
     }

     // waiter::ready is called when listener can accept connections.
     void ready() override {
         container_.connect(srv_->url());
     }

     void on_connection_open(proton::connection& c) override {
         c.open_receiver("messages", proton::receiver_options().credit_window(0));
     }

     void on_receiver_open(proton::receiver &r) override {
         receiver_ = r;
         next_idle_ = proton::make_work(&tester::first_idle, this);
         proton::work call_on_server_idle(make_work(&tester::on_server_idle, this));
         srv_->notify_on_idle(r.work_queue(), call_on_server_idle);
         r.add_credit(initial_credit_);
     }

     void on_message(proton::delivery &d, proton::message &m) override {
         received_++;
         d.accept();
     }

     void run() {
         container_.run(); // Single threaded to avoid locks and barriers.
     }

     void server_available(int available) {
         // If multithreaded, locking would be required.
         srv_->available(available);
     }

     void on_server_idle() {
         next_idle_();
     }

     void fail(const std::string &msg, int rcv) {
         // Call from work_queue.  Remember the exception later.
         std::ostringstream os;
         os << msg << rcv;
         fail_msg_ = os.str();
         receiver_.connection().close();
     }

     void on_connection_close(proton::connection& c) override {
         if (!fail_msg_.empty())
             FAIL(fail_msg_);
     }

     virtual void first_idle() = 0;

   protected:
     proton::internal::pn_unique_ptr<server_connection_handler> srv_;
     proton::container container_;
     proton::receiver receiver_;
     proton::work next_idle_;
     std::string fail_msg_;
     int received_;
     int initial_credit_;
 };


 class basic_credit_tester : public tester {
   public:
     basic_credit_tester() { initial_credit_ = 3; }

     void first_idle() override {
         if (received_ != 3) {
             fail(FAIL_MSG("messages received should be 3 not "), received_);
             return;
         }
         next_idle_ = proton::make_work(&basic_credit_tester::second_idle, this);
         server_available(2);
         receiver_.add_credit(3);
     }

     void second_idle() {
         if (received_ != 5) {
             fail(FAIL_MSG("messages received should be 5 not "), received_);
             return;
         }
         next_idle_ = proton::make_work(&basic_credit_tester::third_idle, this);
         server_available(10);
         receiver_.add_credit(1);
     }

     void third_idle() {
         if (received_ != 7) {
             fail(FAIL_MSG("messages received should be 7 not "), received_);
             return;
         }
         // passed
         receiver_.connection().close();
     }
 };


 int test_basic_credit() {
     basic_credit_tester().run();
     return 0;
 }


 class drain_credit_tester : public tester {
     int drain_finishes_;

   public:
     drain_credit_tester() : drain_finishes_(0) { initial_credit_ = 10; }

     void on_receiver_drain_finish(proton::receiver &r) override {
         drain_finishes_++;
     }

     void first_idle() override {
         if (received_ != 10) {
             fail(FAIL_MSG("messages received should be 10 not "), received_);
             return;
         }
         next_idle_ = proton::make_work(&drain_credit_tester::second_idle, this);
         server_available(10);
         receiver_.add_credit(15);
         receiver_.drain();
     }

     void second_idle() {
         if (received_ != 20) {
             fail(FAIL_MSG("messages received should be 20 not "), received_);
             return;
         }
         if (drain_finishes_ != 1) {
             fail(FAIL_MSG("drain finish callbacks should be 1, not: "), drain_finishes_);
             return;
         }
         if (receiver_.credit() != 0) {
             fail(FAIL_MSG("credit not returned on drain, remaining: "), receiver_.credit());
             return;
         }
         next_idle_ = proton::make_work(&drain_credit_tester::third_idle, this);
         server_available(5);
         receiver_.add_credit(10);
     }

     void third_idle() {
         if (received_ != 25) {
             fail(FAIL_MSG("messages received should be 20 not "), received_);
             return;
         }
         if (receiver_.credit() != 5) {
             fail(FAIL_MSG("incorrect credit after drain, should be 5, not "), receiver_.credit());
             return;
         }
         next_idle_ = proton::make_work(&drain_credit_tester::fourth_idle, this);
         server_available(3);
         receiver_.add_credit(1);
     }

     void fourth_idle() {
         if (received_ != 28) {
             fail(FAIL_MSG("messages received should be 28 not "), received_);
             return;
         }
         if (receiver_.credit() != 3) {
             fail(FAIL_MSG("incorrect credit, should be 3, not "), receiver_.credit());
             return;
         }
         next_idle_ = proton::make_work(&drain_credit_tester::fifth_idle, this);
         server_available(1);
         receiver_.drain();
     }

     void fifth_idle() {
         if (received_ != 29) {
             fail(FAIL_MSG("messages received should be 29 not "), received_);
             return;
         }
         if (drain_finishes_ != 2) {
             fail(FAIL_MSG("drain finish callbacks should be 2, not: "), drain_finishes_);
             return;
         }
         if (receiver_.credit() != 0) {
             fail(FAIL_MSG("second drain credit failed, should be 0, not "), receiver_.credit());
             return;
         }
         // passed
         receiver_.connection().close();
     }
 };

 int test_drain_credit() {
     drain_credit_tester().run();
     return 0;
 }


 }  // namespace


 int main(int argc, char** argv) {
     int failed = 0;
     RUN_ARGV_TEST(failed, test_basic_credit());
     RUN_ARGV_TEST(failed, test_drain_credit());
     return failed;
 }
	/*
	* 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.
	*/

	#include "test_bits.hpp"
	#include "proton/connection.hpp"
	#include "proton/connection_options.hpp"
	#include "proton/container.hpp"
	#include "proton/delivery.hpp"
	#include "proton/error_condition.hpp"
	#include "proton/listen_handler.hpp"
	#include "proton/listener.hpp"
	#include "proton/message.hpp"
	#include "proton/messaging_handler.hpp"
	#include "proton/receiver_options.hpp"
	#include "proton/transport.hpp"
	#include "proton/work_queue.hpp"

	#include "proton/internal/pn_unique_ptr.hpp"

	#include <cstdlib>
	#include <ctime>
	#include <string>
	#include <cstdio>
	#include <sstream>

	namespace {

	// Wait for N things to be done.
	class waiter {
	size_t count;
	public:
	waiter(size_t n) : count(n) {}
	void done() { if (--count == 0) ready(); }
	virtual void ready() = 0;
	};

	class server_connection_handler : public proton::messaging_handler {

	struct listen_handler : public proton::listen_handler {
	proton::connection_options opts;
	std::string url;
	waiter& listen_waiter;

	listen_handler(proton::messaging_handler& h, waiter& w) : listen_waiter(w) {
	opts.handler(h);
	}

	void on_open(proton::listener& l) override {
	std::ostringstream o;
	o << "//:" << l.port(); // Connect to the actual listening port
	url = o.str();
	// Schedule rather than call done() direct to ensure serialization
	l.container().schedule(proton::duration::IMMEDIATE,
	proton::make_work(&waiter::done, &listen_waiter));
	}

	proton::connection_options on_accept(proton::listener&) override { return opts; }
	};

	proton::listener listener_;
	proton::sender sender_;
	int expect_;
	bool closing_;
	int available_;
	int acked_;
	listen_handler listen_handler_;
	proton::work_queue *notify_wq_;
	proton::work notify_work_;

	void close (proton::connection &c) {
	if (closing_) return;

	c.close(proton::error_condition("amqp:connection:forced", "Failover testing"));
	closing_ = true;
	}

	public:
	server_connection_handler(proton::container& c, int a, waiter& w)
	: expect_(0), closing_(false), available_(a), acked_(0), listen_handler_(*this, w), notify_wq_(0)
	{
	listener_ = c.listen("//:0", listen_handler_);
	}

	std::string url() const {
	if (listen_handler_.url.empty()) throw std::runtime_error("no url");
	return listen_handler_.url;
	}

	void on_connection_open(proton::connection &c) override {
	// Only listen for a single connection
	listener_.stop();
	c.open();
	}

	void on_sender_open(proton::sender &s) override {
	s.open();
	sender_ = s;
	}

	void on_sendable(proton::sender &s) override {
	send_available_messages(s);
	}

	void send_available_messages(proton::sender &s) {
	bool draining = s.draining();
	while (available_ && s.credit() > 0) {
	s.send(proton::message("hello"));
	available_--;
	expect_++;
	}
	if (draining && !available_ && s.credit()) {
	s.return_credit(); // return the rest
	}
	}

	void notify_idle() {
	notify_wq_->add(notify_work_);
	}

	void on_tracker_accept(proton::tracker& d) override {
	acked_++;
	if (acked_ == expect_ && (available_ == 0 \|\| d.sender().credit() == 0))
	notify_idle();
	}

	void on_transport_error(proton::transport & ) override {
	// If we get an error then (try to) stop the listener
	// - this will stop the listener if we didn't already accept a connection
	listener_.stop();
	}

	void notify_on_idle(proton::work_queue &wq, proton::work &w) { notify_wq_ = &wq; notify_work_ = w;}

	void available(int i) {
	available_ = i;
	}
	};

	class tester : public proton::messaging_handler, public waiter {
	public:
	tester() : waiter(1), container_(*this, "credit_tester"),
	received_(0), initial_credit_(0) {}

	void on_container_start(proton::container &c) override {
	srv_.reset(new server_connection_handler(c, 100000, *this));
	}

	// waiter::ready is called when listener can accept connections.
	void ready() override {
	container_.connect(srv_->url());
	}

	void on_connection_open(proton::connection& c) override {
	c.open_receiver("messages", proton::receiver_options().credit_window(0));
	}

	void on_receiver_open(proton::receiver &r) override {
	receiver_ = r;
	next_idle_ = proton::make_work(&tester::first_idle, this);
	proton::work call_on_server_idle(make_work(&tester::on_server_idle, this));
	srv_->notify_on_idle(r.work_queue(), call_on_server_idle);
	r.add_credit(initial_credit_);
	}

	void on_message(proton::delivery &d, proton::message &m) override {
	received_++;
	d.accept();
	}

	void run() {
	container_.run(); // Single threaded to avoid locks and barriers.
	}

	void server_available(int available) {
	// If multithreaded, locking would be required.
	srv_->available(available);
	}

	void on_server_idle() {
	next_idle_();
	}

	void fail(const std::string &msg, int rcv) {
	// Call from work_queue. Remember the exception later.
	std::ostringstream os;
	os << msg << rcv;
	fail_msg_ = os.str();
	receiver_.connection().close();
	}

	void on_connection_close(proton::connection& c) override {
	if (!fail_msg_.empty())
	FAIL(fail_msg_);
	}

	virtual void first_idle() = 0;

	protected:
	proton::internal::pn_unique_ptr<server_connection_handler> srv_;
	proton::container container_;
	proton::receiver receiver_;
	proton::work next_idle_;
	std::string fail_msg_;
	int received_;
	int initial_credit_;
	};


	class basic_credit_tester : public tester {
	public:
	basic_credit_tester() { initial_credit_ = 3; }

	void first_idle() override {
	if (received_ != 3) {
	fail(FAIL_MSG("messages received should be 3 not "), received_);
	return;
	}
	next_idle_ = proton::make_work(&basic_credit_tester::second_idle, this);
	server_available(2);
	receiver_.add_credit(3);
	}

	void second_idle() {
	if (received_ != 5) {
	fail(FAIL_MSG("messages received should be 5 not "), received_);
	return;
	}
	next_idle_ = proton::make_work(&basic_credit_tester::third_idle, this);
	server_available(10);
	receiver_.add_credit(1);
	}

	void third_idle() {
	if (received_ != 7) {
	fail(FAIL_MSG("messages received should be 7 not "), received_);
	return;
	}
	// passed
	receiver_.connection().close();
	}
	};


	int test_basic_credit() {
	basic_credit_tester().run();
	return 0;
	}


	class drain_credit_tester : public tester {
	int drain_finishes_;

	public:
	drain_credit_tester() : drain_finishes_(0) { initial_credit_ = 10; }

	void on_receiver_drain_finish(proton::receiver &r) override {
	drain_finishes_++;
	}

	void first_idle() override {
	if (received_ != 10) {
	fail(FAIL_MSG("messages received should be 10 not "), received_);
	return;
	}
	next_idle_ = proton::make_work(&drain_credit_tester::second_idle, this);
	server_available(10);
	receiver_.add_credit(15);
	receiver_.drain();
	}

	void second_idle() {
	if (received_ != 20) {
	fail(FAIL_MSG("messages received should be 20 not "), received_);
	return;
	}
	if (drain_finishes_ != 1) {
	fail(FAIL_MSG("drain finish callbacks should be 1, not: "), drain_finishes_);
	return;
	}
	if (receiver_.credit() != 0) {
	fail(FAIL_MSG("credit not returned on drain, remaining: "), receiver_.credit());
	return;
	}
	next_idle_ = proton::make_work(&drain_credit_tester::third_idle, this);
	server_available(5);
	receiver_.add_credit(10);
	}

	void third_idle() {
	if (received_ != 25) {
	fail(FAIL_MSG("messages received should be 20 not "), received_);
	return;
	}
	if (receiver_.credit() != 5) {
	fail(FAIL_MSG("incorrect credit after drain, should be 5, not "), receiver_.credit());
	return;
	}
	next_idle_ = proton::make_work(&drain_credit_tester::fourth_idle, this);
	server_available(3);
	receiver_.add_credit(1);
	}

	void fourth_idle() {
	if (received_ != 28) {
	fail(FAIL_MSG("messages received should be 28 not "), received_);
	return;
	}
	if (receiver_.credit() != 3) {
	fail(FAIL_MSG("incorrect credit, should be 3, not "), receiver_.credit());
	return;
	}
	next_idle_ = proton::make_work(&drain_credit_tester::fifth_idle, this);
	server_available(1);
	receiver_.drain();
	}

	void fifth_idle() {
	if (received_ != 29) {
	fail(FAIL_MSG("messages received should be 29 not "), received_);
	return;
	}
	if (drain_finishes_ != 2) {
	fail(FAIL_MSG("drain finish callbacks should be 2, not: "), drain_finishes_);
	return;
	}
	if (receiver_.credit() != 0) {
	fail(FAIL_MSG("second drain credit failed, should be 0, not "), receiver_.credit());
	return;
	}
	// passed
	receiver_.connection().close();
	}
	};

	int test_drain_credit() {
	drain_credit_tester().run();
	return 0;
	}


	} // namespace


	int main(int argc, char** argv) {
	int failed = 0;
	RUN_ARGV_TEST(failed, test_basic_credit());
	RUN_ARGV_TEST(failed, test_drain_credit());
	return failed;
	}