view release on metacpan or  search on metacpan

xs/async_base.c  view on Meta::CPAN

    }
    
    #undef _fetch_assert
    #undef pseudo_multi_begin
    #undef pseduo_multi_maybe_add
    #undef pseudo_multi_end
    #undef pseudo_perform
}


static inline void
extract_async_options(PLCBA_t *async, AV *options)
{
    #define _assert_get_cv(idxbase, target, diemsg) \
        if( (tmpsv = av_fetch(options, PLCBA_CTORIDX_ ## idxbase, 0)) == NULL \
            || SvTYPE(*tmpsv) == SVt_NULL) { \
            die("Must have '%s' callback", diemsg); \
        } \
        SvREFCNT_inc(*tmpsv); \
        async->target = *tmpsv;
    
    SV **tmpsv;
    
    _assert_get_cv(CBEVMOD, cv_evmod, "update_event");
    _assert_get_cv(CBERR, cv_err, "error");
    _assert_get_cv(CBWAITDONE, cv_waitdone, "waitdone");
    _assert_get_cv(CBTIMERMOD, cv_timermod, "update_timer");
    
    if( (tmpsv = av_fetch(options, PLCBA_CTORIDX_BLESS_EVENT, 0)) ) {
        if(SvTRUE(*tmpsv)) {
            async->event_stash = gv_stashpv(PLCBA_EVENT_CLASS, 0);
        }
    }
    
    #undef _assert_get_cv
}

SV *
PLCBA_construct(const char *pkg, AV *options)
{
    PLCBA_t *async;
    char *host, *username, *password, *bucket;
    libcouchbase_t instance;
    SV *blessed_obj;
    
    Newxz(async, 1, PLCBA_t);
    
    extract_async_options(async, options);
    
    plcb_ctor_conversion_opts(&async->base, options);
    
    plcb_ctor_cbc_opts(options, &host, &username, &password, &bucket);
    instance = libcouchbase_create(host, username, password, bucket,
                                   plcba_make_io_opts(async));
    
    if(!instance) {
        die("Couldn't create instance!");
    }
    
    plcb_ctor_init_common(&async->base, instance, options);
    plcba_setup_callbacks(async);
    async->base_rv = newRV_inc(newSViv(PTR2IV(&(async->base))));
    
    blessed_obj = newSV(0);
    sv_setiv(newSVrv(blessed_obj, pkg), PTR2IV(async));
    return blessed_obj;
}

/*called from perl when an event arrives*/
void
PLCBA_HaveEvent(const char *pkg, short flags, SV *opaque)
{
    /*TODO: optmize this to take an arrayref, and maybe configure ourselves for
     event loops which have a different calling convention, e.g. POE*/
    
    PLCBA_c_event *cevent;
    //sv_dump(opaque);
    
    cevent = NUM2PTR(PLCBA_c_event*, SvIV(opaque));
    cevent->c.handler(cevent->fd, flags, cevent->c.arg);
}

void
PLCBA_connect(SV *self)
{
    libcouchbase_t instance;
    PLCBA_t *async;
    PLCB_t *base;
    libcouchbase_error_t err;
    
    _mk_common_vars(self, instance, base, async);
    if( (err = libcouchbase_connect(instance)) != LIBCOUCHBASE_SUCCESS) {
        die("Problem with initial connection: %s (%d)",
            libcouchbase_strerror(instance, err), err);
    }
    libcouchbase_wait(instance);
}

void
PLCBA_DESTROY(SV *self)
{
    libcouchbase_t instance;
    PLCBA_t *async;
    PLCB_t *base;
    
    #define _DEC_AND_NULLIFY(fld) \
        if(async->fld) { SvREFCNT_dec(async->fld); async->fld = NULL; }
    
    _DEC_AND_NULLIFY(base_rv);
    _DEC_AND_NULLIFY(cv_evmod);
    _DEC_AND_NULLIFY(cv_timermod);
    _DEC_AND_NULLIFY(cv_err);
    _DEC_AND_NULLIFY(cv_waitdone);
    
    #undef _DEC_AND_NULLIFY
    
    /*cleanup our base object. This will also cause the events to be destroyed*/
    plcb_cleanup(&async->base);
    Safefree(async);
}