Fetch
view release on metacpan or search on metacpan
include/fetch/ft_abi.h view on Meta::CPAN
#ifndef FT_ABI_H
#define FT_ABI_H
/* Fetch-side implementation of the shared C ABI (fetch_abi.h). Included by
* Fetch.xs AFTER ft_ua.h / ft_future.h, so ft_ua_of, ft_follow and the hmf_*
* future helpers are in scope. Everything here is private to Fetch's
* translation unit; consumers reach it only through the FETCH_ABI table
* returned by Fetch::_abi_ptr. */
#include "fetch_abi.h"
/* Construct a Fetch UA from flat key/value SV pairs (kv[0]=key, kv[1]=val,...);
* the shared body of Fetch->new and the ABI ua_new. Blesses into `cls`. */
static SV *ft_ua_new(pTHX_ const char *cls, SV **kv, int nkv) {
ft_ua *ua;
SV *loop_arg = NULL, *headers_arg = NULL, *agent_arg = NULL, *jar_arg = NULL;
int have_keep = 0, keep = 1, have_verify = 0, verify = 1;
int have_maxr = 0, maxr = 5, pool_size = 32, simple = 0;
double timeout = 0.0;
int i;
for (i = 0; i + 1 < nkv; i += 2) {
const char *k = SvPV_nolen(kv[i]);
SV *v = kv[i + 1];
if (strEQ(k, "loop")) loop_arg = v;
else if (strEQ(k, "headers")) headers_arg = v;
else if (strEQ(k, "agent")) agent_arg = v;
else if (strEQ(k, "cookie_jar")) jar_arg = v;
else if (strEQ(k, "keep_alive")) { have_keep = 1; keep = SvTRUE(v) ? 1 : 0; }
else if (strEQ(k, "tls_verify")) { have_verify = 1; verify = SvTRUE(v) ? 1 : 0; }
else if (strEQ(k, "max_redirects")) { have_maxr = 1; maxr = (int)SvIV(v); }
else if (strEQ(k, "timeout")) timeout = SvNV(v);
else if (strEQ(k, "pool_size")) pool_size = (int)SvIV(v);
else if (strEQ(k, "simple_response")) simple = SvTRUE(v) ? 1 : 0;
}
Newxz(ua, 1, ft_ua);
ua->loop = ft_resolve_loop(aTHX_ loop_arg);
if (ft_obj_can(aTHX_ ua->loop, "install_await")) {
dSP;
ENTER; SAVETMPS; PUSHMARK(SP); XPUSHs(ua->loop); PUTBACK;
call_method("install_await", G_DISCARD);
FREETMPS; LEAVE;
}
/* Hyperman-direct mode: when the adapter is Fetch::Loop::Hyperman and
* Hyperman's C ABI resolves (ft_hm.h), connections drive fd interest and
* deadlines straight through the table. Absent/old Hyperman leaves both
* NULL and the Perl _ft_arm/_ft_timer seam is used as before. */
if (sv_isobject(ua->loop)
&& sv_derived_from(ua->loop, "Fetch::Loop::Hyperman")
&& SvTYPE(SvRV(ua->loop)) == SVt_PVHV) {
const hm_abi *A = ft_hm(aTHX);
SV **e = hv_fetchs((HV *)SvRV(ua->loop), "loop", 0);
if (A && e && *e && sv_isobject(*e)
&& sv_derived_from(*e, "Hyperman::Loop")) {
ua->hm_loop = A->loop_of_sv(aTHX_ *e);
ua->hm = A;
/* C await: bridge + run_until instead of the Perl AWAIT sub */
ft_hm_install_await(aTHX_ ua->loop, ua->hm_loop);
}
}
ua->keep_alive = have_keep ? keep : 1;
ua->simple_response = simple;
if (jar_arg && SvOK(jar_arg)) {
if (SvROK(jar_arg)) ua->cookie_jar = SvREFCNT_inc(jar_arg);
else if (SvTRUE(jar_arg)) ua->cookie_jar = ft_load_new(aTHX_ "Fetch::CookieJar", NULL);
}
{
AV *hav = newAV();
if (headers_arg && SvOK(headers_arg))
ft_hdr_pairs_into(aTHX_ hav, headers_arg);
ua->headers = sv_bless(newRV_noinc((SV *)hav),
gv_stashpv("Fetch::Headers", GV_ADD));
}
if (agent_arg && SvOK(agent_arg)) {
ua->agent = newSVsv(agent_arg);
} else {
SV *ver = get_sv("Fetch::VERSION", 0);
ua->agent = newSVpvf("Fetch/%s", (ver && SvOK(ver)) ? SvPV_nolen(ver) : "0");
}
ua->tls_verify = have_verify ? verify : 1;
ua->max_redirects = have_maxr ? maxr : 5;
ua->timeout = timeout;
if (ua->keep_alive) {
ft_pool *p = ft_pool_new(pool_size > 0 ? pool_size : 32);
if (!p) { SvREFCNT_dec(ua->loop); Safefree(ua); croak("Fetch: out of memory"); }
ua->pool = sv_bless(newRV_noinc(newSViv(PTR2IV(p))),
gv_stashpv("Fetch::_Pool", GV_ADD));
}
return sv_bless(newRV_noinc(newSViv(PTR2IV(ua))), gv_stashpv(cls, GV_ADD));
}
/* A second agent over this one's connection pool and loop, with some options
* replaced.
*
* The case this exists for is a per-request cookie jar. A jar on a long-lived
* agent is shared by every request that agent serves, which leaks between them
* the moment the cookies identify an end user rather than the application; and
* building a whole fresh agent per request to avoid that throws away the
* keep-alive pool and re-resolves the loop adapter, which is most of what
* new() costs. A clone gives the caller its own jar over the same pool.
*
* Sharing is safe because DESTROY only releases references: the pool and the
* loop live until the last agent holding them goes. `loop` and `pool_size`
* cannot be overridden - they are the things being shared, and a clone on a
* different loop would be driving the parent's parked connections from the
* wrong place. */
static SV *ft_ua_clone(pTHX_ SV *self, SV **kv, int nkv) {
ft_ua *src = ft_ua_of(aTHX_ self);
ft_ua *ua;
SV *headers_arg = NULL, *agent_arg = NULL, *jar_arg = NULL;
int have_jar = 0, have_keep = 0, keep = 0, have_verify = 0, verify = 0;
int have_maxr = 0, maxr = 0, have_simple = 0, simple = 0, have_to = 0;
double timeout = 0.0;
const char *cls;
int i;
if (!(SvROK(self) && SvOBJECT(SvRV(self))))
croak("Fetch->clone: not a Fetch user agent");
cls = HvNAME(SvSTASH(SvRV(self)));
for (i = 0; i + 1 < nkv; i += 2) {
const char *k = SvPV_nolen(kv[i]);
SV *v = kv[i + 1];
if (strEQ(k, "cookie_jar")) { have_jar = 1; jar_arg = v; }
else if (strEQ(k, "headers")) headers_arg = v;
else if (strEQ(k, "agent")) agent_arg = v;
else if (strEQ(k, "keep_alive")) { have_keep = 1; keep = SvTRUE(v) ? 1 : 0; }
else if (strEQ(k, "tls_verify")) { have_verify = 1; verify = SvTRUE(v) ? 1 : 0; }
else if (strEQ(k, "max_redirects")) { have_maxr = 1; maxr = (int)SvIV(v); }
else if (strEQ(k, "timeout")) { have_to = 1; timeout = SvNV(v); }
else if (strEQ(k, "simple_response")){ have_simple = 1; simple = SvTRUE(v) ? 1 : 0; }
else if (strEQ(k, "loop") || strEQ(k, "pool_size"))
croak("Fetch->clone: '%s' cannot be overridden on a clone; the "
"loop and the connection pool are what it shares", k);
}
Newxz(ua, 1, ft_ua);
/* shared with the parent, by reference */
ua->loop = src->loop ? SvREFCNT_inc(src->loop) : NULL;
ua->pool = src->pool ? SvREFCNT_inc(src->pool) : NULL;
ua->hm = src->hm;
ua->hm_loop = src->hm_loop;
/* inherited unless overridden */
ua->keep_alive = have_keep ? keep : src->keep_alive;
ua->tls_verify = have_verify ? verify : src->tls_verify;
ua->max_redirects = have_maxr ? maxr : src->max_redirects;
ua->simple_response = have_simple ? simple : src->simple_response;
ua->timeout = have_to ? timeout : src->timeout;
if (have_jar) {
/* an explicit undef/0 means "no jar", which is how a clone drops one */
if (jar_arg && SvOK(jar_arg)) {
if (SvROK(jar_arg)) ua->cookie_jar = SvREFCNT_inc(jar_arg);
else if (SvTRUE(jar_arg)) ua->cookie_jar =
ft_load_new(aTHX_ "Fetch::CookieJar", NULL);
}
}
else ua->cookie_jar = src->cookie_jar ? SvREFCNT_inc(src->cookie_jar) : NULL;
/* Headers are copied, not shared: a clone that mutated a shared list would
* be writing into every other agent's defaults. */
{
AV *hav = newAV();
if (src->headers) ft_hdr_pairs_into(aTHX_ hav, src->headers);
if (headers_arg && SvOK(headers_arg))
ft_hdr_pairs_into(aTHX_ hav, headers_arg);
ua->headers = sv_bless(newRV_noinc((SV *)hav),
gv_stashpv("Fetch::Headers", GV_ADD));
}
ua->agent = (agent_arg && SvOK(agent_arg))
? newSVsv(agent_arg)
: (src->agent ? SvREFCNT_inc(src->agent) : NULL);
return sv_bless(newRV_noinc(newSViv(PTR2IV(ua))), gv_stashpv(cls, GV_ADD));
}
static SV *ft_abi_ua_new(pTHX_ SV **kv, int nkv) {
return ft_ua_new(aTHX_ "Fetch", kv, nkv);
}
/* carried through the upstream future to the completion callback */
typedef struct ft_abi_ctx { fetch_map_cb map; void *ud; } ft_abi_ctx;
/* Fires when the upstream request settles: read the response parts in C, let
* the consumer's `map` shape the resolved value, settle the derived future
* with it. Mirrors ft_redirect_cb's ownership (next lives in the closure). */
XS_INTERNAL(ft_abi_complete_cb);
XS_INTERNAL(ft_abi_complete_cb) {
dXSARGS;
hm_clos *cl = hm_clos_of(aTHX_ cv);
SV *f, *next, *val, *err = NULL;
ft_abi_ctx *ctx;
int ok = 0, status = 0;
AV *headers = NULL;
SV *body = NULL;
IV st;
if (!cl || items < 1) XSRETURN_EMPTY;
f = ST(0);
next = cl->a;
ctx = INT2PTR(ft_abi_ctx *, cl->i);
st = hmf_state(aTHX_ f);
if (st == HMF_DONE) {
AV *vals = hmf_values_av(aTHX_ f);
SV **rp = (vals && av_len(vals) >= 0) ? av_fetch(vals, 0, 0) : NULL;
SV *res = rp ? *rp : NULL;
if (res && SvROK(res) && SvTYPE(SvRV(res)) == SVt_PVHV) {
HV *h = (HV *)SvRV(res);
SV **e;
ok = 1;
if ((e = hv_fetchs(h, "status", 0)) && *e) status = (int)SvIV(*e);
if ((e = hv_fetchs(h, "headers", 0)) && *e && SvROK(*e)
&& SvTYPE(SvRV(*e)) == SVt_PVAV) headers = (AV *)SvRV(*e);
if ((e = hv_fetchs(h, "content", 0)) && *e) body = *e;
}
} else if (st == HMF_FAILED) {
AV *vals = hmf_values_av(aTHX_ f);
SV **e = (vals && av_len(vals) >= 0) ? av_fetch(vals, 0, 0) : NULL;
err = e ? *e : NULL;
}
val = ctx->map(aTHX_ ok, status, headers, body, err, ctx->ud);
if (!val) val = newSV(0);
hmf_settle(aTHX_ next, HMF_DONE, &val, 1);
SvREFCNT_dec(val);
Safefree(ctx);
XSRETURN_EMPTY;
}
static SV *ft_abi_request(pTHX_ SV *ua_sv, const char *method, const char *url,
const fetch_hdr *hdrs, int nhdrs,
const char *body, STRLEN blen,
double timeout, int max_redirects,
fetch_map_cb map, void *ud) {
ft_ua *ua = ft_ua_of(aTHX_ ua_sv);
HV *opt = newHV();
SV *f, *next, *cb;
ft_abi_ctx *ctx;
if (nhdrs > 0 && hdrs) {
AV *hav = newAV();
int i;
for (i = 0; i < nhdrs; i++) {
av_push(hav, newSVpvn(hdrs[i].name, hdrs[i].nlen));
av_push(hav, newSVpvn(hdrs[i].val, hdrs[i].vlen));
}
(void)hv_stores(opt, "headers", newRV_noinc((SV *)hav));
}
if (body) (void)hv_stores(opt, "body", newSVpvn(body, blen));
if (timeout > 0) (void)hv_stores(opt, "timeout", newSVnv(timeout));
f = ft_follow(aTHX_ ua_sv, ua, method, url, opt,
max_redirects < 0 ? ua->max_redirects : (IV)max_redirects);
SvREFCNT_dec((SV *)opt);
next = hmf_new(aTHX_ hmf_class_of(aTHX_ f));
Newxz(ctx, 1, ft_abi_ctx);
ctx->map = map;
ctx->ud = ud;
hmf_set_upstream(aTHX_ next, f);
cb = hm_closure(aTHX_ ft_abi_complete_cb, next, NULL, NULL, NULL,
PTR2IV(ctx), 0);
hmf_on_ready(aTHX_ f, cb);
SvREFCNT_dec(cb);
SvREFCNT_dec(f); /* the request keeps f alive */
return next;
}
static void ft_abi_res_parts(pTHX_ SV *res, int *status, AV **headers,
SV **body) {
HV *h;
SV **e;
if (status) *status = 0;
if (headers) *headers = NULL;
if (body) *body = NULL;
if (!res || !SvROK(res) || SvTYPE(SvRV(res)) != SVt_PVHV) return;
h = (HV *)SvRV(res);
if (status && (e = hv_fetchs(h, "status", 0)) && *e) *status = (int)SvIV(*e);
if (headers && (e = hv_fetchs(h, "headers", 0)) && *e && SvROK(*e)
&& SvTYPE(SvRV(*e)) == SVt_PVAV) *headers = (AV *)SvRV(*e);
if (body && (e = hv_fetchs(h, "content", 0)) && *e) *body = *e;
}
/* ---- streaming: wrap the consumer's C callbacks as Fetch's coderefs ------ */
typedef struct ft_abi_sctx {
fetch_on_headers on_headers;
fetch_on_body on_body;
fetch_on_done on_done;
void *ud;
} ft_abi_sctx;
XS_INTERNAL(ft_abi_hdr_tramp);
XS_INTERNAL(ft_abi_hdr_tramp) {
dXSARGS;
hm_clos *cl = hm_clos_of(aTHX_ cv);
ft_abi_sctx *c = cl ? INT2PTR(ft_abi_sctx *, cl->i) : NULL;
int status = 0;
AV *headers = NULL;
if (c && c->on_headers) {
if (items >= 1 && SvOK(ST(0))) status = (int)SvIV(ST(0));
if (items >= 2 && SvROK(ST(1)) && SvTYPE(SvRV(ST(1))) == SVt_PVAV)
headers = (AV *)SvRV(ST(1));
c->on_headers(aTHX_ status, headers, c->ud);
}
XSRETURN_EMPTY;
}
XS_INTERNAL(ft_abi_body_tramp);
XS_INTERNAL(ft_abi_body_tramp) {
dXSARGS;
hm_clos *cl = hm_clos_of(aTHX_ cv);
ft_abi_sctx *c = cl ? INT2PTR(ft_abi_sctx *, cl->i) : NULL;
if (c && c->on_body && items >= 1) {
STRLEN l;
const char *p = SvPV_const(ST(0), l);
c->on_body(aTHX_ p, l, c->ud);
}
XSRETURN_EMPTY;
}
XS_INTERNAL(ft_abi_done_tramp);
XS_INTERNAL(ft_abi_done_tramp) {
dXSARGS;
hm_clos *cl = hm_clos_of(aTHX_ cv);
( run in 2.242 seconds using v1.01-cache-2.11-cpan-14f38c9f855 )