view release on metacpan or  search on metacpan

xs/socket_connection.xs  view on Meta::CPAN

    SvREFCNT_dec(json_sv);
}
OUTPUT:
    RETVAL

void
decode_frames(class, data_sv)
    SV *class
    SV *data_sv
PPCODE:
{
    AV *result = newAV();
    I32 ri, result_count;

    if (SvOK(data_sv)) {
        const char *buf;
        STRLEN buf_len;
        STRLEN consumed = 0;

        buf = SvPV(data_sv, buf_len);

        while (buf_len - consumed >= 4) {
            const unsigned char *p =
                (const unsigned char *)(buf + consumed);
            UV frame_len = ((UV)p[0] << 24) | ((UV)p[1] << 16) |
                           ((UV)p[2] << 8)  | (UV)p[3];

            if (buf_len - consumed < 4 + frame_len) break;

            /* JSON decode payload */
            {
                SV *msg_sv = NULL;
                dSP;
                int count;
                ENTER;
                SAVETMPS;
                PUSHMARK(SP);
                XPUSHs(sv_2mortal(newSVpvn(
                    buf + consumed + 4, frame_len)));
                PUTBACK;
                count = call_pv(
                    "Chandra::Socket::Connection::_xs_json_decode",
                    G_SCALAR);
                SPAGAIN;
                if (count > 0) msg_sv = newSVsv(POPs);
                PUTBACK;
                FREETMPS;
                LEAVE;

                if (msg_sv && SvOK(msg_sv)) {
                    av_push(result, msg_sv);
                } else {
                    if (msg_sv) SvREFCNT_dec(msg_sv);
                }
            }

            consumed += 4 + frame_len;
        }
    }

    /* Push all decoded messages onto Perl stack */
    result_count = av_len(result) + 1;
    for (ri = 0; ri < result_count; ri++) {
        SV **svp = av_fetch(result, ri, 0);
        if (svp) XPUSHs(sv_2mortal(SvREFCNT_inc(*svp)));
    }
    SvREFCNT_dec((SV *)result);
}

int
_xs_do_send(socket_sv, msg_sv)
    SV *socket_sv
    SV *msg_sv
