Function bdb_queue_consume_int

bdb/queue.c:1740–2089

}

static int bdb_queue_consume_int(bdb_state_type *bdb_state, tran_type *intran,
                                 int consumer, const void *prevfnd, int *bdberr)
{
    struct bdb_queue_found item;
    uint8_t hdrbuf[QUEUE_HDR_LEN];
    uint8_t *p_buf = hdrbuf, *p_buf_end = (p_buf + QUEUE_HDR_LEN);
    uint8_t *p_item_buf, *p_item_buf_end;
    size_t fragn;
    int rc;
    tran_type *tran;
    DBT dbt_key, dbt_data;
    db_recno_t recno;
    db_recno_t *recnos;
    struct bdb_queue_header hdr;

    if (gbl_rowlocks) {
        get_physical_transaction(bdb_state, intran, &tran, 0);
    } else
        tran = intran;

    if (!bdb_state->read_write) {
        *bdberr = BDBERR_READONLY;
        return -1;
    }

    if (!prevfnd) {
        *bdberr = BDBERR_BADARGS;
        return -1;
    }

    if (consumer < 0 || consumer >= BDBQUEUE_MAX_CONSUMERS) {
        *bdberr = BDBERR_BADARGS;
        return -1;
    }

    *bdberr = BDBERR_NOERROR;

    p_item_buf = (uint8_t *)prevfnd;
    p_item_buf_end = (p_item_buf + QUEUE_FOUND_LEN);

    if (!(p_item_buf =
              (uint8_t *)queue_found_get(&item, p_item_buf, p_item_buf_end))) {
        logmsg(LOGMSG_ERROR, "%s line %d: queue_found_get returns NULL\n", __func__,
                __LINE__);
        *bdberr = BDBERR_MISC;
        return -1;
    }

    /* Find the first fragment (header only). */
    recnos = (db_recno_t *)p_item_buf;
    recno = ntohl(recnos[0]);
    bzero(&dbt_key, sizeof(dbt_key));
    dbt_key.size = sizeof(db_recno_t);
    dbt_key.ulen = sizeof(db_recno_t);
    dbt_key.data = &recno;
    dbt_key.flags = DB_DBT_USERMEM;

    bzero(&dbt_data, sizeof(dbt_data));