refactor data.go
diff --git a/data.go b/data.go index aeb0203..eca1491 100644 --- a/data.go +++ b/data.go
@@ -516,6 +516,54 @@ return err } +func (dbc *dbManager) getClusterCount() (numApidClusters int, err error) { + rows, err := dbc.db.Query("SELECT COUNT(*) FROM edgex_apid_cluster") + if err != nil { + return + } + defer rows.Close() + rows.Next() + err = rows.Scan(&numApidClusters) + return +} + +func (dbc *dbManager) alterClusterTable() (err error) { + _, err = dbc.db.Exec("ALTER TABLE edgex_apid_cluster ADD COLUMN last_sequence text DEFAULT ''") + if err.Error() == "duplicate column name: last_sequence" { + return nil + } + return +} + +func (dbc *dbManager) writeTransaction(changes *common.ChangeList) bool { + tx, err := dbc.getDb().Begin() + if err != nil { + log.Panicf("Error processing ChangeList: %v", err) + } + defer tx.Rollback() + var ok bool + for _, change := range changes.Changes { + switch change.Operation { + case common.Insert: + ok = dbc.insert(change.Table, []common.Row{change.NewRow}, tx) + case common.Update: + ok = dbc.update(change.Table, []common.Row{change.OldRow}, []common.Row{change.NewRow}, tx) + case common.Delete: + ok = dbc.deleteRowsFromTable(change.Table, []common.Row{change.OldRow}, tx) + } + if !ok { + log.Error("Sql Operation error. Operation rollbacked") + return ok + } + } + err = tx.Commit() + if err != nil { + log.Errorf("Error processing ChangeList: %v", err) + return false + } + return ok +} + /* * generates a random uuid (mix of timestamp & crypto random string) */
diff --git a/data_test.go b/data_test.go index 11a4a6c..e1bf8dc 100644 --- a/data_test.go +++ b/data_test.go
@@ -19,37 +19,44 @@ . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" "sort" + "sync" ) var _ = Describe("data access tests", func() { - + var testCount int + var testDbMan *dbManager BeforeEach(func() { - db := getDB() + testCount += 1 + testDbMan = &dbManager{ + dbMux: sync.RWMutex{}, + } + + db := testDbMan.getDb() //all tests in this file operate on the api_product table. Create the necessary tables for this here - getDB().Exec("CREATE TABLE _transicator_tables " + + db.Exec("CREATE TABLE _transicator_tables " + "(tableName varchar not null, columnName varchar not null, " + "typid integer, primaryKey bool);") - getDB().Exec("DELETE from _transicator_tables") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','id',2950,1)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','tenant_id',1043,1)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','description',1043,0)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','api_resources',1015,0)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','approval_type',1043,0)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','scopes',1015,0)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','proxies',1015,0)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','environments',1015,0)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_at',1114,1)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_by',1043,0)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_at',1114,1)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_by',1043,0)") - getDB().Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','_change_selector',1043,0)") + db.Exec("DELETE from _transicator_tables") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','id',2950,1)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','tenant_id',1043,1)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','description',1043,0)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','api_resources',1015,0)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','approval_type',1043,0)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','scopes',1015,0)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','proxies',1015,0)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','environments',1015,0)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_at',1114,1)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_by',1043,0)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_at',1114,1)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_by',1043,0)") + db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','_change_selector',1043,0)") - getDB().Exec("CREATE TABLE kms_api_product (id text,tenant_id text,name text, description text, " + + db.Exec("CREATE TABLE kms_api_product (id text,tenant_id text,name text, description text, " + "api_resources text,approval_type text,scopes text,proxies text, environments text," + "created_at blob, created_by text,updated_at blob,updated_by text,_change_selector text, " + "primary key (id,tenant_id,created_at,updated_at));") - getDB().Exec("DELETE from kms_api_product") + db.Exec("DELETE from kms_api_product") setDB(db) initDB(db)
diff --git a/init.go b/init.go index 3ad221e..d8226de 100644 --- a/init.go +++ b/init.go
@@ -22,6 +22,7 @@ "time" "github.com/30x/apid-core" + "sync" ) const ( @@ -102,7 +103,9 @@ }, } - dbMan = &dbManager{} + dbMan = &dbManager{ + dbMux: sync.RWMutex{}, + } // set up default database err := dbMan.initDefaultDb()
diff --git a/listener.go b/listener.go index c269b8f..76ec295 100644 --- a/listener.go +++ b/listener.go
@@ -52,71 +52,27 @@ func (lm *listenerManager) processSqliteSnapshot(db apid.DB) { - var numApidClusters int - apidClusters, err := db.Query("SELECT COUNT(*) FROM edgex_apid_cluster") - if err != nil { - log.Panicf("Unable to read database: %s", err.Error()) - } - apidClusters.Next() - err = apidClusters.Scan(&numApidClusters) - if err != nil { - log.Panicf("Unable to read database: %s", err.Error()) + if count, err := lm.dbm.getClusterCount(); err != nil || count != 1 { + log.Panicf("Illegal state for apid_cluster. Must be a single row. %v", err) } - if numApidClusters != 1 { - log.Panic("Illegal state for apid_cluster. Must be a single row.") - } - - _, err = db.Exec("ALTER TABLE edgex_apid_cluster ADD COLUMN last_sequence text DEFAULT ''") - if err != nil { - if err.Error() == "duplicate column name: last_sequence" { - return - } else { - log.Error("[[" + err.Error() + "]]") - log.Panicf("Unable to create last_sequence column on DB. Unrecoverable error ", err) - } + if err := lm.dbm.alterClusterTable(); err != nil { + log.Panicf("Unable to create last_sequence column on DB. Unrecoverable error %v", err) } } func (lm *listenerManager) processChangeList(changes *common.ChangeList) bool { - ok := false - - tx, err := lm.dbm.getDb().Begin() - if err != nil { - log.Panicf("Error processing ChangeList: %v", err) - return ok - } - defer tx.Rollback() - log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes)) for _, change := range changes.Changes { if change.Table == LISTENER_TABLE_APID_CLUSTER { log.Panicf("illegal operation: %s for %s", change.Operation, change.Table) } - switch change.Operation { - case common.Insert: - ok = lm.dbm.insert(change.Table, []common.Row{change.NewRow}, tx) - case common.Update: - if change.Table == LISTENER_TABLE_DATA_SCOPE { - log.Panicf("illegal operation: %s for %s", change.Operation, change.Table) - } - ok = lm.dbm.update(change.Table, []common.Row{change.OldRow}, []common.Row{change.NewRow}, tx) - case common.Delete: - ok = lm.dbm.deleteRowsFromTable(change.Table, []common.Row{change.OldRow}, tx) - } - if !ok { - log.Error("Sql Operation error. Operation rollbacked") - return ok + if change.Operation == common.Update && change.Table == LISTENER_TABLE_DATA_SCOPE { + log.Panicf("illegal operation: %s for %s", change.Operation, change.Table) } } - err = tx.Commit() - if err != nil { - log.Panicf("Error processing ChangeList: %v", err) - return false - } - - return ok + return lm.dbm.writeTransaction(changes) }
diff --git a/managerInterfaces.go b/managerInterfaces.go index 53a4492..e406784 100644 --- a/managerInterfaces.go +++ b/managerInterfaces.go
@@ -58,6 +58,9 @@ getApidInstanceInfo() (info apidInstanceInfo, err error) getLastSequence() string updateLastSequence(lastSequence string) error + getClusterCount() (numApidClusters int, err error) + alterClusterTable() (err error) + writeTransaction(*common.ChangeList) bool } type listenerManagerInterface interface {
diff --git a/token.go b/token.go index 991463c..e21a10d 100644 --- a/token.go +++ b/token.go
@@ -51,6 +51,9 @@ invalidateDone: make(chan bool), isClosed: &isClosedInt, dbm: dbm, + proxyServerBaseUri: config.GetString(configProxyServerBaseURI), + clientId: config.GetString(configConsumerKey), + clientSecret: config.GetString(configConsumerSecret), } return t } @@ -67,6 +70,9 @@ invalidateDone chan bool dbm dbManagerInterface notNewInstanceId bool + proxyServerBaseUri string + clientId string + clientSecret string } func (t *tokenManager) start() { @@ -144,10 +150,10 @@ func (t *tokenManager) retrieveNewToken() { log.Debug("Getting OAuth token...") - uriString := config.GetString(configProxyServerBaseURI) + uriString := t.proxyServerBaseUri uri, err := url.Parse(uriString) if err != nil { - log.Panicf("unable to parse uri config '%s' value: '%s': %v", configProxyServerBaseURI, uriString, err) + log.Panicf("unable to parse uri value: '%s': %v", uriString, err) } uri.Path = path.Join(uri.Path, "/accesstoken") @@ -158,8 +164,8 @@ return func(_ chan bool) error { form := url.Values{} form.Set("grant_type", "client_credentials") - form.Add("client_id", config.GetString(configConsumerKey)) - form.Add("client_secret", config.GetString(configConsumerSecret)) + form.Add("client_id", t.clientId) + form.Add("client_secret", t.clientSecret) req, err := http.NewRequest("POST", uri.String(), bytes.NewBufferString(form.Encode())) req.Header.Set("Content-Type", "application/x-www-form-urlencoded; param=value") req.Header.Set("display_name", apidInfo.InstanceName) @@ -219,7 +225,6 @@ } } t.token = &token - config.Set(configBearerToken, token.AccessToken) return nil }
diff --git a/token_test.go b/token_test.go index 4a0c3e6..e75b54d 100644 --- a/token_test.go +++ b/token_test.go
@@ -221,37 +221,57 @@ func (d *dummyDbMan) initDefaultDb() error { return nil } + func (d *dummyDbMan) setDb(apid.DB) {} + func (d *dummyDbMan) getDb() apid.DB { return nil } + func (d *dummyDbMan) insert(tableName string, rows []common.Row, txn *sql.Tx) bool { return true } + func (d *dummyDbMan) deleteRowsFromTable(tableName string, rows []common.Row, txn *sql.Tx) bool { return true } + func (d *dummyDbMan) update(tableName string, oldRows, newRows []common.Row, txn *sql.Tx) bool { return true } + func (d *dummyDbMan) getPkeysForTable(tableName string) ([]string, error) { return nil, nil } + func (d *dummyDbMan) findScopesForId(configId string) []string { return nil } + func (d *dummyDbMan) getDefaultDb() (apid.DB, error) { return nil, nil } + func (d *dummyDbMan) updateApidInstanceInfo() error { return nil } + func (d *dummyDbMan) getApidInstanceInfo() (info apidInstanceInfo, err error) { return } + func (d *dummyDbMan) getLastSequence() string { return "" } + func (d *dummyDbMan) updateLastSequence(lastSequence string) error { return nil } + +func (d *dummyDbMan) getClusterCount() (int, error) { + return 1, nil +} + +func (d *dummyDbMan) alterClusterTable() error { + return nil +}