peer_webrtc.c

#include <neat.h>
#include "util.h"

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <sys/time.h>

/**********************************************************************

    WebRTC peer in neat
    * connect to peer and send and receive DATA

    peer_webrtc [OPTIONS] SESSION
    -n : number of requests/flows
    -R : receive buffer size in byte
    -v : log level (0 .. 2)

**********************************************************************/

#ifndef timersub
#define timersub(tvp, uvp, vvp)                                 \
    do {                                                        \
        (vvp)->tv_sec = (tvp)->tv_sec - (uvp)->tv_sec;          \
        (vvp)->tv_usec = (tvp)->tv_usec - (uvp)->tv_usec;       \
        if ((vvp)->tv_usec < 0) {                               \
            (vvp)->tv_sec--;                                    \
            (vvp)->tv_usec += 1000000;                          \
        }                                                       \
    } while (0)
#endif

static uint32_t config_rcv_buffer_size      = 10240;
static uint32_t config_snd_buffer_size      = 1024;
static uint32_t config_message_count        = 0;
static uint32_t config_runtime_max          = 0;
static uint16_t config_chargen_offset       = 0;
static uint16_t config_port                 = 23232;
static uint16_t config_log_level            = 1;
static uint16_t config_num_flows            = 1;
static uint16_t config_max_flows            = 100;
static uint16_t config_active               = 0;

static char          *config_property        = "\
{\
    \"transport\": [\
        {\
            \"value\": \"WEBRTC\",\
            \"precedence\": 1\
        }\
    ]\
}";

static uint32_t flows_active = 0;

struct tneat_flow_direction {
    unsigned char *buffer;
    uint32_t calls;
    uint32_t bytes;
    struct timeval tv_first;
    struct timeval tv_last;
};

struct tneat_flow {
    struct tneat_flow_direction rcv;
    struct tneat_flow_direction snd;
};

static neat_error_code on_close(struct neat_flow_operations *opCB);
static neat_error_code on_writable(struct neat_flow_operations *opCB);

/*
    print usage
*/
static void
print_usage()
{
    if (config_log_level >= 2) {
        fprintf(stderr, "%s()\n", __func__);
    }

    printf("tneat [OPTIONS] [HOST]\n");
    printf("\t- l \tsize for each message in byte (%d)\n", config_snd_buffer_size);
    printf("\t- n \tmax number of messages to send (%d)\n", config_message_count);
    printf("\t- p \tport [receive on|send to] (%d)\n", config_port);
    printf("\t- P \tneat properties (%s)\n", config_property);
    printf("\t- R \treceive buffer in byte (%d)\n", config_rcv_buffer_size);
    printf("\t- T \tmax runtime in seconds (%d)\n", config_runtime_max);
    printf("\t- v \tlog level 0..3 (%d)\n", config_log_level);
}

static neat_error_code
on_error(struct neat_flow_operations *opCB)
{
    fprintf(stderr, "%s\n", __func__);
    neat_stop_event_loop(opCB->ctx);
    return NEAT_OK;
}

static neat_error_code
on_parameters(struct neat_flow_operations *opCB)
{
    fprintf(stderr, "%s\n", __func__);
    printf("LocalParameters: %s\n", (char *)opCB->userData);
   /* char *remote = NULL;
    printf("Got local parameters from WebRTC. Now send them to signalling server\n");
    remote = strdup((char *)opCB->userData);
    if (neat_send_remote_parameters(opCB->ctx, opCB->flow, remote) != NEAT_OK) {
        free(remote);
    }*/
    free(opCB->userData);
    opCB->userData = NULL;
    return NEAT_OK;
}

static neat_error_code
on_readable(struct neat_flow_operations *opCB)
{
    struct tneat_flow *tnf = opCB->userData;
    uint32_t buffer_filled;
   // struct timeval diff_time;
    neat_error_code code;
  //  char buffer_filesize_human[32];
  //  double time_elapsed;
printf("peer_webrtc: on_readable\n");
    if (config_log_level >= 2) {
        fprintf(stderr, "%s()\n", __func__);
    }

    code = neat_read(opCB->ctx, opCB->flow, tnf->rcv.buffer, config_rcv_buffer_size, &buffer_filled, NULL, 0);
    if (code) {
        if (code == NEAT_ERROR_WOULD_BLOCK) {
            fprintf(stderr, "%s - neat_read warning: NEAT_ERROR_WOULD_BLOCK\n", __func__);
            return NEAT_OK;
        } else {
            fprintf(stderr, "%s - neat_read error: code %d\n", __func__, (int)code);
            return on_error(opCB);
        }
    }

    if (buffer_filled > 0) {
        // we got data!
        if (tnf->rcv.calls == 0) {
            gettimeofday(&(tnf->rcv.tv_first), NULL);
        }
        tnf->rcv.calls++;
        tnf->rcv.bytes += buffer_filled;
        gettimeofday(&(tnf->rcv.tv_last), NULL);

        if (config_log_level >= 2) {
            printf("%s: neat_read - # %u - %d byte\n", opCB->label, tnf->rcv.calls, buffer_filled);
          //  if (config_log_level >= 4) {
                fwrite(tnf->rcv.buffer, sizeof(char), buffer_filled, stdout);
                printf("\n");
           // }
        }
    // peer disconnected
    }

    return NEAT_OK;
}

