Merge pull request #177 from libp2p/0.7.0

Release 0.7.0
* Add the ability to pre-handle any inbound or outbound stream (#170)
* Check the full causal chain when handling exceptions (#173)
* Update gossip scoring APIs: allow topic scoring params to be updated after startup (#175)

Author: mbaxter

README.md

@@ -69,14 +69,14 @@ Builds are published to JCenter. Maven Central mirrors JCenter, but updates can
        jcenter()
    }
 
-   implementation 'io.libp2p:jvm-libp2p-minimal:0.6.4-RELEASE'
+   implementation 'io.libp2p:jvm-libp2p-minimal:0.7.0-RELEASE'
 ```
 ### Using Maven
 ``` 
   <dependency>
     <groupId>io.libp2p</groupId>
     <artifactId>jvm-libp2p-minimal</artifactId>
-    <version>0.6.4-RELEASE</version>
+    <version>0.7.0-RELEASE</version>
     <type>pom</type>
   </dependency>
 ```

build.gradle.kts

@@ -13,7 +13,7 @@ import java.nio.file.Paths
 // ./gradlew bintrayUpload -PbintrayUser=<user> -PbintrayApiKey=<api-key>
 
 group = "io.libp2p"
-version = "0.6.4-RELEASE"
+version = "0.7.0-RELEASE"
 description = "a minimal implementation of libp2p for the jvm"
 
 plugins {

src/main/java/io/libp2p/core/dsl/HostBuilder.java

@@ -3,7 +3,7 @@
 import io.libp2p.core.Host;
 import io.libp2p.core.crypto.PrivKey;
 import io.libp2p.core.multistream.ProtocolBinding;
-import io.libp2p.core.mux.StreamMuxer;
+import io.libp2p.core.mux.StreamMuxerProtocol;
 import io.libp2p.core.security.SecureChannel;
 import io.libp2p.core.transport.Transport;
 import io.libp2p.transport.ConnectionUpgrader;
@@ -48,7 +48,7 @@ public final HostBuilder secureChannel(
 
     @SafeVarargs
     public final HostBuilder muxer(
-            Supplier<StreamMuxer>... muxers) {
+            Supplier<StreamMuxerProtocol>... muxers) {
         muxers_.addAll(Arrays.asList(muxers));
         return this;
     }
@@ -79,7 +79,7 @@ public Host build() {
                     b.getSecureChannels().add(sc::apply)
                 );
                 muxers_.forEach(m ->
-                    b.getMuxers().add(m::get)
+                    b.getMuxers().add(m.get())
                 );
                 b.getProtocols().addAll(protocols_);
                 listenAddresses_.forEach(a ->
@@ -92,7 +92,7 @@ public Host build() {
     private DefaultMode defaultMode_;
     private List<Function<ConnectionUpgrader, Transport>> transports_ = new ArrayList<>();
     private List<Function<PrivKey, SecureChannel>> secureChannels_ = new ArrayList<>();
-    private List<Supplier<StreamMuxer>> muxers_ = new ArrayList<>();
+    private List<Supplier<StreamMuxerProtocol>> muxers_ = new ArrayList<>();
     private List<ProtocolBinding<?>> protocols_ = new ArrayList<>();
     private List<String> listenAddresses_ = new ArrayList<>();
 }

src/main/kotlin/io/libp2p/core/ConnectionHandler.kt

@@ -5,7 +5,7 @@ import io.libp2p.etc.BroadcastConnectionHandler
 /**
  * The same as [P2PChannelHandler] with the [Connection] specialized [P2PChannel]
  */
-interface ConnectionHandler {
+fun interface ConnectionHandler {
 
     fun handleConnection(conn: Connection)
 
@@ -19,9 +19,6 @@ interface ConnectionHandler {
         }
         fun createBroadcast(handlers: List<ConnectionHandler> = listOf()): Broadcast =
             BroadcastConnectionHandler().also { it += handlers }
-
-        fun createStreamHandlerInitializer(streamHandler: StreamHandler<*>) =
-            create { it.muxerSession().inboundStreamHandler = streamHandler }
     }
 
     interface Broadcast : ConnectionHandler, MutableList<ConnectionHandler>

src/main/kotlin/io/libp2p/core/Host.kt

@@ -41,7 +41,7 @@ interface Host {
      * List of all streams opened at the moment across all the [Connection]s
      * Please note that this list is updated asynchronously so the streams upon receiving
      * of this list can be already closed or not yet completely initialized
-     * To be synchronously notified on stream creation use [addStreamHandler] and
+     * To be synchronously notified on stream creation use [addStreamVisitor] and
      * use [Stream.closeFuture] to be synchronously notified on stream close
      */
     val streams: List<Stream>
@@ -59,17 +59,20 @@ interface Host {
     fun stop(): CompletableFuture<Void>
 
     /**
-     * Adds a handler which is notified when a new [Stream] is created
-     * Note that this is just a hook to be informed on a stream creation
-     * and no actual [Stream.nettyChannel] initialization should happen here.
-     * Refer to [addProtocolHandler] to setup a specific protocol handler
+     * Add the [ChannelVisitor] which would be invoked prior to any protocol [StreamHandler]s on any
+     * created inbound or outbound [Stream]
+     * The [streamVisitor] is free to setup any handlers on a [Stream] however those handlers
+     * should be careful to propagate any events up/down the Netty pipeline and not modify
+     * [ByteBuf]s to keep protocol [StreamHandler]s functioning as expected
      */
-    fun addStreamHandler(handler: StreamHandler<*>)
+    fun addStreamVisitor(streamVisitor: ChannelVisitor<Stream>)
 
     /**
-     * Removes the handler added with [addStreamHandler]
+     * Removes the visitor added with [addStreamVisitor]
+     * Please note that removing a visitor doesn't affect any Netty handlers installed by the visitor
+     * on any streams created before
      */
-    fun removeStreamHandler(handler: StreamHandler<*>)
+    fun removeStreamVisitor(streamVisitor: ChannelVisitor<Stream>)
 
     /**
      * Adds a new supported protocol 'on the fly'

src/main/kotlin/io/libp2p/core/P2PChannelHandler.kt

@@ -1,23 +1,39 @@
 package io.libp2p.core
 
+import io.libp2p.etc.BroadcastChannelVisitor
 import java.util.concurrent.CompletableFuture
 
 /**
  * The central entry point for every protocol which is responsible for initializing [P2PChannel]
  */
-interface P2PChannelHandler<out TController> {
+fun interface P2PChannelHandler<TController> {
 
     /**
      * Should initialize the underlying Netty [io.netty.channel.Channel] **synchronously**
      * and **on the calling thread**
      * Returns the [Future] which is completed with the protocol [TController]
      * when all necessary protocol negotiations are done.
      */
-    fun initChannel(ch: P2PChannel): CompletableFuture<out TController>
+    fun initChannel(ch: P2PChannel): CompletableFuture<TController>
 
-    fun toStreamHandler(): StreamHandler<TController> = object : StreamHandler<TController> {
-        override fun handleStream(stream: Stream): CompletableFuture<out TController> {
-            return initChannel(stream)
-        }
+    @JvmDefault
+    fun toStreamHandler(): StreamHandler<TController> = StreamHandler { stream -> initChannel(stream) }
+}
+
+fun interface ChannelVisitor<TChannel : P2PChannel> {
+
+    fun visit(channel: TChannel)
+
+    @JvmDefault
+    fun toChannelHandler(): P2PChannelHandler<Unit> = P2PChannelHandler {
+        visit(it as TChannel)
+        CompletableFuture.completedFuture(Unit)
+    }
+
+    interface Broadcast<TChannel : P2PChannel> : ChannelVisitor<TChannel>, MutableList<ChannelVisitor<TChannel>>
+
+    companion object {
+        fun <TChannel : P2PChannel> createBroadcast(vararg handlers: ChannelVisitor<TChannel>) =
+            BroadcastChannelVisitor<TChannel>().also { it += handlers }
     }
 }

src/main/kotlin/io/libp2p/core/StreamHandler.kt

@@ -1,6 +1,5 @@
 package io.libp2p.core
 
-import io.libp2p.etc.BroadcastStreamHandler
 import java.util.concurrent.CompletableFuture
 
 /**
@@ -22,28 +21,7 @@ data class StreamPromise<T>(
 /**
  * The same as [P2PChannelHandler] with the [Stream] specialized [P2PChannel]
  */
-interface StreamHandler<out TController> {
+fun interface StreamHandler<TController> {
 
-    fun handleStream(stream: Stream): CompletableFuture<out TController>
-
-    companion object {
-
-        fun create(fn: (Stream) -> Unit) = object : StreamHandler<Unit> {
-            override fun handleStream(stream: Stream): CompletableFuture<out Unit> {
-                fn(stream)
-                return CompletableFuture.completedFuture(Unit)
-            }
-        }
-
-        fun <T> create(channelHandler: P2PChannelHandler<T>) = object : StreamHandler<T> {
-            override fun handleStream(stream: Stream): CompletableFuture<out T> {
-                return channelHandler.initChannel(stream)
-            }
-        }
-
-        fun createBroadcast(vararg handlers: StreamHandler<*>) =
-            BroadcastStreamHandler().also { it += handlers }
-    }
-
-    interface Broadcast : StreamHandler<Any>, MutableList<StreamHandler<*>>
+    fun handleStream(stream: Stream): CompletableFuture<TController>
 }

src/main/kotlin/io/libp2p/core/dsl/Builders.kt

@@ -2,24 +2,29 @@ package io.libp2p.core.dsl
 
 import identify.pb.IdentifyOuterClass
 import io.libp2p.core.AddressBook
+import io.libp2p.core.ChannelVisitor
+import io.libp2p.core.Connection
 import io.libp2p.core.ConnectionHandler
 import io.libp2p.core.Host
-import io.libp2p.core.StreamHandler
+import io.libp2p.core.P2PChannel
+import io.libp2p.core.Stream
 import io.libp2p.core.crypto.KEY_TYPE
 import io.libp2p.core.crypto.PrivKey
 import io.libp2p.core.crypto.generateKeyPair
 import io.libp2p.core.multiformats.Multiaddr
-import io.libp2p.core.multistream.Multistream
+import io.libp2p.core.multistream.MultistreamProtocol
+import io.libp2p.core.multistream.MultistreamProtocolDebug
+import io.libp2p.core.multistream.MultistreamProtocolV1
 import io.libp2p.core.multistream.ProtocolBinding
 import io.libp2p.core.mux.StreamMuxer
 import io.libp2p.core.mux.StreamMuxerDebug
+import io.libp2p.core.mux.StreamMuxerProtocol
 import io.libp2p.core.security.SecureChannel
 import io.libp2p.core.transport.Transport
 import io.libp2p.etc.types.lazyVar
 import io.libp2p.etc.types.toProtobuf
 import io.libp2p.host.HostImpl
 import io.libp2p.host.MemoryAddressBook
-import io.libp2p.mux.mplex.MplexStreamMuxer
 import io.libp2p.network.NetworkImpl
 import io.libp2p.protocol.IdentifyBinding
 import io.libp2p.security.secio.SecIoSecureChannel
@@ -31,8 +36,6 @@ import io.netty.handler.logging.LoggingHandler
 
 typealias TransportCtor = (ConnectionUpgrader) -> Transport
 typealias SecureChannelCtor = (PrivKey) -> SecureChannel
-typealias StreamMuxerCtor = () -> StreamMuxer
-typealias ProtocolCtor = () -> ProtocolBinding<*>
 typealias IdentityFactory = () -> PrivKey
 
 class HostConfigurationException(message: String) : RuntimeException(message)
@@ -58,6 +61,10 @@ open class Builder {
     protected open val connectionHandlers = ConnectionHandlerBuilder()
     protected open val network = NetworkConfigBuilder()
     protected open val debug = DebugBuilder()
+    var multistreamProtocol: MultistreamProtocol = MultistreamProtocolV1
+    var secureMultistreamProtocol: MultistreamProtocol by lazyVar { multistreamProtocol }
+    var muxerMultistreamProtocol: MultistreamProtocol by lazyVar { multistreamProtocol }
+    var streamMultistreamProtocol: MultistreamProtocol by lazyVar { multistreamProtocol }
 
     /**
      * Sets an identity for this host. If unset, libp2p will default to a random identity.
@@ -125,23 +132,35 @@ open class Builder {
             if (identity.factory == null) identity.random()
             if (transports.values.isEmpty()) transports { add(::TcpTransport) }
             if (secureChannels.values.isEmpty()) secureChannels { add(::SecIoSecureChannel) }
-            if (muxers.values.isEmpty()) muxers { add(::MplexStreamMuxer) }
+            if (muxers.values.isEmpty()) muxers { add(StreamMuxerProtocol.Mplex) }
         }
 
-        val privKey = identity.factory!!()
+        if (debug.beforeSecureHandler.handlers.isNotEmpty()) {
+            (secureMultistreamProtocol as? MultistreamProtocolDebug)?.also {
+                val broadcast = ChannelVisitor.createBroadcast(*debug.beforeSecureHandler.handlers.toTypedArray())
+                secureMultistreamProtocol = it.copyWithHandlers(preHandler = broadcast.toChannelHandler())
+            } ?: throw IllegalStateException("beforeSecureHandler can't be installed as MultistreamProtocol doesn't support debugging interface: ${secureMultistreamProtocol.javaClass}")
+        }
 
-        val secureChannels = secureChannels.values.map { it(privKey) }
-        val muxers = muxers.values.map { it() }
+        if (debug.afterSecureHandler.handlers.isNotEmpty()) {
+            (muxerMultistreamProtocol as? MultistreamProtocolDebug)?.also {
+                val broadcast = ChannelVisitor.createBroadcast(*debug.afterSecureHandler.handlers.toTypedArray())
+                muxerMultistreamProtocol = it.copyWithHandlers(preHandler = broadcast.toChannelHandler())
+            } ?: throw IllegalStateException("afterSecureHandler can't be installed as MultistreamProtocol doesn't support debugging interface: ${muxerMultistreamProtocol.javaClass}")
+        }
 
-        muxers.mapNotNull { it as? StreamMuxerDebug }.forEach { it.muxFramesDebugHandler = debug.muxFramesHandler.handler }
+        val streamVisitors = ChannelVisitor.createBroadcast<Stream>()
+        (streamMultistreamProtocol as? MultistreamProtocolDebug)?.also {
+            val broadcastPre =
+                ChannelVisitor.createBroadcast(*(debug.streamPreHandler.handlers + (streamVisitors as ChannelVisitor<Stream>)).toTypedArray())
+            val broadcast = ChannelVisitor.createBroadcast(*debug.streamHandler.handlers.toTypedArray())
+            streamMultistreamProtocol =
+                it.copyWithHandlers(broadcastPre.toChannelHandler(), broadcast.toChannelHandler())
+        } ?: throw IllegalStateException("streamPreHandler or streamHandler can't be installed as MultistreamProtocol doesn't support debugging interface: ${streamMultistreamProtocol.javaClass}")
 
-        val upgrader = ConnectionUpgrader(secureChannels, muxers).apply {
-            beforeSecureHandler = debug.beforeSecureHandler.handler
-            afterSecureHandler = debug.afterSecureHandler.handler
-        }
+        val privKey = identity.factory!!()
 
-        val transports = transports.values.map { it(upgrader) }
-        val addressBook = addressBook.impl
+        val secureChannels = secureChannels.values.map { it(privKey) }
 
         protocols.values.mapNotNull { (it as? IdentifyBinding) }.map { it.protocol }.find { it.idMessage == null }?.apply {
             // initializing Identify with appropriate values
@@ -156,16 +175,23 @@ open class Builder {
             }
         }
 
-        val protocolsMultistream: Multistream<Any> = Multistream.create(protocols.values)
-        val broadcastStreamHandler = StreamHandler.createBroadcast()
-        val allStreamHandlers = StreamHandler.createBroadcast(
-            protocolsMultistream.toStreamHandler(), broadcastStreamHandler
-        )
+        val muxers = muxers.map { it.createMuxer(streamMultistreamProtocol, protocols.values) }
+
+        if (debug.muxFramesHandler.handlers.isNotEmpty()) {
+            val broadcast = ChannelVisitor.createBroadcast(*debug.muxFramesHandler.handlers.toTypedArray())
+            muxers.mapNotNull { it as? StreamMuxerDebug }.forEach {
+                it.muxFramesDebugHandler = broadcast
+            }
+        }
+
+        val upgrader = ConnectionUpgrader(secureMultistreamProtocol, secureChannels, muxerMultistreamProtocol, muxers)
+
+        val transports = transports.values.map { it(upgrader) }
+        val addressBook = addressBook.impl
 
         val connHandlerProtocols = protocols.values.mapNotNull { it as? ConnectionHandler }
         val broadcastConnHandler = ConnectionHandler.createBroadcast(
-            listOf(ConnectionHandler.createStreamHandlerInitializer(allStreamHandlers)) +
-                connHandlerProtocols +
+            connHandlerProtocols +
                 connectionHandlers.values
         )
         val networkImpl = NetworkImpl(transports, broadcastConnHandler)
@@ -175,9 +201,9 @@ open class Builder {
             networkImpl,
             addressBook,
             network.listen.map { Multiaddr(it) },
-            protocolsMultistream,
+            protocols.values,
             broadcastConnHandler,
-            broadcastStreamHandler
+            streamVisitors
         )
     }
 }
@@ -203,7 +229,7 @@ class AddressBookBuilder {
 
 class TransportsBuilder : Enumeration<TransportCtor>()
 class SecureChannelsBuilder : Enumeration<SecureChannelCtor>()
-class MuxersBuilder : Enumeration<StreamMuxerCtor>()
+class MuxersBuilder : Enumeration<StreamMuxerProtocol>()
 class ProtocolsBuilder : Enumeration<ProtocolBinding<Any>>()
 class ConnectionHandlerBuilder : Enumeration<ConnectionHandler>()
 
@@ -212,24 +238,36 @@ class DebugBuilder {
      * Injects the [ChannelHandler] to the wire closest point.
      * Could be primarily useful for security handshake debugging/monitoring
      */
-    val beforeSecureHandler = DebugHandlerBuilder("wire.sec.before")
+    val beforeSecureHandler = DebugHandlerBuilder<Connection>("wire.sec.before")
     /**
      * Injects the [ChannelHandler] right after the connection cipher
      * to handle plain wire messages
      */
-    val afterSecureHandler = DebugHandlerBuilder("wire.sec.after")
+    val afterSecureHandler = DebugHandlerBuilder<Connection>("wire.sec.after")
     /**
      * Injects the [ChannelHandler] right after the [StreamMuxer] pipeline handler
      * It intercepts [io.libp2p.mux.MuxFrame] instances
      */
-    val muxFramesHandler = DebugHandlerBuilder("wire.mux.frames")
+    val muxFramesHandler = DebugHandlerBuilder<Connection>("wire.mux.frames")
+
+    val streamPreHandler = DebugHandlerBuilder<Stream>("wire.stream.pre")
+
+    val streamHandler = DebugHandlerBuilder<Stream>("wire.stream")
 }
 
-class DebugHandlerBuilder(var name: String) {
-    var handler: ChannelHandler? = null
+class DebugHandlerBuilder<TChannel : P2PChannel>(var name: String) {
+    val handlers = mutableListOf<ChannelVisitor<TChannel>>()
+
+    fun addHandler(handler: ChannelVisitor<TChannel>) {
+        handlers += handler
+    }
+
+    fun addNettyHandler(handler: ChannelHandler) {
+        addHandler { it.pushHandler(handler) }
+    }
 
-    fun setLogger(level: LogLevel, loggerName: String = name) {
-        handler = LoggingHandler(loggerName, level)
+    fun addLogger(level: LogLevel, loggerName: String = name) {
+        addNettyHandler(LoggingHandler(loggerName, level))
     }
 }