[#482] reputation/router: Move router to ./common

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>

pkg/services/reputation/common/router/calls.go

@@ -0,0 +1,132 @@
+package router
+
+import (
+	"sync"
+
+	"github.com/nspcc-dev/neofs-node/pkg/services/reputation"
+	"github.com/nspcc-dev/neofs-node/pkg/services/reputation/common"
+	"go.uber.org/zap"
+)
+
+type routeContext struct {
+	common.Context
+
+	passedRoute []ServerInfo
+}
+
+// NewRouteContext wraps the main context of value passing with its traversal route and epoch.
+func NewRouteContext(ctx common.Context, passed []ServerInfo) common.Context {
+	return &routeContext{
+		Context:     ctx,
+		passedRoute: passed,
+	}
+}
+
+type trustWriter struct {
+	router *Router
+
+	ctx *routeContext
+
+	routeMtx sync.RWMutex
+	mServers map[string]common.Writer
+}
+
+// InitWriter initializes and returns Writer that sends each value to its next route point.
+//
+// If ctx was created by NewRouteContext, then the traversed route is taken into account,
+// and the value will be sent to its continuation. Otherwise, the route will be laid
+// from scratch and the value will be sent to its primary point.
+//
+// After building a list of remote points of the next leg of the route, the value is sent
+// sequentially to all of them. If any transmissions (even all) fail, an error will not
+// be returned.
+//
+// Close of the composed Writer calls Close method on each internal Writer generated in
+// runtime and never returns an error.
+//
+// Always returns nil error.
+func (r *Router) InitWriter(ctx common.Context) (common.Writer, error) {
+	var (
+		routeCtx *routeContext
+		ok       bool
+	)
+
+	if routeCtx, ok = ctx.(*routeContext); !ok {
+		routeCtx = &routeContext{
+			Context:     ctx,
+			passedRoute: []ServerInfo{r.localSrvInfo},
+		}
+	}
+
+	return &trustWriter{
+		router:   r,
+		ctx:      routeCtx,
+		mServers: make(map[string]common.Writer),
+	}, nil
+}
+
+func (w *trustWriter) Write(t reputation.Trust) error {
+	w.routeMtx.Lock()
+	defer w.routeMtx.Unlock()
+
+	route, err := w.router.routeBuilder.NextStage(w.ctx.Epoch(), t, w.ctx.passedRoute)
+	if err != nil {
+		return err
+	} else if len(route) == 0 {
+		route = []ServerInfo{nil}
+	}
+
+	for _, remoteInfo := range route {
+		var endpoint string
+
+		if remoteInfo != nil {
+			endpoint = remoteInfo.Address()
+		}
+
+		remoteWriter, ok := w.mServers[endpoint]
+		if !ok {
+			provider, err := w.router.remoteProvider.InitRemote(remoteInfo)
+			if err != nil {
+				w.router.log.Debug("could not initialize writer provider",
+					zap.String("error", err.Error()),
+				)
+
+				continue
+			}
+
+			remoteWriter, err = provider.InitWriter(w.ctx)
+			if err != nil {
+				w.router.log.Debug("could not initialize writer",
+					zap.String("error", err.Error()),
+				)
+
+				continue
+			}
+
+			w.mServers[endpoint] = remoteWriter
+		}
+
+		err := remoteWriter.Write(t)
+		if err != nil {
+			w.router.log.Debug("could not write the value",
+				zap.String("error", err.Error()),
+			)
+		}
+	}
+
+	return nil
+}
+
+func (w *trustWriter) Close() error {
+	for endpoint, wRemote := range w.mServers {
+		err := wRemote.Close()
+		if err != nil {
+			w.router.log.Debug("could not close remote server writer",
+				zap.String("endpoint", endpoint),
+				zap.String("error", err.Error()),
+			)
+		}
+	}
+
+	return nil
+}

pkg/services/reputation/common/router/deps.go

@@ -0,0 +1,42 @@
+package router
+
+import (
+	"github.com/nspcc-dev/neofs-node/pkg/services/reputation"
+	"github.com/nspcc-dev/neofs-node/pkg/services/reputation/common"
+)
+
+// ServerInfo describes a set of
+// characteristics of a point in a route.
+type ServerInfo interface {
+	// PublicKey returns public key of the node
+	// from the route in a binary representation.
+	PublicKey() []byte
+
+	// Returns network address of the node
+	// in the route.
+	//
+	// Can be empty.
+	Address() string
+}
+
+// Builder groups methods to route values in the network.
+type Builder interface {
+	// NextStage must return next group of route points
+	// for passed epoch and trust values.
+	// Implementation must take into account already passed route points.
+	//
+	// Empty passed list means being at the starting point of the route.
+	//
+	// Must return empty list and no error if the endpoint of the route is reached.
+	NextStage(epoch uint64, t reputation.Trust, passed []ServerInfo) ([]ServerInfo, error)
+}
+
+// RemoteWriterProvider describes the component
+// for sending values to a fixed route point.
+type RemoteWriterProvider interface {
+	// InitRemote must return WriterProvider to the route point
+	// corresponding to info.
+	//
+	// Nil info matches the end of the route.
+	InitRemote(info ServerInfo) (common.WriterProvider, error)
+}

pkg/services/reputation/common/router/opts.go

@@ -0,0 +1,28 @@
+package router
+
+import (
+	"github.com/nspcc-dev/neofs-node/pkg/util/logger"
+	"go.uber.org/zap"
+)
+
+// Option sets an optional parameter of Router.
+type Option func(*options)
+
+type options struct {
+	log *logger.Logger
+}
+
+func defaultOpts() *options {
+	return &options{
+		log: zap.L(),
+	}
+}
+
+// WithLogger returns Option to specify logging component.
+func WithLogger(l *logger.Logger) Option {
+	return func(o *options) {
+		if l != nil {
+			o.log = l
+		}
+	}
+}

pkg/services/reputation/common/router/router.go

@@ -0,0 +1,80 @@
+package router
+
+import (
+	"fmt"
+
+	"github.com/nspcc-dev/neofs-node/pkg/util/logger"
+)
+
+// Prm groups the required parameters of the Router's constructor.
+//
+// All values must comply with the requirements imposed on them.
+// Passing incorrect parameter values will result in constructor
+// failure (error or panic depending on the implementation).
+type Prm struct {
+	// Characteristics of the local node's server.
+	//
+	// Must not be nil.
+	LocalServerInfo ServerInfo
+
+	// Component for sending values to a fixed route point.
+	//
+	// Must not be nil.
+	RemoteWriterProvider RemoteWriterProvider
+
+	// Route planner.
+	//
+	// Must not be nil.
+	Builder Builder
+}
+
+// Router represents component responsible for routing
+// local trust values over the network.
+//
+// For each fixed pair (node peer, epoch) there is a
+// single value route on the network. Router provides the
+// interface for writing values to the next point of the route.
+//
+// For correct operation, Router must be created using
+// the constructor (New) based on the required parameters
+// and optional components. After successful creation,
+// the Router is immediately ready to work through API.
+type Router struct {
+	log *logger.Logger
+
+	remoteProvider RemoteWriterProvider
+
+	routeBuilder Builder
+
+	localSrvInfo ServerInfo
+}
+
+const invalidPrmValFmt = "invalid parameter %s (%T):%v"
+
+func panicOnPrmValue(n string, v interface{}) {
+	panic(fmt.Sprintf(invalidPrmValFmt, n, v, v))
+}
+
+func New(prm Prm, opts ...Option) *Router {
+	switch {
+	case prm.RemoteWriterProvider == nil:
+		panicOnPrmValue("RemoteWriterProvider", prm.RemoteWriterProvider)
+	case prm.Builder == nil:
+		panicOnPrmValue("Builder", prm.Builder)
+	case prm.LocalServerInfo == nil:
+		panicOnPrmValue("LocalServerInfo", prm.LocalServerInfo)
+	}
+
+	o := defaultOpts()
+
+	for i := range opts {
+		opts[i](o)
+	}
+
+	return &Router{
+		log:            o.log,
+		remoteProvider: prm.RemoteWriterProvider,
+		routeBuilder:   prm.Builder,
+		localSrvInfo:   prm.LocalServerInfo,
+	}
+}

pkg/services/reputation/common/router/util.go

@@ -0,0 +1,39 @@
+package router
+
+import (
+	"bytes"
+	"errors"
+
+	"github.com/nspcc-dev/neofs-node/pkg/services/reputation"
+)
+
+var errWrongRoute = errors.New("wrong route")
+
+// CheckRoute checks if the route is a route correctly constructed by the builder for value a.
+//
+// Returns nil if route is correct, otherwise an error clarifying the inconsistency.
+func CheckRoute(builder Builder, epoch uint64, t reputation.Trust, route []ServerInfo) error {
+	for i := 1; i < len(route); i++ {
+		servers, err := builder.NextStage(epoch, t, route[:i])
+		if err != nil {
+			return err
+		} else if len(servers) == 0 {
+			break
+		}
+
+		found := false
+
+		for j := range servers {
+			if bytes.Equal(servers[j].PublicKey(), route[i].PublicKey()) {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			return errWrongRoute
+		}
+	}
+
+	return nil
+}