[asterisk-commits] tilghman: branch tilghman/odbc_tx_support r117400 - in /team/tilghman/odbc_tx...

SVN commits to the Asterisk project asterisk-commits at lists.digium.com
Tue May 20 22:59:11 CDT 2008


Author: tilghman
Date: Tue May 20 22:59:11 2008
New Revision: 117400

URL: http://svn.digium.com/view/asterisk?view=rev&rev=117400
Log:
Move the transaction stuff into res_odbc, as it makes more sense there

Modified:
    team/tilghman/odbc_tx_support/funcs/func_odbc.c
    team/tilghman/odbc_tx_support/include/asterisk/res_odbc.h
    team/tilghman/odbc_tx_support/res/res_odbc.c

Modified: team/tilghman/odbc_tx_support/funcs/func_odbc.c
URL: http://svn.digium.com/view/asterisk/team/tilghman/odbc_tx_support/funcs/func_odbc.c?view=diff&rev=117400&r1=117399&r2=117400
==============================================================================
--- team/tilghman/odbc_tx_support/funcs/func_odbc.c (original)
+++ team/tilghman/odbc_tx_support/funcs/func_odbc.c Tue May 20 22:59:11 2008
@@ -63,42 +63,7 @@
 	struct ast_custom_function *acf;
 };
 
-static const char *isolation2text(int iso)
-{
-	if (iso == 0) {
-		return "read_committed";
-	} else if (iso == 1) {
-		return "read_uncommitted";
-	} else if (iso == 2) {
-		return "serializable";
-	} else if (iso == 3) {
-		return "repeatable_read";
-	} else {
-		return "unknown";
-	}
-}
-
-static int text2isolation(const char *txt)
-{
-	if (strncasecmp(txt, "read_", 5) == 0) {
-		if (strncasecmp(txt + 5, "c", 1) == 0) {
-			return 0;
-		} else if (strncasecmp(txt + 5, "u", 1) == 0) {
-			return 1;
-		} else {
-			return -1;
-		}
-	} else if (strncasecmp(txt, "ser", 3) == 0) {
-		return 2;
-	} else if (strncasecmp(txt, "rep", 3) == 0) {
-		return 3;
-	} else {
-		return -1;
-	}
-}
-
 static void odbc_datastore_free(void *data);
-static void odbc_txn_free(void *data);
 
 struct ast_datastore_info odbc_info = {
 	.type = "FUNC_ODBC",
@@ -116,108 +81,6 @@
 	AST_LIST_HEAD(, odbc_datastore_row);
 	char names[0];
 };
