1 files changed, 118 insertions, 0 deletions
diff --git a/src/ufo-roof-read-socket.c b/src/ufo-roof-read-socket.c
new file mode 100644
index 0000000..b72e9d0
--- /dev/null
+++ b/src/ufo-roof-read-socket.c
@@ -0,0 +1,118 @@
+#define _GNU_SOURCE
+
+#include <stdio.h>
+#include <unistd.h>
+#include <errno.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+
+#include "glib.h"
+
+#include "ufo-roof.h"
+#include "ufo-roof-read-socket.h"
+
+typedef struct {
+    UfoRoofReadInterface iface;
+
+    UfoRoofConfig *cfg;
+    int socket;
+} UfoRoofReadSocket;
+
+static void ufo_roof_read_socket_free(UfoRoofReadInterface *iface) {
+    UfoRoofReadSocket *reader = (UfoRoofReadSocket*)iface;
+    
+    if (reader) {
+        if (reader->socket >= 0) 
+            close(reader->socket);
+        free(reader);
+    }
+}
+
+static guint ufo_roof_read_socket(UfoRoofReadInterface *iface, void *buf, GError **error) {
+    struct timespec timeout_ts;
+
+    UfoRoofReadSocket *reader = (UfoRoofReadSocket*)iface;
+    UfoRoofConfig *cfg = reader->cfg;
+
+    struct mmsghdr msg[cfg->max_packets];
+    struct iovec msgvec[cfg->max_packets];
+
+    timeout_ts.tv_sec = cfg->network_timeout / 1000000;
+    timeout_ts.tv_nsec = 1000 * (cfg->network_timeout % 1000000);
+
+        // FIXME: Is it optimal? Auto-tune max_packets? Combine read & build?
+    memset(msg, 0, sizeof(msg));
+    memset(msgvec, 0, sizeof(msgvec));
+    for (int i = 0; i < cfg->max_packets; i++) {
+        msgvec[i].iov_base = buf + i * cfg->max_packet_size;
+        msgvec[i].iov_len = cfg->max_packet_size;
+        msg[i].msg_hdr.msg_iov = &msgvec[i];
+        msg[i].msg_hdr.msg_iovlen = 1;
+    }
+
+    int packets = recvmmsg(reader->socket, msg, reader->cfg->max_packets, MSG_WAITFORONE, &timeout_ts);
+    if (packets < 0) roof_network_error_with_retval(error, 0, "recvmmsg failed, error %i", errno);
+
+        // FIXME: Shall we verify packets consistency here? We can check at least the sizes...
+
+    return (guint)packets;
+}
+
+
+UfoRoofReadInterface *ufo_roof_read_socket_new(UfoRoofConfig *cfg, guint id, GError **error) {
+    int err;
+    int port = cfg->port + id;
+    char port_str[16];
+    const char *addr_str = "0.0.0.0";
+    struct addrinfo sockaddr_hints;
+    struct addrinfo *sockaddr_info;
+
+    UfoRoofReadSocket *reader = (UfoRoofReadSocket*)calloc(1, sizeof(UfoRoofReadSocket));
+    if (!reader) roof_new_error(error, "Can't allocate UfoRoofReadSocket");
+
+    reader->cfg = cfg;
+    reader->iface.close = ufo_roof_read_socket_free;
+    reader->iface.read =ufo_roof_read_socket;
+    
+    snprintf(port_str, sizeof(port_str), "%d", port);
+    port_str[sizeof(port_str) / sizeof(port_str[0]) - 1] = '\0';
+
+    memset(&sockaddr_hints, 0, sizeof(sockaddr_hints));
+    if (!strncmp(cfg->protocol, "udp", 3)) {
+        sockaddr_hints.ai_family = AF_UNSPEC;
+        sockaddr_hints.ai_socktype = SOCK_DGRAM;
+        sockaddr_hints.ai_protocol = IPPROTO_UDP;
+    } else if (!strncmp(cfg->protocol, "tcp", 3)) {
+        sockaddr_hints.ai_family = AF_UNSPEC;
+        sockaddr_hints.ai_socktype = SOCK_STREAM;
+        sockaddr_hints.ai_protocol = IPPROTO_TCP;
+    } else {
+        roof_new_error(error, "Unsupported protocol (%s)", cfg->protocol);
+    }
+
+    err = getaddrinfo(addr_str, port_str, &sockaddr_hints, &sockaddr_info);
+    if (err || !sockaddr_info) {
+        free(reader);
+        roof_new_error(error, "Invalid address (%s) or port (%s)", addr_str, port_str);
+    }
+    
+    reader->socket = socket(sockaddr_info->ai_family, sockaddr_info->ai_socktype | SOCK_CLOEXEC, sockaddr_info->ai_protocol);
+    if(reader->socket == -1) {
+        freeaddrinfo(sockaddr_info);
+        free(reader);
+        roof_new_error(error, "Can't create socket (%s) for address (%s) on port (%s)", cfg->protocol, addr_str, port_str);
+    }
+
+    err = bind(reader->socket, sockaddr_info->ai_addr, sockaddr_info->ai_addrlen);
+    if(err != 0) {
+        freeaddrinfo(sockaddr_info);
+        close(reader->socket);
+        free(reader);
+        roof_new_error(error, "Error (%i) binding socket (%s) for address (%s) on port (%s)", err, cfg->protocol, addr_str, port_str);
+    }
+
+    freeaddrinfo(sockaddr_info);
+
+    return (UfoRoofReadInterface*)reader;
+}