Use defer to complete/rollback txn.
diff --git a/data.go b/data.go
index fd4111c..c0f7a31 100644
--- a/data.go
+++ b/data.go
@@ -54,6 +54,9 @@
log.Errorf("initDB(): Unable to get DB tx err: {%v}", err)
return err
}
+
+ defer completeTxn(tx, err)
+
_, err = tx.Exec(`
CREATE TABLE IF NOT EXISTS APID (
instance_id text,
@@ -63,16 +66,9 @@
);
`)
if err != nil {
- rollbackTxn(tx)
log.Errorf("initDB(): Unable to tx exec err: {%v}", err)
return err
}
- err = tx.Commit()
- if err != nil {
- rollbackTxn(tx)
- log.Errorf("initDB(): tx commit err: {%v}", err)
- return err
- }
log.Debug("Database tables created.")
return nil
}
@@ -177,6 +173,7 @@
if err == nil && affected != 0 {
log.Debugf("DELETE Success [%s] values=%v", sql, values)
} else if err == nil && affected == 0 {
+ err = errors.New("Entry not found. Nothing to delete")
log.Errorf("Entry not found [%s] values=%v. Nothing to delete.", sql, values)
return false
} else {
@@ -444,16 +441,11 @@
return err
}
+ defer completeTxn(tx, err)
+
_, err = tx.Exec("UPDATE EDGEX_APID_CLUSTER SET last_sequence=?;", lastSequence)
if err != nil {
log.Errorf("UPDATE EDGEX_APID_CLUSTER Failed: %v", err)
- rollbackTxn(tx)
- return err
- }
- err = tx.Commit()
- if err != nil {
- log.Errorf("UPDATE EDGEX_APID_CLUSTER Tx Commit err : %v", err)
- rollbackTxn(tx)
return err
}
log.Debugf("UPDATE EDGEX_APID_CLUSTER Success: %s", lastSequence)
@@ -477,13 +469,13 @@
log.Errorf("getApidInstanceInfo: Unable to get DB tx Err: {%v}", err)
return
}
+ defer completeTxn(tx, err)
err = tx.QueryRow("SELECT instance_id, apid_cluster_id, last_snapshot_info FROM APID LIMIT 1").
Scan(&info.InstanceID, &savedClusterId, &info.LastSnapshot)
if err != nil {
if err != sql.ErrNoRows {
log.Errorf("Unable to retrieve apidInstanceInfo: %v", err)
- tx.Rollback()
return
} else {
// first start - no row, generate a UUID and store it
@@ -505,14 +497,6 @@
info.InstanceID, info.ClusterID, "")
info.LastSnapshot = ""
}
- if err == nil {
- err = tx.Commit()
- if err != nil {
- rollbackTxn(tx)
- }
- } else {
- rollbackTxn(tx)
- }
return
}
@@ -528,6 +512,9 @@
log.Errorf("updateApidInstanceInfo: Unable to get DB tx Err: {%v}", err)
return err
}
+
+ defer completeTxn(tx, err)
+
rows, err := tx.Exec(`
REPLACE
INTO APID (instance_id, apid_cluster_id, last_snapshot_info)
@@ -535,20 +522,12 @@
apidInfo.InstanceID, apidInfo.ClusterID, apidInfo.LastSnapshot)
if err != nil {
log.Errorf("updateApidInstanceInfo: Tx Exec Err: {%v}", err)
- rollbackTxn(tx)
return err
}
+
n, err := rows.RowsAffected()
if err == nil && n == 0 {
err = errors.New("no rows affected")
- rollbackTxn(tx)
- } else if err == nil {
- err = tx.Commit()
- if err != nil {
- rollbackTxn(tx)
- }
- } else {
- rollbackTxn(tx)
}
return err
diff --git a/listener.go b/listener.go
index 8635a81..265ab67 100644
--- a/listener.go
+++ b/listener.go
@@ -58,8 +58,16 @@
}
}
-func rollbackTxn (tx apid.Tx) {
- err := tx.Rollback()
+func completeTxn (tx apid.Tx, err error) {
+ if err == nil {
+ err = tx.Commit()
+ if err == nil {
+ log.Debugf("Transaction committed successfully")
+ return
+ }
+ log.Errorf("Transaction commit failed with error : {%v}", err)
+ }
+ err = tx.Rollback()
if err != nil {
log.Panicf("Unable to rollback Transaction. DB in inconsistent state. Err {%v}", err)
}
@@ -78,6 +86,8 @@
log.Panicf("Unable to read database: {%s}", err.Error())
}
+ defer completeTxn(tx, err)
+
if numApidClusters != 1 {
log.Panic("Illegal state for apid_cluster. Must be a single row.")
}
@@ -85,17 +95,11 @@
_, err = tx.Exec("ALTER TABLE edgex_apid_cluster ADD COLUMN last_sequence text DEFAULT ''")
if err != nil {
if err.Error() == "duplicate column name: last_sequence" {
- rollbackTxn(tx)
return
} else {
log.Panicf("Unable to create last_sequence column on DB. Error {%v}", err.Error())
}
}
- err = tx.Commit()
- if err != nil {
- rollbackTxn(tx)
- return
- }
}
func processChangeList(changes *common.ChangeList) bool {
@@ -107,7 +111,7 @@
log.Panicf("Error processing ChangeList: %v", err)
return ok
}
- defer tx.Rollback()
+ defer completeTxn(tx, err)
log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes))
@@ -132,11 +136,5 @@
}
}
- err = tx.Commit()
- if err != nil {
- log.Panicf("Error processing ChangeList: %v", err)
- return false
- }
-
return ok
}
