subscriber.cpp

#include <arpa/inet.h>
#include <ctype.h>
#include <errno.h>
#include <netdb.h>
#include <netinet/in.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/poll.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/select.h>
#include <vector>
#include <string>

#include "common.h"
#include "helpers.h"

#define MAX_CONNECTIONS 5

void print_info(chat_packet packet) {
    if (packet.data_type == 0) {
        printf("%s - INT - %d\n", packet.topic, *((int *)packet.payload));
    } else if (packet.data_type == 1) {
        printf("%s - SHORT_REAL - %.2f\n", packet.topic, *((float *)packet.payload));
    } else if (packet.data_type == 2) {
        printf("%s - FLOAT - %f\n", packet.topic, *((double *)packet.payload));
    } else if (packet.data_type == 3) {
        printf("%s - STRING - %s\n", packet.topic, packet.payload);
    }
}

void run_client(int sockfd, char *id_client) {
    char buf[1551];

    chat_packet recv_packet;
    chat_packet send_packet;
    memset(&recv_packet, 0, sizeof(chat_packet));
    memset(&send_packet, 0, sizeof(chat_packet));

    struct pollfd poll_fds[2];
    poll_fds[0].fd = STDIN_FILENO;
    poll_fds[0].events = POLLIN;
    poll_fds[1].fd = sockfd;
    poll_fds[1].events = POLLIN;

    while (1) {
        int rc = poll(poll_fds, 2, -1);
        DIE(rc < 0, "poll");

        if (poll_fds[0].revents & POLLIN) {
            // receive a keyboard input from user
            memset(buf, 0, 1551);
            fgets(buf, sizeof(buf), stdin);
            buf[strcspn(buf, "\n")] = 0;

            if (strncmp(buf, "subscribe ", 10) == 0) {
                // build a packet with the subscribe message
                char *topic = buf + strlen("subscribe ");
                strncpy(send_packet.topic, topic, 50);
                memcpy(send_packet.payload, "subscribe", 9);

                protocol send_protocol;
                memset(&send_protocol, 0, sizeof(protocol));
                send_protocol.length = sizeof(chat_packet);
                memcpy(send_protocol.payload, &send_packet, send_protocol.length);

                rc = send_all(sockfd, &send_protocol, send_protocol.length + sizeof(int));
                DIE(rc < 0, "send");
                printf("Subscribed to topic %s\n", topic);
            } else if (strncmp(buf, "unsubscribe ", 12) == 0) {
                // build a packet with the unsubscribe message
                char *topic = buf + strlen("unsubscribe ");
                strncpy(send_packet.topic, topic, 50);
                memcpy(send_packet.payload, "unsubscribe", 11);

                protocol send_protocol;
                memset(&send_protocol, 0, sizeof(protocol));
                send_protocol.length = sizeof(chat_packet);
                memcpy(send_protocol.payload, &send_packet, send_protocol.length);

                rc = send_all(sockfd, &send_protocol, send_protocol.length + sizeof(int));
                DIE(rc < 0, "send");
                printf("Unsubscribed from topic %s\n", topic);
            } else if (strncmp(buf, "exit", 4) == 0) {
                // build a packet with the exit message
                strncpy(send_packet.topic, id_client, 50);
                memcpy(send_packet.payload, "exit", 4);

                protocol send_protocol;
                memset(&send_protocol, 0, sizeof(protocol));
                send_protocol.length = sizeof(chat_packet);
                memcpy(send_protocol.payload, &send_packet, send_protocol.length);

                rc = send_all(sockfd, &send_protocol, send_protocol.length + sizeof(int));
                DIE(rc < 0, "send");

                shutdown(sockfd, SHUT_RDWR);
                close(sockfd);
                return;
            }
        }

        if (poll_fds[1].revents & POLLIN) {
            // receive a packet from the server
            protocol recv_protocol;
            memset(&recv_protocol, 0, sizeof(protocol));
            
            rc = recv_all(sockfd, &recv_protocol.length, sizeof(int));
            DIE(rc < 0, "recv");
            rc = recv_all(sockfd, recv_protocol.payload, recv_protocol.length);
            DIE(rc < 0, "recv");
            memcpy(&recv_packet, recv_protocol.payload, recv_protocol.length);

            // check if the server sent an exit message
            if (rc == 0) {
                shutdown(sockfd, SHUT_RDWR);
                close(sockfd);
                return;
            }

            // print its contents
            print_info(recv_packet);
        }
    }
}

int main(int argc, char *argv[]) {
    setvbuf(stdout, NULL, _IONBF, BUFSIZ);
    DIE(argc != 4, "Usage: ./subscriber <ID_CLIENT> <IP_SERVER> <PORT_SERVER>");

    char *id_client = argv[1];
    char *ip_server = argv[2];

    // parse server port as a number
    uint16_t port_server;
    memset(&port_server, 0, sizeof(port_server));
    int rc = sscanf(argv[3], "%hu", &port_server);
    DIE(rc != 1, "Given server_port is invalid");

    // obtain a file descriptor for the socket
    const int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    DIE(sockfd < 0, "socket");

    // complete the server address structure
    struct sockaddr_in serv_addr;
    socklen_t socket_len = sizeof(struct sockaddr_in);
    memset(&serv_addr, 0, socket_len);
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(port_server);
    rc = inet_pton(AF_INET, ip_server, &serv_addr.sin_addr.s_addr);
    DIE(rc <= 0, "inet_pton");

    // connect to server
    DIE(connect(sockfd, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0, "connect");

    // send id_client to server
    protocol send_id;
    memset(&send_id, 0, sizeof(send_id));
    send_id.length = strlen(id_client);
    memcpy(send_id.payload, id_client, send_id.length);
    rc = send_all(sockfd, &send_id, send_id.length + sizeof(int));
    DIE(rc < 0, "send");

    run_client(sockfd, id_client);

    // close the socket
    close(sockfd);

    return 0;
}