Driver: Add general predicate parameter for waitForRx

communication/driver.cpp

@@ -8,7 +8,7 @@
 
 using namespace icsneo;
 
-bool Driver::writeToReadBuffer(const uint8_t* buf, size_t numReceived) {
+bool Driver::pushRx(const uint8_t* buf, size_t numReceived) {
 	bool ret = readBuffer.write(buf, numReceived);
 
 	if(hasRxWaitRequest) {
@@ -18,11 +18,18 @@ bool Driver::writeToReadBuffer(const uint8_t* buf, size_t numReceived) {
 	return ret;
 }
 
-bool Driver::waitForRx(size_t minBytes, std::chrono::milliseconds timeout) {
+bool Driver::waitForRx(size_t limit, std::chrono::milliseconds timeout) {
+	return waitForRx([limit, this]() {
+		return readBuffer.size() >= limit;
+	}, timeout);
+}
+
+bool Driver::waitForRx(std::function<bool()> predicate, std::chrono::milliseconds timeout) {
 	std::unique_lock<std::mutex> lk(rxWaitMutex);
 	hasRxWaitRequest = true;
 
-	auto ret = rxWaitRequestCv.wait_for(lk, timeout, [this, minBytes]{ return readBuffer.size() >= minBytes; });
+	auto ret = rxWaitRequestCv.wait_for(lk, timeout, predicate);
+
 	hasRxWaitRequest = false;
 
 	return ret;

include/icsneo/communication/driver.h

@@ -27,7 +27,8 @@ public:
 	virtual bool isDisconnected() { return disconnected; };
 	virtual bool close() = 0;
 
-	bool waitForRx(size_t minBytes, std::chrono::milliseconds timeout);
+	bool waitForRx(size_t limit, std::chrono::milliseconds timeout);
+	bool waitForRx(std::function<bool()> predicate, std::chrono::milliseconds timeout);
 	bool readWait(std::vector<uint8_t>& bytes, std::chrono::milliseconds timeout = std::chrono::milliseconds(100), size_t limit = 0);
 	bool write(const std::vector<uint8_t>& bytes);
 	virtual bool isEthernet() const { return false; }
@@ -59,7 +60,7 @@ protected:
 	virtual bool writeQueueAlmostFull() { return writeQueue.size_approx() > (writeQueueSize * 3 / 4); }
 	virtual bool writeInternal(const std::vector<uint8_t>& b) { return writeQueue.enqueue(WriteOperation(b)); }
 
-	bool writeToReadBuffer(const uint8_t* buf, size_t numReceived);
+	bool pushRx(const uint8_t* buf, size_t numReceived);
 	RingBuffer readBuffer = RingBuffer(ICSNEO_DRIVER_RINGBUFFER_SIZE);
 	std::atomic<bool> hasRxWaitRequest = false;
 	std::condition_variable rxWaitRequestCv;

platform/ftd3xx.cpp

@@ -136,7 +136,7 @@ void FTD3XX::readTask() {
 		}
 		FT_ReleaseOverlapped(*handle, &overlap);
 		if(received > 0) {
-			writeToReadBuffer(buffer, received);
+			pushRx(buffer, received);
 		}
 	}
 }

platform/posix/cdcacm.cpp

@@ -190,7 +190,7 @@ void CDCACM::readTask() {
 			}
 			std::cout << std::dec << std::endl;
 #endif
-			writeToReadBuffer(readbuf, bytesRead);
+			pushRx(readbuf, bytesRead);
 		} else {
 			if(modeChanging) {
 				// We were expecting a disconnect for reenumeration

platform/posix/firmio.cpp

@@ -242,7 +242,7 @@ void FirmIO::readTask() {
 
 				// Translate the physical address back to our virtual address space
 				uint8_t* addr = reinterpret_cast<uint8_t*>(msg.payload.data.addr - PHY_ADDR_BASE + vbase);
-				while (!writeToReadBuffer(addr, msg.payload.data.len)) {
+				while (!pushRx(addr, msg.payload.data.len)) {
 					std::this_thread::sleep_for(std::chrono::milliseconds(1)); // back-off so reading thread can empty the buffer
 					if (closing || isDisconnected()) {
 						break;

platform/posix/ftdi.cpp

@@ -213,7 +213,7 @@ void FTDI::readTask() {
 			} else
 				report(APIEvent::Type::FailedToRead, APIEvent::Severity::EventWarning);
 		} else
-			writeToReadBuffer(readbuf, readBytes);
+			pushRx(readbuf, readBytes);
 	}
 }
 

platform/posix/pcap.cpp

@@ -284,7 +284,7 @@ void PCAP::readTask() {
 			PCAP* driver = reinterpret_cast<PCAP*>(obj);
 			if(driver->ethPacketizer.inputUp({data, data + header->caplen})) {
 				const auto bytes = driver->ethPacketizer.outputUp();
-				driver->writeToReadBuffer(bytes.data(), bytes.size());
+				driver->pushRx(bytes.data(), bytes.size());
 			}
 		}, (uint8_t*)this);
 	}

platform/tcp.cpp

@@ -533,7 +533,7 @@ void TCP::readTask() {
 	uint8_t readbuf[READ_BUFFER_SIZE];
 	while(!closing) {
 		if(const auto received = ::recv(*socket, (char*)readbuf, READ_BUFFER_SIZE, 0); received > 0) {
-			writeToReadBuffer(readbuf, received);
+			pushRx(readbuf, received);
 		} else {
 			timeout.tv_sec = 0;
 			timeout.tv_usec = 50'000;

platform/windows/pcap.cpp

@@ -278,7 +278,7 @@ void PCAP::readTask() {
 
 		if(ethPacketizer.inputUp({data, data + header->caplen})) {
 			const auto bytes = ethPacketizer.outputUp();
-			writeToReadBuffer(bytes.data(), bytes.size());
+			pushRx(bytes.data(), bytes.size());
 		}
 	}
 }

platform/windows/vcp.cpp

@@ -390,7 +390,7 @@ void VCP::readTask() {
 				if(ReadFile(detail->handle, readbuf, READ_BUFFER_SIZE, nullptr, &detail->overlappedRead)) {
 					if(GetOverlappedResult(detail->handle, &detail->overlappedRead, &bytesRead, FALSE)) {
 						if(bytesRead)
-							writeToReadBuffer(readbuf, bytesRead);
+							pushRx(readbuf, bytesRead);
 					}
 					continue;
 				}
@@ -413,7 +413,7 @@ void VCP::readTask() {
 				auto ret = WaitForSingleObject(detail->overlappedRead.hEvent, 100);
 				if(ret == WAIT_OBJECT_0) {
 					if(GetOverlappedResult(detail->handle, &detail->overlappedRead, &bytesRead, FALSE)) {
-						writeToReadBuffer(readbuf, bytesRead);
+						pushRx(readbuf, bytesRead);
 						state = LAUNCH;
 					} else
 						report(APIEvent::Type::FailedToRead, APIEvent::Severity::Error);