Merge pull request #6 from 30x/all-handlers
All handlers
diff --git a/apigeeSync_suite_test.go b/apigeeSync_suite_test.go
index 92c6aff..ef8a1f6 100644
--- a/apigeeSync_suite_test.go
+++ b/apigeeSync_suite_test.go
@@ -1,12 +1,389 @@
-package apidApigeeSync_test
+package apidApigeeSync
import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"testing"
+ "github.com/30x/apid"
+ "github.com/30x/apid/factory"
+ "io/ioutil"
+ "net/http/httptest"
+ "os"
+ "encoding/json"
+ "net/http"
+ "github.com/apigee-labs/transicator/common"
+ "time"
+ "strconv"
)
+var (
+ tmpDir string
+ testServer *httptest.Server
+ testRouter apid.Router
+)
+
+const testScope = "bootstrap"
+
+var _ = BeforeSuite(func(done Done) {
+ var phase int
+
+ apid.Initialize(factory.DefaultServicesFactory())
+
+ config := apid.Config()
+
+ var err error
+ tmpDir, err = ioutil.TempDir("", "api_test")
+ Expect(err).NotTo(HaveOccurred())
+ config.Set("local_storage_path", tmpDir)
+
+ testRouter = apid.API().Router()
+ testServer = httptest.NewServer(testRouter)
+
+ config.Set(configProxyServerBaseURI, testServer.URL)
+ config.Set(configSnapServerBaseURI, testServer.URL)
+ config.Set(configChangeServerBaseURI, testServer.URL)
+ config.Set(configApidClusterId, "apid_config_scope_0")
+ config.Set(configName, "testhost")
+
+ config.Set(configSnapshotProtocol, "json")
+ config.Set(configApidClusterId, testScope)
+ config.Set(configConsumerKey, "XXXXXXX")
+ config.Set(configConsumerSecret, "YYYYYYY")
+
+ // mock upstream testServer
+ testRouter.HandleFunc("/accesstoken", func(w http.ResponseWriter, req *http.Request) {
+ defer GinkgoRecover()
+
+ Expect(req.Method).To(Equal("POST"))
+ Expect(req.Header.Get("Content-Type")).To(Equal("application/x-www-form-urlencoded; param=value"))
+
+ err := req.ParseForm()
+ Expect(err).NotTo(HaveOccurred())
+ Expect(req.Form.Get("grant_type")).To(Equal("client_credentials"))
+ Expect(req.Header.Get("status")).To(Equal("ONLINE"))
+ Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap"))
+ Expect(req.Header.Get("display_name")).To(Equal("testhost"))
+
+ var plugInfo []pluginDetail
+ plInfo := []byte(req.Header.Get("plugin_details"))
+ err = json.Unmarshal(plInfo, &plugInfo)
+ Expect(err).NotTo(HaveOccurred())
+
+ Expect(plugInfo[0].Name).To(Equal("apidApigeeSync"))
+ Expect(plugInfo[0].SchemaVersion).To(Equal("0.0.2"))
+
+ res := oauthTokenResp{}
+ res.AccessToken = "accesstoken"
+ body, err := json.Marshal(res)
+ Expect(err).NotTo(HaveOccurred())
+ w.Write(body)
+
+ }).Methods("POST")
+
+ testRouter.HandleFunc("/snapshots", func(w http.ResponseWriter, req *http.Request) {
+ defer GinkgoRecover()
+
+ q := req.URL.Query()
+
+ if phase == 0 {
+ phase = 1
+ Expect(q.Get("scope")).To(Equal(testScope))
+ Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap"))
+
+ apidcfgItem := common.Row{}
+ apidcfgItems := []common.Row{}
+ apidcfgItemCh := common.Row{}
+ apidcfgItemsCh := []common.Row{}
+ scv := &common.ColumnVal{
+ Value: testScope,
+ Type: 1,
+ }
+ apidcfgItem["id"] = scv
+ scv = &common.ColumnVal{
+ Value: testScope,
+ Type: 1,
+ }
+ apidcfgItem["_change_selector"] = 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["_change_selector"] = scv
+
+ scv = &common.ColumnVal{
+ Value: testScope,
+ Type: 1,
+ }
+ apidcfgItemCh["apid_cluster_id"] = scv
+
+ scv = &common.ColumnVal{
+ Value: "ert452",
+ Type: 1,
+ }
+ apidcfgItemCh["scope"] = scv
+
+ {
+ scv = &common.ColumnVal{
+ Value: "att",
+ Type: 1,
+ }
+ apidcfgItemCh["org"] = scv
+
+ }
+ {
+ scv = &common.ColumnVal{
+ Value: "prod",
+ Type: 1,
+ }
+ apidcfgItemCh["env"] = scv
+ }
+
+ apidcfgItemsCh = append(apidcfgItemsCh, apidcfgItemCh)
+
+ res := &common.Snapshot{}
+ res.SnapshotInfo = "snapinfo1"
+
+ res.Tables = []common.Table{
+ {
+ Name: "edgex.apid_cluster",
+ Rows: apidcfgItems,
+ },
+ {
+ Name: "edgex.data_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("ert452"))
+ 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
+ }
+
+ }).Methods("GET")
+
+ testRouter.HandleFunc("/changes", func(w http.ResponseWriter, req *http.Request) {
+ defer GinkgoRecover()
+
+ if req.URL.Query().Get("since") == "lastSeq_01" {
+ go func() {
+ block, err := strconv.Atoi(req.URL.Query().Get("block"))
+ Expect(err).NotTo(HaveOccurred())
+ time.Sleep(time.Duration(block) * time.Second)
+ w.WriteHeader(http.StatusNotModified)
+ }()
+ return
+ }
+
+ Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap"))
+ q := req.URL.Query()
+ Expect(q.Get("snapshot")).To(Equal("snapinfo1"))
+ scparams := q["scope"]
+ Expect(scparams).To(ContainElement("ert452"))
+ Expect(scparams).To(ContainElement("bootstrap"))
+
+ 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: testScope,
+ Type: 1,
+ }
+ mpItems["apid_cluster_id"] = scv
+
+ scv = &common.ColumnVal{
+ Value: "ert452",
+ Type: 1,
+ }
+ mpItems["scope"] = scv
+ {
+ scv = &common.ColumnVal{
+ Value: "att",
+ Type: 1,
+ }
+ mpItems["org"] = scv
+ }
+ {
+ scv = &common.ColumnVal{
+ Value: "prod",
+ Type: 1,
+ }
+ mpItems["env"] = scv
+ }
+
+ res.Changes = []common.Change{
+ {
+ Table: "edgex.data_scope",
+ NewRow: mpItems,
+ Operation: 1,
+ },
+ }
+ body, err := json.Marshal(res)
+ Expect(err).NotTo(HaveOccurred())
+ w.Write(body)
+
+ }).Methods("GET")
+
+ // This is actually the first test :)
+ // Tests that entire bootstrap and set of sync operations work
+ apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
+ defer GinkgoRecover()
+
+ if s, ok := event.(*common.Snapshot); ok {
+
+ Expect(s.SnapshotInfo).Should(Equal("snapinfo1"))
+
+ for _, t := range s.Tables {
+ switch t.Name {
+
+ case "edgex.apid_cluster":
+ Expect(t.Rows).To(HaveLen(1))
+ r := t.Rows[0]
+ var cs, id string
+ r.Get("_change_selector", &cs)
+ r.Get("id", &id)
+
+ Expect(cs).To(Equal("bootstrap"))
+ Expect(id).To(Equal("bootstrap"))
+
+ case "edgex.data_scope":
+ Expect(t.Rows).To(HaveLen(1))
+ r := t.Rows[0]
+
+ var cs, id, clusterID, env, org, scope string
+ r.Get("_change_selector", &cs)
+ r.Get("id", &id)
+ r.Get("apid_cluster_id", &clusterID)
+ r.Get("env", &env)
+ r.Get("org", &org)
+ r.Get("scope", &scope)
+
+ Expect(id).To(Equal("apid_config_scope_id_0"))
+ Expect(cs).To(Equal("apid_config_scope_id_0"))
+ Expect(clusterID).To(Equal("bootstrap"))
+ Expect(env).To(Equal("prod"))
+ Expect(org).To(Equal("att"))
+ Expect(scope).To(Equal("ert452"))
+
+ case "kms.api_product":
+ Expect(t.Rows).To(HaveLen(0))
+
+ default:
+ Fail("invalid table: " + t.Name)
+ }
+ }
+
+ } else if cl, ok := event.(*common.ChangeList); ok {
+
+ Expect(cl.LastSequence).To(Equal("lastSeq_01"))
+ Expect(cl.Changes).To(HaveLen(1))
+
+ c := cl.Changes[0]
+ Expect(c.Table).To(Equal("edgex.data_scope"))
+ Expect(c.Operation).To(Equal(common.Insert))
+
+ Expect(c.NewRow).ToNot(BeNil())
+
+ var id, clusterID, env, org, scope string
+ c.NewRow.Get("id", &id)
+ c.NewRow.Get("apid_cluster_id", &clusterID)
+ c.NewRow.Get("env", &env)
+ c.NewRow.Get("org", &org)
+ c.NewRow.Get("scope", &scope)
+
+ Expect(id).To(Equal("apid_config_scope_id_1"))
+ Expect(clusterID).To(Equal("bootstrap"))
+ Expect(env).To(Equal("prod"))
+ Expect(org).To(Equal("att"))
+ Expect(scope).To(Equal("ert452"))
+
+ events.ListenFunc(apid.EventDeliveredSelector, func(e apid.Event) {
+ defer GinkgoRecover()
+
+ // allow other handler to execute to insert last_sequence
+ time.Sleep(50 * time.Millisecond)
+ var seq string
+ err = getDB().
+ QueryRow("SELECT last_sequence FROM APID_CLUSTER LIMIT 1;").
+ Scan(&seq)
+
+ Expect(err).NotTo(HaveOccurred())
+ Expect(seq).To(Equal("lastSeq_01"))
+
+ close(done)
+ })
+ }
+ })
+
+ apid.InitializePlugins()
+})
+
+var _ = BeforeEach(func() {
+ apid.Events().Close()
+
+ token = ""
+ downloadDataSnapshot = false
+ downloadBootSnapshot = false
+ changeFinished = false
+ lastSequence = ""
+
+ _, err := getDB().Exec("DELETE FROM APID_CLUSTER")
+ Expect(err).NotTo(HaveOccurred())
+ _, err = getDB().Exec("DELETE FROM DATA_SCOPE")
+ Expect(err).NotTo(HaveOccurred())
+
+ db, err := data.DB()
+ Expect(err).NotTo(HaveOccurred())
+ _, err = db.Exec("DELETE FROM APID")
+ Expect(err).NotTo(HaveOccurred())
+})
+
+
+var _ = AfterSuite(func() {
+ apid.Events().Close()
+ if testServer != nil {
+ testServer.Close()
+ }
+ os.RemoveAll(tmpDir)
+})
+
func TestApigeeSync(t *testing.T) {
RegisterFailHandler(Fail)
RunSpecs(t, "ApigeeSync Suite")
diff --git a/apigee_sync.go b/apigee_sync.go
index a3de787..be087ec 100644
--- a/apigee_sync.go
+++ b/apigee_sync.go
@@ -14,43 +14,44 @@
)
var token string
-var tokenActive, downloadDataSnapshot, downloadBootSnapshot, chfin bool
+var downloadDataSnapshot, downloadBootSnapshot, changeFinished bool
var lastSequence string
-var gsnapshotInfo string
-func add_headers(req *http.Request) {
- req.Header.Add("Authorization", "Bearer "+token)
- req.Header.Set("apid_instance_id", guuid)
- req.Header.Set("apid_cluster_Id", gapidConfigId)
+func addHeaders(req *http.Request) {
+ req.Header.Add("Authorization", "Bearer " + token)
+ req.Header.Set("apid_instance_id", apidInfo.InstanceID)
+ req.Header.Set("apid_cluster_Id", apidInfo.ClusterID)
req.Header.Set("updated_at_apid", time.Now().Format(time.RFC3339))
}
-func donehandler(e apid.Event) {
- if rsp, ok := e.(apid.EventDeliveryEvent); ok {
- if rsp.Description == "event complete" {
- if ev, ok := rsp.Event.(*common.Snapshot); ok {
- if downloadBootSnapshot == false {
- downloadBootSnapshot = true
- log.Debug("Updated bootstrap SnapshotInfo")
- } else {
- gsnapshotInfo = ev.SnapshotInfo
- downloadDataSnapshot = true
- log.Debug("Updated data SnapshotInfo")
- }
- } else if ev, ok := rsp.Event.(*common.ChangeList); ok {
+func postPluginDataDelivery(e apid.Event) {
+
+ if ede, ok := e.(apid.EventDeliveryEvent); ok {
+
+ if ev, ok := ede.Event.(*common.ChangeList); ok {
+ if lastSequence != ev.LastSequence {
lastSequence = ev.LastSequence
- status := persistChange(lastSequence)
- if status == false {
- log.Fatal("Unable to update Sequence in DB")
+ err := persistChange(lastSequence)
+ if err != nil {
+ log.Panic("Unable to update Sequence in DB")
}
- chfin = true
+ }
+ changeFinished = true
+
+ } else if _, ok := ede.Event.(*common.Snapshot); ok {
+ if downloadBootSnapshot == false {
+ downloadBootSnapshot = true
+ log.Debug("Updated bootstrap SnapshotInfo")
+ } else {
+ downloadDataSnapshot = true
+ log.Debug("Updated data SnapshotInfo")
}
}
}
}
/*
- * Helper function that sleeps for N seconds, if comm. with change agent
+ * Helper function that sleeps for N seconds if comm with change agent
* fails. The retry interval gradually is incremented each time it fails
* till it reaches the Polling Int time, and after which it constantly
* retries at the polling time interval
@@ -61,10 +62,13 @@
pollInterval := config.GetInt(configPollInterval)
for {
startTime := time.Second
- _ = pollChangeAgent() // todo: handle error
+ err := pollChangeAgent()
+ if err != nil {
+ log.Debugf("Error connecting to changeserver: %v", err)
+ }
endTime := time.Second
// Gradually increase retry interval, and max at some level
- if endTime-startTime <= 1 {
+ if endTime - startTime <= 1 {
if times < pollInterval {
times++
} else {
@@ -87,7 +91,7 @@
func pollChangeAgent() error {
if downloadDataSnapshot != true {
- log.Warning("Waiting for snapshot download to complete")
+ log.Warn("Waiting for snapshot download to complete")
return errors.New("Snapshot download in progress...")
}
changesUri, err := url.Parse(config.GetString(configChangeServerBaseURI))
@@ -101,10 +105,10 @@
* 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 = findapidConfigInfo("lastSequence")
+ lastSequence = findApidConfigInfo(lastSequence)
for {
log.Debug("polling...")
- if tokenActive == false {
+ if token == "" {
/* token not valid?, get a new token */
status := getBearerToken()
if status == false {
@@ -113,7 +117,7 @@
}
/* Find the scopes associated with the config id */
- scopes := findScopesforId(gapidConfigId)
+ scopes := findScopesForId(apidInfo.ClusterID)
v := url.Values{}
/* Sequence added to the query if available */
@@ -130,16 +134,16 @@
for _, scope := range scopes {
v.Add("scope", scope)
}
- v.Add("scope", gapidConfigId)
- v.Add("snapshot", gsnapshotInfo)
+ v.Add("scope", apidInfo.ClusterID)
+ v.Add("snapshot", apidInfo.LastSnapshot)
changesUri.RawQuery = v.Encode()
uri := changesUri.String()
- log.Info("Fetching changes: ", uri)
+ log.Debugf("Fetching changes: %s", uri)
/* If error, break the loop, and retry after interval */
client := &http.Client{}
req, err := http.NewRequest("GET", uri, nil)
- add_headers(req)
+ addHeaders(req)
r, err := client.Do(req)
if err != nil {
log.Errorf("change agent comm error: %s", err)
@@ -149,12 +153,11 @@
/* If the call is not Authorized, update flag */
if r.StatusCode != http.StatusOK {
if r.StatusCode == http.StatusUnauthorized {
- tokenActive = false
+ token = ""
log.Errorf("Token expired? Unauthorized request.")
}
r.Body.Close()
- log.Errorf("Get Changes request failed with Resp err: %d",
- r.StatusCode)
+ log.Errorf("Get Changes request failed with Resp err: %d", r.StatusCode)
return err
}
@@ -162,14 +165,22 @@
err = json.NewDecoder(r.Body).Decode(&resp)
r.Body.Close()
if err != nil {
- log.Errorf("JSON Response Data not parsable: [%s] ", err)
+ log.Errorf("JSON Response Data not parsable: %v", err)
return err
}
+ if lastSequence != resp.LastSequence {
+ lastSequence = resp.LastSequence
+ err := persistChange(lastSequence)
+ if err != nil {
+ log.Panic("Unable to update Sequence in DB")
+ }
+ }
+
/* If valid data present, Emit to plugins */
if len(resp.Changes) > 0 {
- chfin = false
- events.ListenFunc(apid.EventDeliveredSelector, donehandler)
+ changeFinished = false
+ events.ListenFunc(apid.EventDeliveredSelector, postPluginDataDelivery)
events.Emit(ApigeeSyncEventSelector, &resp)
/*
* The plugins should have finished what they are doing.
@@ -178,18 +189,26 @@
* (Should there be a configurable Fudge factor?) FIXME
*/
for count := 0; count < 1000; count++ {
- if chfin == false {
- log.Info("Waiting for plugins to complete...")
+ if changeFinished == false {
+ log.Debug("Waiting for plugins to complete...")
time.Sleep(time.Duration(count) * 100 * time.Millisecond)
} else {
break
}
}
- if chfin == false {
- log.Fatal("Never got ack from plugins. Investigate..")
+ if changeFinished == false {
+ log.Panic("Never got ack from plugins. Investigate.")
}
} else {
- log.Info("No Changes detected for Scopes ", scopes)
+ log.Debugf("No Changes detected for Scopes: %s", scopes)
+
+ if lastSequence != resp.LastSequence {
+ lastSequence = resp.LastSequence
+ err := persistChange(lastSequence)
+ if err != nil {
+ log.Panic("Unable to update Sequence in DB")
+ }
+ }
}
}
}
@@ -200,26 +219,27 @@
*/
func getBearerToken() bool {
- log.Info("Getting a Bearer token.")
+ log.Debug("Getting a Bearer token.")
uri, err := url.Parse(config.GetString(configProxyServerBaseURI))
if err != nil {
log.Error(err)
return false
}
uri.Path = path.Join(uri.Path, "/accesstoken")
- tokenActive = false
+
+ token = ""
form := url.Values{}
form.Set("grant_type", "client_credentials")
form.Add("client_id", config.GetString(configConsumerKey))
form.Add("client_secret", config.GetString(configConsumerSecret))
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", ginstName)
- req.Header.Set("apid_instance_id", guuid)
- req.Header.Set("apid_cluster_Id", gapidConfigId)
+ req.Header.Set("display_name", apidInfo.InstanceName)
+ req.Header.Set("apid_instance_id", apidInfo.InstanceID)
+ req.Header.Set("apid_cluster_Id", apidInfo.ClusterID)
req.Header.Set("status", "ONLINE")
req.Header.Set("created_at_apid", time.Now().Format(time.RFC3339))
- req.Header.Set("plugin_details", gpgInfo)
+ req.Header.Set("plugin_details", apidPluginDetails)
client := &http.Client{}
resp, err := client.Do(req)
@@ -245,8 +265,7 @@
return false
}
token = oauthResp.AccessToken
- tokenActive = true
- log.Info("Got a new Bearer token.")
+ log.Debug("Got a new Bearer token.")
return true
}
@@ -267,7 +286,7 @@
func Redirect(req *http.Request, via []*http.Request) error {
req.Header.Add("Authorization", "Bearer "+token)
- req.Header.Add("org", gapidConfigId)
+ req.Header.Add("org", apidInfo.ClusterID)
return nil
}
@@ -283,28 +302,25 @@
* If there is already previous data in sqlite, don't fetch
* again from snapshot server.
*/
-func DownloadSnapshots() {
+func bootstrap() {
- /*
- * Skip Downloading snapshot, if there is already a snapshot
- * available from previous run of APID
- */
- gsnapshotInfo = findapidConfigInfo("snapshotInfo")
- if gsnapshotInfo != "" {
+ // Skip Downloading snapshot if there is already a snapshot available from previous run of APID
+ if apidInfo.LastSnapshot != "" {
+
downloadDataSnapshot = true
downloadBootSnapshot = true
- log.Infof("Starting on downloaded snapshot: %s", gsnapshotInfo)
+ log.Infof("Starting on downloaded snapshot: %s", apidInfo.LastSnapshot)
- // verify DB is accessible
- _, err := data.DBVersion(gsnapshotInfo)
+ // ensure DB version will be accessible on behalf of dependant plugins
+ _, err := data.DBVersion(apidInfo.LastSnapshot)
if err != nil {
log.Panicf("Database inaccessible: %v", err)
}
- // allow plugins to start immediately on existing database
+ // allow plugins (including this one) to start immediately on existing database
snap := &common.Snapshot{
- SnapshotInfo: gsnapshotInfo,
+ SnapshotInfo: apidInfo.LastSnapshot,
}
events.Emit(ApigeeSyncEventSelector, snap)
@@ -312,7 +328,7 @@
}
/* Phase 1 */
- DownloadSnapshot()
+ downloadSnapshot()
/*
* Give some time for all the plugins to process the Downloaded
@@ -332,20 +348,22 @@
log.Debug("Proceeding with existing Sqlite data")
} else if downloadBootSnapshot == true {
log.Debug("Proceed to download Snapshot for data scopes")
- DownloadSnapshot()
+ downloadSnapshot()
} else {
- log.Fatal("Snapshot for bootscope failed")
+ log.Panic("Snapshot for bootscope failed")
}
}
-func DownloadSnapshot() {
+func downloadSnapshot() {
+
+ log.Debugf("downloadSnapshot")
var scopes []string
/* Get the bearer token */
status := getBearerToken()
if status == false {
- log.Fatal("Unable to get Bearer token or is Invalid")
+ log.Panic("Unable to get Bearer token or is Invalid")
}
snapshotUri, err := url.Parse(config.GetString(configSnapServerBaseURI))
if err != nil {
@@ -353,12 +371,12 @@
}
if downloadBootSnapshot == false {
- scopes = append(scopes, (gapidConfigId))
+ scopes = append(scopes, apidInfo.ClusterID)
} else {
- scopes = findScopesforId(gapidConfigId)
+ scopes = findScopesForId(apidInfo.ClusterID)
}
if scopes == nil {
- log.Fatal("Scope cannot be found to download snapshot")
+ log.Panic("Scope cannot be found to download snapshot")
}
/* Frame and send the snapshot request */
snapshotUri.Path = path.Join(snapshotUri.Path, "/snapshots")
@@ -369,13 +387,13 @@
}
snapshotUri.RawQuery = v.Encode()
uri := snapshotUri.String()
- log.Info("Snapshot Download : ", uri)
+ log.Info("Snapshot Download: ", uri)
client := &http.Client{
CheckRedirect: Redirect,
}
req, err := http.NewRequest("GET", uri, nil)
- add_headers(req)
+ addHeaders(req)
/* Set the transport protocol type based on conf file input */
if config.GetString(configSnapshotProtocol) == "json" {
@@ -397,11 +415,10 @@
if err != nil {
if downloadBootSnapshot == false {
- log.Fatal("JSON Response Data not parsable: ", err)
+ log.Fatalf("JSON Response Data not parsable: %v", err)
} else {
-
/*
- * If the data set is empty, allow it to proceed, as changeserver
+ * If the data set is empty, allow it to proceed, as change server
* will feed data. Since Bootstrapping has passed, it has the
* Bootstrap config id to function.
*/
@@ -412,7 +429,7 @@
if r.StatusCode == 200 {
log.Info("Emit Snapshot response to plugins")
- events.ListenFunc(apid.EventDeliveredSelector, donehandler)
+ events.ListenFunc(apid.EventDeliveredSelector, postPluginDataDelivery)
events.Emit(ApigeeSyncEventSelector, &resp)
} else {
@@ -420,88 +437,3 @@
}
}
-
-/*
- * For the given apidConfigId, this function will retrieve all the scopes
- * associated with it
- */
-func findScopesforId(configId string) (scopes []string) {
-
- var scope string
- db, err := data.DB()
- if err != nil {
- log.Errorf("DB open Error: %s", err)
- return nil
- }
-
- rows, err := db.Query("select scope from DATA_SCOPE where apid_cluster_id = $1", configId)
- if err != nil {
- log.Errorf("Failed to query DATA_SCOPE. Err: %s", err)
- return nil
- }
- defer rows.Close()
- for rows.Next() {
- rows.Scan(&scope)
- scopes = append(scopes, scope)
- }
- return scopes
-}
-
-/*
- * Retrieve SnapshotInfo for the given apidConfigId from apid_config table
- */
-func findapidConfigInfo(qparam string) (info string) {
-
- db, err := data.DB()
- if err != nil {
- log.Errorf("DB open Error: %s", err)
- return ""
- }
- query := "select " + qparam + " from APID_CLUSTER"
- rows, err := db.Query(query)
- if err != nil {
- log.Errorf("Failed to query APID_CLUSTER. Err: %s", err)
- return ""
- }
- defer rows.Close()
- for rows.Next() {
- rows.Scan(&info)
- }
- return info
-}
-
-/*
- * Persist the last change Id each time a change has been successfully
- * processed by the plugin(s)
- */
-func persistChange(lastChange string) bool {
- db, err := data.DB()
- if err != nil {
- log.Errorf("DB open Error: %s", err)
- return false
- }
- txn, err := db.Begin()
- if err != nil {
- log.Error("Unable to create Sqlite transaction")
- return false
- }
- prep, err := txn.Prepare("UPDATE APID_CLUSTER SET lastSequence=$1;")
- if err != nil {
- log.Error("UPDATE APID_CLUSTER Failed: ", err)
- return false
- }
- defer prep.Close()
- s := txn.Stmt(prep)
- _, err = s.Exec(lastChange)
- s.Close()
- if err != nil {
- log.Error("UPDATE DATA_SCOPE Failed: ", err)
- txn.Rollback()
- return false
- } else {
- log.Info("UPDATE DATA_SCOPE Success: (", lastChange, ")")
- txn.Commit()
- return true
- }
-
-}
diff --git a/apigee_sync_test.go b/apigee_sync_test.go
index 507aa99..dea238c 100644
--- a/apigee_sync_test.go
+++ b/apigee_sync_test.go
@@ -1,314 +1,29 @@
package apidApigeeSync
import (
- "encoding/json"
- "github.com/30x/apid"
- "github.com/30x/apid/factory"
- "github.com/apigee-labs/transicator/common"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
- "io/ioutil"
- "net/http"
- "net/http/httptest"
- "os"
- "time"
+ "github.com/30x/apid"
+ "github.com/apigee-labs/transicator/common"
)
-var _ = Describe("api", func() {
+var _ = Describe("listener", func() {
- var server *httptest.Server
- var plugInfo []pluginDetail
+ It("should bootstrap from local DB if present", func(done Done) {
- BeforeSuite(func() {
- apid.Initialize(factory.DefaultServicesFactory())
- })
+ Expect(apidInfo.LastSnapshot).NotTo(BeEmpty())
- AfterSuite(func() {
- apid.Events().Close()
- server.Close()
- })
+ apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
+ defer GinkgoRecover()
- It("perform sync round-trip", func(done Done) {
- scount := 0
- phase := 0
- scope := "bootstrap"
- key := "XXXXXXX"
- secret := "YYYYYYY"
+ if s, ok := event.(*common.Snapshot); ok {
+ Expect(s.SnapshotInfo).Should(Equal(apidInfo.LastSnapshot))
+ Expect(s.Tables).To(BeNil())
- // mock upstream server
- server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-
- // first request is for a token
- 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"))
-
- err := req.ParseForm()
- Expect(err).NotTo(HaveOccurred())
- Expect(req.Form.Get("grant_type")).To(Equal("client_credentials"))
- Expect(req.Header.Get("status")).To(Equal("ONLINE"))
- Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap"))
- Expect(req.Header.Get("display_name")).To(Equal("testhost"))
- plinfo := []byte(req.Header.Get("plugin_details"))
- err = json.Unmarshal(plinfo, &plugInfo)
- Expect(err).NotTo(HaveOccurred())
- Expect(plugInfo[0].Name).To(Equal("apidApigeeSync"))
- Expect(plugInfo[0].SchemaVersion).To(Equal("0.0.2"))
-
- res := oauthTokenResp{}
- res.AccessToken = "accesstoken"
- body, err := json.Marshal(res)
- Expect(err).NotTo(HaveOccurred())
- w.Write(body)
- return
+ close(done)
}
+ })
- // next requests are for changes
- if req.URL.Path == "/snapshots" {
- Expect(req.Method).To(Equal("GET"))
- q := req.URL.Query()
-
- if phase == 0 {
- phase = 1
- Expect(q.Get("scope")).To(Equal(scope))
- Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap"))
-
- 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["_change_selector"] = 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["_change_selector"] = scv
-
- scv = &common.ColumnVal{
- Value: scope,
- Type: 1,
- }
- apidcfgItemCh["apid_cluster_id"] = scv
-
- scv = &common.ColumnVal{
- Value: "ert452",
- Type: 1,
- }
- apidcfgItemCh["scope"] = scv
-
- {
- scv = &common.ColumnVal{
- Value: "att",
- Type: 1,
- }
- apidcfgItemCh["org"] = scv
-
- }
- {
- scv = &common.ColumnVal{
- Value: "prod",
- Type: 1,
- }
- apidcfgItemCh["env"] = scv
- }
-
- apidcfgItemsCh = append(apidcfgItemsCh, apidcfgItemCh)
-
- res := &common.Snapshot{}
- res.SnapshotInfo = "snapinfo1"
-
- res.Tables = []common.Table{
- {
- Name: "edgex.apid_cluster",
- Rows: apidcfgItems,
- },
- {
- Name: "edgex.data_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("ert452"))
- 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
- }
-
- }
- // next requests are for changes
- if req.URL.Path == "/changes" {
- Expect(req.Method).To(Equal("GET"))
- Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap"))
- q := req.URL.Query()
- Expect(q.Get("snapshot")).To(Equal("snapinfo1"))
- scparams := q["scope"]
- Expect(scparams).To(ContainElement("ert452"))
- Expect(scparams).To(ContainElement("bootstrap"))
-
- 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_cluster_id"] = scv
-
- scv = &common.ColumnVal{
- Value: "ert452",
- Type: 1,
- }
- mpItems["scope"] = scv
- {
- scv = &common.ColumnVal{
- Value: "att",
- Type: 1,
- }
- mpItems["org"] = scv
- }
- {
- scv = &common.ColumnVal{
- Value: "prod",
- Type: 1,
- }
- mpItems["env"] = scv
- }
-
- res.Changes = []common.Change{
- {
- Table: "edgex.data_scope",
- NewRow: mpItems,
- Operation: 1,
- },
- }
- body, err := json.Marshal(res)
- Expect(err).NotTo(HaveOccurred())
- w.Write(body)
- return
- }
- Fail("should not reach")
- }))
-
- config = apid.Config()
- config.Set(configProxyServerBaseURI, server.URL)
- config.Set(configSnapServerBaseURI, server.URL)
- config.Set(configChangeServerBaseURI, server.URL)
- config.Set(configApidClusterId, "apid_config_scope_0")
- config.Set(configName, "testhost")
-
- config.Set(configSnapshotProtocol, "json")
- config.Set(configApidClusterId, scope)
- config.Set(configConsumerKey, key)
- config.Set(configConsumerSecret, secret)
-
- // set up temporary test database
- tmpDir, err := ioutil.TempDir("", "apigee_sync_test")
- Expect(err).NotTo(HaveOccurred())
- defer os.RemoveAll(tmpDir)
-
- config.Set("data_path", tmpDir)
-
- // start process - plugin will automatically start polling
- apid.InitializePlugins()
-
- 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 data_scope as scp INNER JOIN apid_cluster as ap WHERE scp.apid_cluster_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 data_scope as scp INNER JOIN apid_cluster as ap WHERE scp.apid_cluster_id = ap.id").Scan(&scount)
- Expect(err).NotTo(HaveOccurred())
- Expect(scount).Should(Equal(2))
- close(done)
- } else {
- Fail("Unexpected event")
- }
- }
-
- },
- }
-
- donehandler := func(e apid.Event) {
- if rsp, ok := e.(apid.EventDeliveryEvent); ok {
- Expect(rsp.Description).Should(Equal("event complete"))
- } else {
- Fail("Unexpected event")
- }
- }
- apid.Events().Listen(ApigeeSyncEventSelector, h)
- events.ListenFunc(apid.EventDeliveredSelector, donehandler)
-
+ bootstrap()
})
})
-
-type test_handler struct {
- description string
- f func(event apid.Event)
-}
-
-func (t *test_handler) String() string {
- return t.description
-}
-
-func (t *test_handler) Handle(event apid.Event) {
- t.f(event)
-}
diff --git a/data.go b/data.go
new file mode 100644
index 0000000..e93971b
--- /dev/null
+++ b/data.go
@@ -0,0 +1,318 @@
+package apidApigeeSync
+
+import (
+ "database/sql"
+ "github.com/30x/apid"
+ "sync"
+ "fmt"
+ "crypto/rand"
+ "errors"
+)
+
+var (
+ unsafeDB apid.DB
+ dbMux sync.RWMutex
+)
+
+type dataApidCluster struct {
+ ChangeSelector, ID, Name, OrgAppName, CreatedBy, UpdatedBy, Description string
+ Updated, Created string
+}
+
+type dataDataScope struct {
+ ChangeSelector, ID, ClusterID, Scope, Org, Env, CreatedBy, UpdatedBy string
+ Updated, Created string
+}
+
+/*
+This plugin uses 2 databases:
+1. The default DB is used for APID table.
+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(`
+ CREATE TABLE IF NOT EXISTS APID (
+ instance_id text,
+ last_snapshot_info text,
+ PRIMARY KEY (instance_id)
+ );
+ CREATE TABLE IF NOT EXISTS APID_CLUSTER (
+ id text,
+ name text,
+ description text,
+ umbrella_org_app_name text,
+ created text,
+ created_by text,
+ updated text,
+ updated_by text,
+ _change_selector text,
+ last_sequence text,
+ PRIMARY KEY (id)
+ );
+ CREATE TABLE IF NOT EXISTS DATA_SCOPE (
+ id text,
+ apid_cluster_id text,
+ scope text,
+ org text,
+ env text,
+ created text,
+ created_by text,
+ updated text,
+ updated_by text,
+ _change_selector text,
+ PRIMARY KEY (id, apid_cluster_id)
+ );
+ `)
+ if err != nil {
+ return err
+ }
+
+ log.Debug("Database tables created.")
+ return nil
+}
+
+func getDB() apid.DB {
+ dbMux.RLock()
+ db := unsafeDB
+ dbMux.RUnlock()
+ return db
+}
+
+func setDB(db apid.DB) {
+ dbMux.Lock()
+ unsafeDB = db
+ dbMux.Unlock()
+}
+
+func insertApidCluster(dac dataApidCluster, txn *sql.Tx) error {
+
+ log.Debugf("inserting into APID_CLUSTER: %v", dac)
+
+ stmt, err := txn.Prepare(`
+ INSERT INTO APID_CLUSTER
+ (id, _change_selector, name, umbrella_org_app_name,
+ created, created_by, updated, updated_by,
+ description)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9);
+ `)
+ if err != nil {
+ log.Errorf("prepare insert into APID_CLUSTER transaction Failed: %v", err)
+ return err
+ }
+ defer stmt.Close()
+
+ _, err = stmt.Exec(
+ dac.ID, dac.ChangeSelector, dac.Name, dac.OrgAppName,
+ dac.Created, dac.CreatedBy, dac.Updated, dac.UpdatedBy,
+ dac.Description)
+
+ if err != nil {
+ log.Errorf("insert APID_CLUSTER failed: %v", err)
+ }
+
+ return err
+}
+
+func insertDataScope(ds dataDataScope, txn *sql.Tx) error {
+
+ log.Debugf("insert DATA_SCOPE: %v", ds)
+
+ stmt, err := txn.Prepare(`
+ INSERT INTO DATA_SCOPE
+ (id, apid_cluster_id, scope, org,
+ env, created, created_by, updated,
+ updated_by, _change_selector)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10);
+ `)
+ if err != nil {
+ log.Errorf("insert DATA_SCOPE failed: %v", err)
+ return err
+ }
+ defer stmt.Close()
+
+ _, err = stmt.Exec(
+ ds.ID, ds.ClusterID, ds.Scope, ds.Org,
+ ds.Env, ds.Created, ds.CreatedBy, ds.Updated,
+ ds.UpdatedBy, ds.ChangeSelector)
+
+ if err != nil {
+ log.Errorf("insert DATA_SCOPE failed: %v", err)
+ return err
+ }
+
+ return nil
+}
+
+func deleteDataScope(ds dataDataScope, txn *sql.Tx) error {
+
+ log.Debugf("delete DATA_SCOPE: %v", ds)
+
+ stmt, err := txn.Prepare("DELETE FROM DATA_SCOPE WHERE id=$1 and apid_cluster_id=$2")
+ if err != nil {
+ log.Errorf("update DATA_SCOPE failed: %v", err)
+ return err
+ }
+ defer stmt.Close()
+
+ _, err = stmt.Exec(ds.ID, ds.ClusterID)
+
+ if err != nil {
+ log.Errorf("delete DATA_SCOPE failed: %v", err)
+ return err
+ }
+
+ return nil
+}
+
+/*
+ * For the given apidConfigId, this function will retrieve all the scopes
+ * associated with it
+ */
+func findScopesForId(configId string) (scopes []string) {
+
+ log.Debugf("findScopesForId: %s", configId)
+
+ var scope string
+ db := getDB()
+
+ rows, err := db.Query("select scope from DATA_SCOPE where apid_cluster_id = $1", configId)
+ if err != nil {
+ log.Errorf("Failed to query DATA_SCOPE: %v", err)
+ return
+ }
+ defer rows.Close()
+ for rows.Next() {
+ rows.Scan(&scope)
+ scopes = append(scopes, scope)
+ }
+
+ log.Debugf("scopes: %v", scopes)
+ return
+}
+
+/*
+ * Retrieve SnapshotInfo for the given apidConfigId from apid_config table
+ */
+func findApidConfigInfo(qparam string) (info string) {
+
+ log.Debugf("findApidConfigInfo: %s", qparam)
+
+ db := getDB()
+
+ rows, err := db.Query("select ? from APID_CLUSTER", qparam)
+ if err != nil {
+ log.Errorf("Failed to query APID_CLUSTER: %v", err)
+ return
+ }
+ defer rows.Close()
+ for rows.Next() {
+ rows.Scan(&info)
+ }
+
+ log.Debugf("info: %s", info)
+
+ return
+}
+
+/*
+ * Persist the last change Id each time a change has been successfully
+ * processed by the plugin(s)
+ */
+func persistChange(lastChange string) error {
+
+ log.Debugf("persistChange: %s", lastChange)
+
+ db := getDB()
+
+ stmt, err := db.Prepare("UPDATE APID_CLUSTER SET last_sequence=$1;")
+ if err != nil {
+ log.Errorf("UPDATE APID_CLUSTER Failed: %v", err)
+ return err
+ }
+ defer stmt.Close()
+
+ _, err = stmt.Exec(lastChange)
+ if err != nil {
+ log.Errorf("UPDATE DATA_SCOPE Failed: %v", err)
+ return err
+ }
+
+ log.Infof("UPDATE DATA_SCOPE Success: %s", lastChange)
+
+ return nil
+}
+
+func getApidInstanceInfo() (info apidInstanceInfo, err error) {
+
+ // always use default database for this
+ var db apid.DB
+ db, err = data.DB()
+ if err != nil {
+ return
+ }
+
+ err = db.QueryRow("SELECT instance_id, last_snapshot_info FROM APID LIMIT 1").
+ Scan(&info.InstanceID, &info.LastSnapshot)
+ if err != nil {
+ if err != sql.ErrNoRows {
+ log.Errorf("Unable to retrieve apidInstanceInfo: %v", err)
+ return
+ } else {
+ // first start - no row, generate a UUID and store it
+ err = nil
+ info.InstanceID = generateUUID()
+
+ db.Exec("INSERT INTO APID (instance_id) VALUES (?)", info.InstanceID)
+ }
+ }
+
+ // if name not explicitly configured, just use InstanceID
+ config.SetDefault(configName, info.InstanceID)
+ info.InstanceName = config.GetString(configName)
+
+ // not stored in DB
+ info.ClusterID = config.GetString(configApidClusterId)
+
+ return
+}
+
+func updateApidInstanceInfo() error {
+
+ // always use default database for this
+ db, err := data.DB()
+ if err != nil {
+ return err
+ }
+
+ rows, err := db.Exec(`
+ INSERT OR REPLACE
+ INTO APID (instance_id, last_snapshot_info)
+ VALUES (?, ?)`,
+ apidInfo.InstanceID, apidInfo.LastSnapshot)
+ if err != nil {
+ return err
+ }
+ n, err := rows.RowsAffected()
+ if err == nil && n == 0 {
+ err = errors.New("no rows affected")
+ }
+
+ return err
+}
+
+/*
+ * generates a random uuid (mix of timestamp & crypto random string)
+ */
+func generateUUID() string {
+
+ buff := make([]byte, 16)
+ numRead, err := rand.Read(buff)
+ if numRead != len(buff) || err != nil {
+ panic(err)
+ }
+ /* uuid v4 spec */
+ buff[6] = (buff[6] | 0x40) & 0x4F
+ buff[8] = (buff[8] | 0x80) & 0xBF
+ return fmt.Sprintf("%x-%x-%x-%x-%x", buff[0:4], buff[4:6], buff[6:8], buff[8:10], buff[10:])
+}
diff --git a/init.go b/init.go
index d143dde..931a9ec 100644
--- a/init.go
+++ b/init.go
@@ -1,7 +1,6 @@
package apidApigeeSync
import (
- "crypto/rand"
"encoding/json"
"fmt"
"github.com/30x/apid"
@@ -21,47 +20,81 @@
)
var (
- log apid.LogService
- config apid.ConfigService
- data apid.DataService
- events apid.EventsService
- gapidConfigId string
- guuid string
- ginstName string
- gpgInfo string
+ log apid.LogService
+ config apid.ConfigService
+ data apid.DataService
+ events apid.EventsService
+
+ apidInfo apidInstanceInfo
+ apidPluginDetails string
)
+type apidInstanceInfo struct {
+ InstanceID, InstanceName, ClusterID, LastSnapshot string
+}
+
type pluginDetail struct {
Name string `json:"name"`
SchemaVersion string `json:"schemaVer"`
}
-/*
- * generates a random uuid
- */
-func generate_uuid() string {
- buff := make([]byte, 16)
- numRead, err := rand.Read(buff)
- if numRead != len(buff) || err != nil {
- panic(err)
- }
- /* uuid v4 spec */
- buff[6] = (buff[6] | 0x40) & 0x4F
- buff[8] = (buff[8] | 0x80) & 0xBF
- return fmt.Sprintf("%x-%x-%x-%x-%x", buff[0:4], buff[4:6], buff[6:8], buff[8:10], buff[10:])
-}
-
func init() {
apid.RegisterPlugin(initPlugin)
}
+func initPlugin(services apid.Services) (apid.PluginData, error) {
+ log = services.Log().ForModule("apigeeSync")
+ log.Debug("start init")
+
+ config = services.Config()
+ config.SetDefault(configPollInterval, 120)
+
+ data = services.Data()
+ events = services.Events()
+
+ /* This callback function will get called, once all the plugins are
+ * initialized (not just this plugin). This is needed because,
+ * downloadSnapshots/changes etc have to begin to be processed only
+ * after all the plugins are initialized
+ */
+ events.ListenFunc(apid.SystemEventsSelector, postInitPlugins)
+
+ // check for required values
+ for _, key := range []string{configProxyServerBaseURI, configConsumerKey, configConsumerSecret,
+ configSnapServerBaseURI, configChangeServerBaseURI} {
+ if !config.IsSet(key) {
+ return pluginData, fmt.Errorf("Missing required config value: %s", key)
+ }
+ }
+
+ // set up default database
+ db, err := data.DB()
+ if err != nil {
+ log.Panicf("Unable to access DB: %v", err)
+ }
+ err = initDB(db)
+ if err != nil {
+ log.Panicf("Unable to initialize DB: %v", err)
+ }
+ setDB(db)
+
+ apidInfo, err = getApidInstanceInfo()
+ if err != nil {
+ log.Panicf("Unable to get apid instance info: %v", err)
+ }
+
+ log.Debug("end init")
+
+ return pluginData, nil
+}
+
+// Plugins have all initialized, gather their info and start the ApigeeSync downloads
func postInitPlugins(event apid.Event) {
var plinfoDetails []pluginDetail
if pie, ok := event.(apid.PluginsInitializedEvent); ok {
-
/*
* Store the plugin details in the heap. Needed during
- * Bearer token generation request
+ * Bearer token generation request.
*/
for _, plugin := range pie.Plugins {
name := plugin.Name
@@ -76,17 +109,17 @@
}
if plinfoDetails == nil {
log.Panicf("No Plugins registered!")
- } else {
- pgInfo, err := json.Marshal(plinfoDetails)
- if err != nil {
- log.Panic("Unable to masrhal plugin data", err)
- }
- gpgInfo = (string(pgInfo[:]))
}
+ pgInfo, err := json.Marshal(plinfoDetails)
+ if err != nil {
+ log.Panicf("Unable to marshal plugin data: %v", err)
+ }
+ apidPluginDetails = string(pgInfo[:])
+
log.Debug("start post plugin init")
- /* call to Download Snapshot info */
- go DownloadSnapshots()
+
+ go bootstrap()
/* Begin Looking for changes periodically */
log.Debug("starting update goroutine")
@@ -97,91 +130,3 @@
}
}
-func initPlugin(services apid.Services) (apid.PluginData, error) {
- log = services.Log().ForModule("apigeeSync")
- log.Debug("start init")
-
- config = services.Config()
- data = services.Data()
- events = services.Events()
- guuid = findapidConfigInfo("instance_id")
- if guuid == "" {
- guuid = generate_uuid()
- }
-
- /* If The Instance has no name configured, just re-use UUID */
- ginstName = config.GetString(configName)
- if ginstName == "" {
- ginstName = guuid
- }
-
- /* This callback function will get called, once all the plugins are
- * initialized (not just this plugin). This is needed because,
- * DownloadSnapshots/Changes etc have to begin to be processed only
- * after all the plugins are initialized
- */
- events.ListenFunc(apid.SystemEventsSelector, postInitPlugins)
-
- config.SetDefault(configPollInterval, 120)
- gapidConfigId = config.GetString(configApidClusterId)
- db, err := data.DB()
- if err != nil {
- log.Panic("Unable to access DB", err)
- }
-
- // check for required values
- for _, key := range []string{configProxyServerBaseURI, configConsumerKey, configConsumerSecret, configSnapServerBaseURI, configChangeServerBaseURI} {
- if !config.IsSet(key) {
- return pluginData, fmt.Errorf("Missing required config value: %s", key)
- }
- }
-
- var count int
- row := db.QueryRow("SELECT count(*) FROM sqlite_master WHERE type='table' AND name='apid_cluster' COLLATE NOCASE;")
- if err := row.Scan(&count); err != nil {
- log.Panic("Unable to setup database", err)
- }
- if count == 0 {
- createTables(db)
- }
-
- log.Debug("end init")
-
- return pluginData, nil
-}
-
-func createTables(db apid.DB) {
- _, err := db.Exec(`
-CREATE TABLE apid_cluster (
- id text,
- instance_id text,
- name text,
- description text,
- umbrella_org_app_name text,
- created int64,
- created_by text,
- updated int64,
- updated_by text,
- _change_selector text,
- snapshotInfo text,
- lastSequence text,
- PRIMARY KEY (id)
-);
-CREATE TABLE data_scope (
- id text,
- apid_cluster_id text,
- scope text,
- org text,
- env text,
- created int64,
- created_by text,
- updated int64,
- updated_by text,
- _change_selector text,
- PRIMARY KEY (id)
-);
-`)
- if err != nil {
- log.Panic("Unable to initialize DB", err)
- }
-}
diff --git a/listener.go b/listener.go
index c5f154a..41b7a8e 100644
--- a/listener.go
+++ b/listener.go
@@ -1,11 +1,15 @@
package apidApigeeSync
import (
- "database/sql"
"github.com/30x/apid"
"github.com/apigee-labs/transicator/common"
)
+const (
+ LISTENER_TABLE_APID_CLUSTER = "edgex.apid_cluster"
+ LISTENER_TABLE_DATA_SCOPE = "edgex.data_scope"
+)
+
type handler struct {
}
@@ -15,180 +19,150 @@
func (h *handler) Handle(e apid.Event) {
- res := true
-
- db, err := data.DB()
- if err != nil {
- panic("Unable to access Sqlite DB")
- }
-
- txn, err := db.Begin()
- if err != nil {
- log.Error("Unable to create Sqlite transaction")
- return
- }
-
- snapData, ok := e.(*common.Snapshot)
- if ok {
- res = processSnapshot(snapData, txn)
+ if changeSet, ok := e.(*common.ChangeList); ok {
+ processChangeList(changeSet)
+ } else if snapShot, ok := e.(*common.Snapshot); ok {
+ processSnapshot(snapShot)
} else {
- changeSet, ok := e.(*common.ChangeList)
- if ok {
- res = processChange(changeSet, txn)
- } else {
- log.Fatal("Received invalid event: %v", e)
- }
+ log.Errorf("Received invalid event. Ignoring. %v", e)
}
- if res == true {
- txn.Commit()
- } else {
- txn.Rollback()
- }
- return
}
-func processSnapshot(snapshot *common.Snapshot, txn *sql.Tx) bool {
+func processSnapshot(snapshot *common.Snapshot) {
- log.Debugf("Process Snapshot data")
- res := true
+ log.Debugf("Snapshot received. Switching to DB version: %s", snapshot.SnapshotInfo)
- for _, payload := range snapshot.Tables {
-
- switch payload.Name {
- case "edgex.apid_cluster":
- res = insertApidCluster(payload.Rows, txn, snapshot.SnapshotInfo)
- case "edgex.data_scope":
- res = insertDataScopes(payload.Rows, txn)
- }
- if res == false {
- log.Error("Error encountered in Downloading Snapshot for ApidApigeeSync")
- return res
- }
+ db, err := data.DBVersion(snapshot.SnapshotInfo)
+ if err != nil {
+ log.Panicf("Unable to access database: %v", err)
}
- return res
-}
-func processChange(changes *common.ChangeList, txn *sql.Tx) bool {
+ err = initDB(db)
+ if err != nil {
+ log.Panicf("Unable to initialize database: %v", err)
+ }
- log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes))
- var rows []common.Row
- res := true
+ tx, err := db.Begin()
+ if err != nil {
+ log.Panicf("Error starting transaction: %v", err)
+ }
+ defer tx.Rollback()
- for _, payload := range changes.Changes {
- rows = nil
- switch payload.Table {
- case "edgex.data_scope":
- switch payload.Operation {
- case common.Insert:
- rows = append(rows, payload.NewRow)
- res = insertDataScopes(rows, txn)
+ for _, table := range snapshot.Tables {
+
+ switch table.Name {
+ case LISTENER_TABLE_APID_CLUSTER:
+ if len(table.Rows) != 1 {
+ log.Panic("Illegal state for apid_cluster. Must be a single row.")
+ }
+ ac := makeApidClusterFromRow(table.Rows[0])
+ err := insertApidCluster(ac, tx)
+ if err != nil {
+ log.Panic("Snapshot update failed: %v", err)
+ }
+
+ case LISTENER_TABLE_DATA_SCOPE:
+ for _, row := range table.Rows {
+ ds := makeDataScopeFromRow(row)
+ err := insertDataScope(ds, tx)
+ if err != nil {
+ log.Panic("Snapshot update failed: %v", err)
+ }
}
}
- if res == false {
- log.Error("Sql Operation error. Operation rollbacked")
- return res
- }
}
- return res
-}
-/*
- * INSERT INTO APP_CREDENTIAL op
- */
-func insertApidCluster(rows []common.Row, txn *sql.Tx, snapInfo string) bool {
-
- var scope, id, name, orgAppName, createdBy, updatedBy, Description string
- var updated, created int64
-
- prep, err := txn.Prepare("INSERT INTO APID_CLUSTER (id, instance_id, _change_selector, name, umbrella_org_app_name, created, created_by, updated, updated_by, snapshotInfo)VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10);")
+ err = tx.Commit()
if err != nil {
- log.Error("INSERT APID_CLUSTER Failed: ", err)
- return false
+ log.Panicf("Error committing Snapshot change: %v", err)
}
- defer prep.Close()
- for _, ele := range rows {
- ele.Get("id", &id)
- ele.Get("_change_selector", &scope)
- ele.Get("name", &name)
- ele.Get("umbrella_org_app_name", &orgAppName)
- ele.Get("created", &created)
- ele.Get("created_by", &createdBy)
- ele.Get("updated", &updated)
- ele.Get("updated_by", &updatedBy)
- ele.Get("description", &Description)
-
- s := txn.Stmt(prep)
- _, err = s.Exec(
- id,
- guuid,
- scope,
- name,
- orgAppName,
- created,
- createdBy,
- updated,
- updatedBy,
- snapInfo)
- s.Close()
- if err != nil {
- log.Error("INSERT APID_CLUSTER Failed: ", id, ", ", scope, ")", err)
- return false
- } else {
- log.Info("INSERT APID_CLUSTER Success: (", id, ", ", scope, ")")
- }
- }
- return true
-}
-
-/*
- * INSERT INTO APP_CREDENTIAL op
- */
-func insertDataScopes(rows []common.Row, txn *sql.Tx) bool {
-
- var id, scopeId, apiConfigId, scope, createdBy, updatedBy, org, env string
- var created, updated int64
-
- prep, err := txn.Prepare("INSERT INTO DATA_SCOPE (id, _change_selector, apid_cluster_id, scope, org, env, created, created_by, updated, updated_by)VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10);")
+ apidInfo.LastSnapshot = snapshot.SnapshotInfo
+ err = updateApidInstanceInfo()
if err != nil {
- log.Error("INSERT DATA_SCOPE Failed: ", err)
- return false
+ log.Panicf("Unable to update instance info: %v", err)
}
- defer prep.Close()
- for _, ele := range rows {
+ setDB(db)
+ log.Debugf("Snapshot processed: %s", snapshot.SnapshotInfo)
+}
- ele.Get("id", &id)
- ele.Get("_change_selector", &scopeId)
- ele.Get("apid_cluster_id", &apiConfigId)
- ele.Get("scope", &scope)
- ele.Get("org", &org)
- ele.Get("env", &env)
- ele.Get("created", &created)
- ele.Get("created_by", &createdBy)
- ele.Get("updated", &updated)
- ele.Get("updated_by", &updatedBy)
+func processChangeList(changes *common.ChangeList) {
- s := txn.Stmt(prep)
- _, err = s.Exec(
- id,
- scopeId,
- apiConfigId,
- scope,
- org,
- env,
- created,
- createdBy,
- updated,
- updatedBy)
- s.Close()
+ tx, err := getDB().Begin()
+ if err != nil {
+ log.Panicf("Error processing ChangeList: %v", err)
+ }
+ defer tx.Rollback()
- if err != nil {
- log.Error("INSERT DATA_SCOPE Failed: ", id, ", ", scope, ")", err)
- return false
- } else {
- log.Info("INSERT DATA_SCOPE Success: (", id, ", ", scope, ")")
+ log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes))
+
+ for _, change := range changes.Changes {
+ switch change.Table {
+ case "edgex.apid_cluster":
+ switch change.Operation {
+ case common.Delete:
+ // todo: shut down apid, delete databases, scorch the earth!
+ log.Panicf("illegal operation: %s for %s", change.Operation, change.Table)
+ default:
+ log.Panicf("illegal operation: %s for %s", change.Operation, change.Table)
+ }
+ case "edgex.data_scope":
+ switch change.Operation {
+ case common.Insert:
+ ds := makeDataScopeFromRow(change.NewRow)
+ err = insertDataScope(ds, tx)
+ case common.Delete:
+ ds := makeDataScopeFromRow(change.OldRow)
+ deleteDataScope(ds, tx)
+ default:
+ // common.Update is not allowed
+ log.Panicf("illegal operation: %s for %s", change.Operation, change.Table)
+ }
+ }
+ if err != nil{
+ log.Panicf("Error processing ChangeList: %v", err)
}
}
- return true
+
+ err = tx.Commit()
+ if err != nil {
+ log.Panicf("Error processing ChangeList: %v", err)
+ }
}
+
+func makeApidClusterFromRow(row common.Row) dataApidCluster {
+
+ dac := dataApidCluster{}
+
+ row.Get("id", &dac.ID)
+ row.Get("_change_selector", &dac.ChangeSelector)
+ row.Get("name", &dac.Name)
+ row.Get("umbrella_org_app_name", &dac.OrgAppName)
+ row.Get("created", &dac.Created)
+ row.Get("created_by", &dac.CreatedBy)
+ row.Get("updated", &dac.Updated)
+ row.Get("updated_by", &dac.UpdatedBy)
+ row.Get("description", &dac.Description)
+
+ return dac
+}
+
+func makeDataScopeFromRow(row common.Row) dataDataScope {
+
+ ds := dataDataScope{}
+
+ row.Get("id", &ds.ID)
+ row.Get("_change_selector", &ds.ChangeSelector)
+ row.Get("apid_cluster_id", &ds.ClusterID)
+ row.Get("scope", &ds.Scope)
+ row.Get("org", &ds.Org)
+ row.Get("env", &ds.Env)
+ row.Get("created", &ds.Created)
+ row.Get("created_by", &ds.CreatedBy)
+ row.Get("updated", &ds.Updated)
+ row.Get("updated_by", &ds.UpdatedBy)
+
+ return ds
+}
+
diff --git a/listener_test.go b/listener_test.go
new file mode 100644
index 0000000..02d255d
--- /dev/null
+++ b/listener_test.go
@@ -0,0 +1,338 @@
+package apidApigeeSync
+
+import (
+ . "github.com/onsi/ginkgo"
+ . "github.com/onsi/gomega"
+
+ "github.com/apigee-labs/transicator/common"
+)
+
+var _ = Describe("listener", func() {
+
+ handler := handler{}
+
+ Context("ApigeeSync snapshot event", func() {
+
+ It("should set DB to appropriate version", func() {
+
+ event := common.Snapshot{
+ SnapshotInfo: "test_snapshot",
+ Tables: []common.Table{},
+ }
+
+ handler.Handle(&event)
+
+ Expect(apidInfo.LastSnapshot).To(Equal(event.SnapshotInfo))
+
+ expectedDB, err := data.DBVersion(event.SnapshotInfo)
+ Expect(err).NotTo(HaveOccurred())
+
+ Expect(getDB() == expectedDB).Should(BeTrue())
+ })
+
+ It("should fail if zero apid_cluster rows", func() {
+
+ event := common.Snapshot{
+ SnapshotInfo: "test_snapshot_fail",
+ Tables: []common.Table{
+ {
+ Name: LISTENER_TABLE_APID_CLUSTER,
+ Rows: []common.Row{},
+ },
+ },
+ }
+
+ Expect(func() { handler.Handle(&event) }).To(Panic())
+ })
+
+ It("should fail if more than one apid_cluster rows", func() {
+
+ event := common.Snapshot{
+ SnapshotInfo: "test_snapshot_fail",
+ Tables: []common.Table{
+ {
+ Name: LISTENER_TABLE_APID_CLUSTER,
+ Rows: []common.Row{{}, {}},
+ },
+ },
+ }
+
+ Expect(func() { handler.Handle(&event) }).To(Panic())
+ })
+
+ It("should process a valid Snapshot", func() {
+
+ event := common.Snapshot{
+ SnapshotInfo: "test_snapshot_valid",
+ Tables: []common.Table{
+ {
+ Name: LISTENER_TABLE_APID_CLUSTER,
+ Rows: []common.Row{
+ {
+ "id": &common.ColumnVal{Value: "i"},
+ "_change_selector": &common.ColumnVal{Value: "c"},
+ "name": &common.ColumnVal{Value: "n"},
+ "umbrella_org_app_name": &common.ColumnVal{Value: "o"},
+ "created": &common.ColumnVal{Value: "c"},
+ "created_by": &common.ColumnVal{Value: "c"},
+ "updated": &common.ColumnVal{Value: "u"},
+ "updated_by": &common.ColumnVal{Value: "u"},
+ "description": &common.ColumnVal{Value: "d"},
+ },
+ },
+ },
+ {
+ Name: LISTENER_TABLE_DATA_SCOPE,
+ Rows: []common.Row{
+ {
+ "id": &common.ColumnVal{Value: "i"},
+ "_change_selector": &common.ColumnVal{Value: "c"},
+ "apid_cluster_id": &common.ColumnVal{Value: "a"},
+ "scope": &common.ColumnVal{Value: "s"},
+ "org": &common.ColumnVal{Value: "o"},
+ "env": &common.ColumnVal{Value: "e"},
+ "created": &common.ColumnVal{Value: "c"},
+ "created_by": &common.ColumnVal{Value: "c"},
+ "updated": &common.ColumnVal{Value: "u"},
+ "updated_by": &common.ColumnVal{Value: "u"},
+ },
+ },
+ },
+ },
+ }
+
+ handler.Handle(&event)
+
+ info, err := getApidInstanceInfo()
+ Expect(err).NotTo(HaveOccurred())
+
+ Expect(info.LastSnapshot).To(Equal(event.SnapshotInfo))
+
+ db := getDB()
+
+ // apid Cluster
+ var dcs []dataApidCluster
+
+ rows, err := db.Query(`
+ SELECT id, name, description, umbrella_org_app_name,
+ created, created_by, updated, updated_by,
+ _change_selector
+ FROM APID_CLUSTER`)
+ Expect(err).NotTo(HaveOccurred())
+ defer rows.Close()
+
+ c := dataApidCluster{}
+ for rows.Next() {
+ rows.Scan(&c.ID, &c.Name, &c.Description, &c.OrgAppName,
+ &c.Created, &c.CreatedBy, &c.Updated, &c.UpdatedBy,
+ &c.ChangeSelector)
+ dcs = append(dcs, c)
+ }
+
+ Expect(len(dcs)).To(Equal(1))
+ dc := dcs[0]
+
+ Expect(dc.ID).To(Equal("i"))
+ Expect(dc.Name).To(Equal("n"))
+ Expect(dc.Description).To(Equal("d"))
+ Expect(dc.OrgAppName).To(Equal("o"))
+ Expect(dc.Created).To(Equal("c"))
+ Expect(dc.CreatedBy).To(Equal("c"))
+ Expect(dc.Updated).To(Equal("u"))
+ Expect(dc.UpdatedBy).To(Equal("u"))
+ Expect(dc.ChangeSelector).To(Equal("c"))
+
+ // Data Scope
+ var dds []dataDataScope
+
+ rows, err = db.Query(`
+ SELECT id, apid_cluster_id, scope, org,
+ env, created, created_by, updated,
+ updated_by, _change_selector
+ FROM DATA_SCOPE`)
+ Expect(err).NotTo(HaveOccurred())
+ defer rows.Close()
+
+ d := dataDataScope{}
+ for rows.Next() {
+ rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org,
+ &d.Env, &d.Created, &d.CreatedBy, &d.Updated,
+ &d.UpdatedBy, &d.ChangeSelector)
+ dds = append(dds, d)
+ }
+
+ Expect(len(dds)).To(Equal(1))
+ ds := dds[0]
+
+ Expect(ds.ID).To(Equal("i"))
+ Expect(ds.Org).To(Equal("o"))
+ Expect(ds.Env).To(Equal("e"))
+ Expect(ds.Scope).To(Equal("s"))
+ Expect(ds.Created).To(Equal("c"))
+ Expect(ds.CreatedBy).To(Equal("c"))
+ Expect(ds.Updated).To(Equal("u"))
+ Expect(ds.UpdatedBy).To(Equal("u"))
+ Expect(ds.ChangeSelector).To(Equal("c"))
+ })
+ })
+
+ Context("ApigeeSync change event", func() {
+
+ Context(LISTENER_TABLE_APID_CLUSTER, func() {
+
+ It("insert event should panic", func() {
+
+ event := common.ChangeList{
+ LastSequence: "test",
+ Changes: []common.Change{
+ {
+ Operation: common.Insert,
+ Table: LISTENER_TABLE_APID_CLUSTER,
+ },
+ },
+ }
+
+ Expect(func() { handler.Handle(&event) }).To(Panic())
+ })
+
+ It("update event should panic", func() {
+
+ event := common.ChangeList{
+ LastSequence: "test",
+ Changes: []common.Change{
+ {
+ Operation: common.Update,
+ Table: LISTENER_TABLE_APID_CLUSTER,
+ },
+ },
+ }
+
+ Expect(func() { handler.Handle(&event) }).To(Panic())
+ })
+
+ PIt("delete event should kill all the things!")
+ })
+
+ Context(LISTENER_TABLE_DATA_SCOPE, func() {
+
+ It("insert event should add", func() {
+ event := common.ChangeList{
+ LastSequence: "test",
+ Changes: []common.Change{
+ {
+ Operation: common.Insert,
+ Table: LISTENER_TABLE_DATA_SCOPE,
+ NewRow: common.Row{
+ "id": &common.ColumnVal{Value: "i"},
+ "_change_selector": &common.ColumnVal{Value: "c"},
+ "apid_cluster_id": &common.ColumnVal{Value: "a"},
+ "scope": &common.ColumnVal{Value: "s"},
+ "org": &common.ColumnVal{Value: "o"},
+ "env": &common.ColumnVal{Value: "e"},
+ "created": &common.ColumnVal{Value: "c"},
+ "created_by": &common.ColumnVal{Value: "c"},
+ "updated": &common.ColumnVal{Value: "u"},
+ "updated_by": &common.ColumnVal{Value: "u"},
+ },
+ },
+ },
+ }
+
+ handler.Handle(&event)
+
+ var dds []dataDataScope
+
+ rows, err := getDB().Query(`
+ SELECT id, apid_cluster_id, scope, org,
+ env, created, created_by, updated,
+ updated_by, _change_selector
+ FROM DATA_SCOPE`)
+ Expect(err).NotTo(HaveOccurred())
+ defer rows.Close()
+
+ d := dataDataScope{}
+ for rows.Next() {
+ rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org,
+ &d.Env, &d.Created, &d.CreatedBy, &d.Updated,
+ &d.UpdatedBy, &d.ChangeSelector)
+ dds = append(dds, d)
+ }
+
+ Expect(len(dds)).To(Equal(1))
+ ds := dds[0]
+
+ Expect(ds.ID).To(Equal("i"))
+ Expect(ds.Org).To(Equal("o"))
+ Expect(ds.Env).To(Equal("e"))
+ Expect(ds.Scope).To(Equal("s"))
+ Expect(ds.Created).To(Equal("c"))
+ Expect(ds.CreatedBy).To(Equal("c"))
+ Expect(ds.Updated).To(Equal("u"))
+ Expect(ds.UpdatedBy).To(Equal("u"))
+ Expect(ds.ChangeSelector).To(Equal("c"))
+ })
+
+ It("delete event should delete", func() {
+ insert := common.ChangeList{
+ LastSequence: "test",
+ Changes: []common.Change{
+ {
+ Operation: common.Insert,
+ Table: LISTENER_TABLE_DATA_SCOPE,
+ NewRow: common.Row{
+ "id": &common.ColumnVal{Value: "i"},
+ "_change_selector": &common.ColumnVal{Value: "c"},
+ "apid_cluster_id": &common.ColumnVal{Value: "a"},
+ "scope": &common.ColumnVal{Value: "s"},
+ "org": &common.ColumnVal{Value: "o"},
+ "env": &common.ColumnVal{Value: "e"},
+ "created": &common.ColumnVal{Value: "c"},
+ "created_by": &common.ColumnVal{Value: "c"},
+ "updated": &common.ColumnVal{Value: "u"},
+ "updated_by": &common.ColumnVal{Value: "u"},
+ },
+ },
+ },
+ }
+
+ handler.Handle(&insert)
+
+ delete := common.ChangeList{
+ LastSequence: "test",
+ Changes: []common.Change{
+ {
+ Operation: common.Delete,
+ Table: LISTENER_TABLE_DATA_SCOPE,
+ OldRow: insert.Changes[0].NewRow,
+ },
+ },
+ }
+
+ handler.Handle(&delete)
+
+ var nRows int
+ err := getDB().QueryRow("SELECT count(id) FROM DATA_SCOPE").Scan(&nRows)
+ Expect(err).NotTo(HaveOccurred())
+
+ Expect(nRows).To(Equal(0))
+ })
+
+ It("update event should panic", func() {
+
+ event := common.ChangeList{
+ LastSequence: "test",
+ Changes: []common.Change{
+ {
+ Operation: common.Update,
+ Table: LISTENER_TABLE_DATA_SCOPE,
+ },
+ },
+ }
+
+ Expect(func() { handler.Handle(&event) }).To(Panic())
+ })
+
+ })
+
+ })
+})
diff --git a/payload.go b/payload.go
deleted file mode 100644
index 1728922..0000000
--- a/payload.go
+++ /dev/null
@@ -1,55 +0,0 @@
-package apidApigeeSync
-
-type Payload struct {
- Email string `json:"email"`
- FirstName string `json:"firstName"`
- LastName string `json:"lastName"`
- UserName string `json:"userName"`
- Organization string `json:"organizationName"`
- Status string `json:"status"`
- CreatedAt int64 `json:"createdAt"`
- CreatedBy string `json:"createdBy"`
- LastModifiedAt int64 `json:"lastModifiedAt"`
- LastModifiedBy string `json:"lastModifiedBy"`
- AppId string `json:"appId"`
- AppFamily string `json:"appFamily"`
- ConsumerSecret string `json:"consumerSecret"`
- IssuedAt int64 `json:"issuedAt"`
- DeveloperId string `json:"developerId"`
- CallbackUrl string `json:"callbackUrl"`
- AppName string `json:"name"`
- ApiProducts []Apip `json:"apiProducts"`
- Environments []string `json:"environments"`
- Resources []string `json:"apiResources"`
- URL string `json:"url"`
- Type int `json:"type"`
- ParentId string `json:"parentId"`
- Etag string `json:"etag"`
- Customtag string `json:"customtag"`
- Manifest string `json:"manifest"`
-}
-
-type DataPayload struct {
- EntityIdentifier string `json:"entityIdentifier"`
- EntityType string `json:"entityType"`
- Operation string `json:"operation"`
- PldCont Payload `json:"entityPayload"`
-}
-
-type ChangePayload struct {
- LastMsId int64 `json:"_id"`
- Ts int64 `json:"_ts"`
- Tags []string `json:"tags"`
- Data DataPayload `json:"data"`
-}
-
-type ChangeSet struct {
- AtStart bool `json:"atStart"`
- AtEnd bool `json:"atEnd"`
- Changes []ChangePayload `json:"changes"`
-}
-
-type Apip struct {
- ApiProduct string `json:"apiproduct"`
- Status string `json:"status"`
-}
