[dtn-security] Fwd: Streaming Video with dtn2 and vlc

[Carlos Neves <[e-mail]>]

Streaming Video with dtn2 and vlc
Our team is trying compare dtn2 and Ibr-dtn performance in video streaming
to write a paper about this technologies.
I have troubles to send a stream video in dtn2, because dtnsend does not
work as i wanted for the flag "f" (file).
Changing apps/dtnrecv.c, and do the /configure , make , make install again
, now i can send videos from a computer to another.
You have any solution to do stream of video or sending files without
changing dtnrecv.c original file?
I'm using dtnd version 2.9.0

To receive the stream i'm using:
receiver: dtnrecv dtn://carlos.dtn/r | vlc -
It works for remote communications with other computers.


You have a copy above with the modifications we have made.
dtnrecv.c
/*
 *    Copyright 2004-2006 Intel Corporation
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

#ifdef HAVE_CONFIG_H
#  include <dtn-config.h>
#endif

#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <strings.h>
#include <string.h>
#include <stdlib.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "dtn_api.h"

#define BUFSIZE 16
#define BLOCKSIZE 8192
#define COUNTER_MAX_DIGITS 9

// Find the maximum commandline length
#ifdef __FreeBSD__
/* Needed for PATH_MAX, Linux doesn't need it */
#include <sys/syslimits.h>
#endif

#ifndef PATH_MAX
/* A conservative fallback */
#define PATH_MAX 1024
#endif

const char *progname;

// Daemon connection
int api_IP_set = 0;
char * api_IP = "127.0.0.1";
short api_port = 5010;

int   verbose           = 0;        // verbose output
int   doAppAcks         = 0;            // Application ack of received
bundles
int   quiet             = 0;        // quiet output
char* endpoint        = NULL;     // endpoint for registration
dtn_reg_id_t regid    = DTN_REGID_NONE;// registration id
int   expiration    = 30;         // registration expiration time
int   count             = 0;            // exit after count bundles received
int   failure_action    = DTN_REG_DEFER;// registration delivery failure
action
char* failure_script    = "";         // script to exec
int   replay_action    = DTN_REPLAY_NEW;// registration delivery replay
semantics
int   register_only    = 0;        // register and quit
int   change        = 0;        // change existing registration
int   unregister    = 0;        // remove existing registration
int   recv_timeout    = -1;        // timeout to dtn_recv call
int   no_find_reg    = 0;        // omit call to dtn_find_registration
char filename[PATH_MAX];        // Destination filename for file xfers
dtn_bundle_payload_location_t bundletype = DTN_PAYLOAD_MEM;