-
-static struct ast_datastore_info txn_info = {
-	.type = "FUNC_ODBC_Transaction",
-	.destroy = odbc_txn_free,
-};
-
-struct odbc_txn_frame {
-	AST_LIST_ENTRY(odbc_txn_frame) list;
-	enum { ODBC_STATUS_TXCLOSED = 0, ODBC_STATUS_TXOPEN = 1 } status; /*!< Status of this transaction */
-	struct odbc_obj *obj;                                             /*!< Database handle within which transacted statements are run */
-	unsigned int active:1;                                            /*!< Is this record the current active transaction within the channel? */
-	char name[0];                                                     /*!< Name of this transaction ID */
-};
-
-static struct odbc_txn_frame *find_transaction(struct ast_channel *chan, struct odbc_obj *obj, const char *name, int active)
-{
-	struct ast_datastore *txn_store = ast_channel_datastore_find(chan, &odbc_info, NULL);
-	AST_LIST_HEAD(, odbc_txn_frame) *oldlist;
-	struct odbc_txn_frame *txn;
-
-	if (txn_store) {
-		oldlist = txn_store->data;
-	} else {
-		/* Need to create a new datastore */
-		if (!(txn_store = ast_channel_datastore_alloc(&odbc_info, NULL))) {
-			ast_log(LOG_ERROR, "Unable to allocate a new datastore.  Cannot create a new transaction.\n");
-			return NULL;
-		}
-
-		if (!(oldlist = ast_calloc(1, sizeof(*oldlist)))) {
-			ast_log(LOG_ERROR, "Unable to allocate datastore list head.  Cannot create a new transaction.\n");
-			ast_channel_datastore_free(txn_store);
-			return NULL;
-		}
-
-		txn_store->data = oldlist;
-		AST_LIST_HEAD_INIT(oldlist);
-		ast_channel_datastore_add(chan, txn_store);
-	}
-
-	AST_LIST_LOCK(oldlist);
-
-	/* Scanning for an object is *fast*.  Scanning for a name is much slower. */
-	if (obj != NULL || active == 1) {
-		AST_LIST_TRAVERSE(oldlist, txn, list) {
-			if (txn->obj == obj || txn->active) {
-				AST_LIST_UNLOCK(oldlist);
-				return txn;
-			}
-		}
-	}
-
-	if (name != NULL) {
-		AST_LIST_TRAVERSE(oldlist, txn, list) {
-			if (!strcasecmp(txn->name, name)) {
-				AST_LIST_UNLOCK(oldlist);
-				return txn;
-			}
-		}
-	}
-
-	/* Nothing found, create one */
-	if (name && obj && (txn = ast_calloc(1, sizeof(*txn) + strlen(name) + 1))) {
-		struct odbc_txn_frame *otxn;
-
-		strcpy(txn->name, name); /* SAFE */
-		txn->obj = obj;
-
-		/* On creation, the txn becomes active, and all others inactive */
-		AST_LIST_TRAVERSE(oldlist, otxn, list) {
-			otxn->active = 0;
-		}
-		AST_LIST_INSERT_TAIL(oldlist, txn, list);
-	}
-	AST_LIST_UNLOCK(oldlist);
-
-	return NULL;
-}
-
-static int mark_transaction_active(struct ast_channel *chan, struct odbc_obj *obj)
-{
-	struct ast_datastore *txn_store = ast_channel_datastore_find(chan, &odbc_info, NULL);
-	AST_LIST_HEAD(, odbc_txn_frame) *oldlist;
-	struct odbc_txn_frame *active, *txn;
-
-	if (!txn_store) {
-		return -1;
-	}
-
-	oldlist = txn_store->data;
-	AST_LIST_LOCK(oldlist);
-	AST_LIST_TRAVERSE(oldlist, txn, list) {
-		if (txn->obj == obj) {
-			txn->active = 1;
-			active = txn;
-		} else {
-			txn->active = 0;
-		}
-	}
-	AST_LIST_UNLOCK(oldlist);
-	return active ? 0 : -1;
-}
 
 AST_LIST_HEAD_STATIC(queries, acf_odbc_query);
 
@@ -660,38 +523,6 @@
 	return 0;
 }
 
-static int acf_transaction_read(struct ast_channel *chan, const char *cmd, char *data, char *buf, size_t len)
-{
-	/* TODO */
-	return 0;
-}
-
-static int acf_transaction_write(struct ast_channel *chan, const char *cmd, char *s, const char *value)
-{
-	AST_DECLARE_APP_ARGS(args,
-		AST_APP_ARG(property);
-		AST_APP_ARG(opt);
-	);
-
-	AST_STANDARD_APP_ARGS(args, s);
-	if (strcasecmp(args.property, "transaction") == 0) {
-		/* Set active transaction */
-
-		return 0;
-	} else if (strcasecmp(args.property, "forcecommit") == 0) {
-		/* Set what happens when an uncommitted transaction ends without explicit Commit or Rollback */
-
-		return 0;
-	} else if (strcasecmp(args.property, "isolation") == 0) {
-		/* How do uncommitted transactions affect reads? */
-
-		return 0;
-	} else {
-		ast_log(LOG_ERROR, "Unknown property: '%s'\n", args.property);
-		return -1;
-	}
-}
-
 static struct ast_custom_function escape_function = {
 	.name = "SQL_ESC",
 	.synopsis = "Escapes single ticks for use in SQL statements",
@@ -702,31 +533,6 @@
 "SELECT foo FROM bar WHERE baz='${SQL_ESC(${ARG1})}'\n",
 	.read = acf_escape,
 	.write = NULL,
-};
-
-static struct ast_custom_function odbc_function = {
-	.name = "ODBC",
-	.synopsis = "Sets/retrieves properties supporting database transactions",
-	.syntax = "ODBC(<property>[,<argument>])",
-	.desc =
-"The ODBC() function allows setting several properties to influence how a\n"
-"connected database processes transactions.  The possible properties are as\n"
-"follows:\n"
-"  transaction - gets or sets the active transaction ID.  If set, and the\n"
-"                transaction ID does not exist and a database name is specified\n"
-"                as an argument, it will be created.\n"
-"  forcecommit - controls whether a transaction will be automatically committed\n"
-"                when the channel hangs up.  Defaults to 0.  If a transaction\n"
-"                ID is specified in the optional argument, the property will be\n"
-"                applied to that ID, otherwise to the current active ID.\n"
-"  isolation   - controls the data isolation on uncommitted transactions.  May\n"
-"                be one of the following: read_committed, read_uncommitted,\n"
-"                repeatable_read, or serializable.  Defaults to the database\n"
-"                setting in res_odbc.conf or read_committed if not specified.\n"
-"                If a transaction ID is specified as an optional argument, it\n"
-"                will be applied to that ID, otherwise the current active ID.\n",
-	.read = acf_transaction_read,
-	.write = acf_transaction_write,
 };
 
 static int acf_fetch(struct ast_channel *chan, const char *cmd, char *data, char *buf, size_t len)

