pip/tests/thread/pigrabberbase_test.cpp

#include "pibase_macros.h"
#include "pigrabberbase.h"
#include "pisystemtime.h"
#include "pithread.h"
#include "pitime.h"

#include "gtest/gtest.h"
#include <atomic>
#include <thread>

namespace {

class FakeGrabber: public PIGrabberBase<int> {
	PIOBJECT_SUBCLASS(FakeGrabber, PIGrabberBase<int>)

public:
	std::atomic_int open_count{0};
	std::atomic_bool open_ok{true};
	int open_fail_times = 0; // fail first N open attempts

	std::atomic_bool data_available{false};
	int ready_value = 7;
	std::atomic_int poll_count{0};
	std::atomic_int record_count{0};

protected:
	bool openInternal() override {
		open_count++;
		if (open_fail_times > 0) {
			open_fail_times--;
			return false;
		}
		return open_ok;
	}

	void closeInternal() override {}

	int get(int & val) override {
		poll_count++;
		if (!data_available) return 1; // no item ready yet
		val = ready_value;
		return 0;
	}

	void record(const int & val) override {
		(void)val;
		record_count++;
	}
};

bool waitForQueSize(PIGrabberBase<int> & g, int size, double timeout_s = 5.) {
	PITimeMeasurer tm;
	while (g.queSize() < size && tm.elapsed_s() < timeout_s)
		piMSleep(1);
	return g.queSize() >= size;
}

} // namespace


TEST(PIGrabberBase_Basic, CaptureAndLast) {
	FakeGrabber g;
	g.data_available = true;
	g.ready_value    = 7;

	std::atomic_int ready_events(0);
	CONNECTL(&g, dataReady, [&ready_events]() { ready_events++; });

	g.start();
	ASSERT_TRUE(waitForQueSize(g, 5));

	EXPECT_EQ(g.last(), 7);
	EXPECT_GE(ready_events.load(), 5);

	g.stopGrabber();
	EXPECT_FALSE(g.isRunning());
	EXPECT_FALSE(g.isOpened());

	// queue is frozen now, counters must match exactly
	EXPECT_EQ(g.diag().state().received_packets, (ullong)g.queSize());
	EXPECT_EQ(g.dequeue(), 7);
}


TEST(PIGrabberBase_Wake, WakeUpBeatsPollDelay) {
	FakeGrabber g;
	g.data_available = false;
	g.setPollDelay(PISystemTime::fromMilliseconds(1000));

	g.start();
	piMSleep(100);
	ASSERT_EQ(g.queSize(), 0); // loop is in its 1 second wait

	g.ready_value    = 5;
	g.data_available = true;
	g.wakeUp();

	PITimeMeasurer tm;
	while (g.queSize() < 1 && tm.elapsed_s() < 2.)
		piMSleep(1);
	EXPECT_GE(g.queSize(), 1);
	EXPECT_EQ(g.last(), 5);
	EXPECT_LT(tm.elapsed_m(), 500.); // woke up long before the 1 second poll interval

	g.stopGrabber();
	EXPECT_FALSE(g.isRunning());
}


TEST(PIGrabberBase_Stop, FastStopDuringLongPollWait) {
	FakeGrabber g;
	g.data_available = false;
	g.setPollDelay(PISystemTime::fromMilliseconds(2000));

	g.start();
	piMSleep(100); // loop is now waiting up to 2 seconds

	PITimeMeasurer tm;
	g.stopGrabber();
	EXPECT_LT(tm.elapsed_m(), 500.); // stop interrupts the wait, does not wait for it
	EXPECT_FALSE(g.isRunning());
}


TEST(PIGrabberBase_Open, RetryUntilSuccess) {
	FakeGrabber g;
	g.open_fail_times = 2; // first two opens fail, the third succeeds
	g.data_available  = true;

	g.start();

	PITimeMeasurer tm;
	while (!g.isOpened() && tm.elapsed_s() < 3.)
		piMSleep(1);
	EXPECT_TRUE(g.isOpened());
	EXPECT_LE(tm.elapsed_s(), 1.); // two 200 ms retry waits
	EXPECT_EQ(g.open_count.load(), 3);

	ASSERT_TRUE(waitForQueSize(g, 1));
	EXPECT_EQ(g.last(), g.ready_value);

	g.stopGrabber();
	EXPECT_FALSE(g.isRunning());
}


TEST(PIGrabberBase_Open, FastStopDuringOpenRetry) {
	FakeGrabber g;
	g.open_ok = false; // opens always fail, loop retries every 200 ms

	g.start();
	piMSleep(100); // loop is in the 200 ms open-retry wait

	PITimeMeasurer tm;
	g.stopGrabber();
	EXPECT_LT(tm.elapsed_m(), 150.); // stop interrupts the retry wait
	EXPECT_FALSE(g.isRunning());
	EXPECT_FALSE(g.isOpened());
}


TEST(PIGrabberBase_Recording, StartStopRecord) {
	FakeGrabber g;
	g.data_available = true;

	g.start();
	ASSERT_TRUE(waitForQueSize(g, 2));

	EXPECT_FALSE(g.isRecording());
	g.startRecord("test");
	ASSERT_TRUE(g.isRecording());

	PITimeMeasurer tm;
	while (g.record_count.load() < 1 && tm.elapsed_s() < 5.)
		piMSleep(1);
	EXPECT_GE(g.record_count.load(), 1);

	g.stopRecord();
	EXPECT_FALSE(g.isRecording());

	g.stopGrabber();
	EXPECT_FALSE(g.isRunning());
}


TEST(PIGrabberBase_Close, RestartReopens) {
	FakeGrabber g;
	g.data_available = true;

	g.start();
	ASSERT_TRUE(waitForQueSize(g, 1));
	EXPECT_EQ(g.last(), g.ready_value);

	// stop the data flow, then let the loop settle before clearing
	g.data_available = false;
	piMSleep(20);
	g.restart(); // clear queue and close
	EXPECT_FALSE(g.isOpened());
	EXPECT_TRUE(g.isEmpty());

	// the running loop must reopen the grabber on its own
	PITimeMeasurer tm;
	while (!g.isOpened() && tm.elapsed_s() < 3.)
		piMSleep(1);
	EXPECT_TRUE(g.isOpened());

	g.stopGrabber();
	EXPECT_FALSE(g.isRunning());
}


TEST(PIGrabberBase_PollDelay, GetterSetter) {
	FakeGrabber g;
	EXPECT_DOUBLE_EQ(g.pollDelay().toMilliseconds(), (double)PIP_MIN_MSLEEP);

	g.setPollDelay(PISystemTime::fromMilliseconds(55));
	EXPECT_DOUBLE_EQ(g.pollDelay().toMilliseconds(), 55.);
}