refactor: Refactor msgmux.Router to use actor pkg

Author: gijswijs

go.mod

@@ -64,6 +64,8 @@ require (
 	pgregory.net/rapid v1.2.0
 )
 
+require github.com/lightningnetwork/lnd/actor v0.0.2 // indirect
+
 require (
 	dario.cat/mergo v1.0.1 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect

go.sum

@@ -370,6 +370,8 @@ github.com/lightninglabs/neutrino/cache v1.1.2 h1:C9DY/DAPaPxbFC+xNNEI/z1SJY9GS3
 github.com/lightninglabs/neutrino/cache v1.1.2/go.mod h1:XJNcgdOw1LQnanGjw8Vj44CvguYA25IMKjWFZczwZuo=
 github.com/lightninglabs/protobuf-go-hex-display v1.33.0-hex-display h1:Y2WiPkBS/00EiEg0qp0FhehxnQfk3vv8U6Xt3nN+rTY=
 github.com/lightninglabs/protobuf-go-hex-display v1.33.0-hex-display/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+github.com/lightningnetwork/lnd/actor v0.0.2 h1:+1Tz715d1iC2WB3Y+qgP4ycRC2HTj/JPOBqK7753dNc=
+github.com/lightningnetwork/lnd/actor v0.0.2/go.mod h1:RKgQPYHLVHuIRdMF/Q8+NPV2Nva3F66ZsULD2+TskKc=
 github.com/lightningnetwork/lnd/cert v1.2.2 h1:71YK6hogeJtxSxw2teq3eGeuy4rHGKcFf0d0Uy4qBjI=
 github.com/lightningnetwork/lnd/cert v1.2.2/go.mod h1:jQmFn/Ez4zhDgq2hnYSw8r35bqGVxViXhX6Cd7HXM6U=
 github.com/lightningnetwork/lnd/clock v1.1.1 h1:OfR3/zcJd2RhH0RU+zX/77c0ZiOnIMsDIBjgjWdZgA0=

msgmux/msg_router.go

@@ -4,9 +4,9 @@ import (
 	"context"
 	"fmt"
 	"maps"
-	"sync"
 
 	"github.com/btcsuite/btcd/btcec/v2"
+	"github.com/lightningnetwork/lnd/actor"
 	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/lnwire"
 )
@@ -73,196 +73,219 @@ type Router interface {
 	Stop()
 }
 
-// sendQuery sends a query to the main event loop, and returns the response.
-func sendQuery[Q any, R any](sendChan chan fn.Req[Q, R], queryArg Q,
-	quit chan struct{}) fn.Result[R] {
+// EndpointsMap is a map of all registered endpoints.
+type EndpointsMap map[EndpointName]Endpoint
 
-	query, respChan := fn.NewReq[Q, R](queryArg)
+// routerActorMsg is a union interface for all router actor messages.
+type routerActorMsg interface {
+	actor.Message
+	isRouterActorMsg()
+}
 
-	if !fn.SendOrQuit(sendChan, query, quit) {
-		return fn.Errf[R]("router shutting down")
-	}
+// registerEndpointMsg is a message to register an endpoint.
+type registerEndpointMsg struct {
+	actor.BaseMessage
+	endpoint Endpoint
+}
 
-	return fn.NewResult(fn.RecvResp(respChan, nil, quit))
+// MessageType returns the type name of the message.
+func (m *registerEndpointMsg) MessageType() string { return "registerEndpointMsg" }
+func (m *registerEndpointMsg) isRouterActorMsg()   {}
+
+// unregisterEndpointMsg is a message to unregister an endpoint.
+type unregisterEndpointMsg struct {
+	actor.BaseMessage
+	name EndpointName
 }
 
-// sendQueryErr is a helper function based on sendQuery that can be used when
-// the query only needs an error response.
-func sendQueryErr[Q any](sendChan chan fn.Req[Q, error], queryArg Q,
-	quitChan chan struct{}) error {
+// MessageType returns the type name of the message.
+func (m *unregisterEndpointMsg) MessageType() string { return "unregisterEndpointMsg" }
+func (m *unregisterEndpointMsg) isRouterActorMsg()   {}
 
-	return fn.ElimEither(
-		sendQuery(sendChan, queryArg, quitChan).Either,
-		fn.Iden, fn.Iden,
-	)
+// routeMsg is a message to route a peer message.
+type routeMsg struct {
+	actor.BaseMessage
+	peerMsg PeerMsg
 }
 
-// EndpointsMap is a map of all registered endpoints.
-type EndpointsMap map[EndpointName]Endpoint
+// MessageType returns the type name of the message.
+func (m *routeMsg) MessageType() string { return "routeMsg" }
+func (m *routeMsg) isRouterActorMsg()   {}
 
-// MultiMsgRouter is a type of message router that is capable of routing new
-// incoming messages, permitting a message to be routed to multiple registered
-// endpoints.
-type MultiMsgRouter struct {
-	startOnce sync.Once
-	stopOnce  sync.Once
+// getEndpointsMsg is a message to get all endpoints.
+type getEndpointsMsg struct {
+	actor.BaseMessage
+}
+
+// MessageType returns the type name of the message.
+func (m *getEndpointsMsg) MessageType() string { return "getEndpointsMsg" }
+func (m *getEndpointsMsg) isRouterActorMsg()   {}
+
+// routerBehavior is the actor behavior for the message router.
+type routerBehavior struct {
+	endpoints map[EndpointName]Endpoint
+}
+
+// Receive handles incoming messages for the router actor.
+func (b *routerBehavior) Receive(ctx context.Context,
+	msg routerActorMsg) fn.Result[any] {
+
+	switch m := msg.(type) {
+	case *registerEndpointMsg:
+		log.Infof("MsgRouter: registering new Endpoint(%s)",
+			m.endpoint.Name())
 
-	// registerChan is the channel that all new endpoints will be sent to.
-	registerChan chan fn.Req[Endpoint, error]
+		if _, ok := b.endpoints[m.endpoint.Name()]; ok {
+			log.Errorf("MsgRouter: rejecting duplicate endpoint: %v",
+				m.endpoint.Name())
+			return fn.Ok[any](ErrDuplicateEndpoint)
+		}
+		b.endpoints[m.endpoint.Name()] = m.endpoint
+		return fn.Ok[any](nil) // nil error
+
+	case *unregisterEndpointMsg:
+		log.Infof("MsgRouter: unregistering Endpoint(%s)", m.name)
+		delete(b.endpoints, m.name)
+		return fn.Ok[any](nil) // nil error
+
+	case *routeMsg:
+		var couldSend bool
+		for _, endpoint := range b.endpoints {
+			if endpoint.CanHandle(m.peerMsg) {
+				log.Tracef("MsgRouter: sending msg %T to endpoint %s",
+					m.peerMsg, endpoint.Name())
+
+				sent := endpoint.SendMessage(ctx, m.peerMsg)
+				couldSend = couldSend || sent
+			}
+		}
 
-	// unregisterChan is the channel that all endpoints that are to be
-	// removed are sent to.
-	unregisterChan chan fn.Req[EndpointName, error]
+		var err error
+		if !couldSend {
+			log.Tracef("MsgRouter: unable to route msg %T",
+				m.peerMsg.Message)
+			err = ErrUnableToRouteMsg
+		}
+		return fn.Ok[any](err)
 
-	// msgChan is the channel that all messages will be sent to for
-	// processing.
-	msgChan chan fn.Req[PeerMsg, error]
+	case *getEndpointsMsg:
+		endpointsCopy := make(EndpointsMap, len(b.endpoints))
+		maps.Copy(b.endpoints, endpointsCopy)
+		return fn.Ok[any](endpointsCopy)
 
-	// endpointsQueries is a channel that all queries to the endpoints map
-	// will be sent to.
-	endpointQueries chan fn.Req[Endpoint, EndpointsMap]
+	default:
+		return fn.Err[any](fmt.Errorf("unknown message type: %T", m))
+	}
+}
 
-	wg   sync.WaitGroup
-	quit chan struct{}
+// MultiMsgRouter is a type of message router that is capable of routing new
+// incoming messages, permitting a message to be routed to multiple registered
+// endpoints.
+type MultiMsgRouter struct {
+	actor *actor.Actor[routerActorMsg, any]
 }
 
 // NewMultiMsgRouter creates a new instance of a peer message router.
 func NewMultiMsgRouter() *MultiMsgRouter {
+	beh := &routerBehavior{
+		endpoints: make(map[EndpointName]Endpoint),
+	}
+	actorCfg := actor.ActorConfig[routerActorMsg, any]{
+		ID:          "msg-router",
+		Behavior:    beh,
+		MailboxSize: 100,
+	}
+	routerActor := actor.NewActor(actorCfg)
+
 	return &MultiMsgRouter{
-		registerChan:    make(chan fn.Req[Endpoint, error]),
-		unregisterChan:  make(chan fn.Req[EndpointName, error]),
-		msgChan:         make(chan fn.Req[PeerMsg, error]),
-		endpointQueries: make(chan fn.Req[Endpoint, EndpointsMap]),
-		quit:            make(chan struct{}),
+		actor: routerActor,
 	}
 }
 
 // Start starts the peer message router.
 func (p *MultiMsgRouter) Start(ctx context.Context) {
 	log.Infof("Starting Router")
 
-	p.startOnce.Do(func() {
-		p.wg.Add(1)
-		go p.msgRouter(ctx)
-	})
+	p.actor.Start()
 }
 
 // Stop stops the peer message router.
 func (p *MultiMsgRouter) Stop() {
 	log.Infof("Stopping Router")
 
-	p.stopOnce.Do(func() {
-		close(p.quit)
-		p.wg.Wait()
-	})
+	p.actor.Stop()
 }
 
 // RegisterEndpoint registers a new endpoint with the router. If a duplicate
 // endpoint exists, an error is returned.
 func (p *MultiMsgRouter) RegisterEndpoint(endpoint Endpoint) error {
-	return sendQueryErr(p.registerChan, endpoint, p.quit)
-}
+	msg := &registerEndpointMsg{endpoint: endpoint}
+	future := p.actor.Ref().Ask(context.Background(), msg)
+	res := future.Await(context.Background())
 
-// UnregisterEndpoint unregisters the target endpoint from the router.
-func (p *MultiMsgRouter) UnregisterEndpoint(name EndpointName) error {
-	return sendQueryErr(p.unregisterChan, name, p.quit)
-}
+	val, err := res.Unpack()
+	if err != nil {
+		return err
+	}
 
-// RouteMsg attempts to route the target message to a registered endpoint. If
-// ANY endpoint could handle the message, then nil is returned.
-func (p *MultiMsgRouter) RouteMsg(msg PeerMsg) error {
-	return sendQueryErr(p.msgChan, msg, p.quit)
-}
+	if typedErr, ok := val.(error); ok {
+		return typedErr
+	}
 
-// Endpoints returns a list of all registered endpoints.
-func (p *MultiMsgRouter) endpoints() fn.Result[EndpointsMap] {
-	return sendQuery(p.endpointQueries, nil, p.quit)
+	return nil
 }
 
-// msgRouter is the main goroutine that handles all incoming messages.
-func (p *MultiMsgRouter) msgRouter(ctx context.Context) {
-	defer p.wg.Done()
-
-	// endpoints is a map of all registered endpoints.
-	endpoints := make(map[EndpointName]Endpoint)
-
-	for {
-		select {
-		// A new endpoint was just sent in, so we'll add it to our set
-		// of registered endpoints.
-		case newEndpointMsg := <-p.registerChan:
-			endpoint := newEndpointMsg.Request
-
-			log.Infof("MsgRouter: registering new "+
-				"Endpoint(%s)", endpoint.Name())
-
-			// If this endpoint already exists, then we'll return
-			// an error as we require unique names.
-			if _, ok := endpoints[endpoint.Name()]; ok {
-				log.Errorf("MsgRouter: rejecting "+
-					"duplicate endpoint: %v",
-					endpoint.Name())
-
-				newEndpointMsg.Resolve(ErrDuplicateEndpoint)
-
-				continue
-			}
-
-			endpoints[endpoint.Name()] = endpoint
-
-			newEndpointMsg.Resolve(nil)
-
-		// A request to unregister an endpoint was just sent in, so
-		// we'll attempt to remove it.
-		case endpointName := <-p.unregisterChan:
-			delete(endpoints, endpointName.Request)
-
-			log.Infof("MsgRouter: unregistering "+
-				"Endpoint(%s)", endpointName.Request)
+// UnregisterEndpoint unregisters the target endpoint from the router.
+func (p *MultiMsgRouter) UnregisterEndpoint(name EndpointName) error {
+	msg := &unregisterEndpointMsg{name: name}
+	future := p.actor.Ref().Ask(context.Background(), msg)
+	res := future.Await(context.Background())
 
-			endpointName.Resolve(nil)
+	val, err := res.Unpack()
+	if err != nil {
+		return err
+	}
 
-		// A new message was just sent in. We'll attempt to route it to
-		// all the endpoints that can handle it.
-		case msgQuery := <-p.msgChan:
-			msg := msgQuery.Request
+	if typedErr, ok := val.(error); ok {
+		return typedErr
+	}
 
-			// Loop through all the endpoints and send the message
-			// to those that can handle it the message.
-			var couldSend bool
-			for _, endpoint := range endpoints {
-				if endpoint.CanHandle(msg) {
-					log.Tracef("MsgRouter: sending "+
-						"msg %T to endpoint %s", msg,
-						endpoint.Name())
+	return nil
+}
 
-					sent := endpoint.SendMessage(ctx, msg)
-					couldSend = couldSend || sent
-				}
-			}
+// RouteMsg attempts to route the target message to a registered endpoint. If
+// ANY endpoint could handle the message, then nil is returned.
+func (p *MultiMsgRouter) RouteMsg(msg PeerMsg) error {
+	future := p.actor.Ref().Ask(context.Background(), &routeMsg{peerMsg: msg})
+	res := future.Await(context.Background())
 
-			var err error
-			if !couldSend {
-				log.Tracef("MsgRouter: unable to route "+
-					"msg %T", msg.Message)
+	val, err := res.Unpack()
+	if err != nil {
+		return err
+	}
 
-				err = ErrUnableToRouteMsg
-			}
+	if typedErr, ok := val.(error); ok {
+		return typedErr
+	}
 
-			msgQuery.Resolve(err)
+	return nil
+}
 
-		// A query for the endpoint state just came in, we'll send back
-		// a copy of our current state.
-		case endpointQuery := <-p.endpointQueries:
-			endpointsCopy := make(EndpointsMap, len(endpoints))
-			maps.Copy(endpointsCopy, endpoints)
+// Endpoints returns a list of all registered endpoints.
+func (p *MultiMsgRouter) endpoints() fn.Result[EndpointsMap] {
+	future := p.actor.Ref().Ask(context.Background(), &getEndpointsMsg{})
+	res := future.Await(context.Background())
 
-			endpointQuery.Resolve(endpointsCopy)
+	val, err := res.Unpack()
+	if err != nil {
+		return fn.Err[EndpointsMap](err)
+	}
 
-		case <-p.quit:
-			return
-		}
+	if endpoints, ok := val.(EndpointsMap); ok {
+		return fn.Ok(endpoints)
 	}
+
+	return fn.Err[EndpointsMap](fmt.Errorf("unexpected response type: %T", val))
 }
 
 // A compile time check to ensure MultiMsgRouter implements the MsgRouter