/*
 * © Copyright 1996-2012 ECMWF.
 *
 * This software is licensed under the terms of the Apache Licence Version 2.0
 * which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
 * In applying this licence, ECMWF does not waive the privileges and immunities
 * granted to it by virtue of its status as an intergovernmental organisation nor
 * does it submit to any jurisdiction.
 */

#include "queue.h"
#include "mars.h"

#include <setjmp.h>
#include <signal.h>

#if mars_client_HAVE_RPC
static XDR xmon;
static netblk blk = {
    0,
};
static udpinfo* udpmon = 0;
#endif

static jmp_buf env;

static void catch_alarm(int sig) {
    longjmp(env, 1);
}

#if mars_client_HAVE_RPC

err encode_netblk(XDR* x, netblk* blk) {
    x->x_op = XDR_ENCODE;

    if (!xdr_netblk(x, blk)) {
        /* marslog(LOG_WARN,"encode_request : xdr_netblk"); */
        return -2;
    }

    if (!xdrrec_endofrecord(x, true)) {
        /* marslog(LOG_WARN,"encode_request : xdrrec_endofrecord"); */
        return -2;
    }

    return 0;
}

err decode_netblk(XDR* x, netblk* blk) {

    bzero(blk, sizeof(netblk));
    x->x_op = XDR_DECODE;

    if (!xdrrec_skiprecord(x)) {
        /* marslog(LOG_WARN,"decode_request : xdrrec_skiprecord"); */
        return -2;
    }

    if (!xdr_netblk(x, blk)) {
        /* marslog(LOG_WARN,"decode_request : xdr_netblk"); */
        return -2;
    }

    return 0;
}

void free_netblk(netblk* blk) {
    xdr_free((xdrproc_t)xdr_netblk, (char*)blk);
    bzero(blk, sizeof(netblk));
}
#endif /* mars_client_HAVE_RPC */

static void flush_queue(int code, void* data) {
    qflush(code);
}

static err queue(int code, request* r, char* cinfo, long ninfo) {
#if mars_client_HAVE_RPC
#ifdef AMS
    return 0;
#endif
#ifndef ECMWF
    return 0;
#else
    static request* q = 0;
    int more;
    int maxtry = 10;
    int try    = 0;

    if (mars.request_id == -1)
        return -1;


    if (setjmp(env) != 0)
        return -1;

    signal(SIGALRM, catch_alarm);
    alarm(20);

    if (!q) {
        q = empty_request("QUEUE");
#ifdef METVIEW
        set_value(q, "PROGRAM", "metview");
#else
        set_value(q, "PROGRAM", "%s", progname());
#endif
        install_exit_proc(flush_queue, 0);
    }

    if (!udpmon) {
        udpmon = udp_call(mars.monhost, mars.monport);
        xdrrec_create(&xmon, 0, 0, (caddr_t)udpmon,
                      (xdrproc)&readudp, (xdrproc)&writeudp);
    }

    more = 1;
    while (more && try++ < maxtry) {
        static int check = 0;
        int ok           = 1;
        struct timeval start;

        bzero(&blk, sizeof(blk));
        set_value(q, "REQID", "%ld", mars.request_id);
        set_value(q, "CINFO", "%s", cinfo ? cinfo : "");
        set_value(q, "NINFO", "%ld", ninfo);
        blk.code  = code;
        q->next   = r;
        blk.req   = q;
        blk.check = ++check;

        more = 0;

        if (code == Q_ENTER || code == Q_SLAVE)
            blk.env = get_environ();

        gettimeofday(&start, 0);

        ok = 1;
        if (encode_netblk(&xmon, &blk) != 0)
            ok = 0;

        while (ok) {
            if (decode_netblk(&xmon, &blk) != 0)
                ok = 0;
            else if (blk.check == check) /* we got the right one ... */
            {
                const char* p;
                int i = 0;
                struct timeval stop;
                struct timeval diff;

                gettimeofday(&stop, 0);

                diff.tv_sec  = stop.tv_sec - start.tv_sec;
                diff.tv_usec = stop.tv_usec - start.tv_usec;

                while (diff.tv_usec < 0) {
                    diff.tv_sec--;
                    diff.tv_usec += 1000000;
                }

                /* update timeout to 100% more */
                udpmon->timeout.tv_sec  = diff.tv_sec * 2;
                udpmon->timeout.tv_usec = diff.tv_usec * 2;

                while (udpmon->timeout.tv_usec >= 1000000) {
                    diff.tv_sec++;
                    diff.tv_usec -= 1000000;
                }

                i = 0;
                while (p = get_value(blk.req, "MESSAGE", i++))
                    marslog(LOG_INFO, "%s", p);

                switch (blk.code) {

                    case Q_ENTER:
                        more = 1;
                        /* try  = 0; */
                        code = Q_ENTER;
                        break;

                    case Q_CONT:
                        break;

                    case Q_WAIT:
                        marslog(LOG_EXIT, "Waiting %d minutes...", blk.check);
                        sleep(blk.check * 60);
                        more = 1;
                        /* try  = 0; */
                        code = Q_CHECK;
                        break;

                    case Q_DIE:
                        marslog(LOG_EXIT, "Request aborted by operators");
                        break;

                    default:
                        marslog(LOG_WARN,
                                "Unexpected code received from queue manager %d", blk.code);
                        break;
                }

                break;
            }
            else if (blk.check < check) /* we receive older replies ....*/
                ;
            else if (blk.check > check) {
                marslog(LOG_WARN, "Something is strange in the queue manager!");
                ok = 0;
            }
            free_netblk(&blk);
        }

        if (!ok) {
            /* increase timeout */
            udpmon->timeout.tv_usec += 1000000 / 2; /* 1/2 sec */
            while (udpmon->timeout.tv_usec >= 1000000) {
                udpmon->timeout.tv_usec -= 1000000;
                udpmon->timeout.tv_sec++;
            }

#if 0
			/* server IP address may have changed */
			if(try > maxtry/2)
			{
				udp_free(udpmon);
				udpmon = udp_call(mars.monhost,mars.monport);
			}

#endif
            /* this is important to do, to get rid of incomplete messages */

            xdr_destroy(&xmon);
            xdrrec_create(&xmon, 0, 0, (caddr_t)udpmon,
                          (xdrproc)&readudp, (xdrproc)&writeudp);
        }
    }

    alarm(0);

    if (try >= maxtry) {
        marslog(LOG_WARN, "Mars queue manager on %s, port %d is not available",
                mars.monhost, mars.monport);
        return -1;
    }

    return 0;
#endif
#else
    return 0;
#endif /* mars_client_HAVE_RPC */
}

err qmonitor(char* fmt, ...) {
#if mars_client_HAVE_RPC
    va_list list;

    char buf[1024];

    va_start(list, fmt);
    vsprintf(buf, fmt, list);
    va_end(list);

    /* call queue */
    return queue(Q_CHECK, 0, buf, 0);
#endif /* mars_client_HAVE_RPC */
}

err qenter(request* r) {
    return queue(Q_ENTER, r, 0, 0);
}

err qleave(void) {
    return queue(Q_LEAVE, 0, 0, 0);
}

err qslave(int port) {
    return queue(Q_SLAVE, 0, 0, port);
}

err qsync(void) {
    return queue(Q_SYNC, 0, 0, 0);
}

err qflush(int code) {
    return queue(Q_FLUSH, 0, 0, code);
}