view release on metacpan or  search on metacpan

src/EV__Kafka.xs  view on Meta::CPAN

        const char *tname; int16_t tnlen;
        n = kf_read_string(p, end, &tname, &tnlen);
        if (n < 0) goto done;
        p += n;
        hv_store(th, "topic", 5, newSVpvn(tname ? tname : "", tname ? tnlen : 0), 0);

        /* partitions: ARRAY */
        if (end - p < 4) goto done;
        int32_t part_count = kf_read_i32(p); p += 4;
        AV *parts_av = (AV*)sv_2mortal((SV*)newAV());
        int32_t j;

        for (j = 0; j < part_count; j++) {
            HV *ph = (HV*)sv_2mortal((SV*)newHV());
            if (end - p < 4) goto done;
            int32_t pid = kf_read_i32(p); p += 4;
            hv_store(ph, "partition", 9, newSViv(pid), 0);

            if (end - p < 2) goto done;
            int16_t err = kf_read_i16(p); p += 2;
            hv_store(ph, "error_code", 10, newSViv(err), 0);

            if (end - p < 8) goto done;
            int64_t base_offset = kf_read_i64(p); p += 8;
            hv_store(ph, "base_offset", 11, newSViv(base_offset), 0);

            /* log_append_time (v2+) */
            if (version >= 2) {
                if (end - p < 8) goto done;
                p += 8;
            }

            /* log_start_offset (v5+) */
            if (version >= 5) {
                if (end - p < 8) goto done;
                p += 8;
            }

            av_push(parts_av, newRV_inc((SV*)ph));
        }
        hv_store(th, "partitions", 10, newRV_inc((SV*)parts_av), 0);
        av_push(topics_av, newRV_inc((SV*)th));
    }

    /* throttle_time_ms (v1+) */
    if (version >= 1 && end - p >= 4) {
        int32_t throttle = kf_read_i32(p); p += 4;
        hv_store(result, "throttle_time_ms", 16, newSViv(throttle), 0);
    }

done:
    hv_store(result, "topics", 6, newRV_inc((SV*)topics_av), 0);
    return sv_2mortal(newRV_noinc((SV*)result));
}

/* ================================================================
 * RecordBatch decoder (for Fetch responses)
 * ================================================================ */

/* Decode records from a RecordBatch, push them as hashrefs onto records_av.
 * Returns number of records decoded, or -1 on error. */
static int kf_decode_record_batch(pTHX_ const char *data, size_t len,
    AV *records_av, int64_t *out_base_offset)
{
    const char *p = data;
    const char *end = data + len;
    int n;

    if (end - p < 12) return -1;
    int64_t base_offset = kf_read_i64(p); p += 8;
    if (out_base_offset) *out_base_offset = base_offset;
    int32_t batch_length = kf_read_i32(p); p += 4;

    /* compare in 64-bit so a huge batch_length cannot wrap the check */
    if (batch_length < 0 || (int64_t)(end - p) < (int64_t)batch_length) return -1;
    const char *batch_end = p + batch_length;

    if (batch_end - p < 9) return -1;
    /* int32_t partition_leader_epoch = kf_read_i32(p); */ p += 4;
    int8_t magic = (int8_t)*p; p += 1;
    if (magic != 2) return -1; /* only support magic=2 (current format) */
    uint32_t expected_crc = (uint32_t)kf_read_i32(p); p += 4;
    /* CRC32C covers the bytes from attributes to end of batch. */
    if (crc32c(p, (size_t)(batch_end - p)) != expected_crc) return -1;

    if (batch_end - p < 36) return -1;
    int16_t attributes = kf_read_i16(p); p += 2;
    int compression_type = attributes & 0x07;
    /* int32_t last_offset_delta = kf_read_i32(p); */ p += 4;
    int64_t first_timestamp = kf_read_i64(p); p += 8;
    /* int64_t max_timestamp = kf_read_i64(p); */ p += 8;
    /* int64_t producer_id = kf_read_i64(p); */ p += 8;
    /* int16_t producer_epoch = kf_read_i16(p); */ p += 2;
    /* int32_t base_sequence = kf_read_i32(p); */ p += 4;

    if (batch_end - p < 4) return -1;
    int32_t record_count = kf_read_i32(p); p += 4;

    /* Decompress if needed */
    const char *rec_data = p;
    const char *rec_end = batch_end;
    char *decompressed = NULL;

    if (compression_type != COMPRESS_NONE && batch_end > p) {
        size_t compressed_len = batch_end - p;
        size_t decomp_cap = compressed_len * 4;
        if (decomp_cap < 4096) decomp_cap = 4096;
        /* set only by a codec branch that fully succeeded (see below) */
        int decompressed_ok = 0;

#ifdef HAVE_ZLIB
        if (compression_type == COMPRESS_GZIP) {
            int zok = 0;
            while (!zok && decomp_cap < 64 * 1024 * 1024) {
                Newx(decompressed, decomp_cap, char);
                z_stream zs;
                Zero(&zs, 1, z_stream);
                int zinit = inflateInit2(&zs, MAX_WBITS + 16);
                if (zinit != Z_OK) {
                    Safefree(decompressed);
                    decompressed = NULL;