Google Git
Sign in
edge / apidApigeeSync / 6849c7a3c7b80d1f493008388901000c94ef9d0b
commit6849c7a3c7b80d1f493008388901000c94ef9d0b[log] [tgz]
authorSundar Ramamoorthy <sramamoorthy73@users.noreply.github.com>Tue Apr 11 19:59:12 2017 -0700
committerScott Ganyo <scott@ganyo.com>Tue Apr 11 19:59:12 2017 -0700
tree8958b3ce92a790775706629977232b359a603521
parentf0b9f21f2f86359ff77cf458c656b46bcd5b7ff1 [diff]
Have a single client connection and reuse (#43)

* Reuse a single client connection.

* Add more connections for the tests.

* Default is to cache conns.

* Declare the client at the package level. It is thread safe anyway.

* remove nil assignment of re-direct.

* Set the re-direct as part of init.

* Remove pointess calling of http.client via a function.
5 files changed
tree: 8958b3ce92a790775706629977232b359a603521
  1. cmd/
  2. .gitignore
  3. .travis.yml
  4. apigee_sync.go
  5. apigee_sync_test.go
  6. apigeeSync_suite_test.go
  7. backoff.go
  8. backoff_test.go
  9. changes.go
  10. cover.sh
  11. data.go
  12. glide.yaml
  13. init.go
  14. init_test.go
  15. LICENSE
  16. listener.go
  17. listener_test.go
  18. mock_server.go
  19. pluginData.go
  20. README.md
  21. snapshot.go
  22. token.go
  23. token_test.go
README.md

apidApigeeSync

This core plugin for apid connects to the Apigee Change Agent and publishes the data changes events onto the apid Event service. It also coordinates DB initialization for plugins on startup.

Configuration

namedescription
apigeesync_poll_intervalint. seconds. default: 5
apigeesync_proxy_server_basestring. url. required.
apigeesync_consumer_keystring. required.
apigeesync_consumer_secretstring. required.
apigeesync_instance_namestring. optional. Display Name for UI

This plugin also populates a configuration item for dependant plugins that may need it:

namedescription
apigeesync_apid_instance_idstring

Event Generated

Startup Procedure

ApigeeSync

  1. Read DB version (Snapshot.SnapshotInfo) from default DB
  2. If version found, emit Snapshot event (using Snapshot.SnapshotInfo, no data)
  3. Ask server for Snapshot
  4. Each time a Snapshot is received
    1. Verify Snapshot.SnapshotInfo is different than current
    2. Stop processing change events
    3. Remove or clean new DB version if it exists
    4. Emit Snapshot event
    5. Wait for plugins to finish processing
    6. Save Snapshot.SnapshotInfo in default DB
    7. Release old DB version
    8. Start processing change events

ApigeeSync-dependent plugins

  1. Initialization
    1. Until receiving first Snapshot message, ApigeeSync-dependent APIs must either:
      1. not register (endpoint will return a 404 by default)
      2. return a 503 until DB is initialized
  2. Upon receiving a snapshot notification (this is a HOT DB upgrade)
    1. Get DB for version (use Snapshot.SnapshotInfo as version)
    2. Create plugin's tables, if needed
    3. Insert any snapshot data into plugin's tables
    4. Set reference to new DB for all data access
    5. If db-dependent services are not exposed yet, expose them

Example plugin code:

var (
  log      apid.LogService   // set in initPlugin
  data     apid.DataService
  unsafeDB apid.DB
  dbMux    sync.RWMutex
)

func init() {
  apid.RegisterPlugin(initPlugin)
}

func initPlugin(services apid.Services) error {
  log = services.Log().ForModule("examplePlugin")
  log.Debug("start init")
  data = services.Data()
  return nil
}

// always use getDB() to safely access current DB
func getDB() apid.DB {
  dbMux.RLock()
  db := unsafeDB
  dbMux.RUnlock()
  return db
}

func setDB(db apid.DB) {
  dbMux.Lock()
  if unsafeDB == nil { // init API on DB initialization (optional)
    go initAPI()
  }
  unsafeDB = db
  dbMux.Unlock()
}

func processSnapshot(snapshot *common.Snapshot) {

  log.Debugf("Snapshot received. Switching to DB version: %s", snapshot.SnapshotInfo)

  db, err := data.DBVersion(snapshot.SnapshotInfo)
  if err != nil {
    log.Panicf("Unable to access database: %v", err)
  }

  // init DB as needed (note: DB may exist, use 'CREATE TABLE IF NOT EXISTS' if not explicitly checking)
  initDB(db)

  for _, table := range snapshot.Tables {
    // populate tables from snapshot...
  }

  // switch to new database 
  setDB(db)
  log.Debug("Snapshot processed")
}