subscribe.go

package cli

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	tpb "code.fbi.h-da.de/cocsn/api/go/gosdn/transport"
	"code.fbi.h-da.de/cocsn/gosdn/nucleus/types"

	"code.fbi.h-da.de/cocsn/gosdn/forks/goarista/gnmi"
	"code.fbi.h-da.de/cocsn/gosdn/nucleus"
	log "github.com/sirupsen/logrus"
)

// Subscribe starts a gNMI subscriber requersting the specified paths on the target and
// logs the response to stdout. Only 'stream' mode with 'sample' operation supported.
func Subscribe(address, username, password, deviceName string, sample, heartbeat int64, args ...string) error {
	sbi := &nucleus.OpenConfig{}
	tOpts := &tpb.TransportOption{
		Address:  address,
		Username: username,
		Password: password,
		Tls:      false,
		TransportOption: &tpb.TransportOption_GnmiTransportOption{
			GnmiTransportOption: &tpb.GnmiTransportOption{
				Compression:     "",
				GrpcDialOptions: nil,
				Token:           "",
				Encoding:        0,
			},
		},
	}

	device, err := nucleus.NewDevice(deviceName, tOpts, sbi)
	if err != nil {
		return err
	}

	opts := &gnmi.SubscribeOptions{
		UpdatesOnly:       false,
		Prefix:            "",
		Mode:              "stream",
		StreamMode:        "sample",
		SampleInterval:    uint64(sample * time.Second.Nanoseconds()),
		SuppressRedundant: false,
		HeartbeatInterval: uint64(heartbeat * time.Second.Nanoseconds()),
		Paths:             gnmi.SplitPaths(args),
		Origin:            "",
		Target:            address,
	}
	done := make(chan os.Signal, 1)
	signal.Notify(done, syscall.SIGILL, syscall.SIGTERM)
	ctx := context.WithValue(context.Background(), types.CtxKeyOpts, opts) //nolint
	go func() {
		if err := device.Transport.Subscribe(ctx); err != nil {
			log.Fatal(err)
		}
	}()
	fmt.Println("awaiting signal")
	<-done
	fmt.Println("exiting")

	return nil
}