Modified: team/tilghman/odbc_tx_support/include/asterisk/res_odbc.h
URL: http://svn.digium.com/view/asterisk/team/tilghman/odbc_tx_support/include/asterisk/res_odbc.h?view=diff&rev=117400&r1=117399&r2=117400
==============================================================================
--- team/tilghman/odbc_tx_support/include/asterisk/res_odbc.h (original)
+++ team/tilghman/odbc_tx_support/include/asterisk/res_odbc.h Tue May 20 22:59:11 2008
@@ -47,7 +47,7 @@
 	unsigned int used:1;            /*!< Is this connection currently in use? */
 	unsigned int up:1;
 	unsigned int tx:1;              /*!< Should this connection be unshared, regardless of the class setting? */
-	unsigned int isolation:2;       /*!< The isolation state of this handle, if it differs from the class */
+	struct odbc_txn_frame *txf;     /*!< Reference back to the transaction frame, if applicable */
 	AST_LIST_ENTRY(odbc_obj) list;
 };
 

Modified: team/tilghman/odbc_tx_support/res/res_odbc.c
URL: http://svn.digium.com/view/asterisk/team/tilghman/odbc_tx_support/res/res_odbc.c?view=diff&rev=117400&r1=117399&r2=117400
==============================================================================
--- team/tilghman/odbc_tx_support/res/res_odbc.c (original)
+++ team/tilghman/odbc_tx_support/res/res_odbc.c Tue May 20 22:59:11 2008
@@ -48,6 +48,7 @@
 #include "asterisk/res_odbc.h"
 #include "asterisk/time.h"
 #include "asterisk/astobj2.h"
