storage_mysql.c

这是一个完全开放的

        if(mysql_query(data->conn, "BEGIN") != 0) {
            log_write(drv->st->sm->log, LOG_ERR, "mysql: sql transaction begin failed: %s", mysql_error(data->conn));
            return st_FAILED;
        }
    }

    if(_st_mysql_put_guts(drv, type, owner, os) != st_SUCCESS) {
        if(data->txn)
            mysql_query(data->conn, "ROLLBACK");
        return st_FAILED;
    }

    if(data->txn)
        if(mysql_query(data->conn, "COMMIT") != 0) {
            log_write(drv->st->sm->log, LOG_ERR, "mysql: sql transaction commit failed: %s", mysql_error(data->conn));
            mysql_query(data->conn, "ROLLBACK");
            return st_FAILED;
        }

    return st_SUCCESS;
}

static st_ret_t _st_mysql_get(st_driver_t drv, const char *type, const char *owner, const char *filter, os_t *os) {
    drvdata_t data = (drvdata_t) drv->private;
    char *cond, *buf = NULL;
    int buflen = 0;
    MYSQL_RES *res;
    int ntuples, nfields, i, j;
    MYSQL_FIELD *fields;
    MYSQL_ROW tuple;
    unsigned long *lengths;
    os_object_t o;
    char *val;
    os_type_t ot;
    int ival;
    nad_t nad;
    char tbuf[128];

    if(mysql_ping(data->conn) != 0) {
        log_write(drv->st->sm->log, LOG_ERR, "mysql: connection to database lost");
        return st_FAILED;
    }

    if(data->prefix != NULL) {
        snprintf(tbuf, sizeof(tbuf), "%s%s", data->prefix, type);
        type = tbuf;
    }

    cond = _st_mysql_convert_filter(drv, owner, filter);
    log_debug(ZONE, "generated filter: %s", cond);

    MYSQL_SAFE(buf, strlen(type) + strlen(cond) + 50, buflen);
    sprintf(buf, "SELECT * FROM `%s` WHERE %s ORDER BY `object-sequence`", type, cond);
    free(cond);

    log_debug(ZONE, "prepared sql: %s", buf);

    if(mysql_query(data->conn, buf) != 0) {
        log_write(drv->st->sm->log, LOG_ERR, "mysql: sql select failed: %s", mysql_error(data->conn));
        free(buf);
        return st_FAILED;
    }
    free(buf);

    res = mysql_store_result(data->conn);
    if(res == NULL) {
        log_write(drv->st->sm->log, LOG_ERR, "mysql: sql result retrieval failed: %s", mysql_error(data->conn));
        return st_FAILED;
    }

    ntuples = mysql_num_rows(res);
    if(ntuples == 0) {
        mysql_free_result(res);
        return st_NOTFOUND;
    }

    log_debug(ZONE, "%d tuples returned", ntuples);

    nfields = mysql_num_fields(res);

    if(nfields == 0) {
        log_debug(ZONE, "weird, tuples were returned but no fields *shrug*");
        mysql_free_result(res);
        return st_NOTFOUND;
    }

    fields = mysql_fetch_fields(res);

    *os = os_new();

    for(i = 0; i < ntuples; i++) {
        o = os_object_new(*os);

        if((tuple = mysql_fetch_row(res)) == NULL)
            break;

        for(j = 0; j < nfields; j++) {
            if(strcmp(fields[j].name, "collection-owner") == 0 || strcmp(fields[j].name, "object-sequence") == 0)
                continue;

            if(tuple[j] == NULL)
                continue;

            lengths = mysql_fetch_lengths(res);

            switch(fields[j].type) {
                case FIELD_TYPE_TINY:   /* tinyint */
                    ot = os_type_BOOLEAN;
                    break;

                case FIELD_TYPE_LONG:   /* integer */
                    ot = os_type_INTEGER;
                    break;

                case FIELD_TYPE_BLOB:   /* text */
                    ot = os_type_STRING;
                    break;

                default:
                    log_debug(ZONE, "unknown field type %d, ignoring it", fields[j].type);
                    continue;
            }

            val = tuple[j];

            switch(ot) {
                case os_type_BOOLEAN:
                    ival = (val[0] == '0') ? 0 : 1;
                    os_object_put(o, fields[j].name, &ival, ot);
                    break;

                case os_type_INTEGER:
                    ival = atoi(val);
                    os_object_put(o, fields[j].name, &ival, ot);
                    break;

                case os_type_STRING:
                    os_object_put(o, fields[j].name, val, os_type_STRING);
                    break;
            }
        }
    }

    mysql_free_result(res);

    return st_SUCCESS;
}