CODE:
{
    SV *json_enc = get_sv("Chandra::Socket::Connection::_xs_json", 0);
    SV *payload = NULL;
    STRLEN payload_len, frame_len;
    unsigned char len_buf[4];
    SV *frame;

    /* $payload = $_xs_json->encode($msg) */
    {
        dSP;
        int count;
        ENTER; SAVETMPS;
        PUSHMARK(SP);
        XPUSHs(json_enc);
        XPUSHs(msg_sv);
        PUTBACK;
        count = call_method("encode", G_SCALAR);
        SPAGAIN;
        payload = (count > 0) ? newSVsv(POPs) : newSVpvs("");
        PUTBACK;
        FREETMPS; LEAVE;
    }

    payload_len = SvCUR(payload);

    /* pack('N', length) . payload - build frame in C */
    len_buf[0] = (payload_len >> 24) & 0xFF;
    len_buf[1] = (payload_len >> 16) & 0xFF;
    len_buf[2] = (payload_len >> 8) & 0xFF;
    len_buf[3] = payload_len & 0xFF;

    frame = newSVpvn((char *)len_buf, 4);
    sv_catsv(frame, payload);
    SvREFCNT_dec(payload);
    frame_len = SvCUR(frame);

    /* Ignore SIGPIPE during syswrite - portable across OS */
    {
        SigpipeGuard spg;
        SV *written_sv;
        sigpipe_ignore(&spg);

        /* $socket->syswrite($frame) */
        {
            dSP;
            int count;
            ENTER; SAVETMPS;

xs/socket_connection.xs  view on Meta::CPAN

                }
            } else if (!SvOK(read_result)) {
                /* Check errno for EAGAIN/EWOULDBLOCK */
                int err_no = errno;
#ifdef EAGAIN
                if (err_no == EAGAIN) { PUTBACK; FREETMPS; LEAVE; SvREFCNT_dec(read_buf); goto parse_frames; }
#endif
#ifdef EWOULDBLOCK
                if (err_no == EWOULDBLOCK) { PUTBACK; FREETMPS; LEAVE; SvREFCNT_dec(read_buf); goto parse_frames; }
#endif
                (void)hv_stores(hv, "_connected", newSViv(0));
                SvREFCNT_dec(read_buf);
                PUTBACK; FREETMPS; LEAVE;
                goto return_messages;
            } else if (SvIV(read_result) == 0) {
                (void)hv_stores(hv, "_connected", newSViv(0));
                SvREFCNT_dec(read_buf);
                PUTBACK; FREETMPS; LEAVE;
                goto return_messages;
            }

            PUTBACK; FREETMPS; LEAVE;
        }

        SvREFCNT_dec(read_buf);
    }

    parse_frames:
    /* Parse length-prefixed frames from buffer */
    buf_svp = hv_fetchs(hv, "_buf", 0);
    if (buf_svp && SvOK(*buf_svp)) {
        SV *json_enc = get_sv("Chandra::Socket::Connection::_xs_json", 0);

        while (SvCUR(*buf_svp) >= 4) {
            const unsigned char *p = (const unsigned char *)SvPVX(*buf_svp);
            UV frame_len = ((UV)p[0] << 24) | ((UV)p[1] << 16) |
                           ((UV)p[2] << 8) | (UV)p[3];

            if (frame_len > 16 * 1024 * 1024) {
                warn("Chandra::Socket::Connection: frame too large (%" UVuf " bytes), disconnecting\n", frame_len);
                (void)hv_stores(hv, "_connected", newSViv(0));
                (void)hv_stores(hv, "_buf", newSVpvs(""));
                goto return_messages;
            }

            if (SvCUR(*buf_svp) < 4 + frame_len) break;

            /* Extract payload and advance buffer */
            {
                SV *payload = newSVpvn(SvPVX(*buf_svp) + 4, (STRLEN)frame_len);
                SV *remaining = newSVpvn(
                    SvPVX(*buf_svp) + 4 + frame_len,
                    SvCUR(*buf_svp) - 4 - (STRLEN)frame_len);
                sv_setsv(*buf_svp, remaining);
                SvREFCNT_dec(remaining);

                /* JSON decode */
                {
                    dSP;
                    int count;
                    SV *decoded;
                    ENTER; SAVETMPS;
                    PUSHMARK(SP);
                    XPUSHs(json_enc);
                    XPUSHs(payload);
                    PUTBACK;
                    count = call_method("decode", G_SCALAR | G_EVAL);
                    SPAGAIN;
                    if (count > 0 && !SvTRUE(ERRSV)) {
                        decoded = newSVsv(POPs);
                        av_push(messages, decoded);
                    } else {
                        warn("Chandra::Socket::Connection: malformed JSON frame\n");
                        if (SvTRUE(ERRSV)) sv_setpvs(ERRSV, "");
                    }
                    PUTBACK;
                    FREETMPS; LEAVE;
                }

                SvREFCNT_dec(payload);
            }
        }
    }

    return_messages:
    msg_count = av_len(messages) + 1;
    for (mi = 0; mi < msg_count; mi++) {
        SV **svp = av_fetch(messages, mi, 0);
        if (svp) XPUSHs(sv_2mortal(SvREFCNT_inc(*svp)));
    }
    SvREFCNT_dec((SV *)messages);
}

SV *
_xs_json_encode(data_sv)
    SV *data_sv
CODE:
{
    SV *json_enc = get_sv("Chandra::Socket::Connection::_xs_json", 0);
    dSP;
    int count;
    ENTER; SAVETMPS;
    PUSHMARK(SP);
    XPUSHs(json_enc);
    XPUSHs(data_sv);
    PUTBACK;
    count = call_method("encode", G_SCALAR);
    SPAGAIN;
    RETVAL = (count > 0) ? newSVsv(POPs) : newSVpvs("");
    PUTBACK;
    FREETMPS; LEAVE;
}
OUTPUT:
    RETVAL

SV *
_xs_json_decode(str_sv)
    SV *str_sv
CODE:
{
    SV *json_enc = get_sv("Chandra::Socket::Connection::_xs_json", 0);
    dSP;
    int count;
    ENTER; SAVETMPS;
    PUSHMARK(SP);
    XPUSHs(json_enc);
    XPUSHs(str_sv);
    PUTBACK;
    count = call_method("decode", G_SCALAR | G_EVAL);
    SPAGAIN;
    if (count > 0 && !SvTRUE(ERRSV)) {