#include "piethernet.h"
#include "piliterals_time.h"
#include "pitime.h"

#include "gtest/gtest.h"
#include <arpa/inet.h>
#include <atomic>
#include <cstring>
#include <functional>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

static bool wait_for(std::function<bool()> cond, PISystemTime timeout) {
	PITimeMeasurer tm;
	while (tm.elapsed() < timeout) {
		if (cond()) return true;
		piMinSleep();
	}
	return cond();
}

// writeDevice() retries EAGAIN while the peer does not drain its receive
// buffer; the retry loop must be interruptible, so stopAndWait() on a
// busy writer must return.
TEST(PIEthernet, WriteThreadStopsWhilePeerNotReading) {
	constexpr uint16_t port = 10299;

	// a raw listener that accepts the connection but never reads it
	int ls                  = socket(AF_INET, SOCK_STREAM, 0);
	ASSERT_GE(ls, 0);
	int on = 1;
	setsockopt(ls, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
	sockaddr_in sa;
	memset(&sa, 0, sizeof(sa));
	sa.sin_family = AF_INET;
	sa.sin_port   = htons(port);
	inet_pton(AF_INET, "127.0.0.1", &sa.sin_addr);
	ASSERT_EQ(0, bind(ls, (sockaddr *)&sa, sizeof(sa)));
	ASSERT_EQ(0, listen(ls, 1));

	// on the heap: if the stop below hangs, the destructor would wait for
	// the stuck thread too, so on failure the device is deliberately
	// leaked (the test process is about to exit anyway)
	PIEthernet * eth = new PIEthernet(PIEthernet::TCP_Client);
	// connect() only flags the device, the read thread performs the actual
	// TCP connection; the read thread also provides the stopping state
	// that the EAGAIN retry loop in writeDevice() observes
	ASSERT_TRUE(eth->connect(PINetworkAddress("127.0.0.1", port)));
	eth->startThreadedRead();
	eth->startThreadedWrite();

	int acc = accept(ls, nullptr, nullptr);
	ASSERT_GE(acc, 0);
	ASSERT_TRUE(wait_for([&]() { return eth->isConnected(); }, 5_s)) << "client did not connect";

	// well above the loopback kernel buffer capacity (tcp_wmem + tcp_rmem
	// ~10 MB), so the send buffer is guaranteed to fill up and the write
	// thread to reach the EAGAIN path in writeDevice()
	const size_t MB = 1024 * 1024;
	PIByteArray big(8 * MB, 0x42);
	for (int i = 0; i < 4; ++i)
		eth->writeThreaded(big);

	// let the write thread fill the buffers and jam on the send
	wait_for([] { return false; }, 2_s);

	std::atomic_bool done{false};
	PIThread stopper;
	stopper.start([&] {
		eth->stopAndWait();
		done.store(true);
	});
	ASSERT_TRUE(wait_for([&]() { return done.load(); }, 15_s)) << "stopAndWait() hung on a stalled write";
	stopper.stopAndWait();
	delete eth;
	::close(acc);
	::close(ls);
}