Merge pull request #183 from libp2p/0.8.0

Release 0.8.0

README.md

@@ -61,26 +61,48 @@ do not impact the ability to hold communications with other libp2p processes.
 
 ## Adding as a dependency to your project:
 
-Builds are published to JCenter. Maven Central mirrors JCenter, but updates can take some time to appear so if possible, pull directly from JCenter.
+[![Hosted By: Cloudsmith](https://img.shields.io/badge/OSS%20hosting%20by-cloudsmith-blue?logo=cloudsmith&style=flat-square)](https://cloudsmith.com) 
+Hosting of artefacts is graciously provided by [Cloudsmith](https://cloudsmith.com).
 
 ### Using Gradle
 ```
    repositories {
-       jcenter()
+    maven { url "https://dl.cloudsmith.io/public/libp2p/jvm-libp2p/maven/" }
    }
 
-   implementation 'io.libp2p:jvm-libp2p-minimal:0.7.0-RELEASE'
+   implementation 'io.libp2p:jvm-libp2p-minimal:0.8.0-RELEASE'
 ```
 ### Using Maven
+Add the repository to the `dependencyManagement` section of the pom file:
+```
+<repositories>
+  <repository>
+    <id>libp2p-jvm-libp2p</id>
+    <url>https://dl.cloudsmith.io/public/libp2p/jvm-libp2p/maven/</url>
+    <releases>
+      <enabled>true</enabled>
+      <updatePolicy>always</updatePolicy>
+    </releases>
+    <snapshots>
+      <enabled>true</enabled>
+      <updatePolicy>always</updatePolicy>
+    </snapshots>
+  </repository>
+</repositories>
+```
+
+And then add jvm-libp2p as a dependency:
 ``` 
   <dependency>
     <groupId>io.libp2p</groupId>
     <artifactId>jvm-libp2p-minimal</artifactId>
-    <version>0.7.0-RELEASE</version>
+    <version>0.8.0-RELEASE</version>
     <type>pom</type>
   </dependency>
 ```
 
+
+
 ## Building the project 
 
 To build the library you will need just 

build.gradle.kts

@@ -1,19 +1,18 @@
 import com.google.protobuf.gradle.proto
 import com.google.protobuf.gradle.protobuf
 import com.google.protobuf.gradle.protoc
-import com.jfrog.bintray.gradle.BintrayExtension
 import org.gradle.api.tasks.testing.logging.TestExceptionFormat
 import org.jetbrains.dokka.gradle.DokkaTask
 import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 import java.net.URL
 import java.nio.file.Files
 import java.nio.file.Paths
 
-// To publish the release artifact to JFrog Bintray repo run the following :
-// ./gradlew bintrayUpload -PbintrayUser=<user> -PbintrayApiKey=<api-key>
+// To publish the release artifact to CloudSmith repo run the following :
+// ./gradlew publish -PcloudsmithUser=<user> -PcloudsmithApiKey=<api-key>
 
 group = "io.libp2p"
-version = "0.7.0-RELEASE"
+version = "0.8.0-RELEASE"
 description = "a minimal implementation of libp2p for the jvm"
 
 plugins {
@@ -25,7 +24,6 @@ plugins {
 
     `maven`
     `maven-publish`
-    id("com.jfrog.bintray") version "1.8.1"
     id("org.jetbrains.dokka") version "0.9.18"
 }
 
@@ -213,36 +211,27 @@ val dokkaJar by tasks.creating(Jar::class) {
 publishing {
     repositories {
         maven {
-            // change to point to your repo, e.g. http://my.org/repo
-            url = uri("$buildDir/repo")
+            name = "cloudsmith"
+            url = uri("https://api-g.cloudsmith.io/maven/libp2p/jvm-libp2p")
+            credentials {
+                username = findProperty("cloudsmithUser")
+                password = findProperty("cloudsmithApiKey")
+            }
         }
     }
     publications {
         register("mavenJava", MavenPublication::class) {
             from(components["java"])
             artifact(sourcesJar.get())
             artifact(dokkaJar)
+            groupId = "io.libp2p"
+            artifactId = project.name
         }
     }
 }
 
 fun findProperty(s: String) = project.findProperty(s) as String?
 
-bintray {
-    user = findProperty("bintrayUser")
-    key = findProperty("bintrayApiKey")
-    publish = true
-    setPublications("mavenJava")
-    setConfigurations("archives")
-    pkg(delegateClosureOf<BintrayExtension.PackageConfig> {
-        userOrg = "libp2p"
-        repo = "jvm-libp2p"
-        name = "io.libp2p"
-        setLicenses("Apache-2.0", "MIT")
-        vcsUrl = "https://github.com/libp2p/jvm-libp2p"
-    })
-}
-
 val compileKotlin: KotlinCompile by tasks
 compileKotlin.kotlinOptions {
     languageVersion = "1.4"

src/main/kotlin/io/libp2p/etc/util/netty/mux/AbstractMuxHandler.kt

@@ -23,6 +23,7 @@ abstract class AbstractMuxHandler<TData>() :
     private val activeFuture = CompletableFuture<Void>()
     private var closed = false
     protected abstract val inboundInitializer: MuxChannelInitializer<TData>
+    private val pendingReadComplete = mutableSetOf<MuxId>()
 
     override fun handlerAdded(ctx: ChannelHandlerContext) {
         super.handlerAdded(ctx)
@@ -54,13 +55,19 @@ abstract class AbstractMuxHandler<TData>() :
 
     protected fun childRead(id: MuxId, msg: TData) {
         val child = streamMap[id] ?: throw ConnectionClosedException("Channel with id $id not opened")
+        pendingReadComplete += id
         child.pipeline().fireChannelRead(msg)
     }
 
+    override fun channelReadComplete(ctx: ChannelHandlerContext) {
+        pendingReadComplete.forEach { streamMap[it]?.pipeline()?.fireChannelReadComplete() }
+        pendingReadComplete.clear()
+    }
+
     abstract fun onChildWrite(child: MuxChannel<TData>, data: TData): Boolean
 
     protected fun onRemoteOpen(id: MuxId) {
-        val initializer = inboundInitializer ?: throw InternalErrorException("Illegal state: inbound stream handler is not set up yet")
+        val initializer = inboundInitializer
         val child = createChild(
             id,
             initializer,

src/test/kotlin/io/libp2p/mux/MultiplexHandlerTest.kt

@@ -24,6 +24,7 @@ import io.netty.channel.ChannelInboundHandlerAdapter
 import io.netty.channel.DefaultChannelId
 import io.netty.handler.logging.LogLevel
 import io.netty.handler.logging.LoggingHandler
+import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertFalse
 import org.junit.jupiter.api.Assertions.assertThrows
@@ -83,6 +84,40 @@ class MultiplexHandlerTest {
         assertEquals("66", childHandlers[0].inboundMessages.last())
     }
 
+    @Test
+    fun `test that readComplete event is fired to child channel`() {
+        openStream(12)
+
+        assertThat(childHandlers[0].readCompleteEventCount).isZero()
+
+        writeStream(12, "22")
+
+        assertThat(childHandlers[0].readCompleteEventCount).isEqualTo(1)
+
+        writeStream(12, "23")
+
+        assertThat(childHandlers[0].readCompleteEventCount).isEqualTo(2)
+    }
+
+    @Test
+    fun `test that readComplete event is fired to reading channels only`() {
+        openStream(12)
+        openStream(13)
+
+        assertThat(childHandlers[0].readCompleteEventCount).isZero()
+        assertThat(childHandlers[1].readCompleteEventCount).isZero()
+
+        writeStream(12, "22")
+
+        assertThat(childHandlers[0].readCompleteEventCount).isEqualTo(1)
+        assertThat(childHandlers[1].readCompleteEventCount).isEqualTo(0)
+
+        writeStream(13, "23")
+
+        assertThat(childHandlers[0].readCompleteEventCount).isEqualTo(1)
+        assertThat(childHandlers[1].readCompleteEventCount).isEqualTo(1)
+    }
+
     @Test
     fun twoStreamsInterleaved() {
         openStream(12)
@@ -222,6 +257,7 @@ class MultiplexHandlerTest {
     class TestHandler : ChannelInboundHandlerAdapter() {
         val inboundMessages = mutableListOf<String>()
         var ctx: ChannelHandlerContext? = null
+        var readCompleteEventCount = 0
 
         override fun channelInactive(ctx: ChannelHandlerContext?) {
             println("MultiplexHandlerTest.channelInactive")
@@ -246,6 +282,7 @@ class MultiplexHandlerTest {
         }
 
         override fun channelReadComplete(ctx: ChannelHandlerContext?) {
+            readCompleteEventCount++
             println("MultiplexHandlerTest.channelReadComplete")
         }