Reliable Publisher Sample

/*
 * Copyright 2000 by Sun Microsystems, Inc.
 * All rights reserved
 *
 * rpub : simple *reliable* interactive asynchronous publisher.
 *     It is designed to be used in combination with rsub,
 *     the reliable subscriber.
 *
 * Syntax:
 *   rpub host port
 */
#include <stdlib.h>
#include <stdio.h> 

#include "pasdisp.h"
#include "publisher.h" 

static pas_dispatcher_t *disp = NULL;
static publisher_t *_publisher = NULL;
static int _shutdown = 0;
static renl_t *_renl;
static void _read_stdin(); 

static void _exit_usage()
{
    printf("\nUsage:\nrpub host port\n");
    exit(5);
}
static void _exit_error(const char *msg)
{
    printf("%s\n", msg);
    exit(1);
}
static void _call_shutdown()
{
    _shutdown = 1;
    pas_shutdown(disp);
}
static void _renl_create_cb(void *arg, int rc, void *ignored)
{
    (void *)arg;
    (void *)ignored;
    if (!_publisher)
     {
        printf("Failed to create RENL - status %d\n", rc);
        _call_shutdown();
        return;
    }
    _read_stdin();
    return;
}
static void _publisher_new_cb(void *arg, int rc, void *enc)
{
    _publisher = (publisher_t *)enc;
    (void *)arg;
    if (!_publisher)
    {
        printf("Failed to create publisher - status %d\n", rc);
        _call_shutdown();
        return;
    }
    renl_create_publisher(_publisher, "renl_id", NULL,
                              _renl_create_cb,NULL);
    return;
}
static void _recv_ack(void *arg, int rc, void *ignored)
{
    (void *)ignored;
    if (rc < 0)
     {
        printf("Acknowledgment Timeout\n");
    } else if ( rc == 0) {
        printf("Acknowledgment Received\n");
    }
    fflush (stdout);
    _read_stdin();
    free(arg);
    return;
 }
static void _read_stdin()
{
    static char input[1024];
    printf("rpub> ");
    fflush(stdout);
    while (!_shutdown)
     {
        if ( !fgets(input, sizeof(input), stdin) )
        {
            continue;
        } else {
            char *message;
            unsigned int message_len;
            input[strlen(input) - 1] = 0; /* Strip off the \n */
            if (*input == ’.’ && input[1] == 0)
             {
                publisher_delete(_publisher);
                _call_shutdown();
                break;
            }
            message = strdup(input);
            message_len = strlen(message); 

            /* five seconds timeout */
            publish(_publisher, "enp://siroe.com/xyz",
                     message, message_len,
                     NULL, _recv_ack, message, 5000);
            return;
        }
    }
    return;
}
main(int argc, char **argv)
{
    unsigned short port = 7997;
    char host[256];
    if (argc < 2) _exit_usage();
    if (*(argv[1]) == ’0’)
     {
        strcpy(host, "127.0.0.1");
    } else {
        strcpy(host, argv[1]);
    }
    if (argc > 2)
     {
        port = (unsigned short)atoi(argv[2]);
    }
    disp = pas_dispatcher_new(NULL);
    if (disp == NULL) _exit_error("Can’t create publisher");
    publisher_new_a(disp, NULL, host, port, _publisher_new_cb,
                      NULL);
    pas_dispatch(disp);
    _shutdown = 1;
    pas_dispatcher_delete(disp);
    exit(0);
}