static neat_error_code
on_all_written(struct neat_flow_operations *opCB)
{
    struct tneat_flow *tnf = opCB->userData;
    struct timeval now, diff_time;
    double time_elapsed;

printf("peer_webrtc: on_all_written\n");
    if (config_log_level >= 2) {
        fprintf(stderr, "%s()\n", __func__);
    }

    gettimeofday(&now, NULL);
    timersub(&(tnf->snd.tv_last), &(tnf->snd.tv_first), &diff_time);
    time_elapsed = diff_time.tv_sec + (double)diff_time.tv_usec/1000000.0;
printf("sndCalls=%d message_count=%d time=%f config_runtime_max=%d\n", tnf->snd.calls, config_message_count, time_elapsed, config_runtime_max);
    // runtime- or message-limit reached
    if ((config_runtime_max > 0 && time_elapsed >= config_runtime_max) ||
        (config_message_count > 0 && tnf->snd.calls >= config_message_count)) {
        neat_close(opCB->ctx, opCB->flow);
    } else {
        opCB->on_writable = on_writable;
        opCB->on_all_written = NULL;
        neat_set_operations(opCB->ctx, opCB->flow, opCB);
    }
    return NEAT_OK;
}

/*
    send *config_message_size* chars to peer
*/
static neat_error_code
on_writable(struct neat_flow_operations *opCB)
{
    struct tneat_flow *tnf = opCB->userData;
    neat_error_code code;
printf("peer_webrtc: on_writable\n");
    if (config_log_level >= 2) {
        fprintf(stderr, "%s()\n", __func__);
    }

    // record first send call
    if (tnf->snd.calls == 0) {
        gettimeofday(&(tnf->snd.tv_first), NULL);
    }

    // set callbacks
    opCB->on_writable = NULL;
    opCB->on_all_written = on_all_written;
    neat_set_operations(opCB->ctx, opCB->flow, opCB);

    // increase stats
    tnf->snd.calls++;
    tnf->snd.bytes += config_snd_buffer_size;
    gettimeofday(&(tnf->snd.tv_last), NULL);

    // every message contains a different payload (i++)
    config_chargen_offset = (config_chargen_offset+1) % 72;
    memset(tnf->snd.buffer, 33 + config_chargen_offset, config_snd_buffer_size);

    if (config_log_level >= 2) {
        printf("%s: neat_write - # %u - %d byte\n", opCB->label, tnf->snd.calls, config_snd_buffer_size);
        if (config_log_level >= 4) {
            printf("neat_write - content\n");
            fwrite(tnf->snd.buffer, sizeof(char), config_snd_buffer_size, stdout);
            printf("\n");
        }
    }

    code = neat_write(opCB->ctx, opCB->flow, tnf->snd.buffer, config_snd_buffer_size, NULL, 0);

    if (code != NEAT_OK) {
        fprintf(stderr, "%s - neat_write error: code %d\n", __func__, (int)code);
        return on_error(opCB);
    }

    return NEAT_OK;
}


/*
    Connection established - set callbacks and reset statistics
*/
static neat_error_code
on_connected(struct neat_flow_operations *opCB)
{
    struct tneat_flow *tnf = NULL;
 printf("!!!!!!!Connected!!!!!!\n");
    if (config_log_level >= 1) {
        fprintf(stderr, "%s() - connection established\n", __func__);
    }

    if ((opCB->userData = calloc(1, sizeof(struct tneat_flow))) == NULL) {
        fprintf(stderr, "%s - could not allocate tneat_flow\n", __func__);
        exit(EXIT_FAILURE);
    }

    tnf = opCB->userData;

    if ((tnf->snd.buffer = malloc(config_snd_buffer_size)) == NULL) {
        fprintf(stderr, "%s - could not allocate send buffer\n", __func__);
        exit(EXIT_FAILURE);
    }

    if ((tnf->rcv.buffer = malloc(config_rcv_buffer_size)) == NULL) {
        fprintf(stderr, "%s - could not allocate receive buffer\n", __func__);
        exit(EXIT_FAILURE);
    }

    // reset stats
    tnf->snd.calls = 0;
    tnf->snd.bytes = 0;
    tnf->rcv.calls = 0;
    tnf->rcv.bytes = 0;

    // set callbacks
    opCB->on_readable = on_readable;
    if (config_active) {
        opCB->on_writable = on_writable;
        printf("set on_writable\n");
    } else {
        opCB->on_writable = NULL;
        printf("set writable = NULL\n");
    }

    flows_active++;
    neat_set_operations(opCB->ctx, opCB->flow, opCB);

    return NEAT_OK;
}

