grpc.go

package shared

import (
	"bytes"
	"errors"
	"io"

	"golang.org/x/net/context"

	pb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/plugin"
	ppb "code.fbi.h-da.de/danet/gosdn/api/go/gosdn/pnd"
	gpb "github.com/openconfig/gnmi/proto/gnmi"
	"github.com/sirupsen/logrus"
)

type byteSize float64

// constants representing human friendly data sizes as per https://www.socketloop.com/tutorials/golang-how-to-declare-kilobyte-megabyte-gigabyte-terabyte-and-so-on
const (
	_           = iota // ignore first value by assigning to blank identifier
	KB byteSize = 1 << (10 * iota)
	MB
)

type DeviceModelClient struct{ client pb.PluginClient }

func (m *DeviceModelClient) Unmarshal(json []byte, path *gpb.Path) error {
	_, err := m.client.Unmarshal(context.Background(), &pb.UnmarshalRequest{
		Json: json,
		Path: path,
	})
	return err
}

func (m *DeviceModelClient) SetNode(path *gpb.Path, value *gpb.TypedValue) error {
	_, err := m.client.SetNode(context.Background(), &pb.SetNodeRequest{
		Path:  path,
		Value: value,
	})
	return err
}

func (m *DeviceModelClient) GetNode(path *gpb.Path) ([]*gpb.Notification, error) {
	resp, err := m.client.GetNode(context.Background(), &pb.GetNodeRequest{
		Path: path,
	})
	return resp.GetNodes(), err
}

func (m *DeviceModelClient) DeleteNode(path *gpb.Path) error {
	_, err := m.client.DeleteNode(context.Background(), &pb.DeleteNodeRequest{
		Path: path,
	})
	return err
}

func (m *DeviceModelClient) Model() ([]byte, error) {
	resp, err := m.client.Model(context.Background(), &pb.ModelRequest{})
	return resp.Json, err
}

func (m *DeviceModelClient) Diff(original, modified []byte) (*gpb.Notification, error) {
	resp, err := m.client.Diff(context.Background(), &pb.DiffRequest{
		Original: original,
		Modified: modified,
	})
	return resp.GetNotification(), err
}

func (m *DeviceModelClient) ValidateChange(operation ppb.ApiOperation, path *gpb.Path, value []byte) ([]byte, error) {
	resp, err := m.client.ValidateChange(context.Background(), &pb.ValidateChangeRequest{
		Operation: operation,
		Path:      path,
		Value:     value,
	})
	return resp.GetModel(), err
}

func (m *DeviceModelClient) SchemaTreeGzip() ([]byte, error) {
	schemaTreeGzipClient, err := m.client.SchemaTreeGzip(context.Background(), &pb.SchemaTreeGzipRequest{})
	if err != nil {
		return nil, err
	}

	sTreeBytes := []byte{}

	for {
		payload, err := schemaTreeGzipClient.Recv()
		if err != nil {
			if errors.Is(err, io.EOF) {
				break
			}
			logrus.Error(err)

			closeErr := schemaTreeGzipClient.CloseSend()
			if closeErr != nil {
				return nil, err
			}

			return nil, err
		}
		sTreeBytes = append(sTreeBytes, payload.Chunk...)
	}

	return sTreeBytes, nil
}

type DeviceModelServer struct {
	Impl DeviceModel
	pb.UnimplementedPluginServer
}

func (m *DeviceModelServer) Unmarshal(
	ctx context.Context,
	req *pb.UnmarshalRequest) (*pb.UnmarshalResponse, error) {
	err := m.Impl.Unmarshal(req.GetJson(), req.GetPath())
	if err != nil {
		return &pb.UnmarshalResponse{Valid: false}, err
	}
	return &pb.UnmarshalResponse{Valid: true}, err
}

func (m *DeviceModelServer) SetNode(
	ctx context.Context,
	req *pb.SetNodeRequest) (*pb.SetNodeResponse, error) {
	err := m.Impl.SetNode(req.GetPath(), req.GetValue())
	return &pb.SetNodeResponse{Valid: true}, err
}

func (m *DeviceModelServer) GetNode(
	ctx context.Context,
	req *pb.GetNodeRequest) (*pb.GetNodeResponse, error) {
	nodes, err := m.Impl.GetNode(req.GetPath())
	return &pb.GetNodeResponse{Nodes: nodes}, err
}

func (m *DeviceModelServer) DeleteNode(
	ctx context.Context,
	req *pb.DeleteNodeRequest) (*pb.DeleteNodeResponse, error) {
	err := m.Impl.DeleteNode(req.GetPath())
	return &pb.DeleteNodeResponse{Valid: true}, err
}

func (m *DeviceModelServer) Model(
	ctx context.Context,
	req *pb.ModelRequest) (*pb.ModelResponse, error) {
	model, err := m.Impl.Model()
	return &pb.ModelResponse{Json: model}, err
}

func (m *DeviceModelServer) Diff(
	ctx context.Context,
	req *pb.DiffRequest) (*pb.DiffResponse, error) {
	notification, err := m.Impl.Diff(req.GetOriginal(), req.GetModified())
	return &pb.DiffResponse{Notification: notification}, err
}

func (m *DeviceModelServer) ValidateChange(
	ctx context.Context,
	req *pb.ValidateChangeRequest) (*pb.ValidateChangeResponse, error) {
	model, err := m.Impl.ValidateChange(req.GetOperation(), req.GetPath(), req.GetValue())
	return &pb.ValidateChangeResponse{Model: model}, err
}

func (m *DeviceModelServer) SchemaTreeGzip(
	req *pb.SchemaTreeGzipRequest,
	stream pb.Plugin_SchemaTreeGzipServer) error {
	buffer := make([]byte, int(MB))
	schema, err := m.Impl.SchemaTreeGzip()
	if err != nil {
		return err
	}

	schemaReader := bytes.NewReader(schema)

	for {
		n, err := schemaReader.Read(buffer)
		if err != nil {
			if errors.Is(err, io.EOF) {
				logrus.Println(err)
			}
			break
		}
		logrus.WithField("n", n).Trace("read bytes")
		payload := &pb.Payload{Chunk: buffer[:n]}
		err = stream.Send(payload)
		if err != nil {
			return err
		}
	}

	return nil
}