Channels that are open for reactive streams are Closeable

Author: elizarov

reactive/kotlinx-coroutines-reactive/src/main/kotlin/kotlinx/coroutines/experimental/reactive/Channel.kt

@@ -21,17 +21,18 @@ import kotlinx.coroutines.experimental.channels.ReceiveChannel
 import org.reactivestreams.Publisher
 import org.reactivestreams.Subscriber
 import org.reactivestreams.Subscription
+import java.io.Closeable
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater
 
 /**
  * Return type for [Publisher.open] that can be used to [receive] elements from the
  * subscription and to manually [close] it.
  */
-public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T> {
+public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T>, Closeable {
     /**
      * Closes this subscription channel.
      */
-    public fun close()
+    public override fun close()
 }
 
 /**

reactive/kotlinx-coroutines-rx1/src/main/kotlin/kotlinx/coroutines/experimental/rx1/RxChannel.kt

@@ -21,17 +21,18 @@ import kotlinx.coroutines.experimental.channels.ReceiveChannel
 import rx.Observable
 import rx.Subscriber
 import rx.Subscription
+import java.io.Closeable
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater
 
 /**
  * Return type for [Observable.open] that can be used to [receive] elements from the
  * subscription and to manually [close] it.
  */
-public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T> {
+public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T>, Closeable {
     /**
      * Closes this subscription channel.
      */
-    public fun close()
+    public override fun close()
 }
 
 /**

reactive/kotlinx-coroutines-rx2/src/main/kotlin/kotlinx/coroutines/experimental/rx2/RxChannel.kt

@@ -22,16 +22,17 @@ import io.reactivex.Observer
 import io.reactivex.disposables.Disposable
 import kotlinx.coroutines.experimental.channels.LinkedListChannel
 import kotlinx.coroutines.experimental.channels.ReceiveChannel
+import java.io.Closeable
 
 /**
  * Return type for [Observable.open] that can be used to [receive] elements from the
  * subscription and to manually [close] it.
  */
-public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T> {
+public interface SubscriptionReceiveChannel<out T> : ReceiveChannel<T>, Closeable {
     /**
      * Closes this subscription channel.
      */
-    public fun close()
+    public override fun close()
 }
 
 /**