tcpStream.cpp

Go to the documentation of this file.

#include <arpa/inet.h>
#include <cerrno>
#include <cstdint>
#include <cstdio>
#include <fcntl.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
 
#include "osefTypes.h"
#include "tlvCommon.h"
 
#include "tcpStream.h"
 
const char *TcpStreamReader::default_ip = "192.168.2.2";
const uint16_t TcpStreamReader::default_port = 11120;
 
// Open socket with relevant settings for our usage
void open_socket(int &socket_fd)
{
        int fd, fd_flags;
 
        fd = socket(AF_INET, SOCK_STREAM, 0);
        if (fd < 0) {
                printf("Error: cannot create socket\n");
                return;
        }
 
        // Lose the pesky "Address already in use" error message
        int yes = 1;
        if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0)
                printf("Error: failed allowing address reuse\n");
 
        socket_fd = fd;
}
 
// Close socket
void close_socket(int &socket_fd)
{
        if (socket_fd < 0)
                return;
 
        while (close(socket_fd))
                continue;
 
        socket_fd = -1;
}
 
// Read specified bytes from socket and discard them
void emptySocket(int socket_fd, int bytes_remaining)
{
        constexpr int size_temp = 1000;
        uint8_t temp[size_temp];
        while (bytes_remaining > 0) {
                unsigned read_size = (bytes_remaining >= size_temp) ? size_temp : bytes_remaining;
                int n = read(socket_fd, temp, read_size);
                if (n >= 0)
                        bytes_remaining -= n;
        }
}
 
// To be repeatedly called to read socket in order to get a frame in the
// specified buffer
int readSocket(int socket_fd, uint8_t *buffer, const size_t buffer_size, int &frame_write_index)
{
        // frame_write_index indicates the number of bytes already read
        uint8_t *dest = buffer + frame_write_index;
 
        // If we haven't finished to read the tlv header yet, attempt to read it, else
        // continue to read the tlv value
        Tlv::tlv_s *tlv = (Tlv::tlv_s *)buffer;
        int bytes_to_read =
                (frame_write_index < (int)sizeof(Tlv::header_s)) ? sizeof(Tlv::header_s) : Tlv::getSize(*tlv);
        int n = read(socket_fd, dest, bytes_to_read - frame_write_index);
 
        // Handle special return codes
        if (n == -1 && (errno == EAGAIN || errno == ECONNREFUSED)) {
                return 2;
 
        } else if (n == 0) {
                return 0;
 
        } else if (n == -1) {
                printf("Error in reading socket (errno = %d)\n", errno);
                return -1;
        }
 
        // If no error occurred, increment the write index by the number of bytes read
        frame_write_index += n;
 
        // Check type to mitigate retro-compatibility issues (parsed type will be '4'
        // if server uses former format, which is non-osef)
        if (frame_write_index == sizeof(Tlv::header_s) && tlv->getType() != OSEF_TYPE_TIMESTAMPED_DATA) {
                printf("Receive malformated frame: tlv type (%d) is wrong\n", tlv->getType());
                return -1;
        }
 
        // If the frame is oversized for the buffer, ignore it and delete
        // corresponding data in incoming socket
        if (frame_write_index == sizeof(Tlv::header_s) && Tlv::getSize(*tlv) > buffer_size) {
                printf("Incoming frame too large : %lu (max allocated size %lu)\n", Tlv::getSize(*tlv), buffer_size);
                emptySocket(socket_fd, (int)(Tlv::getSize(*tlv) - sizeof(Tlv::header_s)));
                frame_write_index = 0;
                return 2;
        }
 
        // The frame is not complete yet, come back soon!
        if (frame_write_index < (int)sizeof(Tlv::header_s) || frame_write_index < (int)Tlv::getSize(*tlv))
                return 2;
 
        // The frame is complete
        return 1;
}
 
// Connect to the stream on the specified address and port
// returns:
//  - 1 on sucess
//  - 0 on non-fatal failure (ALB unavailable, processing is not started yet?)
//  - -1 on fatal failure (invalid IP)
int TcpStreamReader::connectToALB(const char *ipv4, const uint16_t port)
{
        int ret;
        struct in_addr addr;
 
        if (socket_fd == -1)
                open_socket(socket_fd);
 
        // Parse IPv4 string
        ret = inet_pton(AF_INET, ipv4, (uint8_t *)&addr);
        if (ret <= 0) {
                printf("Cannot parse ALB IPv4");
                return -1;
        }
 
        // Copy result to sockaddr we want to connect
        struct sockaddr_in alb = {};
        alb.sin_family = AF_INET;
        alb.sin_addr = addr;
        alb.sin_port = htons(port);
 
        ret = connect(socket_fd, (struct sockaddr *)&alb, sizeof(alb));
 
        if (ret < 0) {
                if (errno == ECONNREFUSED)
                        return 0; // ALB unavailable (processing not started yet?)
                else
                        return -1; // Unhandled error
        }
 
        printf("Connected to %s:%d\n", ipv4, port);
        return 1;
}
 
// Poll the socket until a new frame is read
// returns:
//  - 1 on sucess (buffer now contains a valid frame)
//  - -1 on fatal failure (corrupted stream)
int TcpStreamReader::getNextFrame(uint8_t *buffer, const size_t buffer_size)
{
        int frame_write_index = 0;
        while (true) {
                int ret = readSocket(socket_fd, buffer, buffer_size, frame_write_index);
                if (ret == 2) {
                        continue;
                } else
                        return ret;
        }
}
 
// Disconnect from ALB
// returns:
// - 0 on success
// - <0 on failure (not connected or cannot close socket)
int TcpStreamReader::disconnectfromALB()
{
        int ret = 0;
 
        if (socket_fd < 0)
                return -1;
 
        ret = close(socket_fd);
        if (ret < 0)
                printf("Error closing socket\n");
 
        socket_fd = -1;
 
        return ret;
}