myxce/src/controller/queue.c

#include <errno.h>
#include <stdlib.h>
#include <string.h>

#include "cluster.h"
#include "xce/cluster_def.h"
#include "xce/types.h"

void queue_fini(queue_t* q)
{
    if (q != NULL) {
        free(q);
    }
}

queue_t* create_queue(cluster_t* c, xce_queue_t* p)
{
    queue_t  q1;
    queue_t* q;

    memcpy(&(q1.queue), p, sizeof(q1.queue));

    q = create_obj_by_typename(&(c->objmgr), "xqueue", &q1);
    return q;
}

int build_queues(void* c)
{
    cluster_t* cls = (cluster_t*)c;
    int        i;

    if (cls == NULL) {
        return -EINVAL;
    }

    if (cls->myconf == NULL) {
        return -EINVAL;
    }

    for (i = 0; i < cls->myconf->cluster_config->nqueues; i++) {
        xce_queue_t* q = cls->myconf->cluster_config->queues[i];

        create_queue(cls, q);
    }

    return 0;
}

int queue_ctrl(cluster_t* c, xce_queuectrl_t* ctrl)
{
    queue_t* q  = NULL;
    int      rc = 0;

    if (!c || !ctrl || !ctrl->queue || *ctrl->queue == '\0') {
        TRACE("%s invalid arguments.", __func__);
        return -EINVAL;
    }
    TRACE("%s queue_ctrl queue %s", __func__, ctrl->queue);
    q = qmgr_lockq(&(c->qmgr), ctrl->queue, OBJREF_UPDATE);
    if (q == NULL) {
        PERROR("qmgr_lockq");
        return -ENOENT;
    }

    switch (ctrl->action) {
    case QUEUE_OPEN:
        if (!(q->queue.status & QUEUE_STAT_CLOSED)) {
            goto clean;
        }
        q->queue.status &= ~QUEUE_STAT_CLOSED;
        break;
    case QUEUE_CLOSE:
        if (q->queue.status & QUEUE_STAT_CLOSED) {
            goto clean;
        }
        q->queue.status |= QUEUE_STAT_CLOSED;
        break;
    case QUEUE_INACTIVE:
        if (q->queue.status & QUEUE_STAT_INACT_ADMIN) {
            goto clean;
        }
        q->queue.status |= QUEUE_STAT_INACT_ADMIN;
        break;
    case QUEUE_ACTIVE:
        if (!(q->queue.status & QUEUE_STAT_INACT_ADMIN)) {
            goto clean;
        }
        q->queue.status &= ~QUEUE_STAT_INACT_ADMIN;
        break;
    default:
        rc = -1;
        break;
    }
    if (rc == 0) {
        xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE,
                      HOOK_AFTER_EVENT, HOOK_DETAIL_QUEUE_UPDATE, c, q);
    }
clean:
    obj_deref(q);
    return rc;
}

int do_queuecontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int         rc = 0;
    xce_reply_t reply;

    init_reply_by_req(&reply, request);

    rc               = queue_ctrl(c, request->body.queuectrlmsg);
    reply.status     = rc;
    reply.data_which = RequestReply_data_nodata;

    xce_reply(ep, c->myhost, &reply);
    return rc;
}

int get_queue_exec(cluster_t* c, char* queue, char** pre_exec, char** post_exec)
{
    queue_t* q = NULL;

    TRACE("entering %s", __func__);

    if ((c == NULL) || (queue == NULL) || (pre_exec == NULL)
        || (post_exec == NULL)) {
        return -EINVAL;
    }

    TRACE("%s queue %s", __func__, queue);

    q = qmgr_lockq(&(c->qmgr), queue, OBJREF_READ);
    if (q == NULL) {
        PERROR("qmgr_lockq");
        return -ENOENT;
    }

    if (q->queue.pre_exec != NULL) {
        *pre_exec = create_obj_by_typename(&(c->objmgr), "xstring",
                                           q->queue.pre_exec);
    } else {
        *pre_exec = NULL;
    }

    if (q->queue.post_exec != NULL) {
        *post_exec = create_obj_by_typename(&(c->objmgr), "xstring",
                                            q->queue.post_exec);
    } else {
        *post_exec = NULL;
    }

    obj_deref(q);

    return 0;
}