Add test for FdObserver::whenUrgentDataAvailable

c++/src/kj/async-unix-test.c++

@@ -26,7 +26,9 @@
 #include <unistd.h>
 #include <fcntl.h>
 #include <sys/types.h>
+#include <sys/socket.h>
 #include <sys/stat.h>
+#include <netinet/in.h>
 #include <kj/compat/gtest.h>
 #include <pthread.h>
 #include <algorithm>
@@ -453,6 +455,55 @@ TEST(AsyncUnixTest, WriteObserver) {
   EXPECT_TRUE(writable);
 }
 
+#ifdef __linux__
+TEST(AsyncUnixTest, UrgentObserver) {
+  // Verify that FdObserver correctly detects availability of out-of-band data.
+  // Availability of out-of-band data is implementation-specific.
+	// Linux's TCP/IP stack supports out-of-band messages for TCP sockets, which is used for this
+  // test. Other platforms appear to lack generally available interfaces that could be used for
+  // generating a POLLPRI event, so we disable the test for all platforms but Linux.
+
+  UnixEventPort port;
+  EventLoop loop(port);
+  WaitScope waitScope(loop);
+  int tmpFd;
+
+  // Spawn a TCP server on localhost:44444
+  KJ_SYSCALL(tmpFd = socket(AF_INET, SOCK_STREAM, 0));
+  kj::AutoCloseFd serverFd(tmpFd);
+  sockaddr_in saddr = {0};
+  saddr.sin_family = AF_INET;
+  saddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+  saddr.sin_port = htons(44444);
+  KJ_SYSCALL(bind(serverFd, reinterpret_cast<sockaddr*>(&saddr), sizeof(saddr)));
+  KJ_SYSCALL(listen(serverFd, 1));
+
+  // Accept one connection, send one OOB message, wait for a response
+  Thread thread([&]() {
+    sockaddr_in caddr;
+    socklen_t caddrLen = sizeof(caddr);
+    int tmpFd;
+    KJ_SYSCALL(tmpFd = accept(serverFd, reinterpret_cast<sockaddr*>(&caddr), &caddrLen));
+    kj::AutoCloseFd clientFd(tmpFd);
+    delay();
+    char c = 'x';
+    KJ_SYSCALL(send(clientFd, &c, 1, MSG_OOB));
+    KJ_SYSCALL(recv(clientFd, &c, 1, 0));
+  });
+
+  KJ_SYSCALL(tmpFd = socket(AF_INET, SOCK_STREAM, 0));
+  kj::AutoCloseFd clientFd(tmpFd);
+  UnixEventPort::FdObserver observer(port, clientFd, UnixEventPort::FdObserver::OBSERVE_URGENT);
+  KJ_SYSCALL(connect(clientFd, reinterpret_cast<sockaddr*>(&saddr), sizeof(saddr)));
+
+  observer.whenUrgentDataAvailable().wait(waitScope);
+
+  // Dummy message for unblocking the server thread
+  char c = 'x';
+  KJ_SYSCALL(send(clientFd, &c, 1, 0));
+}
+#endif
+
 TEST(AsyncUnixTest, SteadyTimers) {
   captureSignals();
   UnixEventPort port;