Merge pull request #1509 from lightninglabs/rqf-response-matching

rfq+rpcserver: match event to order

Author: Roasbeef

rfq/manager.go

@@ -1116,6 +1116,24 @@ func (q *PeerAcceptedBuyQuoteEvent) Timestamp() time.Time {
 	return q.timestamp.UTC()
 }
 
+// MatchesOrder checks if the sell quote matches the provided order.
+func (q *PeerAcceptedBuyQuoteEvent) MatchesOrder(order BuyOrder) bool {
+	if q.Request.AssetSpecifier != order.AssetSpecifier {
+		return false
+	}
+
+	// If the order has no peer, we accept equality just based on the
+	// specifier.
+	if order.Peer.IsNone() {
+		return true
+	}
+
+	// If a peer is specified, ensure it matches the event's peer.
+	return fn.MapOptionZ(order.Peer, func(vertex route.Vertex) bool {
+		return q.Peer == vertex
+	})
+}
+
 // Ensure that the PeerAcceptedBuyQuoteEvent struct implements the Event
 // interface.
 var _ fn.Event = (*PeerAcceptedBuyQuoteEvent)(nil)
@@ -1197,6 +1215,24 @@ func (q *PeerAcceptedSellQuoteEvent) Timestamp() time.Time {
 	return q.timestamp.UTC()
 }
 
+// MatchesOrder checks if the sell quote matches the provided order.
+func (q *PeerAcceptedSellQuoteEvent) MatchesOrder(order SellOrder) bool {
+	if q.Request.AssetSpecifier != order.AssetSpecifier {
+		return false
+	}
+
+	// If the order has no peer, we accept equality just based on the
+	// specifier.
+	if order.Peer.IsNone() {
+		return true
+	}
+
+	// If a peer is specified, ensure it matches the event's peer.
+	return fn.MapOptionZ(order.Peer, func(vertex route.Vertex) bool {
+		return q.Peer == vertex
+	})
+}
+
 // Ensure that the PeerAcceptedSellQuoteEvent struct implements the Event
 // interface.
 var _ fn.Event = (*PeerAcceptedSellQuoteEvent)(nil)

rpcserver.go

@@ -6853,8 +6853,26 @@ func (r *rpcServer) AddAssetBuyOrder(ctx context.Context,
 	timeout := time.After(time.Second * time.Duration(req.TimeoutSeconds))
 
 	for {
+		type targetEventType = *rfq.PeerAcceptedBuyQuoteEvent
 		select {
 		case event := <-eventSubscriber.NewItemCreated.ChanOut():
+			acceptedQuote, ok := event.(targetEventType)
+			if !ok {
+				rpcsLog.Debugf("Received event of type %T "+
+					"but expected accepted sell quote, "+
+					"skipping", event)
+
+				continue
+			}
+
+			if !acceptedQuote.MatchesOrder(*buyOrder) {
+				rpcsLog.Debugf("Received event of type %T "+
+					"but order doesn't match, skipping",
+					event)
+
+				continue
+			}
+
 			resp, err := rfq.NewAddAssetBuyOrderResponse(event)
 			if err != nil {
 				return nil, fmt.Errorf("error marshalling "+
@@ -7027,8 +7045,26 @@ func (r *rpcServer) AddAssetSellOrder(ctx context.Context,
 	timeout := time.After(time.Second * time.Duration(req.TimeoutSeconds))
 
 	for {
+		type targetEventType = *rfq.PeerAcceptedSellQuoteEvent
 		select {
 		case event := <-eventSubscriber.NewItemCreated.ChanOut():
+			acceptedQuote, ok := event.(targetEventType)
+			if !ok {
+				rpcsLog.Debugf("Received event of type %T "+
+					"but expected accepted sell quote, "+
+					"skipping", event)
+
+				continue
+			}
+
+			if !acceptedQuote.MatchesOrder(*sellOrder) {
+				rpcsLog.Debugf("Received event of type %T "+
+					"but order doesn't match, skipping",
+					event)
+
+				continue
+			}
+
 			resp, err := rfq.NewAddAssetSellOrderResponse(event)
 			if err != nil {
 				return nil, fmt.Errorf("error marshalling "+