added telemetry concept

cmd/gnmi-telemetry/telemetry.go

+package main
+
+import (
+	"code.fbi.h-da.de/cocsn/gosdn/forks/goarista/gnmi"
+	"code.fbi.h-da.de/cocsn/gosdn/nucleus"
+	schema "code.fbi.h-da.de/cocsn/yang-models/generated/arista"
+	"context"
+	"fmt"
+	"github.com/google/uuid"
+	gpb "github.com/openconfig/gnmi/proto/gnmi"
+	log "github.com/sirupsen/logrus"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+)
+
+func main() {
+	log.SetLevel(log.DebugLevel)
+	sbi := &nucleus.AristaOC{}
+	transport := &nucleus.Gnmi{
+		SetNode: sbi.SetNode(),
+		RespChan: make(chan *gpb.SubscribeResponse),
+	}
+	device := nucleus.Device{
+		Device: &schema.Device{},
+		SBI:    sbi,
+		Config: nucleus.DeviceConfig{
+			Uuid:     uuid.New(),
+			Address:  "[fdfd::ce05]:6030",
+			Username: "admin",
+			Password: "arista",
+		},
+		Transport: transport,
+	}
+	pnd := nucleus.NewPND("openconfig", sbi)
+	if err := pnd.AddDevice(device); err != nil {
+		log.Fatal(err)
+	}
+
+	cfg := &gnmi.Config{
+		Addr:     device.Config.Address,
+		Password: device.Config.Password,
+		Username: device.Config.Username,
+	}
+	ctx := gnmi.NewContext(context.Background(), cfg)
+	ctx = context.WithValue(ctx, "config", cfg)
+
+	paths := []string{"/interfaces/interface/name"}
+
+	opts := &gnmi.SubscribeOptions{
+		UpdatesOnly:       false,
+		Prefix:            "",
+		Mode:              "stream",
+		StreamMode:        "sample",
+		SampleInterval:    uint64(10 * time.Second.Nanoseconds()),
+		SuppressRedundant: false,
+		HeartbeatInterval: uint64(time.Second.Nanoseconds()),
+		Paths:             gnmi.SplitPaths(paths),
+		Origin:            "",
+		Target:            device.Config.Address,
+	}
+	done := make(chan os.Signal, 1)
+	signal.Notify(done, syscall.SIGILL, syscall.SIGTERM)
+	ctx = context.WithValue(ctx, "opts", opts)
+	go func() {
+		if err := transport.Subscribe(ctx); err != nil {
+			log.Fatal(err)
+		}
+	}()
+	fmt.Println("awaiting signal")
+	<-done
+	fmt.Println("exiting")
+}
\ No newline at end of file

nucleus/gnmi_transport.go

@@ -7,10 +7,12 @@ import (
 	"github.com/openconfig/gnmi/proto/gnmi_ext"
 	"github.com/openconfig/goyang/pkg/yang"
 	"github.com/openconfig/ygot/ytypes"
+	log "github.com/sirupsen/logrus"
 )
 
 type Gnmi struct {
 	SetNode func(schema *yang.Entry, root interface{}, path *gpb.Path, val interface{}, opts ...ytypes.SetNodeOpt) error
+	RespChan chan *gpb.SubscribeResponse
 }
 
 func (g *Gnmi) SetConfig(interface{}) error {
@@ -25,7 +27,9 @@ func (g *Gnmi) GetConfig() interface{} {
 // TODO: Convert to meaningfiul calls
 func (g *Gnmi)Get(ctx context.Context, params ...string) (interface{}, error){return  nil, nil}
 func (g *Gnmi)Set(ctx context.Context, params ...string) (interface{}, error){return  nil, nil}
-func (g *Gnmi)Subscribe(ctx context.Context, params ...string) error{return  nil}
+func (g *Gnmi)Subscribe(ctx context.Context, params ...string) error{
+	return g.subscribe(ctx)
+}
 
 func (g *Gnmi)Type() string {
 	return "gnmi"
@@ -94,13 +98,21 @@ func (g *Gnmi) set(ctx context.Context, setOps []*gnmi.Operation,
 }
 
 // Subscribe calls GNMI subscribe
-func (g *Gnmi) subscribe(ctx context.Context, subscribeOptions *gnmi.SubscribeOptions,
-	respChan chan<- *gpb.SubscribeResponse) error {
+func (g *Gnmi) subscribe(ctx context.Context) error {
 	client, err := gnmi.Dial(ctx.Value("config").(*gnmi.Config))
 	if err != nil {
 		return err
 	}
-	return gnmi.SubscribeErr(ctx, client, subscribeOptions, respChan)
+	opts := ctx.Value("opts").(*gnmi.SubscribeOptions)
+	go func() {
+		for  {
+			resp := <- g.RespChan
+			if err := gnmi.LogSubscribeResponse(resp); err != nil {
+				log.Fatal(err)
+			}
+		}
+	}()
+	return gnmi.SubscribeErr(ctx, client, opts, g.RespChan)
 }
 
 // Close calls GNMI close