Updated naming convention

Author: lucassaldanha

libp2p/src/main/kotlin/io/libp2p/pubsub/AbstractRouter.kt

@@ -149,9 +149,6 @@ abstract class AbstractRouter(
         subscribedTopics.forEach {
             partsQueue.addSubscribe(it)
         }
-
-        // TODO end control extension
-
         flushPending(peer)
     }
 
@@ -188,7 +185,9 @@ abstract class AbstractRouter(
             processControl(msg.control, peer)
         }
 
-        if (protocol.supportsExtensions()) {
+        // TODO we need to handle the existence of extension messages more generically (https://github.com/libp2p/jvm-libp2p/issues/441)
+
+        if (protocol.supportsExtensions() && (msg.hasTestExtension() || msg.hasPartial())) {
             processExtensions(msg, peer)
         }
 

libp2p/src/main/kotlin/io/libp2p/pubsub/gossip/Gossip.kt

@@ -12,6 +12,7 @@ import io.libp2p.pubsub.PubsubApiImpl
 import io.libp2p.pubsub.PubsubProtocol
 import io.libp2p.pubsub.gossip.builders.GossipRouterBuilder
 import io.netty.channel.ChannelHandler
+import org.slf4j.LoggerFactory
 import java.util.concurrent.CompletableFuture
 
 class Gossip @JvmOverloads constructor(
@@ -21,6 +22,8 @@ class Gossip @JvmOverloads constructor(
 ) :
     ProtocolBinding<Unit>, ConnectionHandler, PubsubApi by api {
 
+    private val logger = LoggerFactory.getLogger(Gossip::class.java)
+
     fun updateTopicScoreParams(scoreParams: Map<String, GossipTopicScoreParams>) {
         router.score.updateTopicParams(scoreParams)
     }
@@ -62,6 +65,7 @@ class Gossip @JvmOverloads constructor(
     }
 
     override fun initChannel(ch: P2PChannel, selectedProtocol: String): CompletableFuture<out Unit> {
+        logger.trace("Gossip initChannel - selected protocol: {}", selectedProtocol)
         router.addPeerWithDebugHandler(ch as Stream, debugGossipHandler)
         return CompletableFuture.completedFuture(Unit)
     }

libp2p/src/main/kotlin/io/libp2p/pubsub/gossip/GossipExtensionsState.kt

@@ -6,33 +6,33 @@ import pubsub.pb.Rpc
 class GossipExtensionsState {
 
     /*
-        Tracks the peers that we have already sent an extension control message
+        Tracks the peers that we have already sent a control extensions message
      */
-    private val outgoingExtensionControlMsgPeers: MutableSet<PeerId> = mutableSetOf()
+    private val outgoingControlExtensionsMsgPeers: MutableSet<PeerId> = mutableSetOf()
 
     /*
-        Tracks peers that already sent us an extension control message
+        Tracks peers that already sent us a control extensions message
      */
     private val peerExtensionSupportMap: MutableMap<PeerId, Rpc.ControlExtensions> = mutableMapOf()
 
     fun onPeerDisconnected(peer: PeerId) {
-        outgoingExtensionControlMsgPeers.remove(peer)
+        outgoingControlExtensionsMsgPeers.remove(peer)
         peerExtensionSupportMap.remove(peer)
     }
 
-    fun onExtensionControlMessage(ctrlExtensions: Rpc.ControlExtensions, receivedFrom: PeerId) {
+    fun onControlExtensionsMessage(ctrlExtensions: Rpc.ControlExtensions, receivedFrom: PeerId) {
         peerExtensionSupportMap[receivedFrom] = ctrlExtensions
     }
 
     fun registerControlExtensionMessageSentToPeers(peerId: PeerId) {
-        outgoingExtensionControlMsgPeers.add(peerId)
+        outgoingControlExtensionsMsgPeers.add(peerId)
     }
 
     fun peerSupportedExtensions(peerId: PeerId) = peerExtensionSupportMap[peerId]
 
-    fun hasReceivedExtensionControlFrom(peer: PeerId) =
+    fun hasReceivedControlExtensionsFrom(peer: PeerId) =
         peerExtensionSupportMap.contains(peer)
 
-    fun hasSentExtensionControlTo(peer: PeerId) =
-        outgoingExtensionControlMsgPeers.contains(peer)
+    fun hasSentControlExtensionsTo(peer: PeerId) =
+        outgoingControlExtensionsMsgPeers.contains(peer)
 }

libp2p/src/main/kotlin/io/libp2p/pubsub/gossip/GossipRouter.kt

@@ -167,7 +167,7 @@ open class GossipRouter(
         super.onPeerActive(peer)
         eventBroadcaster.notifyConnected(peer.peerId, peer.getRemoteAddress())
         heartbeatTask.hashCode() // force lazy initialization
-        sendExtensionsControl(peer)
+        sendControlExtensions(peer)
     }
 
     override fun notifyUnseenMessage(peer: PeerHandler, msg: PubsubMessage) {
@@ -400,35 +400,56 @@ open class GossipRouter(
     ) {
         logger.trace("Received control extension {}", ctrlExtensions.toString())
 
-        if (gossipExtensionsState.hasReceivedExtensionControlFrom(receivedFrom.peerId)) {
+        if (gossipExtensionsState.hasReceivedControlExtensionsFrom(receivedFrom.peerId)) {
             // TODO Should disconnect peers that send control extension multiple times (https://github.com/libp2p/jvm-libp2p/issues/437)
             logger.trace(
                 "Received another control extension message from peer {}",
                 receivedFrom.peerId
             )
             return
         } else {
-            gossipExtensionsState.onExtensionControlMessage(ctrlExtensions, receivedFrom.peerId)
+            gossipExtensionsState.onControlExtensionsMessage(ctrlExtensions, receivedFrom.peerId)
         }
     }
 
     override fun processExtensions(msg: Rpc.RPC, receivedFrom: PeerHandler) {
         val peerSupportedExtensions =
             gossipExtensionsState.peerSupportedExtensions(receivedFrom.peerId)
-        if (peerSupportedExtensions == null) {
+
+        // TODO Revisit this logic as part of adding feature flags (https://github.com/libp2p/jvm-libp2p/issues/441)
+
+        when {
+            msg.hasTestExtension() && checkPeerExtensionSupport(
+                peerSupportedExtensions,
+                Rpc.ControlExtensions::hasTestExtension
+            ) ->
+                processTestExtensionMessage(msg.testExtension, receivedFrom)
+
+            msg.hasPartial() && checkPeerExtensionSupport(
+                peerSupportedExtensions,
+                Rpc.ControlExtensions::hasPartialMessages
+            ) ->
+                processPartialMessageExtension(msg.partial, receivedFrom)
+        }
+    }
+
+    private fun checkPeerExtensionSupport(
+        peerSavedPreferences: Rpc.ControlExtensions?,
+        checkSupportFunction: (Rpc.ControlExtensions) -> Boolean
+    ): Boolean {
+        if (peerSavedPreferences == null) {
+            return false
+        }
+
+        if (!checkSupportFunction.invoke(peerSavedPreferences)) {
             logger.trace(
-                "Ignoring extension messages from peer {} - did it send an extension control message?",
-                receivedFrom.peerId
+                "Ignoring extension messages from peer {} - did it send an control extensions message?",
+                peerSavedPreferences
             )
-        } else {
-            when {
-                peerSupportedExtensions.hasTestExtension() && msg.hasTestExtension() ->
-                    processTestExtensionMessage(msg.testExtension, receivedFrom)
-
-                peerSupportedExtensions.hasPartialMessages() && msg.hasPartial() ->
-                    processPartialMessageExtension(msg.partial, receivedFrom)
-            }
+            return false
         }
+
+        return true
     }
 
     private fun processTestExtensionMessage(
@@ -582,7 +603,7 @@ open class GossipRouter(
             lastPublished -= topic
         }
 
-        activePeers.forEach { sendExtensionsControl(it) }
+        activePeers.forEach { sendControlExtensions(it) }
     }
 
     override fun unsubscribe(topic: Topic) {
@@ -783,23 +804,25 @@ open class GossipRouter(
         send(peer, iDontWant)
     }
 
-    private fun sendExtensionsControl(peer: PeerHandler) {
+    private fun sendControlExtensions(peer: PeerHandler) {
         if (!this.protocol.supportsExtensions()) {
             logger.trace(
-                "Protocol does not support extensions. Won't send extensions control message."
+                "Protocol does not support extensions. Won't send control extensions message."
             )
             return
         }
 
-        if (gossipExtensionsState.hasSentExtensionControlTo(peer.peerId)) {
+        if (gossipExtensionsState.hasSentControlExtensionsTo(peer.peerId)) {
             logger.trace(
-                "Already sent extension control msg to peer {}. Won't send another one.",
+                "Already sent control extensions msg to peer {}. Won't send another one.",
                 peer.peerId
             )
             return
         }
 
-        pendingRpcParts.getQueue(peer).addExtensionsControl(
+        logger.trace("Sending control extensions message to peer {}", peer.peerId)
+
+        pendingRpcParts.getQueue(peer).addControlExtensions(
             Rpc.ControlExtensions.newBuilder()
                 .setTestExtension(true)
                 .setPartialMessages(true)

libp2p/src/main/kotlin/io/libp2p/pubsub/gossip/GossipRpcPartsQueue.kt

@@ -28,7 +28,7 @@ interface GossipRpcPartsQueue : RpcPartsQueue {
     fun addPrune(topic: Topic, backoffSeconds: Long, backoffPeers: List<PeerId>)
 
     // TODO Need to check if we should handle when control extension and extension messages could be separated by split  (https://github.com/libp2p/jvm-libp2p/issues/440)
-    fun addExtensionsControl(ctrlMessage: Rpc.ControlExtensions)
+    fun addControlExtensions(ctrlMessage: Rpc.ControlExtensions)
 }
 
 /**
@@ -110,7 +110,7 @@ open class DefaultGossipRpcPartsQueue(
         addPart(PrunePart(topic, backoffSeconds, backoffPeers))
     }
 
-    override fun addExtensionsControl(ctrlMessage: Rpc.ControlExtensions) {
+    override fun addControlExtensions(ctrlMessage: Rpc.ControlExtensions) {
         addPart(ControlExtensionPart(ctrlMessage))
     }