Bulk and RPC API support in translib (sonic-net#16)

Bulk and RPC API support in translib

translib/acl_app.go

@@ -21,6 +21,7 @@ package translib
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"reflect"
 	"strconv"
@@ -166,6 +167,11 @@ func (app *AclApp) translateGet(dbs [db.MaxDB]*db.DB) error {
 	return err
 }
 
+func (app *AclApp) translateAction(dbs [db.MaxDB]*db.DB) error {
+    err := errors.New("Not supported")
+    return err
+}
+
 func (app *AclApp) translateSubscribe(dbs [db.MaxDB]*db.DB, path string) (*notificationOpts, *notificationInfo, error) {
 	pathInfo := NewPathInfo(path)
 	notifInfo := notificationInfo{dbno: db.ConfigDB}
@@ -288,6 +294,13 @@ func (app *AclApp) processGet(dbs [db.MaxDB]*db.DB) (GetResponse, error) {
 	return GetResponse{Payload: payload}, err
 }
 
+func (app *AclApp) processAction(dbs [db.MaxDB]*db.DB) (ActionResponse, error) {
+    var resp ActionResponse
+    err := errors.New("Not implemented")
+
+    return resp, err
+}
+
 func (app *AclApp) translateCRUCommon(d *db.DB, opcode int) ([]db.WatchKeys, error) {
 	var err error
 	var keys []db.WatchKeys

translib/app_interface.go

@@ -79,12 +79,14 @@ type appInterface interface {
 	translateReplace(d *db.DB) ([]db.WatchKeys, error)
 	translateDelete(d *db.DB) ([]db.WatchKeys, error)
 	translateGet(dbs [db.MaxDB]*db.DB) error
+	translateAction(dbs [db.MaxDB]*db.DB) error
 	translateSubscribe(dbs [db.MaxDB]*db.DB, path string) (*notificationOpts, *notificationInfo, error)
 	processCreate(d *db.DB) (SetResponse, error)
 	processUpdate(d *db.DB) (SetResponse, error)
 	processReplace(d *db.DB) (SetResponse, error)
 	processDelete(d *db.DB) (SetResponse, error)
 	processGet(dbs [db.MaxDB]*db.DB) (GetResponse, error)
+	processAction(dbs [db.MaxDB]*db.DB) (ActionResponse, error)
 }
 
 //App modules will use this function to register with App interface during boot up

translib/common_app.go

@@ -112,6 +112,11 @@ func (app *CommonApp) translateGet(dbs [db.MaxDB]*db.DB) error {
 	return err
 }
 
+func (app *CommonApp) translateAction(dbs [db.MaxDB]*db.DB) error {
+    err := errors.New("Not supported")
+    return err
+}
+
 func (app *CommonApp) translateSubscribe(dbs [db.MaxDB]*db.DB, path string) (*notificationOpts, *notificationInfo, error) {
 	err := errors.New("Not supported")
 	notifInfo := notificationInfo{dbno: db.ConfigDB}
@@ -205,6 +210,13 @@ func (app *CommonApp) processGet(dbs [db.MaxDB]*db.DB) (GetResponse, error) {
     return GetResponse{Payload: resPayload}, err
 }
 
+func (app *CommonApp) processAction(dbs [db.MaxDB]*db.DB) (ActionResponse, error) {
+    var resp ActionResponse
+    err := errors.New("Not implemented")
+
+    return resp, err
+}
+
 func (app *CommonApp) translateCRUDCommon(d *db.DB, opcode int) ([]db.WatchKeys, error) {
 	var err error
 	var keys []db.WatchKeys

translib/db/db.go

@@ -147,8 +147,9 @@ func(dbNo DBNum) String() string {
 type Options struct {
 	DBNo               DBNum
 	InitIndicator      string
-	TableNameSeparator string
-	KeySeparator       string
+	TableNameSeparator string //Overriden by the DB config file's separator.
+	KeySeparator       string //Overriden by the DB config file's separator.
+	IsWriteDisabled    bool //Indicated if write is allowed
 
 	DisableCVLCheck    bool
 }
@@ -580,18 +581,20 @@ func (d *DB) doWrite(ts * TableSpec, op _txOp, key Key, val interface{}) error {
 	var e error = nil
 	var value Value
 
+	if d.Opts.IsWriteDisabled {
+		glog.Error("doWrite: Write to DB disabled")
+		e = errors.New("Write to DB disabled during this operation")
+		goto doWriteExit
+	}
+
 	switch d.txState {
 	case txStateNone:
-		if glog.V(2) {
-			glog.Info("doWrite: No Transaction.")
-		}
-		break
+		glog.Info("doWrite: No Transaction.")
 	case txStateWatch:
 		if glog.V(2) {
 			glog.Info("doWrite: Change to txStateSet, txState: ", d.txState)
 		}
 		d.txState = txStateSet
-		break
 	case txStateSet:
 		if glog.V(5) {
 			glog.Info("doWrite: Remain in txStateSet, txState: ", d.txState)
@@ -653,7 +656,7 @@ func (d *DB) doWrite(ts * TableSpec, op _txOp, key Key, val interface{}) error {
 
 	// Transaction case.
 
-	glog.Info("doWrite: op: ", op, "  ", key, " : ", value)
+	glog.Info("doWrite: op: ", op, "  ", d.key2redis(ts, key), " : ", value)
 
 	switch op {
 	case txOpHMSet, txOpHDel:
@@ -1099,7 +1102,6 @@ func (d *DB) StartTx(w []WatchKeys, tss []*TableSpec) error {
 	}
 
 	var e error = nil
-	var args []interface{}
 	var ret cvl.CVLRetCode
 
 	//Start CVL session
@@ -1115,6 +1117,44 @@ func (d *DB) StartTx(w []WatchKeys, tss []*TableSpec) error {
 		goto StartTxExit
 	}
 
+	e = d.performWatch(w, tss)
+
+StartTxExit:
+
+	if glog.V(3) {
+		glog.Info("StartTx: End: e: ", e)
+	}
+	return e
+}
+
+func (d *DB) AppendWatchTx(w []WatchKeys, tss []*TableSpec) error {
+	if glog.V(3) {
+		glog.Info("AppendWatchTx: Begin: w: ", w, " tss: ", tss)
+	}
+
+	var e error = nil
+
+	// Validate State
+	if d.txState == txStateNone {
+		glog.Error("AppendWatchTx: Incorrect State, txState: ", d.txState)
+		e = errors.New("Transaction has not started")
+		goto AppendWatchTxExit
+	}
+
+	e = d.performWatch(w, tss)
+
+AppendWatchTxExit:
+
+	if glog.V(3) {
+		glog.Info("AppendWatchTx: End: e: ", e)
+	}
+	return e
+}
+
+func (d *DB) performWatch(w []WatchKeys, tss []*TableSpec) error {
+	var e error
+	var args []interface{}
+
 	// For each watchkey
 	//   If a pattern, Get the keys, appending results to Cmd args.
 	//   Else append keys to the Cmd args
@@ -1133,7 +1173,7 @@ func (d *DB) StartTx(w []WatchKeys, tss []*TableSpec) error {
 
 		redisKeys, e := d.client.Keys(redisKey).Result()
 		if e != nil {
-			glog.Warning("StartTx: Keys: " + e.Error())
+			glog.Warning("performWatch: Keys: " + e.Error())
 			continue
 		}
 		for j := 0; j < len(redisKeys); j++ {
@@ -1148,27 +1188,22 @@ func (d *DB) StartTx(w []WatchKeys, tss []*TableSpec) error {
 	}
 
 	if len(args) == 1 {
-		glog.Warning("StartTx: Empty WatchKeys. Skipping WATCH")
-		goto StartTxSkipWatch
+		glog.Warning("performWatch: Empty WatchKeys. Skipping WATCH")
+		goto SkipWatch
 	}
 
 	// Issue the WATCH
 	_, e = d.client.Do(args...).Result()
 
 	if e != nil {
-		glog.Warning("StartTx: Do: WATCH ", args, " e: ", e.Error())
+		glog.Warning("performWatch: Do: WATCH ", args, " e: ", e.Error())
 	}
 
-StartTxSkipWatch:
+SkipWatch:
 
 	// Switch State
 	d.txState = txStateWatch
 
-StartTxExit:
-
-	if glog.V(3) {
-		glog.Info("StartTx: End: e: ", e)
-	}
 	return e
 }
 

translib/intf_app.go

@@ -245,6 +245,11 @@ func (app *IntfApp) translateGet(dbs [db.MaxDB]*db.DB) error {
 	return err
 }
 
+func (app *IntfApp) translateAction(dbs [db.MaxDB]*db.DB) error {
+    err := errors.New("Not supported")
+    return err
+}
+
 func (app *IntfApp) translateSubscribe(dbs [db.MaxDB]*db.DB, path string) (*notificationOpts, *notificationInfo, error) {
 	app.appDB = dbs[db.ApplDB]
 	pathInfo := NewPathInfo(path)
@@ -470,6 +475,13 @@ func (app *IntfApp) processGet(dbs [db.MaxDB]*db.DB) (GetResponse, error) {
 	return GetResponse{Payload: payload}, err
 }
 
+func (app *IntfApp) processAction(dbs [db.MaxDB]*db.DB) (ActionResponse, error) {
+    var resp ActionResponse
+    err := errors.New("Not implemented")
+
+    return resp, err
+}
+
 /* Checking IP adderss is v4 */
 func validIPv4(ipAddress string) bool {
 	ipAddress = strings.Trim(ipAddress, " ")

translib/lldp_app.go

@@ -132,6 +132,11 @@ func (app *lldpApp) translateGet(dbs [db.MaxDB]*db.DB) error  {
     return err
 }
 
+func (app *lldpApp) translateAction(dbs [db.MaxDB]*db.DB) error {
+    err := errors.New("Not supported")
+    return err
+}
+
 func (app *lldpApp) translateSubscribe(dbs [db.MaxDB]*db.DB, path string) (*notificationOpts, *notificationInfo, error) {
     pathInfo := NewPathInfo(path)
     notifInfo := notificationInfo{dbno: db.ApplDB}
@@ -254,6 +259,13 @@ func (app *lldpApp) processGet(dbs [db.MaxDB]*db.DB) (GetResponse, error)  {
    return GetResponse{Payload:payload}, err
 }
 
+func (app *lldpApp) processAction(dbs [db.MaxDB]*db.DB) (ActionResponse, error) {
+    var resp ActionResponse
+    err := errors.New("Not implemented")
+
+    return resp, err
+}
+
 /** Helper function to populate JSON response for GET request **/
 func (app *lldpApp) getLldpNeighInfoFromInternalMap(ifName *string, ifInfo *ocbinds.OpenconfigLldp_Lldp_Interfaces_Interface) {
 

translib/pfm_app.go

@@ -72,6 +72,11 @@ func (app *PlatformApp) getAppRootObject() (*ocbinds.OpenconfigPlatform_Componen
     return deviceObj.Components
 }
 
+func (app *PlatformApp) translateAction(dbs [db.MaxDB]*db.DB) error {
+    err := errors.New("Not supported")
+    return err
+}
+
 func (app *PlatformApp) translateSubscribe(dbs [db.MaxDB]*db.DB, path string) (*notificationOpts, *notificationInfo, error) {
 
     var err error
@@ -189,6 +194,14 @@ func (app *PlatformApp) processGet(dbs [db.MaxDB]*db.DB) (GetResponse, error)  {
     return GetResponse{Payload: payload}, err
 }
 
+func (app *PlatformApp) processAction(dbs [db.MaxDB]*db.DB) (ActionResponse, error) {
+    var resp ActionResponse
+    err := errors.New("Not implemented")
+
+    return resp, err
+}
+
+
 ///////////////////////////
 
 

translib/sys_app.go

@@ -76,6 +76,11 @@ func (app *SysApp) getAppRootObject() *ocbinds.OpenconfigSystem_System {
 	return deviceObj.System
 }
 
+func (app *SysApp) translateAction(dbs [db.MaxDB]*db.DB) error {
+    err := errors.New("Not supported")
+    return err
+}
+
 func (app *SysApp) translateSubscribe(dbs [db.MaxDB]*db.DB, path string) (*notificationOpts, *notificationInfo, error) {
 	var err error
 
@@ -339,3 +344,11 @@ func (app *SysApp) processGet(dbs [db.MaxDB]*db.DB) (GetResponse, error) {
 	}
 	return GetResponse{Payload: payload}, err
 }
+
+func (app *SysApp) processAction(dbs [db.MaxDB]*db.DB) (ActionResponse, error) {
+    var resp ActionResponse
+    err := errors.New("Not implemented")
+
+    return resp, err
+}
+