static neat_error_code
on_close(struct neat_flow_operations *opCB)
{
    printf("on_close\n");
    struct tneat_flow *tnf = opCB->userData;
    char buffer_filesize_human[32];
    double time_elapsed;
    struct timeval diff_time;

    fprintf(stderr, "%s\n", __func__);

    if (tnf == NULL && flows_active == 0) {
        fprintf(stderr, "%s - stopping event loop\n", __func__);
        neat_stop_event_loop(opCB->ctx);
        return NEAT_OK;
    }

    timersub(&(tnf->rcv.tv_last), &(tnf->rcv.tv_first), &diff_time);
    time_elapsed = diff_time.tv_sec + (double)diff_time.tv_usec/1000000.0;

    printf("flow closed - read statistics\n");
    printf("\tbytes\t\t: %u\n", tnf->rcv.bytes);
    printf("\trcv-calls\t: %u\n", tnf->rcv.calls);
    printf("\tduration\t: %.2fs\n", time_elapsed);
    printf("\tbandwidth\t: %s/s\n", filesize_human(tnf->rcv.bytes/time_elapsed, buffer_filesize_human, sizeof(buffer_filesize_human)));

    // write print statistics
    timersub(&(tnf->snd.tv_last), &(tnf->snd.tv_first), &diff_time);
    time_elapsed = diff_time.tv_sec + (double)diff_time.tv_usec/1000000.0;

    printf("flow closed - write statistics\n");
    printf("\tbytes\t\t: %u\n", tnf->snd.bytes);
    printf("\tsnd-calls\t: %u\n", tnf->snd.calls);
    printf("\tduration\t: %.2fs\n", time_elapsed);
    printf("\tbandwidth\t: %s/s\n", filesize_human(tnf->snd.bytes/time_elapsed, buffer_filesize_human, sizeof(buffer_filesize_human)));

    // cleanup
    if (tnf->snd.buffer) {
        free(tnf->snd.buffer);
        tnf->snd.buffer = NULL;
    }

    if (tnf->rcv.buffer) {
        free(tnf->rcv.buffer);
        tnf->rcv.buffer = NULL;
    }

    if (tnf) {
        free(tnf);
        tnf = 0;
    }

    fprintf(stderr, "%s - flow closed OK!\n", __func__);

    // stop event loop if we are active part
    flows_active--;
    printf("active flows left: %d\n", flows_active);
  /*  if (config_active && !flows_active) {
        fprintf(stderr, "%s - stopping event loop\n", __func__);
        neat_stop_event_loop(opCB->ctx);
    }*/

    return NEAT_OK;
}

