Add input pipeline handling to streamer

5 files changed, 257 insertions, 21 deletions

diff --git a/input.c b/input.c
new file mode 100644
index 0000000..6add38d
--- /dev/null
+++ b/input.c
@@ -0,0 +1,143 @@
+/*
+ * Copyright (C) 2023 Mikhail Burakov. This file is part of streamer.
+ *
+ * streamer is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * streamer is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with streamer.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "input.h"
+
+#include <errno.h>
+#include <fcntl.h>
+#include <linux/uhid.h>
+#include <stddef.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include "toolbox/buffer.h"
+#include "toolbox/utils.h"
+
+struct InputHandler {
+  struct Buffer buffer;
+  int uhid_fd;
+};
+
+struct InputHandler* InputHandlerCreate(void) {
+  struct InputHandler* input_handler = malloc(sizeof(struct InputHandler));
+  if (!input_handler) {
+    LOG("Failed to allocate input handler (%s)", strerror(errno));
+    return NULL;
+  }
+  *input_handler = (struct InputHandler){
+      .uhid_fd = -1,
+  };
+
+  BufferCreate(&input_handler->buffer);
+  input_handler->uhid_fd = open("/dev/uhid", O_RDWR);
+  if (input_handler->uhid_fd == -1) {
+    LOG("Failed to open uhid device (%s)", strerror(errno));
+    goto rollback_input_handler;
+  }
+  return input_handler;
+
+rollback_input_handler:
+  free(input_handler);
+  return NULL;
+}
+
+int InputHandlerGetEventsFd(struct InputHandler* input_handler) {
+  return input_handler->uhid_fd;
+}
+
+bool InputHandlerProcessEvents(struct InputHandler* input_handler) {
+  struct uhid_event uhid_event;
+  if (read(input_handler->uhid_fd, &uhid_event, sizeof(uhid_event)) < 0) {
+    LOG("Failed to process uhid events (%s)", strerror(errno));
+    return false;
+  }
+  // TODO(mburakov): Add logging?
+  return true;
+}
+
+bool InputHandlerHandle(struct InputHandler* input_handler, int fd) {
+  switch (BufferAppendFrom(&input_handler->buffer, fd)) {
+    case -1:
+      LOG("Failed to append input data to buffer (%s)", strerror(errno));
+      return false;
+    case 0:
+      LOG("Client closed connection");
+      return false;
+    default:
+      break;
+  }
+
+  for (;;) {
+    struct uhid_event* event = input_handler->buffer.data;
+    if (input_handler->buffer.size < sizeof(event->type)) {
+      // mburakov: Packet type is not yet available.
+      return true;
+    }
+
+    size_t size;
+    switch (event->type) {
+      case UHID_CREATE2:
+        if (input_handler->buffer.size <
+            offsetof(struct uhid_event, u.create2.rd_size) +
+                sizeof(event->u.create2.rd_size)) {
+          // mburakov: Report descriptor size is not yet available.
+          return true;
+        }
+        size = offsetof(struct uhid_event, u.create2.rd_data) +
+               event->u.create2.rd_size;
+        if (input_handler->buffer.size < size) {
+          // mburakov: Report descriptor data is not yet available.
+          return true;
+        }
+        break;
+      case UHID_INPUT2:
+        if (input_handler->buffer.size <
+            offsetof(struct uhid_event, u.input2.size) +
+                sizeof(event->u.input2.size)) {
+          // mburakov: Report size is not yet available.
+          return true;
+        }
+        size =
+            offsetof(struct uhid_event, u.input2.data) + event->u.input2.size;
+        if (input_handler->buffer.size < size) {
+          // mburakov: Report data is not yet available.
+          return true;
+        }
+        break;
+      case UHID_DESTROY:
+        size = sizeof(event->type);
+        break;
+      default:
+        __builtin_unreachable();
+    }
+
+    // mburakov: This write has to be atomic.
+    if (write(input_handler->uhid_fd, event, size) != (ssize_t)size) {
+      LOG("Failed to write uhid event (%s)", strerror(errno));
+      return false;
+    }
+    BufferDiscard(&input_handler->buffer, size);
+  }
+}
+
+void InputHandlerDestroy(struct InputHandler* input_handler) {
+  close(input_handler->uhid_fd);
+  BufferDestroy(&input_handler->buffer);
+  free(input_handler);
+}
diff --git a/input.h b/input.h
new file mode 100644
index 0000000..1de40a6
--- /dev/null
+++ b/input.h
@@ -0,0 +1,31 @@
+/*
+ * Copyright (C) 2023 Mikhail Burakov. This file is part of streamer.
+ *
+ * streamer is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * streamer is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with streamer.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#ifndef STREAMER_INPUT_H_
+#define STREAMER_INPUT_H_
+
+#include <stdbool.h>
+
+struct InputHandler;
+
+struct InputHandler* InputHandlerCreate(void);
+int InputHandlerGetEventsFd(struct InputHandler* input_handler);
+bool InputHandlerProcessEvents(struct InputHandler* input_handler);
+bool InputHandlerHandle(struct InputHandler* input_handler, int fd);
+void InputHandlerDestroy(struct InputHandler* input_handler);
+
+#endif  // STREAMER_INPUT_H_
diff --git a/main.c b/main.c
index 6b32209..121085a 100644
--- a/main.c
+++ b/main.c
@@ -30,6 +30,7 @@
 #include "colorspace.h"
 #include "encode.h"
 #include "gpu.h"
+#include "input.h"
 #include "toolbox/io_muxer.h"
 #include "toolbox/utils.h"
 
@@ -50,8 +51,10 @@ struct Contexts {
   int server_fd;
   struct GpuContext* gpu_context;
   struct CaptureContext* capture_context;
-  struct EncodeContext* encode_context;
+
   int client_fd;
+  struct InputHandler* input_handler;
+  struct EncodeContext* encode_context;
 };
 
 static int CreateServerSocket(const char* arg) {
@@ -88,6 +91,29 @@ rollback_sock:
   return -1;
 }
 
+static void MaybeDropClient(struct Contexts* contexts) {
+  static const struct itimerspec spec = {0};
+  if (contexts->encode_context) {
+    EncodeContextDestroy(contexts->encode_context);
+    contexts->encode_context = NULL;
+  }
+  if (contexts->input_handler) {
+    IoMuxerForget(&contexts->io_muxer,
+                  InputHandlerGetEventsFd(contexts->input_handler));
+    InputHandlerDestroy(contexts->input_handler);
+    contexts->input_handler = NULL;
+  }
+  if (contexts->client_fd != -1) {
+    IoMuxerForget(&contexts->io_muxer, contexts->client_fd);
+    close(contexts->client_fd);
+    contexts->client_fd = -1;
+  }
+  if (timerfd_settime(contexts->timer_fd, 0, &spec, NULL)) {
+    LOG("Failed to disarm timer (%s)", strerror(errno));
+    g_signal = SIGABRT;
+  }
+}
+
 static void OnTimerExpire(void* user) {
   struct Contexts* contexts = user;
   if (!IoMuxerOnRead(&contexts->io_muxer, contexts->timer_fd, &OnTimerExpire,
@@ -146,20 +172,43 @@ static void OnTimerExpire(void* user) {
   }
   return;
 
-drop_client:;
-  static const struct itimerspec spec = {0};
-  if (timerfd_settime(contexts->timer_fd, 0, &spec, NULL)) {
-    LOG("Failed to disarm timer (%s)", strerror(errno));
-    g_signal = SIGABRT;
+drop_client:
+  MaybeDropClient(contexts);
+}
+
+static void OnClientWriting(void* user) {
+  struct Contexts* contexts = user;
+  if (!IoMuxerOnRead(&contexts->io_muxer, contexts->client_fd, &OnClientWriting,
+                     user)) {
+    LOG("Failed to reschedule client reading (%s)", strerror(errno));
+    goto drop_client;
   }
-  if (contexts->encode_context) {
-    EncodeContextDestroy(contexts->encode_context);
-    contexts->encode_context = NULL;
+  if (!InputHandlerHandle(contexts->input_handler, contexts->client_fd)) {
+    LOG("Failed to handle client input");
+    goto drop_client;
   }
-  if (contexts->client_fd != -1) {
-    close(contexts->client_fd);
-    contexts->client_fd = -1;
+  return;
+
+drop_client:
+  MaybeDropClient(contexts);
+}
+
+static void OnInputEvents(void* user) {
+  struct Contexts* contexts = user;
+  if (!IoMuxerOnRead(&contexts->io_muxer,
+                     InputHandlerGetEventsFd(contexts->input_handler),
+                     &OnInputEvents, user)) {
+    LOG("Failed to reschedule events reading (%s)", strerror(errno));
+    goto drop_client;
   }
+  if (!InputHandlerProcessEvents(contexts->input_handler)) {
+    LOG("Failed to process events");
+    goto drop_client;
+  }
+  return;
+
+drop_client:
+  MaybeDropClient(contexts);
 }
 
 static void OnClientConnecting(void* user) {
@@ -184,20 +233,34 @@ static void OnClientConnecting(void* user) {
   }
 
   contexts->client_fd = client_fd;
+  if (!IoMuxerOnRead(&contexts->io_muxer, contexts->client_fd, &OnClientWriting,
+                     user)) {
+    LOG("Failed to schedule client reading (%s)", strerror(errno));
+    goto drop_client;
+  }
+  contexts->input_handler = InputHandlerCreate();
+  if (!contexts->input_handler) {
+    LOG("Failed to create input handler");
+    goto drop_client;
+  }
+  if (!IoMuxerOnRead(&contexts->io_muxer,
+                     InputHandlerGetEventsFd(contexts->input_handler),
+                     &OnInputEvents, user)) {
+    LOG("Failed to schedule events reading (%s)", strerror(errno));
+    goto drop_client;
+  }
   static const struct itimerspec spec = {
       .it_interval.tv_nsec = capture_period,
       .it_value.tv_nsec = capture_period,
   };
   if (timerfd_settime(contexts->timer_fd, 0, &spec, NULL)) {
     LOG("Failed to arm timer (%s)", strerror(errno));
-    goto rollback_client_fd;
+    goto drop_client;
   }
   return;
 
-rollback_client_fd:
-  close(contexts->client_fd);
-  contexts->client_fd = -1;
-  return;
+drop_client:
+  MaybeDropClient(contexts);
 }
 
 int main(int argc, char* argv[]) {
@@ -255,9 +318,7 @@ int main(int argc, char* argv[]) {
       g_signal = SIGABRT;
     }
   }
-
-  if (contexts.encode_context) EncodeContextDestroy(contexts.encode_context);
-  if (contexts.client_fd != -1) close(contexts.client_fd);
+  MaybeDropClient(&contexts);
 
 rollback_capture_context:
   CaptureContextDestroy(contexts.capture_context);
diff --git a/makefile b/makefile
index d1501ce..eb8fa47 100644
--- a/makefile
+++ b/makefile
@@ -1,8 +1,9 @@
 bin:=$(notdir $(shell pwd))
-src:=$(shell ls *.c)
+src:=$(wildcard *.c)
 obj:=$(src:.c=.o)
 
 obj+=\
+	toolbox/buffer.o \
 	toolbox/io_muxer.o
 
 libs:=\
diff --git a/toolbox b/toolbox
-Subproject 809618988f918d0a54cc67bf416e455517cfc8c
+Subproject c5ca05dd86910984cb3ab7e3a17718f7be6ba27