static st_ret_t _st_mysql_delete(st_driver_t drv, const char *type, const char *owner, const char *filter) {
    drvdata_t data = (drvdata_t) drv->private;
    char *cond, *buf = NULL;
    int buflen = 0;
    char tbuf[128];

    if(mysql_ping(data->conn) != 0) {
        log_write(drv->st->sm->log, LOG_ERR, "mysql: connection to database lost");
        return st_FAILED;
    }

    if(data->prefix != NULL) {
        snprintf(tbuf, sizeof(tbuf), "%s%s", data->prefix, type);
        type = tbuf;
    }

    cond = _st_mysql_convert_filter(drv, owner, filter);
    log_debug(ZONE, "generated filter: %s", cond);

    MYSQL_SAFE(buf, strlen(type) + strlen(cond) + 19, buflen);
    sprintf(buf, "DELETE FROM `%s` WHERE %s", type, cond);
    free(cond);

    log_debug(ZONE, "prepared sql: %s", buf);

    if(mysql_query(data->conn, buf) != 0) {
        log_write(drv->st->sm->log, LOG_ERR, "mysql: sql delete failed: %s", mysql_error(data->conn));
        free(buf);
        return st_FAILED;
    }
    free(buf);

    return st_SUCCESS;
}

static st_ret_t _st_mysql_replace(st_driver_t drv, const char *type, const char *owner, const char *filter, os_t os) {
    drvdata_t data = (drvdata_t) drv->private;

    if(mysql_ping(data->conn) != 0) {
        log_write(drv->st->sm->log, LOG_ERR, "mysql: connection to database lost");
        return st_FAILED;
    }

    if(data->txn) {
        if(mysql_query(data->conn, "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE") != 0) {
            log_write(drv->st->sm->log, LOG_ERR, "mysql: sql transaction setup failed: %s", mysql_error(data->conn));
            return st_FAILED;
        }

        if(mysql_query(data->conn, "BEGIN") != 0) {
            log_write(drv->st->sm->log, LOG_ERR, "mysql: sql transaction begin failed: %s", mysql_error(data->conn));
            return st_FAILED;
        }
    }

    if(_st_mysql_delete(drv, type, owner, filter) == st_FAILED) {
        if(data->txn)
            mysql_query(data->conn, "ROLLBACK");
        return st_FAILED;
    }

    if(_st_mysql_put_guts(drv, type, owner, os) == st_FAILED) {
        if(data->txn)
            mysql_query(data->conn, "ROLLBACK");
        return st_FAILED;
    }

    if(data->txn)
        if(mysql_query(data->conn, "COMMIT") != 0) {
            log_write(drv->st->sm->log, LOG_ERR, "mysql: sql transaction commit failed: %s", mysql_error(data->conn));
            mysql_query(data->conn, "ROLLBACK");
            return st_FAILED;
        }

    return st_SUCCESS;
}

static void _st_mysql_free(st_driver_t drv) {
    drvdata_t data = (drvdata_t) drv->private;

    mysql_close(data->conn);

    xhash_free(data->filters);

    free(data);
}

st_ret_t st_mysql_init(st_driver_t drv) {
    char *host, *port, *dbname, *user, *pass;
    MYSQL *conn;
    drvdata_t data;

    host = config_get_one(drv->st->sm->config, "storage.mysql.host", 0);
    port = config_get_one(drv->st->sm->config, "storage.mysql.port", 0);
    dbname = config_get_one(drv->st->sm->config, "storage.mysql.dbname", 0);
    user = config_get_one(drv->st->sm->config, "storage.mysql.user", 0);
    pass = config_get_one(drv->st->sm->config, "storage.mysql.pass", 0);

    if(host == NULL || port == NULL || dbname == NULL || user == NULL || pass == NULL) {
        log_write(drv->st->sm->log, LOG_ERR, "mysql: invalid driver config");
        return st_FAILED;
    }

    conn = mysql_init(NULL);
    if(conn == NULL) {
        log_write(drv->st->sm->log, LOG_ERR, "mysql: unable to allocate database connection state");
        return st_FAILED;
    }

    if(mysql_real_connect(conn, host, user, pass, dbname, atoi(port), NULL, 0) == NULL) {
        log_write(drv->st->sm->log, LOG_ERR, "mysql: connection to database failed: %s", mysql_error(conn));
        mysql_close(conn);
        return 1;
    }

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

    data->conn = conn;

    data->filters = xhash_new(17);

    if(config_get_one(drv->st->sm->config, "storage.mysql.transactions", 0) != NULL)
        data->txn = 1;
    else
        log_write(drv->st->sm->log, LOG_WARNING, "mysql: transactions disabled");

    data->prefix = config_get_one(drv->st->sm->config, "storage.mysql.prefix", 0);

    drv->private = (void *) data;

    drv->add_type = _st_mysql_add_type;
    drv->put = _st_mysql_put;
    drv->get = _st_mysql_get;
    drv->delete = _st_mysql_delete;
    drv->replace = _st_mysql_replace;
    drv->free = _st_mysql_free;

    return st_SUCCESS;
}

#endif