File tree Expand file tree Collapse file tree 1 file changed +12
-7
lines changed
libp2p/src/main/kotlin/io/libp2p/pubsub/gossip Expand file tree Collapse file tree 1 file changed +12
-7
lines changed Original file line number Diff line number Diff line change @@ -416,15 +416,20 @@ open class GossipRouter(
416
416
}
417
417
.flatten()
418
418
}
419
- val list = peers
420
- .filterNot { peerDoesNotWantMessage(it, msg.messageId) }
421
- .map { submitPublishMessage(it, msg) }
422
419
423
420
mCache + = msg
424
- flushAllPending()
425
421
426
- return if (list.isNotEmpty()) {
427
- anyComplete(list)
422
+ return if (peers.isNotEmpty()) {
423
+ val publishedMessages = peers
424
+ .filterNot { peerDoesNotWantMessage(it, msg.messageId) }
425
+ .map { submitPublishMessage(it, msg) }
426
+ if (publishedMessages.isEmpty()) {
427
+ // all peers have sent IDONTWANT for this message id
428
+ CompletableFuture .completedFuture(Unit )
429
+ } else {
430
+ flushAllPending()
431
+ anyComplete(publishedMessages)
432
+ }
428
433
} else {
429
434
completedExceptionally(
430
435
NoPeersForOutboundMessageException (" No peers for message topics ${msg.topics} found" )
@@ -614,7 +619,7 @@ open class GossipRouter(
614
619
.flatten()
615
620
.distinct()
616
621
.minus(receivedFrom)
617
- .forEach { peer -> sendIdontwant(peer , msg.messageId) }
622
+ .forEach { sendIdontwant(it , msg.messageId) }
618
623
}
619
624
620
625
private fun enqueuePrune (peer : PeerHandler , topic : Topic ) {
You can’t perform that action at this time.
0 commit comments