int
main(int argc, char *argv[])
{
    struct neat_ctx *ctx = NULL;
    int i = 0;
    struct neat_tlv options[1];
    char name[20];

    struct neat_flow *flows[config_max_flows];
    struct neat_flow_operations ops[config_max_flows];

    int arg, result;
    char *arg_property = config_property;

    memset(&ops, 0, sizeof(ops));

    result = EXIT_SUCCESS;

    while ((arg = getopt(argc, argv, "l:n:p:P:R:T:v:")) != -1) {
        switch(arg) {
        case 'l':
            config_snd_buffer_size = atoi(optarg);
            if (config_log_level >= 1) {
                printf("option - send buffer size: %d\n", config_snd_buffer_size);
            }
            break;
        case 'n':
            config_message_count = atoi(optarg);
            if (config_log_level >= 1) {
                printf("option - message limit: %d\n", config_message_count);
            }
            break;
        case 'p':
            config_port = atoi(optarg);
            if (config_log_level >= 1) {
                printf("option - port: %d\n", config_port);
            }
            break;
        case 'P':
            if (read_file(optarg, &arg_property) < 0) {
                fprintf(stderr, "Unable to read properties from %s: %s",
                        optarg, strerror(errno));
                result = EXIT_FAILURE;
                goto cleanup;
            }
            if (config_log_level >= 1) {
                printf("option - properties: %s\n", arg_property);
            }
            break;
        case 'R':
            config_rcv_buffer_size = atoi(optarg);
            if (config_log_level >= 1) {
                printf("option - receive buffer size: %d\n", config_rcv_buffer_size);
            }
            break;
        case 'T':
            config_runtime_max = atoi(optarg);
            if (config_log_level >= 1) {
                printf("option - runtime limit: %d\n", config_runtime_max);
            }
            break;
        case 'v':
            config_log_level = atoi(optarg);
            if (config_log_level >= 1) {
                printf("option - log level: %d\n", config_log_level);
            }
            break;
        default:
            print_usage();
            goto cleanup;
            break;
        }
    }


    if (config_port == 0) {
        config_active = 1;
        printf("role: active\n");
    } else {
        config_active = 0;
        printf("role: passive\n");
    }
    if ((ctx = neat_init_ctx()) == NULL) {
        fprintf(stderr, "%s - neat_init_ctx failed\n", __func__);
        result = EXIT_FAILURE;
        goto cleanup;
    }

    if (config_log_level == 0) {
        neat_log_level(ctx, NEAT_LOG_ERROR);
    } else if (config_log_level == 1){
        neat_log_level(ctx, NEAT_LOG_WARNING);
    } else {
        neat_log_level(ctx, NEAT_LOG_DEBUG);
    }

    options[0].tag = NEAT_TAG_CHANNEL_NAME;
    options[0].type = NEAT_TYPE_STRING;

    // create listening flow for accepted new data channels
    struct neat_flow *listening_flow;
    struct neat_flow_operations operation;
    if ((listening_flow = neat_new_flow(ctx)) == NULL) {
        fprintf(stderr, "%s - neat_new_flow failed\n", __func__);
        result = EXIT_FAILURE;
        goto cleanup;
    }

    memset(&operation, 0, sizeof(struct neat_flow_operations));
    operation.on_connected = on_connected;
    operation.on_error     = on_error;
    operation.on_close     = on_close;
    operation.on_parameters = on_parameters;
    char *params = calloc(1, 2048);
    operation.userData     = params;

    if (neat_set_operations(ctx, listening_flow, &operation)) {
        fprintf(stderr, "%s - neat_set_operations failed\n", __func__);
        result = EXIT_FAILURE;
        goto cleanup;
    }

    // set properties
    if (neat_set_property(ctx, listening_flow, arg_property)) {
        fprintf(stderr, "%s - neat_set_property failed\n", __func__);
        result = EXIT_FAILURE;
        goto cleanup;
    }


    // wait for on_connected or on_error to be invoked
    if (neat_accept(ctx, listening_flow, config_port, NULL, 0)) {
        fprintf(stderr, "%s - neat_accept failed\n", __func__);
        result = EXIT_FAILURE;
        goto cleanup;
    }
printf("neat_accept returned\n");

    for (i = 0; i < config_num_flows; i++) {
        if (config_port == 0) {
            sprintf(name, "Channel %d", 2 * i);
        } else {
            sprintf(name, "Channel %d", 2 * i + 1);
        }
        options[0].value.string = name;
        if ((flows[i] = neat_new_flow(ctx)) == NULL) {
            fprintf(stderr, "could not initialize context\n");
            result = EXIT_FAILURE;
            goto cleanup;
        }

        // set properties
        if (neat_set_property(ctx, flows[i], arg_property)) {
            fprintf(stderr, "%s - error: neat_set_property\n", __func__);
            result = EXIT_FAILURE;
             goto cleanup;
        }

        ops[i].on_connected = on_connected;
        ops[i].on_error = on_error;
        ops[i].on_close = on_close;
        ops[i].userData = &result; // allow on_error to modify the result variable
        neat_set_operations(ctx, flows[i], &(ops[i]));

        // wait for on_connected or on_error to be invoked
        if (neat_open(ctx, flows[i], argv[optind], config_port, options, 1) != NEAT_OK) {
            fprintf(stderr, "Could not open flow\n");
            exit(EXIT_FAILURE);
        } else {
            fprintf(stderr, "Opened flow %d\n", i);
           // flows_active++;
        }
    }


    neat_start_event_loop(ctx, NEAT_RUN_DEFAULT);

    // cleanup
cleanup:
printf("cleanup\n");
    if (config_log_level >= 1) {
        printf("freeing ctx bye bye!\n");
    }

    if (ctx != NULL) {
        neat_free_ctx(ctx);
    }
    exit(result);
}