net_integration.cpp

#include "catch.hpp"
#include <ftl/protocol.hpp>
#include <ftl/protocol/self.hpp>
#include <ftl/protocol/node.hpp>
#include <ftl/uri.hpp>
#include <ftl/exception.hpp>

#include <thread>
#include <chrono>

using std::this_thread::sleep_for;
using std::chrono::milliseconds;

// --- Support -----------------------------------------------------------------

static bool try_for(int count, const std::function<bool()> &f) {
	int i=count;
	while (i-- > 0) {
		if (f()) return true;
		sleep_for(milliseconds(10));
	}
	return false;
}

// --- Tests -------------------------------------------------------------------

TEST_CASE("Listen and Connect", "[net]") {
	auto self = ftl::createDummySelf();
	
	self->listen(ftl::URI("tcp://localhost:0")); 

	SECTION("valid tcp connection using ipv4") {
		auto uri = "tcp://127.0.0.1:" + std::to_string(self->getListeningURIs().front().getPort());
		LOG(INFO) << uri;
		auto p = ftl::connectNode(uri);
		REQUIRE( p );
		
		REQUIRE( p->waitConnection(5) );
		
		REQUIRE( self->waitConnections(5) == 1 );
		REQUIRE( ftl::getSelf()->numberOfNodes() == 1);
	}

	SECTION("valid tcp connection using hostname") {
		auto uri = "tcp://localhost:" + std::to_string(self->getListeningURIs().front().getPort());
		auto p = ftl::connectNode(uri);
		REQUIRE( p );
		
		REQUIRE( p->waitConnection(5) );
		
		REQUIRE( self->waitConnections(5) == 1 );
		REQUIRE( ftl::getSelf()->numberOfNodes() == 1);
	}

	SECTION("invalid protocol") {
		bool throws = false;
		try {
			auto p = ftl::connectNode("http://localhost:1234");
		}
		catch (const ftl::exception& ex) {
			ex.ignore();
			throws = true;
		}
		REQUIRE(throws);
	}

	SECTION("automatic reconnect from originating connection") {
		auto uri = "tcp://localhost:" + std::to_string(self->getListeningURIs().front().getPort());

		auto p_connecting = ftl::connectNode(uri);
		REQUIRE(p_connecting);

		REQUIRE(p_connecting->waitConnection(5));
		p_connecting->close(true);

		REQUIRE(p_connecting->status() != ftl::protocol::NodeStatus::kConnected);
		REQUIRE(p_connecting->waitConnection(5));
	}

	SECTION("automatic reconnect from remote termination") {
		auto uri = "tcp://localhost:" + std::to_string(self->getListeningURIs().front().getPort());

		auto p_connecting = ftl::connectNode(uri);
		REQUIRE(p_connecting);

		REQUIRE(p_connecting->waitConnection(5));
		REQUIRE(p_connecting->connectionCount() == 1);
		
		auto nodes = self->getNodes();
		REQUIRE( nodes.size() == 1 );
		for (auto &node : nodes) {
			node->waitConnection(5);
			node->close();
		}

		bool r = try_for(500, [p_connecting]{ return p_connecting->connectionCount() >= 2; });
		REQUIRE( r );
	}

	ftl::protocol::reset();
}

TEST_CASE("Self::onConnect()", "[net]") {
	auto self = ftl::createDummySelf();
	
	self->listen(ftl::URI("tcp://localhost:0")); 

	auto uri = "tcp://localhost:" + std::to_string(self->getListeningURIs().front().getPort());

	SECTION("single valid remote init connection") {
		bool done = false;

		auto h = self->onConnect([&](const std::shared_ptr<ftl::protocol::Node> &p_listening) {
			done = true;
			return true;
		});

		REQUIRE( ftl::connectNode(uri) );

		bool result = try_for(20, [&done]{ return done; });
		REQUIRE( result );
	}

	SECTION("single valid init connection") {
		bool done = false;

		auto h = ftl::getSelf()->onConnect([&](const std::shared_ptr<ftl::protocol::Node> &p_listening) {
			done = true;
			return true;
		});

		REQUIRE( ftl::connectNode(uri)->waitConnection(5) );

		REQUIRE( done );
	}

	ftl::protocol::reset();
}