void
usage()
{
    fprintf(stderr, "usage: %s [opts] <endpoint> \n", progname);
    fprintf(stderr, "options:\n");
    fprintf(stderr, " -A daemon api IP address\n");
    fprintf(stderr, " -B daemon api IP port\n");
    fprintf(stderr, " -a Application acknowledgement of received bundles
(to daemon)\n");
    fprintf(stderr, " -v verbose\n");
    fprintf(stderr, " -q quiet\n");
    fprintf(stderr, " -h help\n");
    fprintf(stderr, " -d <eid|demux_string> endpoint id\n");
    fprintf(stderr, " -r <regid> use existing registration regid\n");
    fprintf(stderr, " -n <count> exit after count bundles received\n");
    fprintf(stderr, " -e <time> registration expiration time in seconds "
            "(default: %d)\n", expiration);
    fprintf(stderr, " -f <defer|drop|exec> failure action\n");
    fprintf(stderr, " -R <new|none|all> replay action\n");
    fprintf(stderr, " -F <script> failure script for exec action\n");
    fprintf(stderr, " -x call dtn_register and immediately exit\n");
    fprintf(stderr, " -c call dtn_change_registration and immediately
exit\n");
    fprintf(stderr, " -u call dtn_unregister and immediately exit\n");
    fprintf(stderr, " -N don't try to find an existing registration\n");
    fprintf(stderr, " -t <timeout> timeout value for call to dtn_recv\n");
    fprintf(stderr, " -o <template> Write out transfers to files using this
template (# chars are\n"
            "replaced with a counter). Example: f##.bin goes to f00.bin,
f01.bin, etc...\n");
}

void
parse_options(int argc, char**argv)
{
    int c, done = 0;

    progname = argv[0];

    memset(filename, 0, sizeof(char) * PATH_MAX);

    while (!done)
    {
        c = getopt(argc, argv, "aA:B:vqhHd:r:e:f:R:F:xn:cuNt:o:");
        switch (c)
        {
        case 'a':
            doAppAcks = 1;
        case 'A':
            api_IP_set = 1;
            api_IP = optarg;
            break;
        case 'B':
            api_port = atoi(optarg);
            break;
        case 'v':
            verbose = 1;
            break;
        case 'q':
            quiet = 1;
            break;
        case 'h':
        case 'H':
            usage();
            exit(0);
            return;
        case 'r':
            regid = atoi(optarg);
            break;
        case 'e':
            expiration = atoi(optarg);
            break;
        case 'f':
            if (!strcasecmp(optarg, "defer")) {
                failure_action = DTN_REG_DEFER;

            } else if (!strcasecmp(optarg, "drop")) {
                failure_action = DTN_REG_DROP;

            } else if (!strcasecmp(optarg, "exec")) {
                failure_action = DTN_REG_EXEC;

            } else {
                fprintf(stderr, "invalid failure action '%s'\n", optarg);
                usage();
                exit(1);
            }
        case 'R':
            if (!strcasecmp(optarg, "new")) {
                replay_action = DTN_REPLAY_NEW;

            } else if (!strcasecmp(optarg, "none")) {
                replay_action = DTN_REPLAY_NONE;

            } else if (!strcasecmp(optarg, "all")) {
                replay_action = DTN_REPLAY_ALL;

            } else {
                fprintf(stderr, "invalid replay action '%s'\n", optarg);
                usage();
                exit(1);
            }
        case 'F':
            failure_script = optarg;
            break;
        case 'x':
            register_only = 1;
            break;
        case 'n':
            count = atoi(optarg);
            break;
        case 'c':
            change = 1;
            break;
        case 'u':
            unregister = 1;
            break;
        case 'N':
            no_find_reg = 1;
            break;
        case 't':
            recv_timeout = atoi(optarg);
            break;
        case 'o':
            strncpy(filename, optarg, PATH_MAX);
            break;
        case -1:
            done = 1;
            break;
        default:
            // getopt already prints an error message for unknown
            // option characters
            usage();
            exit(1);
        }
    }

    endpoint = argv[optind];
    if (!endpoint && (regid == DTN_REGID_NONE)) {
        fprintf(stderr, "must specify either an endpoint or a regid\n");
        usage();
        exit(1);
    }

    if ((change || unregister) && (regid == DTN_REGID_NONE)) {
        fprintf(stderr, "must specify regid when using -%c option\n",
                change ? 'c' : 'u');
        usage();
        exit(1);
    }

    if (failure_action == DTN_REG_EXEC && failure_script == NULL) {
        fprintf(stderr, "failure action EXEC must supply script\n");
        usage();
        exit(1);
    }

    // the default is to use memory transfer mode, but if someone specifies
a
    // filename then we need to tell the API to expect a file
    if ( filename[0] != '\0' )
        bundletype = DTN_PAYLOAD_FILE;

}


static void
print_data(char* buffer, u_int length)
{
    fwrite(buffer,1,length,stdout);
}

/* -----------------------------------------------------------------------
*/
/* Builds the temporary file based on the template given and an integer
 * counter value
 */
int buildfilename(char* template, char* newfilename, int counter )
{
    char counterasstring[COUNTER_MAX_DIGITS];
    char formatstring[10];
    char* startloc;
    char* endloc;
    int templatelen;

    strcpy(newfilename, template);

    endloc = rindex(newfilename, '#');
    /* No template in the filename, just copy it as is */
    if ( endloc == NULL )
        return 0;

    /* Search backwards for the start of the template */
    for ( startloc = endloc; *startloc == '#' && startloc != template;
          startloc -= 1 );

    startloc += 1;

    templatelen = endloc - startloc + 1;
    if ( templatelen > COUNTER_MAX_DIGITS )
        templatelen = COUNTER_MAX_DIGITS;

    sprintf(formatstring, "%%0%dd", templatelen);
    sprintf(counterasstring, formatstring, counter);

    if ( strlen(counterasstring) > (unsigned int)templatelen )
        fprintf(stderr, "Warning: Filename template not large enough "
                "to support counter value %d\n", counter);

    memcpy(startloc, counterasstring, sizeof(char) * templatelen);

    return 0;
}

/* -----------------------------------------------------------------------
*/
/* File transfers suffer considerably from the inability to safely send
 * metadata on the same channel as the file transfer in DTN.  Perhaps we
 * should work around this?
 */
int
handle_file_transfer(dtn_bundle_spec_t spec, dtn_bundle_payload_t payload,
                     int* total_bytes, int counter)
{
    int tempdes;
    int destdes;
    char block[BLOCKSIZE];
    ssize_t bytesread;
    struct stat fileinfo;
    char currentfile[PATH_MAX];

    // Create the filename by searching for ### characters in the given
    // filename and replacing that with an incrementing counter.
    buildfilename(filename, currentfile, counter);

    // Try to rename the old file to the new name to avoid unnecessary
copying
    if (rename(filename, currentfile) == 0) {
        // success!

    } else {
        // Copy the file into place
        tempdes = open(payload.filename.filename_val, O_RDONLY);

        if ( tempdes < 0 )
        {
            fprintf(stderr, "While opening the temporary file for reading
'%s': %s\n",
                    payload.filename.filename_val, strerror(errno));
            exit(1);
        }

        destdes = creat(currentfile, 0644);

        if ( destdes < 0 )
        {
            fprintf(stderr, "While opening output file for writing '%s':
%s\n",
                    filename, strerror(errno));
            exit(1);
        }

        // Duplicate the file
        while ( (bytesread = read(tempdes, block, sizeof(block))) > 0 )
            write(destdes, block, bytesread);

        close(tempdes);
        close(destdes);

        unlink(payload.filename.filename_val);
    }

    if ( stat(currentfile, &fileinfo) == -1 )
    {
        fprintf(stderr, "Unable to stat destination file '%s': %s\n",
                currentfile, strerror(errno));
        return 1;
    }

    *total_bytes += fileinfo.st_size;

    return 0;
}

int
main(int argc, char** argv)
{
    int i;
    u_int k;
    int ret;
    int total_bytes = 0;
    dtn_handle_t handle;
    dtn_endpoint_id_t local_eid;
    dtn_reg_info_t reginfo;
    dtn_bundle_spec_t spec;
    dtn_bundle_payload_t payload;
    int call_bind;

    // force stdout to always be line buffered, even if output is
    // redirected to a pipe or file
    setvbuf(stdout, (char *)NULL, _IOLBF, 0);

    progname = argv[0];

    parse_options(argc, argv);

    int err = 0;
    if (api_IP_set) err = dtn_open_with_IP(api_IP,api_port,&handle);
    else err = dtn_open(&handle);

    if (err != DTN_SUCCESS) {
        fprintf(stderr, "fatal error opening dtn handle: %s\n",
                dtn_strerror(err));
        exit(1);
    }

    // if we're not given a regid, or we're in change mode, we need to
    // build up the reginfo
    memset(&reginfo, 0, sizeof(reginfo));

    if ((regid == DTN_REGID_NONE) || change)
    {
        // if the specified eid starts with '/', then build a local
        // eid based on the configuration of our dtn router plus the
        // demux string. otherwise make sure it's a valid one
        if (endpoint[0] == '/') {
            dtn_build_local_eid(handle, &local_eid, (char *) endpoint);

        } else {
            if (dtn_parse_eid_string(&local_eid, endpoint)) {
                fprintf(stderr, "invalid destination endpoint '%s'\n",
                        endpoint);
                goto err;
            }
        }

        // create a new registration based on this eid
        dtn_copy_eid(&reginfo.endpoint, &local_eid);
        reginfo.regid             = regid;
        reginfo.expiration        = expiration;
        reginfo.flags             = failure_action;
        reginfo.replay_flags      = replay_action;
        reginfo.script.script_val = failure_script;
        reginfo.script.script_len = strlen(failure_script) + 1;
    }

    if (change) {
        if ((ret = dtn_change_registration(handle, regid, &reginfo)) != 0) {
            fprintf(stderr, "error changing registration: %d (%s)\n",
                    ret, dtn_strerror(dtn_errno(handle)));
            goto err;
        }
        goto done;
    }

    if (unregister) {
        if (dtn_unregister(handle, regid) != 0) {
            fprintf(stderr, "error in unregister regid %d: %s\n",
                    regid, dtn_strerror(dtn_errno(handle)));
            goto err;
        }

        goto done;
    }

    // try to see if there is an existing registration that matches
    // the given endpoint, in which case we'll use that one.
    if (regid == DTN_REGID_NONE && ! no_find_reg) {
        if (dtn_find_registration(handle, &local_eid, &regid) != 0) {
            if (dtn_errno(handle) != DTN_ENOTFOUND) {
                fprintf(stderr, "error in find_registration: %s\n",
                        dtn_strerror(dtn_errno(handle)));
                goto err;
            }
        }
        call_bind = 1;
    }

    // if the user didn't give us a registration to use, get a new one
    if (regid == DTN_REGID_NONE) {
        if ((ret = dtn_register(handle, &reginfo, &regid)) != 0) {
            fprintf(stderr, "error creating registration: %d (%s)\n",
                    ret, dtn_strerror(dtn_errno(handle)));
            goto err;
        }

        call_bind = 0;
    } else {
        call_bind = 1;
        dtn_parse_eid_string(&local_eid, "dtn://??/?");
    }

    if (register_only) {
        goto done;
    }

    if (call_bind) {
        // bind the current handle to the found registration
        if (dtn_bind(handle, regid) != 0) {
            fprintf(stderr, "error binding to registration: %s\n",
                    dtn_strerror(dtn_errno(handle)));
            goto err;
        }
    }


    for (i = 0; (count == 0) || (i < count); ++i) {
        memset(&spec, 0, sizeof(spec));
        memset(&payload, 0, sizeof(payload));


        if ((ret = dtn_recv(handle, &spec, bundletype, &payload,
                                                        recv_timeout)) < 0)
        {
            fprintf(stderr, "error getting recv reply: %d (%s)\n",
                    ret, dtn_strerror(dtn_errno(handle)));
            goto err;
        }

    //printf("deb1-ola!");
    //printf("%d",bundletype);
    //printf("%d",DTN_PAYLOAD_FILE);

        // Files need to be handled differently than memory transfers
        if ( bundletype == DTN_PAYLOAD_FILE )
        {
        //printf("deb-in-ola!");
                handle_file_transfer(spec, payload, &total_bytes, i);
                dtn_free_payload(&payload);
                continue;
        }
//    printf("deb2-ola!");
        total_bytes += payload.buf.buf_len;

        if (quiet) {
            dtn_free_payload(&payload);
            if (spec.blocks.blocks_len > 0) {
                free(spec.blocks.blocks_val);
                spec.blocks.blocks_val = NULL;
                spec.blocks.blocks_len = 0;
            }
            if (spec.metadata.metadata_len > 0) {
                free(spec.metadata.metadata_val);
                spec.metadata.metadata_val = NULL;
                spec.metadata.metadata_len = 0;
            }
            continue;
        }

        dtn_extension_block_t *block = spec.blocks.blocks_val;
        for (k = 0; k < spec.blocks.blocks_len; k++) {
            printf("Extension Block %i:\n", k);
            printf("\ttype = %i\n\tflags = %i\n",
                   block[k].type, block[k].flags);
            print_data(block[k].data.data_val, block[k].data.data_len);
        }

        dtn_extension_block_t *meta = spec.metadata.metadata_val;
        for (k = 0; k < spec.metadata.metadata_len; k++) {
            printf("Metadata Extension Block %i:\n", k);
            printf("\ttype = %i\n\tflags = %i\n",
                   meta[k].type, meta[k].flags);
            print_data(meta[k].data.data_val, meta[k].data.data_len);
        }

        print_data(payload.buf.buf_val, payload.buf.buf_len);

        if (doAppAcks) {
            dtn_bundle_id_t id;
            dtn_ack(handle, &spec, &id);
        }

        dtn_free_payload(&payload);
        if (spec.blocks.blocks_len > 0) {
            int i=0;
            for ( i=0; i<spec.blocks.blocks_len; i++ ) {
                free(spec.blocks.blocks_val[i].data.data_val);
            }
            free(spec.blocks.blocks_val);
            spec.blocks.blocks_val = NULL;
            spec.blocks.blocks_len = 0;
        }
        if (spec.metadata.metadata_len > 0) {
            int i=0;
            for ( i=0; i<spec.metadata.metadata_len; i++ ) {
                free(spec.metadata.metadata_val[i].data.data_val);
            }
            free(spec.metadata.metadata_val);
            spec.metadata.metadata_val = NULL;
            spec.metadata.metadata_len = 0;
        }
    }


done:
    dtn_close(handle);
    return 0;

err:
    dtn_close(handle);
    return 1;
}


Respects,
Carlos Neves Abrantes, André Carvalho
Communication Networks Engineering
Instituto Superior Técnico - Portugal