view release on metacpan or  search on metacpan

include/fetch/ft_future.h  view on Meta::CPAN

            if (hm_call_list(aTHX_ NULL, NULL, user, args, n, out)) {
                SV *err = sv_2mortal(newSVsv(ERRSV));
                hmf_settle(aTHX_ next, HMF_FAILED, &err, 1);
                SvREFCNT_dec((SV *)out);
            } else {
                hmf_chain_into(aTHX_ next, out);
            }
        } else {
            AV *copy = newAV();
            SSize_t i;
            for (i = 0; i < n; i++) {
                SV **e = av_fetch(vals, i, 0);
                av_push(copy, e ? newSVsv(*e) : newSV(0));
            }
            hmf_settle_av(aTHX_ next, st, copy);
        }
    } else {
        hmf_cancel(aTHX_ next);
    }
    XSRETURN_EMPTY;
}

/* transform continuation: b = done-mapper, c = fail-mapper */
XS_INTERNAL(hm_xs_transform_cb);
XS_INTERNAL(hm_xs_transform_cb) {
    dXSARGS;
    hm_clos *cl = hm_clos_of(aTHX_ cv);
    SV *f, *next;
    IV st;
    if (!cl || items < 1) XSRETURN_EMPTY;
    f = ST(0);
    next = cl->a;
    st = hmf_state(aTHX_ f);
    if (st == HMF_DONE || st == HMF_FAILED) {
        SV *mapper = (st == HMF_DONE) ? cl->b : cl->c;
        AV *vals = hmf_values_av(aTHX_ f);
        SSize_t n = vals ? av_len(vals) + 1 : 0;
        AV *out = newAV();
        if (mapper && SvOK(mapper)) {
            SV **args = vals ? AvARRAY(vals) : NULL;
            if (hm_call_list(aTHX_ NULL, NULL, mapper, args, n, out)) {
                SV *err = sv_2mortal(newSVsv(ERRSV));
                hmf_settle(aTHX_ next, HMF_FAILED, &err, 1);
                SvREFCNT_dec((SV *)out);
                XSRETURN_EMPTY;
            }
        } else {
            SSize_t i;
            for (i = 0; i < n; i++) {
                SV **e = av_fetch(vals, i, 0);
                av_push(out, e ? newSVsv(*e) : newSV(0));
            }
        }
        hmf_settle_av(aTHX_ next, st, out);
    } else {
        hmf_cancel(aTHX_ next);
    }
    XSRETURN_EMPTY;
}

/* on_done / on_fail filter: b = user cb, i = required state */
XS_INTERNAL(hm_xs_ondone_cb);
XS_INTERNAL(hm_xs_ondone_cb) {
    dXSARGS;
    hm_clos *cl = hm_clos_of(aTHX_ cv);
    SV *f;
    if (!cl || items < 1) XSRETURN_EMPTY;
    f = ST(0);
    if (hmf_state(aTHX_ f) == cl->i) {
        AV *vals = hmf_values_av(aTHX_ f);
        SSize_t n = vals ? av_len(vals) + 1 : 0;
        dSP;
        SSize_t j;
        ENTER; SAVETMPS; PUSHMARK(SP);
        for (j = 0; j < n; j++) {
            SV **e = av_fetch(vals, j, 0);
            XPUSHs(e ? *e : &PL_sv_undef);
        }
        PUTBACK;
        call_sv(cl->b, G_DISCARD);
        FREETMPS; LEAVE;
    }
    XSRETURN_EMPTY;
}

/* ---- convergent combinators --------------------------------------------- *
 * a = result future, b = remaining-count SV, c = subfutures AV ref,
 * i = mode. */
#define HM_COMB_WAIT_ALL  0
#define HM_COMB_WAIT_ANY  1
#define HM_COMB_NEEDS_ALL 2
#define HM_COMB_NEEDS_ANY 3

XS_INTERNAL(hm_xs_comb_cb);
XS_INTERNAL(hm_xs_comb_cb) {
    dXSARGS;
    hm_clos *cl = hm_clos_of(aTHX_ cv);
    SV *g, *res;
    AV *list;
    IV left;
    if (!cl || items < 1) XSRETURN_EMPTY;
    g = ST(0);
    res = cl->a;
    list = (AV *)SvRV(cl->c);
    if (hmf_state(aTHX_ res) != HMF_PENDING) XSRETURN_EMPTY;

    switch (cl->i) {
    case HM_COMB_WAIT_ALL: {
        left = SvIV(cl->b) - 1;
        sv_setiv(cl->b, left);
        if (left == 0) {
            AV *out = newAV();
            SSize_t i, n = av_len(list) + 1;
            for (i = 0; i < n; i++) {
                SV **e = av_fetch(list, i, 0);
                av_push(out, e ? newSVsv(*e) : newSV(0));
            }
            hmf_settle_av(aTHX_ res, HMF_DONE, out);
        }
        break;
    }