/* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. */ /* $Rev$ $Date$ */ /** * PostgreSQL-based database component implementation. */ #define WANT_HTTPD_LOG 1 #include "string.hpp" #include "function.hpp" #include "list.hpp" #include "value.hpp" #include "monad.hpp" #include "parallel.hpp" #include "pgsql.hpp" namespace tuscany { namespace sqldb { /** * Get an item from the database. */ const failable get(const list& params, const pgsql::PGSql& pg) { return pgsql::get(car(params), pg); } /** * Post an item to the database. */ const failable post(const list& params, const pgsql::PGSql& pg) { const value id = append(car(params), mklist(mkuuid())); const failable val = pgsql::post(id, cadr(params), pg); if (!hasContent(val)) return mkfailure(val); return id; } /** * Put an item into the database. */ const failable put(const list& params, const pgsql::PGSql& pg) { const failable val = pgsql::put(car(params), cadr(params), pg); if (!hasContent(val)) return mkfailure(val); return value(content(val)); } /** * Patch an item in the database. */ const failable patch(const list& params, const pgsql::PGSql& pg) { // Read patch value p = assoc("patch", assoc("content", car(cadr(params)))); if (isNil(p)) return mkfailure("Couldn't read patch script"); const string script = cadr(p); debug(script, "pgsql::patch::script"); const lambda(const value&, const pgsql::PGSql&, const string&, const int)> tryPatch = [&tryPatch](const value& key, const pgsql::PGSql& pg, const string& script, const int count) -> const failable { // Begin database transaction const failable brc = pgsql::begin(pg); if (!hasContent(brc)) return mkfailure(brc); // Get existing value from database const failable ival = pgsql::get(key, pg); if (!hasContent(ival) && rcode(ival) != 404) { pgsql::rollback(pg); return mkfailure(ival); } // Apply patch istringstream is(script); scheme::Env env = scheme::setupEnvironment(); const value pval = scheme::evalScript(cons("patch", scheme::quotedParameters(mklist(key, hasContent(ival)? content(ival) : (value)list()))), is, env); if (isNil(pval)) { ostringstream os; os << "Couldn't patch postgresql entry: " << key; return mkfailure(str(os), 404, false); } // Push patched value to database const failable val = pgsql::patch(key, pval, pg); if (!hasContent(val)) { pgsql::rollback(pg); // Retry on a transaction serialization error if (rcode(val) == 409 && count > 0) return tryPatch(key, pg, script, count - 1); return mkfailure(val); } // Commit database transaction const failable crc = pgsql::commit(pg); if (!hasContent(crc)) return mkfailure(crc); return value(content(val)); }; // Try patching the entry and automatically retry a few times on transaction // serialization errors return tryPatch(car(params), pg, script, 5); } /** * Delete an item from the database. */ const failable del(const list& params, const pgsql::PGSql& pg) { const failable val = pgsql::del(car(params), pg); if (!hasContent(val)) return mkfailure(val); return value(content(val)); } /** * Start the component. */ const failable start(const list& params) { // Connect to the configured database and table debug("sqldb::start"); const gc_pool cp(gc_current_pool()); const value conninfo = ((lvvlambda)car(params))(nilListValue); const value table = ((lvvlambda)cadr(params))(nilListValue); const lambda()> newPGSql = [conninfo, table, cp]() -> const gc_ptr { debug("sqldb::newPGSql"); const gc_scoped_pool sp(pool(cp)); return new (gc_new()) pgsql::PGSql(conninfo, table); }; const perthread_ptr pg = *(new (gc_new >()) perthread_ptr(newPGSql)); // Return the component implementation lambda function const lvvlambda applySqldb = [pg](const list& params) -> const value { const value func(car(params)); if (func == "get") return get(cdr(params), *pg); if (func == "post") return post(cdr(params), *pg); if (func == "put") return put(cdr(params), *pg); if (func == "patch") return patch(cdr(params), *pg); if (func == "delete") return del(cdr(params), *pg); return mkfailure(); }; return value(applySqldb); } } } extern "C" { const tuscany::value apply(const tuscany::list& params) { const tuscany::value func(car(params)); if (func == "start") return tuscany::sqldb::start(cdr(params)); return tuscany::mkfailure(); } }