Minor updates to logic, tests.
diff --git a/apigee_sync.go b/apigee_sync.go index 25a590f..40335e2 100644 --- a/apigee_sync.go +++ b/apigee_sync.go
@@ -41,8 +41,8 @@ } else { times = pollInterval } - log.Debugf("sleep for %d secs", time.Duration(times)) - time.Sleep(time.Duration(times) * time.Second) + log.Debugf("Connecting to changeserver...") + time.Sleep(time.Duration(times) * 100 * time.Millisecond) } else { // Reset sleep interval times = 1 @@ -151,6 +151,7 @@ */ func getBearerToken() bool { + log.Info("Getting a Bearer token.") uri, err := url.Parse(config.GetString(configProxyServerBaseURI)) if err != nil { log.Error(err) @@ -168,7 +169,7 @@ client := &http.Client{} resp, err := client.Do(req) if err != nil { - log.Error(err) + log.Error("Unable to Connect to Edge Proxy Server ", err) return false } defer resp.Body.Close() @@ -178,7 +179,7 @@ } body, err := ioutil.ReadAll(resp.Body) if err != nil { - log.Error(err) + log.Error("Unable to read EdgeProxy Sever response ", err) return false } @@ -190,7 +191,7 @@ } token = oauthResp.AccessToken tokenActive = true - log.Info("Got a new token..") + log.Info("Got a new Bearer token.") return true } @@ -251,9 +252,25 @@ if downloadBootSnapshot == false { scopes = append(scopes, (config.GetString(configScopeId))) } else { - scopes = findScopesforId(config.GetString(configScopeId)) - } + /* + * With WAL there is a chance, commits happening might take + * a bit, give some time if another session is in Committing + * the scopes + */ + for count := 0; count < 60; count++ { + scopes = findScopesforId(config.GetString(configScopeId)) + if scopes == nil { + log.Info("User Scopes not found in DB, retry in a bit.") + time.Sleep(time.Duration(count) * 100 * time.Millisecond) + } else { + break + } + } + if scopes == nil { + log.Fatal("Scope cannot be found to download snapshot") + } + } /* Frame and send the snapshot request */ snapshotUri.Path = path.Join(snapshotUri.Path, "/snapshots")
diff --git a/apigee_sync_test.go b/apigee_sync_test.go index b3a2910..d7dbe86 100644 --- a/apigee_sync_test.go +++ b/apigee_sync_test.go
@@ -4,12 +4,14 @@ "encoding/json" "github.com/30x/apid" "github.com/30x/apid/factory" + "github.com/30x/transicator/common" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" "io/ioutil" "net/http" "net/http/httptest" "os" + "time" ) var _ = Describe("api", func() { @@ -26,8 +28,9 @@ }) It("perform sync round-trip", func(done Done) { - count := 0 - org := "test-org" + scount := 0 + phase := 0 + scope := "bootstrap" key := "XXXXXXX" secret := "YYYYYYY" @@ -35,19 +38,13 @@ server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { // first request is for a token - if req.URL.Path == "/v1/edgex/accesstoken" { + if req.URL.Path == "/accesstoken" { Expect(req.Method).To(Equal("POST")) Expect(req.Header.Get("Content-Type")).To(Equal("application/x-www-form-urlencoded; param=value")) - user, pw, ok := req.BasicAuth() - Expect(ok).To(BeTrue()) - Expect(user).To(Equal(key)) - Expect(pw).To(Equal(secret)) - err := req.ParseForm() Expect(err).NotTo(HaveOccurred()) Expect(req.Form.Get("grant_type")).To(Equal("client_credentials")) - Expect(req.Form.Get("org")).To(Equal(org)) res := oauthTokenResp{} res.AccessToken = "accesstoken" @@ -58,25 +55,132 @@ } // next requests are for changes - if req.URL.Path == "/v1/edgex/changeagent/changes" { + if req.URL.Path == "/snapshots" { Expect(req.Method).To(Equal("GET")) + q := req.URL.Query() - // break polling after first time - if count > 0 { - w.WriteHeader(http.StatusUnauthorized) + if phase == 0 { + phase = 1 + Expect(q.Get("scope")).To(Equal(scope)) + + apidcfgItem := common.Row{} + apidcfgItems := []common.Row{} + apidcfgItemCh := common.Row{} + apidcfgItemsCh := []common.Row{} + scv := &common.ColumnVal{ + Value: scope, + Type: 1, + } + apidcfgItem["id"] = scv + scv = &common.ColumnVal{ + Value: scope, + Type: 1, + } + apidcfgItem["_apid_scope"] = scv + apidcfgItems = append(apidcfgItems, apidcfgItem) + + scv = &common.ColumnVal{ + Value: "apid_config_scope_id_0", + Type: 1, + } + apidcfgItemCh["id"] = scv + + scv = &common.ColumnVal{ + Value: "apid_config_scope_id_0", + Type: 1, + } + apidcfgItemCh["_apid_scope"] = scv + + scv = &common.ColumnVal{ + Value: scope, + Type: 1, + } + apidcfgItemCh["apid_config_id"] = scv + + scv = &common.ColumnVal{ + Value: "att~prod", + Type: 1, + } + apidcfgItemCh["scope"] = scv + + apidcfgItemsCh = append(apidcfgItemsCh, apidcfgItemCh) + + res := &common.Snapshot{} + res.SnapshotInfo = "snapinfo1" + + res.Tables = []common.Table{ + { + Name: "edgex.apid_config", + Rows: apidcfgItems, + }, + { + Name: "edgex.apid_config_scope", + Rows: apidcfgItemsCh, + }, + } + + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + w.Write(body) + return + } else { + phase = 2 + Expect(q.Get("scope")).To(Equal("att~prod")) + res := &common.Snapshot{} + res.SnapshotInfo = "snapinfo1" + + apidcfgItems := []common.Row{} + res.Tables = []common.Table{ + { + Name: "kms.api_product", + Rows: apidcfgItems, + }, + } + + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + w.Write(body) return } - count++ + } + // next requests are for changes + if req.URL.Path == "/changes" { + Expect(req.Method).To(Equal("GET")) q := req.URL.Query() - Expect(q.Get("block")).To(Equal("60")) - Expect(q.Get("since")).To(Equal("0")) - Expect(q.Get("tag")).To(Equal("org:" + org)) + Expect(q.Get("snapshot")).To(Equal("snapinfo1")) + scparams := q["scope"] + Expect(scparams).To(ContainElement("att~prod")) + Expect(scparams).To(ContainElement("bootstrap")) - var res = ChangeSet{} - res.Changes = []ChangePayload{ + res := &common.ChangeList{} + + res.LastSequence = "lastSeq_01" + mpItems := common.Row{} + + scv := &common.ColumnVal{ + Value: "apid_config_scope_id_1", + Type: 1, + } + mpItems["id"] = scv + + scv = &common.ColumnVal{ + Value: scope, + Type: 1, + } + mpItems["apid_config_id"] = scv + + scv = &common.ColumnVal{ + Value: "att~test", + Type: 1, + } + mpItems["scope"] = scv + + res.Changes = []common.Change{ { - LastMsId: 10, + Table: "edgex.apid_config_scope", + NewRow: mpItems, + Operation: 1, }, } body, err := json.Marshal(res) @@ -84,26 +188,25 @@ w.Write(body) return } - Fail("should not reach") })) config = apid.Config() config.Set(configProxyServerBaseURI, server.URL) - config.Set(configOrganization, org) + config.Set(configSnapServerBaseURI, server.URL) + config.Set(configChangeServerBaseURI, server.URL) + config.Set(configScopeId, "apid_config_scope_0") + config.Set(configSnapshotProtocol, "json") + config.Set(configScopeId, scope) config.Set(configConsumerKey, key) config.Set(configConsumerSecret, secret) // set up temporary test database - tmpDir, err := ioutil.TempDir("", "sync_test") + tmpDir, err := ioutil.TempDir("", "apigee_sync_test") Expect(err).NotTo(HaveOccurred()) defer os.RemoveAll(tmpDir) config.Set("data_path", tmpDir) - db, err := apid.Data().DB() - Expect(err).NotTo(HaveOccurred()) - _, err = db.Exec("create table change_id (org varchar(255), snapshot_change_id integer, PRIMARY KEY (org))") - Expect(err).NotTo(HaveOccurred()) // start process - plugin will automatically start polling apid.InitializePlugins() @@ -111,21 +214,33 @@ h := &test_handler{ "sync data", func(event apid.Event) { + _, ok := event.(*common.Snapshot) + if ok { + if phase > 1 { + db, err := data.DB() + Expect(err).NotTo(HaveOccurred()) + // verify event data (post snapshot) + err = db.QueryRow("Select count(scp.id) from apid_config_scope as scp INNER JOIN apid_config as ap WHERE scp.apid_config_id = ap.id").Scan(&scount) + Expect(err).NotTo(HaveOccurred()) + Expect(scount).Should(Equal(1)) + } + } else { + // verify event data (post change) + // There should be 2 scopes now + _, ok := event.(*common.ChangeList) + if ok { + time.Sleep(200 * time.Millisecond) + db, err := data.DB() + Expect(err).NotTo(HaveOccurred()) + err = db.QueryRow("Select count(scp.id) from apid_config_scope as scp INNER JOIN apid_config as ap WHERE scp.apid_config_id = ap.id").Scan(&scount) + Expect(err).NotTo(HaveOccurred()) + Expect(scount).Should(Equal(2)) + close(done) + } else { + Fail("Unexpected event") + } + } - // verify event data - changes := event.(ChangeSet) - Expect(len(changes.Changes)).Should(Equal(1)) - Expect(changes.Changes[0].LastMsId).Should(Equal(int64(10))) - - // verify database update - db, err := data.DB() - Expect(err).ShouldNot(HaveOccurred()) - var lastId int64 - err = db.QueryRow("select snapshot_change_id from change_id where org=?", org).Scan(&lastId) - Expect(err).ShouldNot(HaveOccurred()) - Expect(lastId).Should(Equal(int64(10))) - - close(done) }, }
diff --git a/init.go b/init.go index 33e8297..a086b8d 100644 --- a/init.go +++ b/init.go
@@ -102,6 +102,7 @@ updated_by text, _apid_scope text, snapshotInfo text, + lastSequence text, PRIMARY KEY (id) ); CREATE TABLE apid_config_scope (
diff --git a/listener.go b/listener.go index ce6074e..c264cd2 100644 --- a/listener.go +++ b/listener.go
@@ -48,9 +48,7 @@ insertApidConfig(row, db, snapshot.SnapshotInfo) } case "edgex.apid_config_scope": - for _, row := range payload.Rows { - insertApidConfigScope(row, db) - } + insertApidConfigScopes(payload.Rows, db) } } } @@ -58,19 +56,20 @@ func processChange(changes *common.ChangeList) { log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes)) - + var rows []common.Row db, err := data.DB() if err != nil { panic("Unable to access Sqlite DB") } for _, payload := range changes.Changes { - + rows = nil switch payload.Table { case "edgex.apid_config_scope": switch payload.Operation { case 1: - insertApidConfigScope(payload.NewRow, db) + rows = append(rows, payload.NewRow) + insertApidConfigScopes(rows, db) } } } @@ -84,6 +83,14 @@ var scope, id, name, orgAppName, createdBy, updatedBy, Description string var updated, created int64 + prep, err := db.Prepare("INSERT INTO APID_CONFIG (id, _apid_scope, name, umbrella_org_app_name, created, created_by, updated, updated_by, snapshotInfo)VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9);") + if err != nil { + log.Error("INSERT APID_CONFIG Failed: ", err) + return false + } + + txn, err := db.Begin() + ele.Get("id", &id) ele.Get("_apid_scope", &scope) ele.Get("name", &name) @@ -94,12 +101,7 @@ ele.Get("updated_by", &updatedBy) ele.Get("description", &Description) - stmt, err := db.Prepare("INSERT INTO APID_CONFIG (id, _apid_scope, name, umbrella_org_app_name, created, created_by, updated, updated_by, snapshotInfo)VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9);") - if err != nil { - log.Error("INSERT APID_CONFIG Failed: ", err) - return false - } - _, err = stmt.Exec( + _, err = txn.Stmt(prep).Exec( id, scope, name, @@ -112,9 +114,11 @@ if err != nil { log.Error("INSERT APID_CONFIG Failed: ", id, ", ", scope, ")", err) + txn.Rollback() return false } else { log.Info("INSERT APID_CONFIG Success: (", id, ", ", scope, ")") + txn.Commit() return true } @@ -123,38 +127,47 @@ /* * INSERT INTO APP_CREDENTIAL op */ -func insertApidConfigScope(ele common.Row, db *sql.DB) bool { +func insertApidConfigScopes(rows []common.Row, db *sql.DB) bool { var id, scopeId, apiConfigId, scope, createdBy, updatedBy string var created, updated int64 - ele.Get("id", &id) - ele.Get("_apid_scope", &scopeId) - ele.Get("apid_config_id", &apiConfigId) - ele.Get("scope", &scope) - ele.Get("created", &created) - ele.Get("created_by", &createdBy) - ele.Get("updated", &updated) - ele.Get("updated_by", &updatedBy) - - stmt, err := db.Prepare("INSERT INTO APID_CONFIG_SCOPE (id, _apid_scope, apid_config_id, scope, created, created_by, updated, updated_by)VALUES($1,$2,$3,$4,$5,$6,$7,$8);") - - _, err = stmt.Exec( - id, - scopeId, - apiConfigId, - scope, - created, - createdBy, - updated, - updatedBy) - + prep, err := db.Prepare("INSERT INTO APID_CONFIG_SCOPE (id, _apid_scope, apid_config_id, scope, created, created_by, updated, updated_by)VALUES($1,$2,$3,$4,$5,$6,$7,$8);") if err != nil { - log.Error("INSERT APID_CONFIG_SCOPE Failed: ", id, ", ", scope, ")", err) + log.Error("INSERT APID_CONFIG_SCOPE Failed: ", err) return false - } else { - log.Info("INSERT APID_CONFIG_SCOPE Success: (", id, ", ", scope, ")") - return true } + txn, err := db.Begin() + for _, ele := range rows { + + ele.Get("id", &id) + ele.Get("_apid_scope", &scopeId) + ele.Get("apid_config_id", &apiConfigId) + ele.Get("scope", &scope) + ele.Get("created", &created) + ele.Get("created_by", &createdBy) + ele.Get("updated", &updated) + ele.Get("updated_by", &updatedBy) + + _, err = txn.Stmt(prep).Exec( + id, + scopeId, + apiConfigId, + scope, + created, + createdBy, + updated, + updatedBy) + + if err != nil { + log.Error("INSERT APID_CONFIG_SCOPE Failed: ", id, ", ", scope, ")", err) + txn.Rollback() + return false + } else { + log.Info("INSERT APID_CONFIG_SCOPE Success: (", id, ", ", scope, ")") + } + } + txn.Commit() + return true }