storage_db.c

这是一个完全开放的

    drvdata_t data = (drvdata_t) drv->private;
    dbdata_t dbd = xhash_get(data->dbs, type);
    DBC *c;
    DB_TXN *t;
    st_ret_t ret;

    if(os_count(os) == 0)
        return st_SUCCESS;

    ret = _st_db_cursor_new(drv, dbd, &c, &t);
    if(ret != st_SUCCESS)
        return ret;

    ret = _st_db_put_guts(drv, type, owner, os, dbd, c, t);
    if(ret != st_SUCCESS) {
        t->abort(t);
        _st_db_cursor_free(drv, dbd, c, NULL);
        return st_FAILED;
    }

    return _st_db_cursor_free(drv, dbd, c, t);
}

static st_ret_t _st_db_get(st_driver_t drv, const char *type, const char *owner, const char *filter, os_t *os) {
    drvdata_t data = (drvdata_t) drv->private;
    dbdata_t dbd = xhash_get(data->dbs, type);
    DBC *c;
    DB_TXN *t;
    st_ret_t ret;
    DBT key, val;
    st_filter_t f;
    int err;
    os_object_t o;
    char *cfilter;

    ret = _st_db_cursor_new(drv, dbd, &c, &t);
    if(ret != st_SUCCESS)
        return ret;

    f = NULL;
    if(filter != NULL) {
        f = xhash_get(data->filters, filter);
        if(f == NULL) {
            f = storage_filter(filter);
            cfilter = pstrdup(xhash_pool(data->filters), filter);
            xhash_put(data->filters, cfilter, (void *) f);
            pool_cleanup(xhash_pool(data->filters), (pool_cleaner) pool_free, f->p);
        }
    }

    memset(&key, 0, sizeof(DBT));
    memset(&val, 0, sizeof(DBT));

    key.data = (char *) owner;
    key.size = strlen(owner);

    *os = os_new();

    err = c->c_get(c, &key, &val, DB_SET);
    while(err == 0) {
        o = _st_db_object_deserialise(drv, *os, val.data, val.size);

        if(o != NULL && !storage_match(f, o, os))
            os_object_free(o);

        err = c->c_get(c, &key, &val, DB_NEXT_DUP);
    }

    if(err != 0 && err != DB_NOTFOUND) {
        log_write(drv->st->sm->log, LOG_ERR, "db: couldn't move cursor for type %s owner %s in storage db: %s", type, owner, db_strerror(err));
        t->abort(t);
        _st_db_cursor_free(drv, dbd, c, NULL);
        os_free(*os);
        return st_FAILED;
    }

    ret = _st_db_cursor_free(drv, dbd, c, t);
    if(ret != st_SUCCESS) {
        os_free(*os);
        return ret;
    }

    if(os_count(*os) == 0) {
        os_free(*os);
        return st_NOTFOUND;
    }

    return st_SUCCESS;
}

static st_ret_t _st_db_delete_guts(st_driver_t drv, const char *type, const char *owner, const char *filter, dbdata_t dbd, DBC *c, DB_TXN *t) {
    drvdata_t data = (drvdata_t) drv->private;
    DBT key, val;
    st_filter_t f;
    int err;
    os_t os;
    os_object_t o;
    char *cfilter;

    f = NULL;
    if(filter != NULL) {
        f = xhash_get(data->filters, filter);
        if(f == NULL) {
            f = storage_filter(filter);
            cfilter = pstrdup(xhash_pool(data->filters), filter);
            xhash_put(data->filters, cfilter, (void *) f);
            pool_cleanup(xhash_pool(data->filters), (pool_cleaner) pool_free, f->p);
        }
    }

    memset(&key, 0, sizeof(DBT));
    memset(&val, 0, sizeof(DBT));

    key.data = (char *) owner;
    key.size = strlen(owner);

    os = os_new();

    err = c->c_get(c, &key, &val, DB_SET);
    while(err == 0) {
        o = _st_db_object_deserialise(drv, os, val.data, val.size);

        if(o != NULL && storage_match(f, o, os))
            err = c->c_del(c, 0);

        if(err == 0)
            err = c->c_get(c, &key, &val, DB_NEXT_DUP);
    }

    os_free(os);

    if(err != 0 && err != DB_NOTFOUND) {
        log_write(drv->st->sm->log, LOG_ERR, "db: couldn't move cursor for type %s owner %s in storage db: %s", type, owner, db_strerror(err));
        return st_FAILED;
    }

    return st_SUCCESS;
}

