Merge pull request #379 from libp2p/1.2.0

1.2.0 release

build.gradle.kts

@@ -13,15 +13,15 @@ plugins {
 
     id("org.jetbrains.kotlin.jvm") version kotlinVersion apply false
 
-    id("com.github.ben-manes.versions").version("0.48.0")
+    id("com.github.ben-manes.versions").version("0.51.0")
     id("idea")
     id("io.gitlab.arturbosch.detekt").version("1.22.0")
     id("java")
     id("maven-publish")
     id("org.jetbrains.dokka").version("1.9.20")
-    id("com.diffplug.spotless").version("6.21.0")
+    id("com.diffplug.spotless").version("6.25.0")
     id("java-test-fixtures")
-    id("io.spring.dependency-management").version("1.1.3")
+    id("io.spring.dependency-management").version("1.1.6")
 
     id("org.jetbrains.kotlin.android") version kotlinVersion apply false
     id("com.android.application") version "7.4.2" apply false
@@ -37,7 +37,7 @@ configure(
         }
 ) {
     group = "io.libp2p"
-    version = "1.1.1-RELEASE"
+    version = "1.2.0-RELEASE"
 
     apply(plugin = "kotlin")
     apply(plugin = "idea")
@@ -115,7 +115,11 @@ configure(
                     "ktlint_standard_no-wildcard-imports" to "disabled",
                     "ktlint_standard_enum-entry-name-case" to "disabled",
                     "ktlint_standard_trailing-comma-on-call-site" to "disabled",
-                    "ktlint_standard_trailing-comma-on-declaration-site" to "disabled"
+                    "ktlint_standard_trailing-comma-on-declaration-site" to "disabled",
+                    "ktlint_standard_value-parameter-comment" to "disabled",
+                    "ktlint_standard_value-argument-comment" to "disabled",
+                    "ktlint_standard_property-naming" to "disabled",
+                    "ktlint_standard_function-naming" to "disabled"
                 )
             )
         }

examples/android-chatter/build.gradle

