// Copyright (2019-2022) Paul Scherrer Institute
// SPDX-License-Identifier: GPL-3.0-or-later

#include "UdpReceiver.h"

#include <sys/socket.h>
#include <arpa/inet.h>

#include "../common/JFJochException.h"

UdpReceiver::UdpReceiver(uint16_t udp_port_number,
                         ThreadSafeFIFO<Completion> &completion_queue,
                         ThreadSafeFIFO<ProcessWorkRequest> &wr_queue,
                         uint32_t nmodules,
                         volatile bool &in_cancel,
                         uint32_t nthreads) :
        process(completion_queue, wr_queue, nmodules),
        cancel(in_cancel) {
    for (int i = 0; i < nthreads; i++)
        futures.emplace_back(std::async(std::launch::async, &UdpReceiver::Run, this, udp_port_number));
}

UdpReceiver::~UdpReceiver() {
    cancel = true;
    for (auto &iter: futures) {
        if (iter.valid())
            iter.get();
    }
}

void UdpReceiver::Run(uint16_t udp_port_number) {
    int fd = socket(AF_INET, SOCK_DGRAM, 0);
    if (fd < 0)
        throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot create UDP socket");

    timeval timeout{
        .tv_sec = 0,
        .tv_usec = 1000
    };

    if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) <= 0)
        throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot set socket timeout");

    sockaddr_in server_addr {
        .sin_family = AF_INET,
        .sin_port = htons(udp_port_number),
        .sin_addr = { .s_addr = INADDR_ANY }
    }, client_addr{};

    if (bind(fd, (struct sockaddr *) &server_addr, sizeof(server_addr)) <= 0)
        throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot bind to UDP port");

    char buffer[9000];
    socklen_t src_addr_len=sizeof(client_addr);

    while (!cancel) {
        auto count = recvfrom(fd, buffer, sizeof(buffer), 0, (struct sockaddr *) &client_addr, &src_addr_len);

        if (count == sizeof(jf_udp_payload))
            process.ProcessPacket((jf_udp_payload *) buffer, client_addr.sin_addr.s_addr);
        else if ((errno != EAGAIN) && (errno != EWOULDBLOCK) && (count == -1))
                throw JFJochException(JFJochExceptionCategory::UDPError, "Cannot bind to UDP port");
    }

    close(fd);
}