Function register_queue_with_berkdb_and_master

lua/sp.c:4759–4818

}

static int register_queue_with_berkdb_and_master(Lua L, const char *type)
{
    SP sp = getsp(L);
    struct sqlclntstate *clnt = sp->clnt;

    if (sp->consumer) {
        return luaL_error(L, "%s:%s already registered\n", sp->consumer->type, sp->spname);
    }
    if (sp->parent != sp) {
        return luaL_error(L, "attempt to run consumer in child thread");
    }
    if (clnt->dbtran.mode != TRANLEVEL_SOSQL) {
        return luaL_error(L, "%s is only supported under default transaction mode", type);
    }

    dbconsumer_t *consumer;
    size_t sz = dbconsumer_sz(sp->spname);
    new_lua_t_sz(L, consumer, DBTYPES_DBCONSUMER, sz);

    sp->consumer = consumer;
    consumer->type = type;
    consumer->emit_timeoutms = 60000; /* emit times-out after 1 min */
    consumer->osql_max_trans = clnt->osql_max_trans;

    if (lua_gettop(L) == 2) {
        lua_insert(L, 1); /* move dbconsumer to bottom of stack */
        dbconsumer_getargs(L, consumer);
    }

    /* register with master */
    trigger_reg_t *info = &consumer->info;
    info->elect_cookie = ATOMIC_LOAD32(gbl_master_changes);
    info->trigger_cookie = get_id(thedb->bdb_env);
    info->spname_len = strlen(sp->spname);
    memcpy(info->spname, sp->spname, info->spname_len + 1);
    int hostname_len = strlen(gbl_myhostname);
    memcpy(trigger_hostname(info), gbl_myhostname, hostname_len + 1);
    ctrace("%s:%s %016" PRIx64 " register req\n", type, info->spname, info->trigger_cookie);
    int rc = luabb_trigger_register(L, info, consumer->register_timeoutms);
    if (rc != CDB2_TRIG_REQ_SUCCESS) {
        ctrace("%s:%s %016" PRIx64 " register failed rc:%d\n", type, info->spname, info->trigger_cookie, rc);
        force_unregister(L, info);
        if (rc == -2) { /* register timeout */
            sp->consumer = NULL;
            return 0;
        }
        return luaL_error(L, sp->error);
    }
    ctrace("%s:%s %016" PRIx64 " register success\n", type, info->spname, info->trigger_cookie);

    init_fake_ireq(thedb, &consumer->iq);
    consumer->iq.usedb = find_and_lock_queue_table(L);

    /* register with berkdb */
    if (bdb_trigger_subscribe(consumer->iq.usedb->handle, &consumer->cond, &consumer->lock, &consumer->status) != 0) {
        return luaL_error(L, "bdb_trigger_subscribe failed for %s:%s", type, sp->spname);
    }