+#include "asterisk/app.h"
 
 struct odbc_class
 {
@@ -61,8 +62,8 @@
 	unsigned int haspool:1;              /*!< Boolean - TDS databases need this */
 	unsigned int delme:1;                /*!< Purge the class */
 	unsigned int backslash_is_escape:1;  /*!< On this database, the backslash is a native escape sequence */
-	unsigned int commit_uncommitted:1;   /*!< Should uncommitted transactions be auto-committed on handle release? */
-	unsigned int isolation:2;            /*!< Flags for how the DB should deal with data in uncommitted transactions */
+	unsigned int forcecommit:1;          /*!< Should uncommitted transactions be auto-committed on handle release? */
+	unsigned int isolation:2;            /*!< Flags for how the DB should deal with data in other, uncommitted transactions */
 	unsigned int limit;                  /*!< 1023 wasn't enough for some people */
 	unsigned int count;                  /*!< Running count of pooled connections */
 	unsigned int idlecheck;              /*!< Recheck the connection if it is idle for this long */
@@ -74,6 +75,216 @@
 static odbc_status odbc_obj_connect(struct odbc_obj *obj);
 static odbc_status odbc_obj_disconnect(struct odbc_obj *obj);
 static int odbc_register_class(struct odbc_class *class, int connect);
+static void odbc_txn_free(void *data);
+
+static struct ast_datastore_info txn_info = {
+	.type = "ODBC_Transaction",
+	.destroy = odbc_txn_free,
+};
+
+struct odbc_txn_frame {
+	AST_LIST_ENTRY(odbc_txn_frame) list;
+	/*!\brief Status of this transaction
+	 * This is important only when considering what happens when an uncommitted
+	 * transaction is released without an explicit Commit or Rollback.  That
+	 * action is decided by the forcecommit flag, below.
+	 */
+	enum { ODBC_STATUS_TXCLOSED = 0, ODBC_STATUS_TXOPEN = 1 } status;
+	struct ast_channel *owner;
+	struct odbc_obj *obj;        /*!< Database handle within which transacted statements are run */
+	/*!\brief Is this record the current active transaction within the channel?
+	 * Note that the active flag is really only necessary for statements which
+	 * are triggered from the dialplan, as there isn't a direct corelation
+	 * between multiple statements.  Applications wishing to use transactions
+	 * may simply perform each statement on the same odbc_obj, which keeps the
+	 * transaction persistent.
+	 */
+	unsigned int active:1;
+	unsigned int forcecommit:1;     /*!< Should uncommitted transactions be auto-committed on handle release? */
+	unsigned int isolation:2;       /*!< Flags for how the DB should deal with data in other, uncommitted transactions */
+	char name[0];                   /*!< Name of this transaction ID */
+};
+
+static const char *isolation2text(int iso)
+{
+	if (iso == 0) {
+		return "read_committed";
+	} else if (iso == 1) {
+		return "read_uncommitted";
+	} else if (iso == 2) {
+		return "serializable";
+	} else if (iso == 3) {
+		return "repeatable_read";
+	} else {
+		return "unknown";
+	}
+}
+
+static int text2isolation(const char *txt)
+{
+	if (strncasecmp(txt, "read_", 5) == 0) {
+		if (strncasecmp(txt + 5, "c", 1) == 0) {
+			return 0;
+		} else if (strncasecmp(txt + 5, "u", 1) == 0) {
+			return 1;
+		} else {
+			return -1;
+		}
+	} else if (strncasecmp(txt, "ser", 3) == 0) {
+		return 2;
+	} else if (strncasecmp(txt, "rep", 3) == 0) {
+		return 3;
+	} else {
+		return -1;
+	}
+}
+
+static struct odbc_txn_frame *find_transaction(struct ast_channel *chan, struct odbc_obj *obj, const char *name, int active)
+{
+	struct ast_datastore *txn_store;
+	AST_LIST_HEAD(, odbc_txn_frame) *oldlist;
+	struct odbc_txn_frame *txn = NULL;
+
+	if (!chan && obj && obj->txf && obj->txf->owner) {
+		chan = obj->txf->owner;
+	} else {
+		/* No channel == no transaction */
+		return NULL;
+	}
+
+	if ((txn_store = ast_channel_datastore_find(chan, &txn_info, NULL))) {
+		oldlist = txn_store->data;
+	} else {
+		/* Need to create a new datastore */
+		if (!(txn_store = ast_channel_datastore_alloc(&txn_info, NULL))) {
+			ast_log(LOG_ERROR, "Unable to allocate a new datastore.  Cannot create a new transaction.\n");
+			return NULL;
+		}
+
+		if (!(oldlist = ast_calloc(1, sizeof(*oldlist)))) {
+			ast_log(LOG_ERROR, "Unable to allocate datastore list head.  Cannot create a new transaction.\n");
+			ast_channel_datastore_free(txn_store);
+			return NULL;
+		}
+
+		txn_store->data = oldlist;
+		AST_LIST_HEAD_INIT(oldlist);
+		ast_channel_datastore_add(chan, txn_store);
+	}
+
+	AST_LIST_LOCK(oldlist);
+
+	/* Scanning for an object is *fast*.  Scanning for a name is much slower. */
+	if (obj != NULL || active == 1) {
+		AST_LIST_TRAVERSE(oldlist, txn, list) {
+			if (txn->obj == obj || txn->active) {
+				AST_LIST_UNLOCK(oldlist);
+				return txn;
+			}
+		}
+	}
+
+	if (name != NULL) {
+		AST_LIST_TRAVERSE(oldlist, txn, list) {
+			if (!strcasecmp(txn->name, name)) {
+				AST_LIST_UNLOCK(oldlist);
+				return txn;
+			}
+		}
+	}
+
+	/* Nothing found, create one */
+	if (name && obj && (txn = ast_calloc(1, sizeof(*txn) + strlen(name) + 1))) {
+		struct odbc_txn_frame *otxn;
+
+		strcpy(txn->name, name); /* SAFE */
+		txn->obj = obj;
+		txn->isolation = obj->parent->isolation;
+		txn->forcecommit = obj->parent->forcecommit;
+		txn->active = 1;
+
+		/* On creation, the txn becomes active, and all others inactive */
+		AST_LIST_TRAVERSE(oldlist, otxn, list) {
+			otxn->active = 0;
+		}
+		AST_LIST_INSERT_TAIL(oldlist, txn, list);
+	}
+	AST_LIST_UNLOCK(oldlist);
+
+	return txn;
+}
+
+static void release_transaction(struct odbc_txn_frame *tx)
+{
+	struct ast_channel *chan;
+	struct ast_datastore *txn_store;
+	AST_LIST_HEAD(, odbc_txn_frame) *oldlist;
+
+	if (tx && tx->owner) {
+		chan = tx->owner;
+	} else {
+		/* No channel == no transaction */
+		return;
+	}
+
+	if (!(txn_store = ast_channel_datastore_find(chan, &txn_info, NULL))) {
+		return;
+	}
+
+	oldlist = txn_store->data;
+	AST_LIST_LOCK(oldlist);
+	AST_LIST_REMOVE(oldlist, tx, list);
+	AST_LIST_UNLOCK(oldlist);
+	if (tx->obj) {
+		/* Have to remove this link, first, or the destructors will recursively call each other */
+		tx->obj->txf = NULL;
+		ast_odbc_release_obj(tx->obj);
+		tx->obj = NULL;
+	}
+	tx->owner = NULL;
+	ast_free(tx);
+}
+
+static void odbc_txn_free(void *vdata)
+{
+	struct odbc_txn_frame *tx;
+	AST_LIST_HEAD(, odbc_txn_frame) *oldlist = vdata;
+	AST_LIST_LOCK(oldlist);
+	while ((tx = AST_LIST_REMOVE_HEAD(oldlist, list))) {
+		release_transaction(tx);
+	}
+	AST_LIST_UNLOCK(oldlist);
+	AST_LIST_HEAD_DESTROY(oldlist);
+	ast_free(oldlist);
+}
+
+static int mark_transaction_active(struct ast_channel *chan, struct odbc_txn_frame *tx)
+{
+	struct ast_datastore *txn_store;
+	AST_LIST_HEAD(, odbc_txn_frame) *oldlist;
+	struct odbc_txn_frame *active, *txn;
+
+	if (!chan && tx && tx->owner) {
+		chan = tx->owner;
+	}
+
+	if (!(txn_store = ast_channel_datastore_find(chan, &txn_info, NULL))) {
+		return -1;
+	}
+
+	oldlist = txn_store->data;
+	AST_LIST_LOCK(oldlist);
+	AST_LIST_TRAVERSE(oldlist, txn, list) {
+		if (txn == tx) {
+			txn->active = 1;
+			active = txn;
+		} else {
+			txn->active = 0;
+		}
+	}
+	AST_LIST_UNLOCK(oldlist);
+	return active ? 0 : -1;
+}
 
 static void odbc_class_destructor(void *data)
 {
@@ -247,7 +458,7 @@
 	struct ast_variable *v;
 	char *cat;
 	const char *dsn, *username, *password, *sanitysql;
-	int enabled, pooling, limit, bse, commit_uncommitted, isolation;
+	int enabled, pooling, limit, bse, forcecommit, isolation;
 	unsigned int idlecheck;
 	int connect = 0, res = 0;
 	struct ast_flags config_flags = { 0 };
@@ -273,7 +484,7 @@
 			pooling = 0;
 			limit = 0;
 			bse = 1;
-			commit_uncommitted = isolation = 0;
+			forcecommit = isolation = 0;
 			for (v = ast_variable_browse(config, cat); v; v = v->next) {
 				if (!strcasecmp(v->name, "pooling")) {
 					if (ast_true(v->value))
@@ -308,19 +519,12 @@
 					sanitysql = v->value;
 				} else if (!strcasecmp(v->name, "backslash_is_escape")) {
 					bse = ast_true(v->value);
-				} else if (!strcasecmp(v->name, "commit_uncommitted")) {
-					commit_uncommitted = ast_true(v->value);
+				} else if (!strcasecmp(v->name, "forcecommit")) {
+					forcecommit = ast_true(v->value);
 				} else if (!strcasecmp(v->name, "isolation")) {
-					if (!strcasecmp(v->value, "read_committed")) {
+					if ((isolation = text2isolation(v->value)) < 0) {
+						ast_log(LOG_ERROR, "Unrecognized value for 'isolation': '%s' in section '%s'\n", v->value, cat);
 						isolation = 0;
-					} else if (!strcasecmp(v->value, "read_uncommitted")) {
-						isolation = 1;
-					} else if (!strcasecmp(v->value, "repeatable_read")) {
-						isolation = 2;
-					} else if (!strcasecmp(v->value, "serializable")) {
-						isolation = 3;
-					} else {
-						ast_log(LOG_ERROR, "Unrecognized value for 'isolation': '%s' in section '%s'\n", v->value, cat);
 					}
 				}
 			}
@@ -355,7 +559,7 @@
 				}
 
 				new->backslash_is_escape = bse ? 1 : 0;
-				new->commit_uncommitted = commit_uncommitted ? 1 : 0;
+				new->forcecommit = forcecommit ? 1 : 0;
 				new->isolation = isolation;
 				new->idlecheck = idlecheck;
 
@@ -484,20 +688,58 @@
 
 void ast_odbc_release_obj(struct odbc_obj *obj)
 {
-	if (odbc->tx) {
-		/* TODO Find the datastore with this handle and check if the transaction is uncommitted */
-		SQLEndTrans(SQL_HANDLE_DBC, obj->con, obj->parent->commit_uncommitted ? SQL_COMMIT : SQL_ROLLBACK);
+	struct odbc_txn_frame *tx = find_transaction(NULL, obj, NULL, 0);
+
+	if (tx) {
+		SQLEndTran(SQL_HANDLE_DBC, obj->con, obj->parent->forcecommit ? SQL_COMMIT : SQL_ROLLBACK);
 	}
 
 	/* For pooled connections, this frees the connection to be
 	 * reused.  For non-pooled connections, it does nothing. */
 	obj->used = 0;
-	ao2_ref(obj, -1);
+	if (obj->txf) {
+		/* We'll get called a second time, with txf set to NULL */
+		release_transaction(obj->txf);
+	} else {
+		ao2_ref(obj, -1);
+	}
 }
 
 int ast_odbc_backslash_is_escape(struct odbc_obj *obj)
 {
 	return obj->parent->backslash_is_escape;
+}
+
+static int commit_exec(struct ast_channel *chan, void *data)
+{
+	struct odbc_txn_frame *tx;
+	if (ast_strlen_zero(data)) {
+		tx = find_transaction(chan, NULL, NULL, 1);
+	} else {
+		tx = find_transaction(chan, NULL, data, 0);
+	}
+
+	if (tx) {
+		SQLEndTran(SQL_HANDLE_DBC, tx->obj->con, SQL_COMMIT);
+		tx->status = ODBC_STATUS_TXCLOSED;
+	}
+	return 0;
+}
+
+static int rollback_exec(struct ast_channel *chan, void *data)
+{
+	struct odbc_txn_frame *tx;
+	if (ast_strlen_zero(data)) {
+		tx = find_transaction(chan, NULL, NULL, 1);
+	} else {
+		tx = find_transaction(chan, NULL, data, 0);
+	}
+
+	if (tx) {
+		SQLEndTran(SQL_HANDLE_DBC, tx->obj->con, SQL_ROLLBACK);
+		tx->status = ODBC_STATUS_TXCLOSED;
+	}
+	return 0;
 }
 
 struct odbc_obj *ast_odbc_request_obj2(const char *name, struct ast_flags flags)
@@ -548,12 +790,46 @@
 			}
 		}
 	} else if (ast_test_flag(&flags, RES_ODBC_INDEPENDENT_CONNECTION)) {
-		/* Non-pooled connections do not support independent connections */
-		return NULL;
+		/* Non-pooled connections -- but must use a separate connection handle */
+		aoi = ao2_iterator_init(class->obj_container, 0);
+		while ((obj = ao2_iterator_next(&aoi))) {
+			if (obj->tx && !obj->used) {
+				/* Transactional connection */
+				obj->used++;
+				break;
+			}
+			ao2_ref(obj, -1);
+		}
+
+		if (!obj) {
+			class->count++;
+			obj = ao2_alloc(sizeof(*obj), odbc_obj_destructor);
+			if (!obj) {
+				ao2_ref(class, -1);
+				return NULL;
+			}
+			ast_mutex_init(&obj->lock);
+			/* obj inherits the outstanding reference to class */
+			obj->parent = class;
+			if (odbc_obj_connect(obj) == ODBC_FAIL) {
+				ast_log(LOG_WARNING, "Failed to connect to %s\n", name);
+				ao2_ref(obj, -1);
+				obj = NULL;
+				class->count--;
+			} else {
+				obj->used = 1;
+				ao2_link(class->obj_container, obj);
+			}
+		}
 	} else {
 		/* Non-pooled connection: multiple modules can use the same connection. */
 		aoi = ao2_iterator_init(class->obj_container, 0);
 		while ((obj = ao2_iterator_next(&aoi))) {
+			if (obj->tx) {
+				/* Transactional connection; do not return */
+				ao2_ref(obj, -1);
+				continue;
+			}
 			/* Non-pooled connection: if there is an entry, return it */
 			break;
 		}
@@ -662,6 +938,143 @@
 	ast_mutex_unlock(&obj->lock);
 	return ODBC_SUCCESS;
 }