@@ -24,6 +24,7 @@ android {
     packagingOptions {
         exclude 'META-INF/io.netty.versions.properties'
         exclude 'META-INF/INDEX.LIST'
+        exclude 'META-INF/versions/9/OSGI-INF/MANIFEST.MF'
     }
     kotlinOptions {
         jvmTarget = "11"

examples/chatter/src/main/kotlin/io/libp2p/example/chat/ChatProtocol.kt

@@ -19,9 +19,9 @@ typealias OnChatMessage = (PeerId, String) -> Unit
 
 class Chat(chatCallback: OnChatMessage) : ChatBinding(ChatProtocol(chatCallback))
 
-const val protocolId: ProtocolId = "/example/chat/0.1.0"
+const val PROTOCOL_ID: ProtocolId = "/example/chat/0.1.0"
 
-open class ChatBinding(echo: ChatProtocol) : StrictProtocolBinding<ChatController>(protocolId, echo)
+open class ChatBinding(echo: ChatProtocol) : StrictProtocolBinding<ChatController>(PROTOCOL_ID, echo)
 
 open class ChatProtocol(
     private val chatCallback: OnChatMessage

funding.json

@@ -0,0 +1,5 @@
+{
+  "opRetro": {
+    "projectId": "0x0be3a0fa062180bdfbfdefa993b09acd9edcae93ba0d8d5829dd01c138268f40"
+  }
+}

libp2p/build.gradle.kts

@@ -1,6 +1,6 @@
 plugins {
     id("com.google.protobuf").version("0.9.4")
-    id("me.champeau.jmh").version("0.7.1")
+    id("me.champeau.jmh").version("0.7.2")
 }
 
 // https://docs.gradle.org/current/userguide/java_testing.html#ex-disable-publishing-of-test-fixtures-variants

libp2p/src/main/kotlin/io/libp2p/core/multistream/NegotiatedProtocol.kt

@@ -6,7 +6,7 @@ import java.util.concurrent.CompletableFuture
 /**
  * Represents [ProtocolBinding] with exact protocol version which was agreed on
  */
-open class NegotiatedProtocol<TController, TBinding : ProtocolBinding<TController>> (
+open class NegotiatedProtocol<TController, TBinding : ProtocolBinding<TController>>(
     val binding: TBinding,
     val protocol: ProtocolId
 ) {

libp2p/src/main/kotlin/io/libp2p/crypto/Libp2pCrypto.kt

@@ -19,11 +19,11 @@ import org.bouncycastle.crypto.macs.HMac
 import org.bouncycastle.crypto.params.KeyParameter
 
 /**
- * ErrRsaKeyTooSmall is returned when trying to generate or parse an RSA key
+ * ERR_RSA_KEY_TOO_SMALL is returned when trying to generate or parse an RSA key
  * that's smaller than 512 bits. Keys need to be larger enough to sign a 256bit
  * hash so this is a reasonable absolute minimum.
  */
-const val ErrRsaKeyTooSmall = "rsa keys must be >= 512 bits to be useful"
+const val ERR_RSA_KEY_TOO_SMALL = "rsa keys must be >= 512 bits to be useful"
 
 const val RSA_ALGORITHM = "RSA"
 const val SHA_ALGORITHM = "SHA-256"

libp2p/src/main/kotlin/io/libp2p/crypto/keys/Rsa.kt

@@ -16,7 +16,7 @@ import crypto.pb.Crypto
 import io.libp2p.core.Libp2pException
 import io.libp2p.core.crypto.PrivKey
 import io.libp2p.core.crypto.PubKey
-import io.libp2p.crypto.ErrRsaKeyTooSmall
+import io.libp2p.crypto.ERR_RSA_KEY_TOO_SMALL
 import io.libp2p.crypto.KEY_PKCS8
 import io.libp2p.crypto.Libp2pCrypto
 import io.libp2p.crypto.RSA_ALGORITHM
@@ -100,7 +100,7 @@ class RsaPublicKey(private val k: JavaPublicKey) : PubKey(Crypto.KeyType.RSA) {
 @JvmOverloads
 fun generateRsaKeyPair(bits: Int, random: SecureRandom = SecureRandom()): Pair<PrivKey, PubKey> {
     if (bits < 2048) {
-        throw Libp2pException(ErrRsaKeyTooSmall)
+        throw Libp2pException(ERR_RSA_KEY_TOO_SMALL)
     }
 
     val kp: KeyPair = with(

libp2p/src/main/kotlin/io/libp2p/discovery/MDnsDiscovery.kt

@@ -1,23 +1,21 @@
 package io.libp2p.discovery
 
-import io.libp2p.core.Discoverer
-import io.libp2p.core.Host
-import io.libp2p.core.PeerId
-import io.libp2p.core.PeerInfo
-import io.libp2p.core.PeerListener
+import io.libp2p.core.*
 import io.libp2p.core.multiformats.Multiaddr
+import io.libp2p.core.multiformats.MultiaddrComponent
 import io.libp2p.core.multiformats.Protocol
 import io.libp2p.discovery.mdns.AnswerListener
 import io.libp2p.discovery.mdns.JmDNS
 import io.libp2p.discovery.mdns.ServiceInfo
 import io.libp2p.discovery.mdns.impl.DNSRecord
 import io.libp2p.discovery.mdns.impl.constants.DNSRecordType
-import java.net.Inet4Address
-import java.net.Inet6Address
-import java.net.InetAddress
+import java.net.*
+import java.util.*
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.ForkJoinPool
+import java.util.stream.Collectors
+import java.util.stream.Stream
 
 class MDnsDiscovery(
     private val host: Host,
@@ -61,6 +59,10 @@ class MDnsDiscovery(
         newPeerFoundListeners.forEach { it(peerInfo) }
     }
 
+    fun addHandler(h: PeerListener) {
+        newPeerFoundListeners += h
+    }
+
     private fun ipfsDiscoveryInfo(): ServiceInfo {
         return ServiceInfo.create(
             serviceTag,
@@ -87,11 +89,80 @@ class MDnsDiscovery(
         return Integer.parseInt(str)
     }
 
+    /* /ip6/::/tcp/4001 should expand to the following for example:
+    "/ip6/0:0:0:0:0:0:0:1/udp/4001/quic"
+    "/ip4/50.116.48.246/tcp/4001"
+    "/ip4/127.0.0.1/tcp/4001"
+    "/ip6/2600:3c03:0:0:f03c:92ff:fee7:bc1c/tcp/4001"
+    "/ip6/0:0:0:0:0:0:0:1/tcp/4001"
+    "/ip4/50.116.48.246/udp/4001/quic"
+    "/ip4/127.0.0.1/udp/4001/quic"
+    "/ip6/2600:3c03:0:0:f03c:92ff:fee7:bc1c/udp/4001/quic"
+     */
+    fun expandWildcardAddresses(addr: Multiaddr): List<Multiaddr> {
+        // Do not include /p2p or /ipfs components which are superfluous here
+        if (!isWildcard(addr)) {
+            return java.util.List.of(
+                Multiaddr(
+                    addr.components
+                        .stream()
+                        .filter { c: MultiaddrComponent ->
+                            (
+                                c.protocol !== Protocol.P2P &&
+                                    c.protocol !== Protocol.IPFS
+                                )
+                        }
+                        .collect(Collectors.toList())
+                )
+            )
+        }
+        if (addr.has(Protocol.IP4)) return listNetworkAddresses(false, addr)
+        return if (addr.has(Protocol.IP6)) listNetworkAddresses(true, addr) else emptyList()
+    }
+
+    fun listNetworkAddresses(includeIp6: Boolean, addr: Multiaddr): List<Multiaddr> {
+        return try {
+            Collections.list(NetworkInterface.getNetworkInterfaces()).stream()
+                .flatMap { net: NetworkInterface ->
+                    net.interfaceAddresses.stream()
+                        .map { obj: InterfaceAddress -> obj.address }
+                        .filter { ip: InetAddress? -> includeIp6 || ip is Inet4Address }
+                }
+                .map { ip: InetAddress ->
+                    Multiaddr(
+                        Stream.concat(
+                            Stream.of(
+                                MultiaddrComponent(
+                                    if (ip is Inet4Address) Protocol.IP4 else Protocol.IP6,
+                                    ip.address
+                                )
+                            ),
+                            addr.components.stream()
+                                .filter { c: MultiaddrComponent ->
+                                    c.protocol !== Protocol.IP4 && c.protocol !== Protocol.IP6 && c.protocol !== Protocol.P2P && c.protocol !== Protocol.IPFS
+                                }
+                        )
+                            .collect(Collectors.toList())
+                    )
+                }
+                .collect(Collectors.toList())
+        } catch (e: SocketException) {
+            throw RuntimeException(e)
+        }
+    }
+
+    fun isWildcard(addr: Multiaddr): Boolean {
+        val s = addr.toString()
+        return s.contains("/::/") || s.contains("/0:0:0:0/")
+    }
+
     private fun ip4Addresses() = ipAddresses(Protocol.IP4, Inet4Address::class.java)
     private fun ip6Addresses() = ipAddresses(Protocol.IP6, Inet6Address::class.java)
 
     private fun <R> ipAddresses(protocol: Protocol, klass: Class<R>): List<R> {
-        return host.listenAddresses().map {
+        return host.listenAddresses().flatMap {
+            expandWildcardAddresses(it)
+        }.map {
             it.getFirstComponent(protocol)
         }.filterNotNull().map {
             InetAddress.getByAddress(localhost.hostName, it.value)
@@ -112,7 +183,7 @@ class MDnsDiscovery(
                 val aRecords = answers.filter { DNSRecordType.TYPE_A.equals(it.recordType) }
                 val aaaaRecords = answers.filter { DNSRecordType.TYPE_AAAA.equals(it.recordType) }
 
-                if (txtRecord == null || srvRecord == null || aRecords.isEmpty()) {
+                if (txtRecord == null || srvRecord == null || (aRecords.isEmpty() && aaaaRecords.isEmpty())) {
                     return // incomplete answers
                 }
 

libp2p/src/main/kotlin/io/libp2p/etc/util/netty/AbstractChildChannel.kt

@@ -20,7 +20,10 @@ import java.net.SocketAddress
  */
 abstract class AbstractChildChannel(parent: Channel, id: ChannelId?) : AbstractChannel(parent, id) {
     private enum class State {
-        OPEN, ACTIVE, INACTIVE, CLOSED
+        OPEN,
+        ACTIVE,
+        INACTIVE,
+        CLOSED
     }
 
     private val parentCloseFuture = parent.closeFuture()

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

@@ -17,8 +17,6 @@ import java.util.Collections.singletonList
 import java.util.Optional
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.ScheduledExecutorService
-import java.util.function.BiConsumer
-import java.util.function.Consumer
 
 // 1 MB default max message size
 const val DEFAULT_MAX_PUBSUB_MESSAGE_SIZE = 1 shl 20
@@ -223,7 +221,7 @@ abstract class AbstractRouter(
 
         validFuts.forEach { (msg, validationFut) ->
             validationFut.thenAcceptAsync(
-                Consumer { res ->
+                { res ->
                     seenMessages[msg] = Optional.of(res)
                     if (res == ValidationResult.Invalid) notifyUnseenInvalidMessage(peer, msg)
                 },
@@ -247,7 +245,7 @@ abstract class AbstractRouter(
         // broadcast others on completion
         undone.forEach {
             it.second.whenCompleteAsync(
-                BiConsumer { res, err ->
+                { res, err ->
                     when {
                         err != null -> logger.warn("Exception while handling message from peer $peer: ${it.first}", err)
                         res == ValidationResult.Invalid -> logger.debug("Invalid pubsub message from peer $peer: ${it.first}")

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

@@ -6,10 +6,25 @@ enum class PubsubProtocol(val announceStr: ProtocolId) {
 
     Gossip_V_1_0("/meshsub/1.0.0"),
     Gossip_V_1_1("/meshsub/1.1.0"),
+    Gossip_V_1_2("/meshsub/1.2.0"),
     Floodsub("/floodsub/1.0.0");
 
     companion object {
         fun fromProtocol(protocol: ProtocolId) = PubsubProtocol.values().find { protocol == it.announceStr }
             ?: throw NoSuchElementException("No PubsubProtocol found with protocol $protocol")
     }
+
+    /**
+     * https://github.com/libp2p/specs/blob/master/pubsub/gossipsub/gossipsub-v1.1.md#prune-backoff-and-peer-exchange
+     */
+    fun supportsBackoffAndPX(): Boolean {
+        return this == Gossip_V_1_1 || this == Gossip_V_1_2
+    }
+
+    /**
+     * https://github.com/libp2p/specs/blob/master/pubsub/gossipsub/gossipsub-v1.2.md#idontwant-message
+     */
+    fun supportsIDontWant(): Boolean {
+        return this == Gossip_V_1_2
+    }
 }

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

@@ -51,7 +51,9 @@ class SimpleSeenCache<TValue> : SeenCache<TValue> {
     override fun put(msg: PubsubMessage, value: TValue) {
         map[msg.messageId] = value
     }
-    override fun remove(messageId: MessageId) { map -= messageId }
+    override fun remove(messageId: MessageId) {
+        map -= messageId
+    }
 }
 
 class LRUSeenCache<TValue>(val delegate: SeenCache<TValue>, private val maxSize: Int) : SeenCache<TValue> by delegate {

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

@@ -30,13 +30,23 @@ class Gossip @JvmOverloads constructor(
     }
 
     override val protocolDescriptor =
-        if (router.protocol == PubsubProtocol.Gossip_V_1_1) {
-            ProtocolDescriptor(
-                PubsubProtocol.Gossip_V_1_1.announceStr,
-                PubsubProtocol.Gossip_V_1_0.announceStr
-            )
-        } else {
-            ProtocolDescriptor(PubsubProtocol.Gossip_V_1_0.announceStr)
+        when (router.protocol) {
+            PubsubProtocol.Gossip_V_1_2 -> {
+                ProtocolDescriptor(
+                    PubsubProtocol.Gossip_V_1_2.announceStr,
+                    PubsubProtocol.Gossip_V_1_1.announceStr,
+                    PubsubProtocol.Gossip_V_1_0.announceStr
+                )
+            }
+            PubsubProtocol.Gossip_V_1_1 -> {
+                ProtocolDescriptor(
+                    PubsubProtocol.Gossip_V_1_1.announceStr,
+                    PubsubProtocol.Gossip_V_1_0.announceStr
+                )
+            }
+            else -> {
+                ProtocolDescriptor(PubsubProtocol.Gossip_V_1_0.announceStr)
+            }
         }
 
     override fun handleConnection(conn: Connection) {

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

@@ -231,7 +231,24 @@ data class GossipParams(
      * callback to notify outer system to which peers Gossip wants to be connected
      * The second parameter is a signed peer record: https://github.com/libp2p/specs/pull/217
      */
-    val connectCallback: (PeerId, ByteArray) -> Unit = { _: PeerId, _: ByteArray -> }
+    val connectCallback: (PeerId, ByteArray) -> Unit = { _: PeerId, _: ByteArray -> },
+
+    /**
+     * [maxIDontWantMessageIds] is the maximum number of IDONTWANT message ids allowed per heartbeat per peer
+     */
+    val maxIDontWantMessageIds: Int = maxIHaveLength * maxIHaveMessages,
+
+    /**
+     * [iDontWantMinMessageSizeThreshold] controls the minimum size (in bytes) that an incoming message needs to be so that an IDONTWANT message is sent to mesh peers.
+     * The default is 16 KB.
+     */
+    val iDontWantMinMessageSizeThreshold: Int = 16000,
+
+    /**
+     * [iDontWantTTL] Expiry time for cache of received IDONTWANT messages for peers
+     */
+    val iDontWantTTL: Duration = 3.seconds
+
 ) {
     init {
         check(D >= 0, "D should be >= 0")