fix defer rollback transaction issues. (#65)

data.go

diff --git a/data.go b/data.go
index 4376a76..3f50df4 100644
--- a/data.go
+++ b/data.go
@@ -54,7 +54,7 @@
 		log.Errorf("initDB(): Unable to get DB tx err: {%v}", err)
 		return err
 	}
-	defer completeTxn(tx, err)
+	defer tx.Rollback()
 	_, err = tx.Exec(`
 	CREATE TABLE IF NOT EXISTS APID (
 	    instance_id text,
@@ -67,6 +67,10 @@
 		log.Errorf("initDB(): Unable to tx exec err: {%v}", err)
 		return err
 	}
+	if err = tx.Commit(); err != nil {
+		log.Errorf("Error when initDb: %v", err)
+		return err
+	}
 	log.Debug("Database tables created.")
 	return nil
 }
@@ -437,14 +441,17 @@
 		log.Errorf("getApidInstanceInfo: Unable to get DB tx Err: {%v}", err)
 		return err
 	}
-	defer completeTxn(tx, err)
+	defer tx.Rollback()
 	_, err = tx.Exec("UPDATE EDGEX_APID_CLUSTER SET last_sequence=?;", lastSequence)
 	if err != nil {
 		log.Errorf("UPDATE EDGEX_APID_CLUSTER Failed: %v", err)
 		return err
 	}
 	log.Debugf("UPDATE EDGEX_APID_CLUSTER Success: %s", lastSequence)
-	return nil
+	if err = tx.Commit(); err != nil {
+		log.Errorf("Commit error in updateLastSequence: %v", err)
+	}
+	return err
 }
 
 func getApidInstanceInfo() (info apidInstanceInfo, err error) {
@@ -464,7 +471,7 @@
 		log.Errorf("getApidInstanceInfo: Unable to get DB tx Err: {%v}", err)
 		return
 	}
-	defer completeTxn(tx, err)
+	defer tx.Rollback()
 	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 {
@@ -491,6 +498,9 @@
 			info.InstanceID, info.ClusterID, "")
 		info.LastSnapshot = ""
 	}
+	if err = tx.Commit(); err != nil {
+		log.Errorf("Commit error in getApidInstanceInfo: %v", err)
+	}
 	return
 }
 
@@ -506,7 +516,7 @@
 		log.Errorf("updateApidInstanceInfo: Unable to get DB tx Err: {%v}", err)
 		return err
 	}
-	defer completeTxn(tx, err)
+	defer tx.Rollback()
 	rows, err := tx.Exec(`
 		REPLACE
 		INTO APID (instance_id, apid_cluster_id, last_snapshot_info)
@@ -520,6 +530,9 @@
 	if err == nil && n == 0 {
 		err = errors.New("no rows affected")
 	}
+	if err = tx.Commit(); err != nil {
+		log.Errorf("Commit error in updateApidInstanceInfo: %v", err)
+	}
 
 	return err
 }

data_test.go

diff --git a/data_test.go b/data_test.go
index 53b7509..691e887 100644
--- a/data_test.go
+++ b/data_test.go
@@ -15,13 +15,13 @@
 package apidApigeeSync
 
 import (
+	"github.com/apid/apid-core"
 	"github.com/apid/apid-core/data"
 	"github.com/apigee-labs/transicator/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"sort"
 	"strconv"
-	"github.com/apid/apid-core"
 )
 
 var _ = Describe("data access tests", func() {

listener.go

diff --git a/listener.go b/listener.go
index e9f24ad..94befd6 100644
--- a/listener.go
+++ b/listener.go
@@ -30,7 +30,7 @@
 	var prevDb string
 	if apidInfo.LastSnapshot != "" && apidInfo.LastSnapshot != snapshot.SnapshotInfo {
 		log.Debugf("Release snapshot for {%s}. Switching to version {%s}",
-			apidInfo.LastSnapshot , snapshot.SnapshotInfo)
+			apidInfo.LastSnapshot, snapshot.SnapshotInfo)
 		prevDb = apidInfo.LastSnapshot
 	} else {
 		log.Debugf("Process snapshot for version {%s}",
@@ -59,22 +59,6 @@
 	}
 }
 
-
-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)
-	}
-}
-
 func processSqliteSnapshot(db apid.DB) {
 
 	var numApidClusters int
@@ -82,7 +66,7 @@
 	if err != nil {
 		log.Panicf("Unable to open DB txn: {%v}", err.Error())
 	}
-	defer completeTxn(tx, err)
+	defer tx.Rollback()
 	err = tx.QueryRow("SELECT COUNT(*) FROM edgex_apid_cluster").Scan(&numApidClusters)
 	if err != nil {
 		log.Panicf("Unable to read database: {%s}", err.Error())
@@ -100,6 +84,9 @@
 			log.Panicf("Unable to create last_sequence column on DB.  Error {%v}", err.Error())
 		}
 	}
+	if err = tx.Commit(); err != nil {
+		log.Errorf("Error when commit in processSqliteSnapshot: %v", err)
+	}
 }
 
 func processChangeList(changes *common.ChangeList) bool {
@@ -109,9 +96,8 @@
 	tx, err := getDB().Begin()
 	if err != nil {
 		log.Panicf("Error processing ChangeList: %v", err)
-		return ok
 	}
-	defer completeTxn(tx, err)
+	defer tx.Rollback()
 
 	log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes))
 
@@ -133,9 +119,13 @@
 		if !ok {
 			err = errors.New("Sql Operation error. Operation rollbacked")
 			log.Error("Sql Operation error. Operation rollbacked")
-			return ok
+			return false
 		}
 	}
 
-	return ok
+	if err = tx.Commit(); err != nil {
+		log.Errorf("Commit error in processChangeList: %v", err)
+		return false
+	}
+	return true
 }