+
+static int acf_transaction_read(struct ast_channel *chan, const char *cmd, char *data, char *buf, size_t len)
+{
+	AST_DECLARE_APP_ARGS(args,
+		AST_APP_ARG(property);
+		AST_APP_ARG(opt);
+	);
+	struct odbc_txn_frame *tx;
+
+	AST_STANDARD_APP_ARGS(args, data);
+	if (strcasecmp(args.property, "transaction") == 0) {
+		if ((tx = find_transaction(chan, NULL, NULL, 1))) {
+			ast_copy_string(buf, tx->name, len);
+			return 0;
+		}
+	} else if (strcasecmp(args.property, "isolation") == 0) {
+		if (!ast_strlen_zero(args.opt)) {
+			tx = find_transaction(chan, NULL, args.opt, 0);
+		} else {
+			tx = find_transaction(chan, NULL, NULL, 1);
+		}
+		if (tx) {
+			ast_copy_string(buf, isolation2text(tx->isolation), len);
+			return 0;
+		}
+	} else if (strcasecmp(args.property, "forcecommit") == 0) {
+		if (!ast_strlen_zero(args.opt)) {
+			tx = find_transaction(chan, NULL, args.opt, 0);
+		} else {
+			tx = find_transaction(chan, NULL, NULL, 1);
+		}
+		if (tx) {
+			ast_copy_string(buf, tx->forcecommit ? "1" : "0", len);
+			return 0;
+		}
+	}
+	return -1;
+}
+
+static int acf_transaction_write(struct ast_channel *chan, const char *cmd, char *s, const char *value)
+{
+	AST_DECLARE_APP_ARGS(args,
+		AST_APP_ARG(property);
+		AST_APP_ARG(opt);
+	);
+	struct odbc_txn_frame *tx;
+
+	AST_STANDARD_APP_ARGS(args, s);
+	if (strcasecmp(args.property, "transaction") == 0) {
+		/* Set active transaction */
+		struct odbc_obj *obj;
+		if ((tx = find_transaction(chan, NULL, value, 0))) {
+			mark_transaction_active(chan, tx);
+		} else {
+			/* No such transaction, create one */
+			struct ast_flags flags = { RES_ODBC_INDEPENDENT_CONNECTION };
+			if (ast_strlen_zero(args.opt) || !(obj = ast_odbc_request_obj2(args.opt, flags))) {
+				ast_log(LOG_ERROR, "Could not create transaction: invalid database specification '%s'\n", S_OR(args.opt, ""));
+				return -1;
+			}
+			if (!(tx = find_transaction(chan, obj, value, 0))) {
+				return -1;
+			}
+		}
+		return 0;
+	} else if (strcasecmp(args.property, "forcecommit") == 0) {
+		/* Set what happens when an uncommitted transaction ends without explicit Commit or Rollback */
+		if (ast_strlen_zero(args.opt)) {
+			tx = find_transaction(chan, NULL, NULL, 1);
+		} else {
+			tx = find_transaction(chan, NULL, args.opt, 0);
+		}
+		if (ast_true(value)) {
+			tx->forcecommit = 1;
+		} else if (ast_false(value)) {
+			tx->forcecommit = 0;
+		} else {
+			ast_log(LOG_ERROR, "Invalid value for forcecommit: '%s'\n", S_OR(value, ""));
+			return -1;
+		}
+
+		return 0;
+	} else if (strcasecmp(args.property, "isolation") == 0) {
+		/* How do uncommitted transactions affect reads? */
+		int isolation = text2isolation(value);
+		if (ast_strlen_zero(args.opt)) {
+			tx = find_transaction(chan, NULL, NULL, 1);
+		} else {
+			tx = find_transaction(chan, NULL, args.opt, 0);
+		}
+		if (isolation == -1) {
+			ast_log(LOG_ERROR, "Invalid isolation specification: '%s'\n", S_OR(value, ""));
+		} else {
+			tx->isolation = isolation;
+		}
+		return 0;
+	} else {
+		ast_log(LOG_ERROR, "Unknown property: '%s'\n", args.property);
+		return -1;
+	}
+}
+
+static struct ast_custom_function odbc_function = {
+	.name = "ODBC",
+	.synopsis = "Sets/retrieves properties supporting database transactions",
+	.syntax = "ODBC(<property>[,<argument>])",
+	.desc =
+"The ODBC() function allows setting several properties to influence how a\n"
+"connected database processes transactions.  The possible properties are as\n"
+"follows:\n"
+"  transaction - gets or sets the active transaction ID.  If set, and the\n"
+"                transaction ID does not exist and a database name is specified\n"
+"                as an argument, it will be created.\n"
+"  forcecommit - controls whether a transaction will be automatically committed\n"
+"                when the channel hangs up.  Defaults to 0.  If a transaction\n"
+"                ID is specified in the optional argument, the property will be\n"
+"                applied to that ID, otherwise to the current active ID.\n"
+"  isolation   - controls the data isolation on uncommitted transactions.  May\n"
+"                be one of the following: read_committed, read_uncommitted,\n"
+"                repeatable_read, or serializable.  Defaults to the database\n"
+"                setting in res_odbc.conf or read_committed if not specified.\n"
+"                If a transaction ID is specified as an optional argument, it\n"
+"                will be applied to that ID, otherwise the current active ID.\n",
+	.read = acf_transaction_read,
+	.write = acf_transaction_write,
+};
+
+static const char *app_commit = "ODBC_Commit";
+static const char *app_rollback = "ODBC_Rollback";
+static const char *commit_synopsis = "Commit ODBC transaction";
+static const char *rollback_synopsis = "Rollback ODBC transaction";
+static const char commit_descrip[] =
+"  ODBC_Commit([<id>]):\n"
+"Commits the transaction specified by <id> or the active transaction otherwise.\n";
+static const char rollback_descrip[] =
+"  ODBC_Rollback([<id>]:\n"
+"Rollback the transaction specified by <id> or the active transaction otherwise.\n";
 
 static int reload(void)
 {
@@ -739,6 +1152,9 @@
 	if (load_odbc_config() == -1)
 		return AST_MODULE_LOAD_DECLINE;
 	ast_cli_register_multiple(cli_odbc, sizeof(cli_odbc) / sizeof(struct ast_cli_entry));
+	ast_register_application(app_commit, commit_exec, commit_synopsis, commit_descrip);
+	ast_register_application(app_rollback, rollback_exec, rollback_synopsis, rollback_descrip);
+	ast_custom_function_register(&odbc_function);
 	ast_log(LOG_NOTICE, "res_odbc loaded.\n");
 	return 0;
 }




More information about the asterisk-commits mailing list