subscriber.cpp

#include "utils.h"

std::pair<std::vector<char*>, int> parse_input(char *buffer) {
    std::vector<char*> result;
    int length = 0;
    char *token = strtok(buffer, " ");
    while (token != NULL) {
        result.push_back(token);
        length++;
        token = strtok(NULL, " ");
    }
    return {result, length};
}

struct tcp_packet_to_server create_request(command_type type, char *topic, char *id) {
    struct tcp_packet_to_server request;
    request.type = type;
    if (topic != NULL) {
        strcpy(request.topic, topic);
    }
    strcpy(request.id, id);
    return request;
}

void format_message(struct tcp_packet_to_client *message) {
    char ip[INET_ADDRSTRLEN];
    inet_ntop(AF_INET, &message->ip_udp, ip, INET_ADDRSTRLEN);
    uint16_t port = ntohs(message->port_udp);

    char data_type[15];
    char payload[MAX_SIZE_PAYLOAD];
    switch (message->packet.type_of_data) {
        case 0: {
                uint8_t sign_bit = message->packet.data[0];
                uint32_t tmp_value;
                memcpy(&tmp_value, message->packet.data + 1, sizeof(uint32_t));
                int32_t value = ntohl(tmp_value);
                if (sign_bit == 1) {
                    value = -1 * value;
                }
                strcpy(data_type, "INT");
                snprintf(payload, sizeof(payload), "%d", value);
                break;
            }
        case 1: {
                uint16_t tmp_value2;
                memcpy(&tmp_value2, message->packet.data, sizeof(uint16_t));
                double short_value = ntohs(tmp_value2) / 100.0;
                strcpy(data_type, "SHORT_REAL");
                snprintf(payload, sizeof(payload), "%.2f", short_value);
                break;
            }
        case 2: {
                uint8_t float_sign_bit = message->packet.data[0];
                uint32_t float_tmp_value;
                memcpy(&float_tmp_value, message->packet.data + 1, sizeof(uint32_t));
                uint8_t power = message->packet.data[5];
                float float_value = ntohl(float_tmp_value);
                float_value = float_value / (float) pow(10, power);
                if (float_sign_bit == 1) {
                    float_value = -1 * float_value;
                }
                strcpy(data_type, "FLOAT");
                snprintf(payload, sizeof(payload), "%.*f", power, float_value);
                break;
            }
        case 3: {
                strcpy(data_type, "STRING");
                snprintf(payload, sizeof(payload), "%s", message->packet.data);
                break;
            }
        default:
            WRITE_ERROR("Unknown type\n");
            return;
    }
    printf("%s:%hu - %s - %s - %s\n",
            ip, port, message->packet.topic,
            data_type, payload);
}

void run_client(int client_fd, char *id) {
    std::vector<struct pollfd> fds(2, {0, 0, 0});

    fds[0].fd = client_fd;
    fds[0].events = POLLIN;

    fds[1].fd = STDIN_FILENO;
    fds[1].events = POLLIN;

    while (1) {
        int ready = poll(fds.data(), fds.size(), 2);
        if (ready < 0) {
            perror("Error at poll");
            close(client_fd);
            exit(1);
        }

        if (fds[0].revents & POLLIN) {
            char buffer[MAX_SIZE_TOPIC];

            size_t length;
            int bytes = recv_all(client_fd, &length, sizeof(size_t));
            if (bytes < 0) {
                perror("Error ar recv");
            }
            if (bytes == 0) {
                close(client_fd);
                exit(1);
            }
            int received_bytes = recv_all(client_fd, buffer, length);
            if (received_bytes < 0) {
                perror("Error at recv");
                close(client_fd);
                exit(1);
            }
            struct tcp_packet_to_client result = *(struct tcp_packet_to_client *)buffer;
            switch (result.type) {
                case EXIT_CODE:
                    close(client_fd);
                    exit(1);
                case TOPIC:
                    format_message(&result);
                    break;
                default:
                    WRITE_ERROR("Unknown commnand\n");
                    break;
            }
        }

        if (fds[1].revents & POLLIN) {
            char buffer[MAX_SIZE_FROM_READ];
            int bytes = read(STDIN_FILENO, buffer, sizeof(buffer));
            if (bytes <= 0) {
                perror("Error at read");
                close(client_fd);
                exit(1);
            }
            buffer[strcspn(buffer, "\n")] = '\0';
            if (strcmp(buffer, "exit") == 0) {
                struct tcp_packet_to_server request = create_request(EXIT, NULL, id);
                send_all(client_fd, (void *)&request, sizeof(request));
                close(client_fd);
                exit(1);
            }

            auto [words, length] = parse_input(buffer);
            if (strcmp(words[0], "subscribe") == 0) {
                if (length == 2) {
                    struct tcp_packet_to_server request = create_request(SUBSCRIBE, words[1], id);
                    send_all(client_fd, (void *)&request, sizeof(request));
                    printf("Subscribed to topic %s\n", words[1]);
                } else {
                    WRITE_ERROR("This command isn't supported\n");
                }
            } else if (strcmp(words[0], "unsubscribe") == 0) {
                if (length == 2) {
                    struct tcp_packet_to_server request = create_request(UNSUBSCRIBE, words[1], id);
                    send_all(client_fd, (void *)&request, sizeof(request));
                    printf("Unsubscribed from topic %s\n", words[1]);
                } else {
                    WRITE_ERROR("This command isn't supported\n");
                }
            } else {
                WRITE_ERROR("This command isn't supported\n");
            }
        }
    }
}

int main(int argc, char *argv[]) {
    if (argc != 4) {
        WRITE_ERROR("Usage: ./subscriber <ID_CLIENT> <IP_SERVER> <PORT>: Invalid number of arguments\n");
        return 1;
    }
    setvbuf(stdout, NULL, _IONBF, BUFSIZ);

    int client_socket = socket(AF_INET, SOCK_STREAM, 0);
    if (client_socket < 0) {
        perror("Error at socket system call");
        return 1;
    }

    struct sockaddr_in address;
    address.sin_family = AF_INET;
    address.sin_port = htons(atoi(argv[3]));
    if (inet_pton(AF_INET, argv[2], &address.sin_addr) < 0) {
        perror("Error at inet_pton");
        close(client_socket);
        return 1;
    }

    if (connect(client_socket, (struct sockaddr *)&address, sizeof(address)) < 0) {
        perror("Error at connect");
        close(client_socket);
        return 1;
    }

    struct tcp_packet_to_server request;
    strcpy(request.id, argv[1]);
    send_all(client_socket, &request, sizeof(request));

    run_client(client_socket, argv[1]);
    close(client_socket);
    return 0;
}