static st_ret_t _st_db_delete(st_driver_t drv, const char *type, const char *owner, const char *filter) {
    drvdata_t data = (drvdata_t) drv->private;
    dbdata_t dbd = xhash_get(data->dbs, type);
    DBC *c;
    DB_TXN *t;
    st_ret_t ret;

    ret = _st_db_cursor_new(drv, dbd, &c, &t);
    if(ret != st_SUCCESS)
        return ret;

    ret = _st_db_delete_guts(drv, type, owner, filter, dbd, c, t);
    if(ret != st_SUCCESS) {
        t->abort(t);
        _st_db_cursor_free(drv, dbd, c, NULL);
        return st_FAILED;
    }

    return _st_db_cursor_free(drv, dbd, c, t);
}

static st_ret_t _st_db_replace(st_driver_t drv, const char *type, const char *owner, const char *filter, os_t os) {
    drvdata_t data = (drvdata_t) drv->private;
    dbdata_t dbd = xhash_get(data->dbs, type);
    DBC *c;
    DB_TXN *t;
    st_ret_t ret;

    ret = _st_db_cursor_new(drv, dbd, &c, &t);
    if(ret != st_SUCCESS)
        return ret;

    ret = _st_db_delete_guts(drv, type, owner, filter, dbd, c, t);
    if(ret != st_SUCCESS) {
        t->abort(t);
        _st_db_cursor_free(drv, dbd, c, NULL);
        return st_FAILED;
    }

    if(os_count(os) == 0)
        return _st_db_cursor_free(drv, dbd, c, t);

    ret = _st_db_put_guts(drv, type, owner, os, dbd, c, t);
    if(ret != st_SUCCESS) {
        t->abort(t);
        _st_db_cursor_free(drv, dbd, c, NULL);
        return st_FAILED;
    }

    return _st_db_cursor_free(drv, dbd, c, t);
}

static void _st_db_free(st_driver_t drv) {
    drvdata_t data = (drvdata_t) drv->private;
    const char *key;
    dbdata_t dbd;

    if(xhash_iter_first(data->dbs))
        do {
            xhash_iter_get(data->dbs, &key, (void **) &dbd);

            log_debug(ZONE, "closing %s db", key);

            dbd->db->close(dbd->db, 0);
            free(dbd);
        } while(xhash_iter_next(data->dbs));

    xhash_free(data->dbs);

    xhash_free(data->filters);

    free(data);
}

/** panic function */
static void _st_db_panic(DB_ENV *env, int errval) {
    log_t log = (log_t) env->app_private;

    log_write(log, LOG_CRIT, "db: corruption detected! close all jabberd processes and run db_recover");

    exit(2);
}

st_ret_t st_db_init(st_driver_t drv) {
    char *path;
    int err;
    DB_ENV *env;
    drvdata_t data;

    path = config_get_one(drv->st->sm->config, "storage.db.path", 0);
    if(path == NULL) {
        log_write(drv->st->sm->log, LOG_ERR, "db: no path specified in config file");
        return st_FAILED;
    }

    if((err = db_env_create(&env, 0)) != 0) {
        log_write(drv->st->sm->log, LOG_ERR, "db: couldn't create environment: %s", db_strerror(err));
        return st_FAILED;
    }

    if((err = env->set_paniccall(env, _st_db_panic)) != 0) {
        log_write(drv->st->sm->log, LOG_ERR, "db: couldn't set panic call: %s", db_strerror(err));
        return st_FAILED;
    }

    /* store the log context in case we panic */
    env->app_private = drv->st->sm->log;

    if((err = env->open(env, path, DB_INIT_LOCK | DB_INIT_MPOOL | DB_INIT_LOG | DB_INIT_TXN | DB_CREATE, 0)) != 0) {
        log_write(drv->st->sm->log, LOG_ERR, "db: couldn't open environment: %s", db_strerror(err));
        env->close(env, 0);
        return st_FAILED;
    }

    data = (drvdata_t) malloc(sizeof(struct drvdata_st));
    memset(data, 0, sizeof(struct drvdata_st));

    data->env = env;
    data->path = path;

    if(config_get_one(drv->st->sm->config, "storage.db.sync", 0) != NULL)
        data->sync = 1;

    data->dbs = xhash_new(101);

    data->filters = xhash_new(17);

    drv->private = (void *) data;

    drv->add_type = _st_db_add_type;
    drv->put = _st_db_put;
    drv->get = _st_db_get;
    drv->replace = _st_db_replace;
    drv->delete = _st_db_delete;
    drv->free = _st_db_free;

    return st_SUCCESS;
}

#endif