reformat code
diff --git a/apigee_sync.go b/apigee_sync.go index 10745de..26d8047 100644 --- a/apigee_sync.go +++ b/apigee_sync.go
@@ -39,7 +39,7 @@ if apidInfo.LastSnapshot != "" { snapshot := startOnLocalSnapshot(apidInfo.LastSnapshot) - processSnapshot(snapshot) + listenerMan.processSnapshot(snapshot) events.EmitWithCallback(ApigeeSyncEventSelector, snapshot, func(event apid.Event) { apidChangeManager.pollChangeWithBackoff() })
diff --git a/apigee_sync_test.go b/apigee_sync_test.go index d81ec32..02aac01 100644 --- a/apigee_sync_test.go +++ b/apigee_sync_test.go
@@ -303,7 +303,7 @@ */ It("Should be able to handle duplicate snapshot during bootstrap", func() { initializeContext() - apidTokenManager = createSimpleTokenManager() + apidTokenManager = createTokenManager() apidTokenManager.start() apidSnapshotManager = createSnapShotManager() //events.Listen(ApigeeSyncEventSelector, &handler{})
diff --git a/changes.go b/changes.go index 993613f..6724922 100644 --- a/changes.go +++ b/changes.go
@@ -35,15 +35,21 @@ // 0 for pollChangeWithBackoff() not launched, 1 for launched isLaunched *int32 quitChan chan bool + dbm dbManagerInterface + lm listenerManagerInterface + sm snapShotManagerInterface } -func createChangeManager() *pollChangeManager { +func createChangeManager(dbm dbManagerInterface, lm listenerManagerInterface, sm snapShotManagerInterface) *pollChangeManager { isClosedInt := int32(0) isLaunchedInt := int32(0) return &pollChangeManager{ isClosed: &isClosedInt, quitChan: make(chan bool), isLaunched: &isLaunchedInt, + dbm: dbm, + lm: lm, + sm: sm, } } @@ -69,7 +75,7 @@ go func() { c.quitChan <- true apidTokenManager.close() - <-apidSnapshotManager.close() + <-c.sm.close() log.Debug("change manager closed") finishChan <- false }() @@ -80,7 +86,7 @@ go func() { c.quitChan <- true apidTokenManager.close() - <-apidSnapshotManager.close() + <-c.sm.close() log.Debug("change manager closed") finishChan <- true }() @@ -120,7 +126,7 @@ * Check to see if we have lastSequence already saved in the DB, * in which case, it has to be used to prevent re-reading same data */ - lastSequence = getLastSequence() + lastSequence = c.dbm.getLastSequence() for { select { @@ -150,7 +156,7 @@ log.Debug("polling...") /* Find the scopes associated with the config id */ - scopes := findScopesForId(apidInfo.ClusterID) + scopes := c.dbm.findScopesForId(apidInfo.ClusterID) v := url.Values{} /* Sequence added to the query if available */ @@ -252,7 +258,7 @@ /* If valid data present, Emit to plugins */ if len(resp.Changes) > 0 { - processChangeList(&resp) + c.lm.processChangeList(&resp) select { case <-time.After(httpTimeout): log.Panic("Timeout. Plugins failed to respond to changes.") @@ -262,13 +268,17 @@ log.Debugf("No Changes detected for Scopes: %s", scopes) } - updateSequence(resp.LastSequence) + lastSequence = resp.LastSequence + err = c.dbm.updateLastSequence(resp.LastSequence) + if err != nil { + log.Panic("Unable to update Sequence in DB") + } /* * Check to see if there was any change in scope. If found, handle it * by getting a new snapshot */ - newScopes := findScopesForId(apidInfo.ClusterID) + newScopes := c.dbm.findScopesForId(apidInfo.ClusterID) cs := scopeChanged(newScopes, scopes) if cs != nil { return cs @@ -287,9 +297,9 @@ log.Debugf("handleChangeServerError: changeManager has been closed") return } - if c, ok := err.(changeServerError); ok { - log.Debugf("%s. Fetch a new snapshot to sync...", c.Code) - apidSnapshotManager.downloadDataSnapshot() + if ce, ok := err.(changeServerError); ok { + log.Debugf("%s. Fetch a new snapshot to sync...", ce.Code) + c.sm.downloadDataSnapshot() } else { log.Debugf("Error connecting to changeserver: %v", err) } @@ -332,15 +342,6 @@ return seqCurr.Compare(seqPrev) } -func updateSequence(seq string) { - lastSequence = seq - err := updateLastSequence(seq) - if err != nil { - log.Panic("Unable to update Sequence in DB") - } - -} - /* * Returns nil if the two arrays have matching contents */
diff --git a/data.go b/data.go index 74c69df..4f63058 100644 --- a/data.go +++ b/data.go
@@ -28,8 +28,7 @@ ) var ( - unsafeDB apid.DB - dbMux sync.RWMutex + dbMux sync.RWMutex ) type dataApidCluster struct { @@ -48,8 +47,12 @@ 2. The versioned DB is used for APID_CLUSTER & DATA_SCOPE (Currently, the snapshot never changes, but this is future-proof) */ -func initDB(db apid.DB) error { - _, err := db.Exec(` +func (dbc *dbManager) initDefaultDb() error { + db, err := dbc.getDefaultDb() + if err != nil { + return err + } + _, err = db.Exec(` CREATE TABLE IF NOT EXISTS APID ( instance_id text, apid_cluster_id text, @@ -65,21 +68,30 @@ return nil } -func getDB() apid.DB { - dbMux.RLock() - db := unsafeDB - dbMux.RUnlock() - return db +type dbManager struct { + db apid.DB + dbMux sync.RWMutex } -func setDB(db apid.DB) { +func (dbc *dbManager) getDb() apid.DB { + dbc.dbMux.RLock() + defer dbc.dbMux.RUnlock() + return dbc.db +} + +func (dbc *dbManager) getDefaultDb() (apid.DB, error) { + db, err := dataService.DB() + return db, err +} + +func (dbc *dbManager) setDb(db apid.DB) { dbMux.Lock() - unsafeDB = db - dbMux.Unlock() + defer dbMux.Unlock() + dbc.db = db } //TODO if len(rows) > 1000, chunk it up and exec multiple inserts in the txn -func insert(tableName string, rows []common.Row, txn *sql.Tx) bool { +func (dbc *dbManager) insert(tableName string, rows []common.Row, txn *sql.Tx) bool { if len(rows) == 0 { return false @@ -133,8 +145,8 @@ return values } -func _delete(tableName string, rows []common.Row, txn *sql.Tx) bool { - pkeys, err := getPkeysForTable(tableName) +func (dbc *dbManager) deleteRowsFromTable(tableName string, rows []common.Row, txn *sql.Tx) bool { + pkeys, err := dbc.getPkeysForTable(tableName) sort.Strings(pkeys) if len(pkeys) == 0 || err != nil { log.Errorf("DELETE No primary keys found for table. %s", tableName) @@ -200,8 +212,8 @@ } -func update(tableName string, oldRows, newRows []common.Row, txn *sql.Tx) bool { - pkeys, err := getPkeysForTable(tableName) +func (dbc *dbManager) update(tableName string, oldRows, newRows []common.Row, txn *sql.Tx) bool { + pkeys, err := dbc.getPkeysForTable(tableName) if len(pkeys) == 0 || err != nil { log.Errorf("UPDATE No primary keys found for table.", tableName) return false @@ -333,8 +345,8 @@ return sql } -func getPkeysForTable(tableName string) ([]string, error) { - db := getDB() +func (dbc *dbManager) getPkeysForTable(tableName string) ([]string, error) { + db := dbc.getDb() normalizedTableName := normalizeTableName(tableName) sql := "SELECT columnName FROM _transicator_tables WHERE tableName=$1 AND primaryKey ORDER BY columnName;" rows, err := db.Query(sql, normalizedTableName) @@ -367,12 +379,12 @@ * For the given apidConfigId, this function will retrieve all the distinch scopes * associated with it. Distinct, because scope is already a collection of the tenants. */ -func findScopesForId(configId string) (scopes []string) { +func (dbc *dbManager) findScopesForId(configId string) (scopes []string) { log.Debugf("findScopesForId: %s", configId) var scope sql.NullString - db := getDB() + db := dbc.getDb() query := ` SELECT scope FROM edgex_data_scope WHERE apid_cluster_id = $1 @@ -402,9 +414,9 @@ /* * Retrieve SnapshotInfo for the given apidConfigId from apid_config table */ -func getLastSequence() (lastSequence string) { +func (dbc *dbManager) getLastSequence() (lastSequence string) { - err := getDB().QueryRow("select last_sequence from EDGEX_APID_CLUSTER LIMIT 1").Scan(&lastSequence) + err := dbc.getDb().QueryRow("select last_sequence from EDGEX_APID_CLUSTER LIMIT 1").Scan(&lastSequence) if err != nil && err != sql.ErrNoRows { log.Panicf("Failed to query EDGEX_APID_CLUSTER: %v", err) return @@ -418,11 +430,11 @@ * Persist the last change Id each time a change has been successfully * processed by the plugin(s) */ -func updateLastSequence(lastSequence string) error { +func (dbc *dbManager) updateLastSequence(lastSequence string) error { log.Debugf("updateLastSequence: %s", lastSequence) - stmt, err := getDB().Prepare("UPDATE EDGEX_APID_CLUSTER SET last_sequence=$1;") + stmt, err := dbc.getDb().Prepare("UPDATE EDGEX_APID_CLUSTER SET last_sequence=$1;") if err != nil { log.Errorf("UPDATE EDGEX_APID_CLUSTER Failed: %v", err) return err @@ -440,7 +452,7 @@ return nil } -func getApidInstanceInfo() (info apidInstanceInfo, err error) { +func (dbc *dbManager) getApidInstanceInfo() (info apidInstanceInfo, err error) { info.InstanceName = config.GetString(configName) info.ClusterID = config.GetString(configApidClusterId) @@ -482,10 +494,10 @@ return } -func updateApidInstanceInfo() error { +func (dbc *dbManager) updateApidInstanceInfo() error { // always use default database for this - db, err := dataService.DB() + db, err := dbc.getDefaultDb() if err != nil { return err }
diff --git a/init.go b/init.go index 245d4af..e4be5aa 100644 --- a/init.go +++ b/init.go
@@ -50,11 +50,12 @@ events apid.EventsService apidInfo apidInstanceInfo newInstanceID bool - apidTokenManager tokenManager - apidChangeManager changeManager - apidSnapshotManager snapShotManager + apidTokenManager TokenManagerInterface + apidChangeManager changeManagerInterface + apidSnapshotManager snapShotManagerInterface httpclient *http.Client - + dbMan dbManagerInterface + listenerMan listenerManagerInterface /* Set during post plugin initialization * set this as a default, so that it's guaranteed to be valid even if postInitPlugins isn't called */ @@ -102,18 +103,22 @@ }, } - // set up default database - db, err := dataService.DB() - if err != nil { - return fmt.Errorf("Unable to access DB: %v", err) - } - err = initDB(db) - if err != nil { - return fmt.Errorf("Unable to access DB: %v", err) - } - setDB(db) + dbMan = &dbManager{} - apidInfo, err = getApidInstanceInfo() + // set up default database + err := dbMan.initDefaultDb() + if err != nil { + return fmt.Errorf("Unable to access DB: %v", err) + } + + db, err := dbMan.getDefaultDb() + if err != nil { + return fmt.Errorf("Unable to access DB: %v", err) + } + + dbMan.setDb(db) + + apidInfo, err = dbMan.getApidInstanceInfo() if err != nil { return fmt.Errorf("Unable to get apid instance info: %v", err) } @@ -127,9 +132,12 @@ } func createManagers() { - apidSnapshotManager = createSnapShotManager() - apidChangeManager = createChangeManager() - apidTokenManager = createSimpleTokenManager() + listenerMan = &listenerManager{ + dbm: dbMan, + } + apidSnapshotManager = createSnapShotManager(dbMan, listenerMan) + apidChangeManager = createChangeManager(dbMan, listenerMan, apidSnapshotManager) + apidTokenManager = createTokenManager(dbMan) } func checkForRequiredValues() error {
diff --git a/listener.go b/listener.go index 9c002f2..c269b8f 100644 --- a/listener.go +++ b/listener.go
@@ -24,7 +24,11 @@ LISTENER_TABLE_DATA_SCOPE = "edgex.data_scope" ) -func processSnapshot(snapshot *common.Snapshot) { +type listenerManager struct { + dbm dbManagerInterface +} + +func (lm *listenerManager) processSnapshot(snapshot *common.Snapshot) { log.Debugf("Snapshot received. Switching to DB version: %s", snapshot.SnapshotInfo) db, err := dataService.DBVersion(snapshot.SnapshotInfo) @@ -32,21 +36,21 @@ log.Panicf("Unable to access database: %v", err) } - processSqliteSnapshot(db) + lm.processSqliteSnapshot(db) //update apid instance info apidInfo.LastSnapshot = snapshot.SnapshotInfo - err = updateApidInstanceInfo() + err = lm.dbm.updateApidInstanceInfo() if err != nil { log.Panicf("Unable to update instance info: %v", err) } - setDB(db) + lm.dbm.setDb(db) log.Debugf("Snapshot processed: %s", snapshot.SnapshotInfo) } -func processSqliteSnapshot(db apid.DB) { +func (lm *listenerManager) processSqliteSnapshot(db apid.DB) { var numApidClusters int apidClusters, err := db.Query("SELECT COUNT(*) FROM edgex_apid_cluster") @@ -74,11 +78,11 @@ } } -func processChangeList(changes *common.ChangeList) bool { +func (lm *listenerManager) processChangeList(changes *common.ChangeList) bool { ok := false - tx, err := getDB().Begin() + tx, err := lm.dbm.getDb().Begin() if err != nil { log.Panicf("Error processing ChangeList: %v", err) return ok @@ -93,14 +97,14 @@ } switch change.Operation { case common.Insert: - ok = insert(change.Table, []common.Row{change.NewRow}, tx) + 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 = update(change.Table, []common.Row{change.OldRow}, []common.Row{change.NewRow}, tx) + ok = lm.dbm.update(change.Table, []common.Row{change.OldRow}, []common.Row{change.NewRow}, tx) case common.Delete: - ok = _delete(change.Table, []common.Row{change.OldRow}, tx) + ok = lm.dbm.deleteRowsFromTable(change.Table, []common.Row{change.OldRow}, tx) } if !ok { log.Error("Sql Operation error. Operation rollbacked")
diff --git a/managerInterfaces.go b/managerInterfaces.go index 5022bdd..53a4492 100644 --- a/managerInterfaces.go +++ b/managerInterfaces.go
@@ -15,11 +15,13 @@ package apidApigeeSync import ( + "database/sql" + "github.com/30x/apid-core" "github.com/apigee-labs/transicator/common" "net/url" ) -type tokenManager interface { +type TokenManagerInterface interface { getBearerToken() string invalidateToken() error getToken() *OauthToken @@ -28,7 +30,7 @@ start() } -type snapShotManager interface { +type snapShotManagerInterface interface { close() <-chan bool downloadBootSnapshot() storeBootSnapshot(snapshot *common.Snapshot) @@ -37,7 +39,29 @@ downloadSnapshot(isBoot bool, scopes []string, snapshot *common.Snapshot) error } -type changeManager interface { +type changeManagerInterface interface { close() <-chan bool pollChangeWithBackoff() } + +type dbManagerInterface interface { + initDefaultDb() error + setDb(apid.DB) + getDb() apid.DB + insert(tableName string, rows []common.Row, txn *sql.Tx) bool + deleteRowsFromTable(tableName string, rows []common.Row, txn *sql.Tx) bool + update(tableName string, oldRows, newRows []common.Row, txn *sql.Tx) bool + getPkeysForTable(tableName string) ([]string, error) + findScopesForId(configId string) (scopes []string) + getDefaultDb() (apid.DB, error) + updateApidInstanceInfo() error + getApidInstanceInfo() (info apidInstanceInfo, err error) + getLastSequence() string + updateLastSequence(lastSequence string) error +} + +type listenerManagerInterface interface { + processSnapshot(snapshot *common.Snapshot) + processSqliteSnapshot(db apid.DB) + processChangeList(changes *common.ChangeList) bool +}
diff --git a/snapshot.go b/snapshot.go index 3040b24..d7fbacd 100644 --- a/snapshot.go +++ b/snapshot.go
@@ -29,7 +29,7 @@ "time" ) -type simpleSnapShotManager struct { +type snapShotManager struct { // to send quit signal to the downloading thread quitChan chan bool // to mark the graceful close of snapshotManager @@ -38,16 +38,20 @@ isClosed *int32 // make sure close() returns immediately if there's no downloading/processing snapshot isDownloading *int32 + lm listenerManagerInterface + dbm dbManagerInterface } -func createSnapShotManager() *simpleSnapShotManager { +func createSnapShotManager(dbm dbManagerInterface, lm listenerManagerInterface) *snapShotManager { isClosedInt := int32(0) isDownloadingInt := int32(0) - return &simpleSnapShotManager{ + return &snapShotManager{ quitChan: make(chan bool, 1), finishChan: make(chan bool, 1), isClosed: &isClosedInt, isDownloading: &isDownloadingInt, + dbm: dbm, + lm: lm, } } @@ -57,7 +61,7 @@ * use <- close() for blocking close * should only be called by pollChangeManager, because pollChangeManager is dependent on it */ -func (s *simpleSnapShotManager) close() <-chan bool { +func (s *snapShotManager) close() <-chan bool { //has been closed before if atomic.SwapInt32(s.isClosed, 1) == int32(1) { log.Error("snapShotManager: close() called on a closed snapShotManager!") @@ -77,7 +81,7 @@ } // retrieve boot information: apid_config and apid_config_scope -func (s *simpleSnapShotManager) downloadBootSnapshot() { +func (s *snapShotManager) downloadBootSnapshot() { if atomic.SwapInt32(s.isDownloading, 1) == int32(1) { log.Panic("downloadBootSnapshot: only 1 thread can download snapshot at the same time!") } @@ -113,12 +117,12 @@ s.storeBootSnapshot(snapshot) } -func (s *simpleSnapShotManager) storeBootSnapshot(snapshot *common.Snapshot) { - processSnapshot(snapshot) +func (s *snapShotManager) storeBootSnapshot(snapshot *common.Snapshot) { + s.lm.processSnapshot(snapshot) } // use the scope IDs from the boot snapshot to get all the data associated with the scopes -func (s *simpleSnapShotManager) downloadDataSnapshot() { +func (s *snapShotManager) downloadDataSnapshot() { if atomic.SwapInt32(s.isDownloading, 1) == int32(1) { log.Panic("downloadDataSnapshot: only 1 thread can download snapshot at the same time!") } @@ -132,7 +136,7 @@ log.Debug("download Snapshot for data scopes") - scopes := findScopesForId(apidInfo.ClusterID) + scopes := s.dbm.findScopesForId(apidInfo.ClusterID) scopes = append(scopes, apidInfo.ClusterID) snapshot := &common.Snapshot{} err := s.downloadSnapshot(false, scopes, snapshot) @@ -146,7 +150,7 @@ s.storeDataSnapshot(snapshot) } -func (s *simpleSnapShotManager) storeDataSnapshot(snapshot *common.Snapshot) { +func (s *snapShotManager) storeDataSnapshot(snapshot *common.Snapshot) { knownTables = extractTablesFromSnapshot(snapshot) _, err := dataService.DBVersion(snapshot.SnapshotInfo) @@ -154,7 +158,7 @@ log.Panicf("Database inaccessible: %v", err) } - processSnapshot(snapshot) + s.lm.processSnapshot(snapshot) log.Info("Emitting Snapshot to plugins") select { @@ -226,7 +230,7 @@ // a blocking method // will keep retrying with backoff until success -func (s *simpleSnapShotManager) downloadSnapshot(isBoot bool, scopes []string, snapshot *common.Snapshot) error { +func (s *snapShotManager) downloadSnapshot(isBoot bool, scopes []string, snapshot *common.Snapshot) error { // if closed if atomic.LoadInt32(s.isClosed) == int32(1) { log.Warn("Trying to download snapshot with a closed snapShotManager")
diff --git a/token.go b/token.go index 1612025..f10fff8 100644 --- a/token.go +++ b/token.go
@@ -39,10 +39,10 @@ man.close() */ -func createSimpleTokenManager() *simpleTokenManager { +func createTokenManager(dbm dbManagerInterface) *tokenManager { isClosedInt := int32(0) - t := &simpleTokenManager{ + t := &tokenManager{ quitPollingForToken: make(chan bool, 1), closed: make(chan bool), getTokenChan: make(chan bool), @@ -50,11 +50,12 @@ returnTokenChan: make(chan *OauthToken), invalidateDone: make(chan bool), isClosed: &isClosedInt, + dbm: dbm, } return t } -type simpleTokenManager struct { +type tokenManager struct { token *OauthToken isClosed *int32 quitPollingForToken chan bool @@ -64,19 +65,20 @@ refreshTimer <-chan time.Time returnTokenChan chan *OauthToken invalidateDone chan bool + dbm dbManagerInterface } -func (t *simpleTokenManager) start() { +func (t *tokenManager) start() { t.retrieveNewToken() t.refreshTimer = time.After(t.token.refreshIn()) go t.maintainToken() } -func (t *simpleTokenManager) getBearerToken() string { +func (t *tokenManager) getBearerToken() string { return t.getToken().AccessToken } -func (t *simpleTokenManager) maintainToken() { +func (t *tokenManager) maintainToken() { for { select { case <-t.closed: @@ -97,7 +99,7 @@ } // will block until valid -func (t *simpleTokenManager) invalidateToken() error { +func (t *tokenManager) invalidateToken() error { //has been closed if atomic.LoadInt32(t.isClosed) == int32(1) { log.Debug("TokenManager: invalidateToken() called on closed tokenManager") @@ -109,7 +111,7 @@ return nil } -func (t *simpleTokenManager) getToken() *OauthToken { +func (t *tokenManager) getToken() *OauthToken { //has been closed if atomic.LoadInt32(t.isClosed) == int32(1) { log.Debug("TokenManager: getToken() called on closed tokenManager") @@ -123,7 +125,7 @@ * blocking close() of tokenMan */ -func (t *simpleTokenManager) close() { +func (t *tokenManager) close() { //has been closed if atomic.SwapInt32(t.isClosed, 1) == int32(1) { log.Panic("TokenManager: close() has been called before!") @@ -138,7 +140,7 @@ } // don't call externally. will block until success. -func (t *simpleTokenManager) retrieveNewToken() { +func (t *tokenManager) retrieveNewToken() { log.Debug("Getting OAuth token...") uriString := config.GetString(configProxyServerBaseURI) @@ -151,7 +153,7 @@ pollWithBackoff(t.quitPollingForToken, t.getRetrieveNewTokenClosure(uri), func(err error) { log.Errorf("Error getting new token : ", err) }) } -func (t *simpleTokenManager) getRetrieveNewTokenClosure(uri *url.URL) func(chan bool) error { +func (t *tokenManager) getRetrieveNewTokenClosure(uri *url.URL) func(chan bool) error { return func(_ chan bool) error { form := url.Values{} form.Set("grant_type", "client_credentials") @@ -208,7 +210,7 @@ if newInstanceID { newInstanceID = false - err = updateApidInstanceInfo() + err = t.dbm.updateApidInstanceInfo() if err != nil { log.Errorf("unable to unmarshal update apid instance info : %v", string(body), err) return err
diff --git a/token_test.go b/token_test.go index 85b71c7..66cd189 100644 --- a/token_test.go +++ b/token_test.go
@@ -94,7 +94,7 @@ w.Write(body) })) config.Set(configProxyServerBaseURI, ts.URL) - testedTokenManager := createSimpleTokenManager() + testedTokenManager := createTokenManager() testedTokenManager.start() token := testedTokenManager.getToken() @@ -123,7 +123,7 @@ })) config.Set(configProxyServerBaseURI, ts.URL) - testedTokenManager := createSimpleTokenManager() + testedTokenManager := createTokenManager() testedTokenManager.start() token := testedTokenManager.getToken() Expect(token.AccessToken).ToNot(BeEmpty()) @@ -163,7 +163,7 @@ })) config.Set(configProxyServerBaseURI, ts.URL) - testedTokenManager := createSimpleTokenManager() + testedTokenManager := createTokenManager() testedTokenManager.start() testedTokenManager.getToken() @@ -204,7 +204,7 @@ })) config.Set(configProxyServerBaseURI, ts.URL) - testedTokenManager := createSimpleTokenManager() + testedTokenManager := createTokenManager() testedTokenManager.start() testedTokenManager.getToken() testedTokenManager.invalidateToken()