Parsee/src/Extensions/PushEvents.c

#ifdef JANET

#include "Extensions/Internal.h"

typedef struct ExtensionEventReply {
    pthread_mutex_t lock;
    pthread_cond_t cond;

    HashMap *obj;

    Extension *ext;
    bool reply, ack;
} ExtensionEventReply;

static void
OnEventResponse(JanetEVGenericMessage msg)
{
    ExtensionEventReply *reply = msg.argp;
    Extension *ext = reply->ext;
    Janet args[1] = {
        JsonToJanet(reply->obj),
    };
    Janet ret;

    if (TryCall(ext, "on-event", 1, args, &ret))
    {
        pthread_mutex_lock(&reply->lock);
        reply->ack = true;
        reply->reply = janet_truthy(ret);
        pthread_cond_signal(&reply->cond);
        pthread_mutex_unlock(&reply->lock);
        return;
    }
    pthread_mutex_lock(&reply->lock);
    reply->ack = true;
    reply->reply = false;
    pthread_cond_signal(&reply->cond);
    pthread_mutex_unlock(&reply->lock);
}
bool
ExtensionPushEvent(Extensions *ctx, HashMap *obj)
{
    bool veto = false;
    size_t i;

    char *key;
    Extension *val;
    if (!ctx || !obj)
    {
        return false;
    }

    pthread_mutex_lock(&ctx->mtx);
    for (i = 0; IterateReentrant(ctx->extensions, key, val, i); )
    {
        ExtensionEventReply *reply = Malloc(sizeof(*reply));
        JanetEVGenericMessage msg = {
            /* We except a veto boolean. */
            .tag = JANET_STANZA_PUSH,
        };

        pthread_mutex_init(&reply->lock, NULL);
        pthread_cond_init(&reply->cond, NULL);
        reply->ext = val;
        reply->reply = false;
        reply->ack = false;
        reply->obj = obj;
        msg.argp = reply;

        janet_ev_post_event(val->vm, OnEventResponse, msg);
        pthread_mutex_lock(&reply->lock);
        while (!reply->ack)
        {
            pthread_cond_wait(&reply->cond, &reply->lock);
        }
        pthread_mutex_unlock(&reply->lock);
        pthread_mutex_destroy(&reply->lock);
        pthread_cond_destroy(&reply->cond);
        if (reply->reply)
        {
            veto = true;
        }
        Free(reply);
    }
    pthread_mutex_unlock(&ctx->mtx);

    return veto;
}
#endif