WIP: add asset group key field to QuoteRequest

Author: ffranr

chain_bridge.go

@@ -200,6 +200,15 @@ func (l *LndRpcChainBridge) EstimateFee(ctx context.Context,
 	return l.lnd.WalletKit.EstimateFeeRate(ctx, int32(confTarget))
 }
 
+// SubscribeCustomMessages creates a subscription to custom messages received
+// from our peers.
+func (l *LndRpcChainBridge) SubscribeCustomMessages(
+	ctx context.Context) (<-chan lndclient.CustomMessage,
+	<-chan error, error) {
+
+	return l.lnd.Client.SubscribeCustomMessages(ctx)
+}
+
 // A compile time assertion to ensure LndRpcChainBridge meets the
 // tapgarden.ChainBridge interface.
 var _ tapgarden.ChainBridge = (*LndRpcChainBridge)(nil)

config.go

@@ -10,6 +10,7 @@ import (
 	"github.com/lightninglabs/taproot-assets/address"
 	"github.com/lightninglabs/taproot-assets/monitoring"
 	"github.com/lightninglabs/taproot-assets/proof"
+	"github.com/lightninglabs/taproot-assets/rfq"
 	"github.com/lightninglabs/taproot-assets/tapdb"
 	"github.com/lightninglabs/taproot-assets/tapfreighter"
 	"github.com/lightninglabs/taproot-assets/tapgarden"
@@ -122,6 +123,8 @@ type Config struct {
 
 	UniverseFederation *universe.FederationEnvoy
 
+	RfqManager *rfq.Manager
+
 	UniverseStats universe.Telemetry
 
 	// UniversePublicAccess is flag which, If true, and the Universe server

rfq/log.go

@@ -0,0 +1,26 @@
+package rfq
+
+import (
+	"github.com/btcsuite/btclog"
+)
+
+// Subsystem defines the logging code for this subsystem.
+const Subsystem = "RFQ"
+
+// log is a logger that is initialized with no output filters.  This
+// means the package will not perform any logging by default until the caller
+// requests it.
+var log = btclog.Disabled
+
+// DisableLog disables all library log output.  Logging output is disabled
+// by default until UseLogger is called.
+func DisableLog() {
+	UseLogger(btclog.Disabled)
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

rfq/rfq.go

@@ -0,0 +1,124 @@
+package rfq
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/lightninglabs/taproot-assets/fn"
+)
+
+type ManagerCfg struct {
+	PeerMessagePorter PeerMessagePorter
+}
+
+// Manager is a struct that handles RFQ order management.
+type Manager struct {
+	startOnce sync.Once
+	stopOnce  sync.Once
+
+	cfg ManagerCfg
+
+	rfqStreamHandle *StreamHandler
+
+	// ContextGuard provides a wait group and main quit channel that can be
+	// used to create guarded contexts.
+	*fn.ContextGuard
+}
+
+func NewManager(cfg ManagerCfg) (Manager, error) {
+	return Manager{
+		cfg: cfg,
+	}, nil
+}
+
+// Start attempts to start a new RFQ manager.
+func (m *Manager) Start() error {
+	var startErr error
+	m.startOnce.Do(func() {
+		ctx, cancel := m.WithCtxQuitNoTimeout()
+		defer cancel()
+
+		log.Info("Initializing RFQ subsystems")
+		err := m.initSubsystems(ctx)
+		if err != nil {
+			startErr = err
+			return
+		}
+
+		// Start the manager's main event loop in a separate goroutine.
+		m.Wg.Add(1)
+		go func() {
+			defer m.Wg.Done()
+
+			log.Info("Starting RFQ manager main event loop")
+			err = m.mainEventLoop()
+			if err != nil {
+				startErr = err
+				return
+			}
+		}()
+	})
+	return startErr
+}
+
+func (m *Manager) Stop() error {
+	var stopErr error
+
+	m.stopOnce.Do(func() {
+		log.Info("Stopping RFQ manager")
+
+		err := m.rfqStreamHandle.Stop()
+		if err != nil {
+			stopErr = fmt.Errorf("error stopping RFQ stream "+
+				"handler: %w", err)
+			return
+		}
+	})
+
+	return stopErr
+}
+
+func (m *Manager) initSubsystems(ctx context.Context) error {
+	var err error
+
+	// Initialise the RFQ raw message stream handler and start it in a
+	// separate goroutine.
+	m.rfqStreamHandle, err = NewStreamHandler(ctx, m.cfg.PeerMessagePorter)
+	if err != nil {
+		return fmt.Errorf("failed to create RFQ stream handler: %w",
+			err)
+	}
+
+	m.Wg.Add(1)
+	go func() {
+		defer m.Wg.Done()
+
+		log.Info("Starting RFQ stream handler")
+		err = m.rfqStreamHandle.Start()
+		if err != nil {
+			return
+		}
+	}()
+
+	return nil
+}
+
+func (m *Manager) mainEventLoop() error {
+	for {
+		select {
+		// Handle RFQ message stream events.
+		case quoteReq := <-m.rfqStreamHandle.IncomingQuoteRequests.NewItemCreated.ChanOut():
+			log.Debugf("Received RFQ quote request from message "+
+				"stream handler: %v", quoteReq)
+			// TODO(ffranr): send to negotiator (+ price oracle)
+
+		case errStream := <-m.rfqStreamHandle.ErrChan:
+			return fmt.Errorf("error received from RFQ stream "+
+				"handler: %w", errStream)
+
+		case <-m.Quit:
+			return nil
+		}
+	}
+}