/*TEST_CASE("Universe::onDisconnect()", "[net]") {
	Universe a;
	Universe b;

	a.listen(ftl::URI("tcp://localhost:0"));
	auto uri = "tcp://localhost:" + std::to_string(a.getListeningURIs().front().getPort());

	SECTION("single valid remote close") {
		bool done = false;

		a.onDisconnect([&done](Peer *p) {
			done = true;
		});

		Peer *p = b.connect(uri);
		p->waitConnection();
		sleep_for(milliseconds(20));
		p->close();

		REQUIRE( try_for(20, [&done]{ return done; }) );
	}

	SECTION("single valid close") {
		bool done = false;

		b.onDisconnect([&done](Peer *p) {
			done = true;
		});

		Peer *p = b.connect(uri);
		p->waitConnection();
		sleep_for(milliseconds(20));
		p->close();

		REQUIRE( try_for(20, [&done]{ return done; }) );
	}
}

TEST_CASE("Universe::broadcast()", "[net]") {
	Universe a;
	Universe b;
	
	a.listen(ftl::URI("tcp://localhost:0"));
	auto uri = "tcp://localhost:" + std::to_string(a.getListeningURIs().front().getPort());
	
	SECTION("no arguments to no peers") {
		bool done = false;
		a.bind("hello", [&done]() {
			done = true;
		});
		
		b.broadcast("done");
		
		sleep_for(milliseconds(50));
		REQUIRE( !done );
	}
	
	SECTION("no arguments to one peer") {
		b.connect(uri)->waitConnection();
		
		bool done = false;
		a.bind("hello", [&done]() {
			done = true;
		});
		
		b.broadcast("hello");
		
		REQUIRE( try_for(20, [&done]{ return done; }) );
	}
	
	SECTION("one argument to one peer") {
		b.connect(uri)->waitConnection();
		
		int done = 0;
		a.bind("hello", [&done](int v) {
			done = v;
		});
		
		b.broadcast("hello", 676);
		
		REQUIRE( try_for(20, [&done]{ return done == 676; }) );
	}
	
	SECTION("one argument to two peers") {
		Universe c;
		
		b.connect(uri)->waitConnection();
		c.connect(uri)->waitConnection();
		
		int done1 = 0;
		b.bind("hello", [&done1](int v) {
			done1 = v;
		});
		
		int done2 = 0;
		c.bind("hello", [&done2](int v) {
			done2 = v;
		});

		REQUIRE( a.numberOfPeers() == 2 );
		//sleep_for(milliseconds(100)); // NOTE: Binding might not be ready
		
		a.broadcast("hello", 676);
		
		REQUIRE( try_for(20, [&done1, &done2]{ return done1 == 676 && done2 == 676; }) );
	}
}

TEST_CASE("Universe::findAll()", "") {
	Universe a;
	Universe b;
	Universe c;

	a.listen(ftl::URI("tcp://localhost:0"));
	auto uri = "tcp://localhost:" + std::to_string(a.getListeningURIs().front().getPort());

	b.connect(uri)->waitConnection();
	c.connect(uri)->waitConnection();

	SECTION("no values exist") {
		REQUIRE( (c.findAll<int>("test_all").size() == 0) );
	}

	SECTION("one set exists") {
		a.bind("test_all", []() -> std::vector<int> {
			return {3,4,5};
		});

		auto res = c.findAll<int>("test_all");
		REQUIRE( (res.size() == 3) );
		REQUIRE( (res[0] == 3) );
	}

	SECTION("two sets exists") {
		b.bind("test_all", []() -> std::vector<int> {
			return {3,4,5};
		});
		c.bind("test_all", []() -> std::vector<int> {
			return {6,7,8};
		});

		//sleep_for(milliseconds(100)); // NOTE: Binding might not be ready

		auto res = a.findAll<int>("test_all");
		REQUIRE( (res.size() == 6) );
		REQUIRE( (res[0] == 3 || res[0] == 6) );
	}
}

TEST_CASE("Peer::call() __ping__", "") {
	Universe a;
	Universe b;
	Universe c;

	a.listen(ftl::URI("tcp://localhost:0"));
	auto uri = "tcp://localhost:" + std::to_string(a.getListeningURIs().front().getPort());

	auto *p = b.connect(uri);
	p->waitConnection();

	SECTION("single ping") {
		int64_t res = p->call<int64_t>("__ping__");
		REQUIRE((res <= ftl::timer::get_time() && res > 0));
	}

	SECTION("large number of pings") {
		for (int i=0; i<100; ++i) {
			int64_t res = p->call<int64_t>("__ping__");
			REQUIRE(res > 0);
		}
	}

	SECTION("large number of parallel pings") {
		std::atomic<int> count = 0;
		for (int i=0; i<100; ++i) {
			ftl::pool.push([&count, p](int id) {
				int64_t res = p->call<int64_t>("__ping__");
				REQUIRE( res > 0 );
				count++;
			});
		}

		while (count < 100) std::this_thread::sleep_for(milliseconds(5));
	}

	SECTION("single invalid rpc") {
		bool errored = false;
		try {
			int64_t res = p->call<int64_t>("__ping2__");
			REQUIRE( res > 0 );  // Not called or required actually
		} catch (const ftl::exception &e) {
			e.ignore(); // supress log output
			errored = true;
		}

		REQUIRE(errored);
	}
}*/