view release on metacpan or  search on metacpan

Websockets.xs  view on Meta::CPAN

static void capture_request_headers(struct lws *wsi, HV *hv) {
    int i;
    for (i = 0; i < N_REQUEST_HDRS; i++)
        capture_header(wsi, hv, request_hdrs[i].tok,
                       request_hdrs[i].name, request_hdrs[i].nlen);
}

/* Inject all key/value pairs from an HV as HTTP headers via lws.
   Returns -1 if lws rejected a header (client path aborts the handshake);
   server callers ignore the result and simply stop adding. */
static int inject_headers(struct lws *wsi, HV *hv,
                          unsigned char **p, unsigned char *end) {
    HE *entry;
    char kbuf[256];
    hv_iterinit(hv);
    while ((entry = hv_iternext(hv))) {
        I32 klen;
        const char *key = hv_iterkey(entry, &klen);
        SV *val_sv;
        STRLEN vlen;
        const char *val;
        if (klen >= 254) continue;
        val_sv = hv_iterval(hv, entry);
        val = SvPV(val_sv, vlen);
        memcpy(kbuf, key, klen);
        kbuf[klen] = ':';
        kbuf[klen + 1] = '\0';
        if (lws_add_http_header_by_name(wsi, (unsigned char *)kbuf,
                (unsigned char *)val, vlen, p, end))
            return -1;
    }
    return 0;
}

/* Format a wsi pointer as the lookup key for handshake_headers_map.
   Callers must pass a buffer of at least 32 bytes; returns the key length. */
static int wsi_key(char *buf, struct lws *wsi) {
    return snprintf(buf, 32, "%p", (void*)wsi);
}

#define DEBUG_LOG(fmt, ...) do { if (ev_ws_debug) fprintf(stderr, "[EV::WS] " fmt "\n", ##__VA_ARGS__); } while(0)

static void ctx_ref(ev_ws_ctx_t* ctx) {
    ctx->refcnt++;
}

static void ctx_unref(ev_ws_ctx_t* ctx) {
    if (--ctx->refcnt == 0) {
        Safefree(ctx);
    }
}

/* Schedule the next lws housekeeping wake-up.

   lws_service_adjust_timeout returns the ms until lws next needs servicing; 0
   means "service as soon as possible". This timer only paces lws's time-based
   work (connection/handshake timeouts, TLS cert aging, draining buffered rx) --
   socket readability/writability is driven by the per-fd io watcher, not here.
   We deliberately floor the delay at 1ms rather than arming an ev_idle watcher
   on 0: do_lws_service is now non-blocking, so an always-ready idle watcher
   would busy-spin at 100% CPU whenever lws keeps asking for immediate service.
   A 1ms floor lets the loop block briefly so every other EV watcher still
   fires, at negligible latency for this coarse, time-based work. */
static void schedule_timeout(ev_ws_ctx_t* ctx) {
    int delay_ms = lws_service_adjust_timeout(ctx->lws_ctx, 1000, 0);
    double delay_s;

    if (delay_ms < 1) delay_ms = 1;
    delay_s = (double)delay_ms / 1000.0;

    ev_timer_stop(ctx->loop, &ctx->timer);
    ev_timer_set(&ctx->timer, delay_s, 0.);
    ev_timer_start(ctx->loop, &ctx->timer);
}

static void flush_recv_messages(ev_ws_ctx_t* ctx);

static void do_lws_service(ev_ws_ctx_t* ctx) {
    if (ctx && ctx->magic == EV_WS_CTX_MAGIC && ctx->lws_ctx) {
        int alive = 1;
        int* prev_flag = ctx->alive_flag;
        ctx->alive_flag = &alive;
        ctx_ref(ctx);
        /* "Forced service": drive connections that need servicing with no
           pending socket event -- rx already read into lws's buflist, plus any
           due lws_sul timeouts. Counterintuitively a timeout_ms of 0 is NOT
           non-blocking here: lws maps it to its maximum internal poll wait, so
           the old lws_service(ctx, 0) blocked for seconds on an idle connection
           and starved every other EV watcher. A negative timeout_ms clamps the
           wait to 0, servicing only ready + forced-service work and returning at
           once (see the lws_service_adjust_timeout docs). lws_service_fd(ctx,
           NULL) is invalid since lws 3.2 and never drains buflists. Per-fd I/O
           is driven by io_cb via lws_service_fd(&pollfd). */
        lws_service_tsi(ctx->lws_ctx, -1, 0);
        if (alive)
            flush_recv_messages(ctx); /* deliver messages reassembled above */
        if (alive) {
            ctx->alive_flag = prev_flag;
            schedule_timeout(ctx);
        } else if (prev_flag) {
            *prev_flag = 0; /* propagate destruction up the alive_flag chain */
        }
        ctx_unref(ctx);
    }
}

static void timer_cb(EV_P_ ev_timer* w, int revents) {
    (void)loop; (void)revents;
    do_lws_service((ev_ws_ctx_t*)w->data);
}

/* Forward declarations */
static void io_cb(EV_P_ ev_io* w, int revents);
static void add_fd_watcher(ev_ws_ctx_t* ctx, int fd, int events);
static void change_fd_watcher(ev_ws_ctx_t* ctx, int fd, int events);
static void conn_ref(ev_ws_conn_t* conn);
static void conn_unref(ev_ws_conn_t* conn);

static SV* get_conn_sv(ev_ws_conn_t* conn) {
    if (conn->perl_self) {
        SV* rv = newRV_inc(conn->perl_self);