Add return result for stream send() (#21)

Author: buildbreaker

library/src/main/kotlin/build/buf/connect/BidirectionalStreamInterface.kt

@@ -31,11 +31,21 @@ interface BidirectionalStreamInterface<Input, Output> {
      * Send a request to the server over the stream.
      *
      * @param input The request message to send.
+     * @return [Result.success] on send success, [Result.failure] on
+     *         any sends which are not successful.
      */
-    suspend fun send(input: Input)
+    suspend fun send(input: Input): Result<Unit>
 
     /**
-     * Close the stream. No calls to `send()` are valid after calling `close()`.
+     * Close the stream. No calls to [send] are valid after calling [close].
      */
     fun close()
+
+    /**
+     * Determine if the underlying stream is closed.
+     *
+     * @return true if the underlying stream is closed. If the stream is still open,
+     *         this will return false.
+     */
+    fun isClosed(): Boolean
 }

library/src/main/kotlin/build/buf/connect/ClientOnlyStreamInterface.kt

@@ -24,10 +24,18 @@ interface ClientOnlyStreamInterface<Input, Output> {
      *
      * @param input The request message to send.
      */
-    suspend fun send(input: Input)
+    suspend fun send(input: Input): Result<Unit>
 
     /**
-     * Close the stream. No calls to `send()` are valid after calling `close()`.
+     * Close the stream. No calls to [send] are valid after calling [close].
      */
     fun close()
+
+    /**
+     * Determine if the underlying stream is closed.
+     *
+     * @return true if the underlying stream is closed. If the stream is still open,
+     *         this will return false.
+     */
+    fun isClosed(): Boolean
 }

library/src/main/kotlin/build/buf/connect/ConnectError.kt

@@ -18,7 +18,7 @@ import kotlin.reflect.KClass
 
 /**
  * Typed error provided by Connect RPCs that may optionally wrap additional typed custom errors
- * using `details`.
+ * using [details].
  */
 data class ConnectError constructor(
     // The resulting status code.
@@ -35,7 +35,7 @@ data class ConnectError constructor(
 ) : Throwable(message, exception) {
 
     /**
-     * Unpacks values from `self.details` and returns the first matching error, if any.
+     * Unpacks values from [details] and returns the first matching error, if any.
      *
      * @return The unpacked typed error details, if available.
      */
@@ -51,7 +51,7 @@ data class ConnectError constructor(
     }
 
     /**
-     * Creates a new ConnectError with the specified CompletionParser.
+     * Creates a new [ConnectError] with the specified [ErrorDetailParser].
      */
     fun setErrorParser(errorParser: ErrorDetailParser): ConnectError {
         return ConnectError(

library/src/main/kotlin/build/buf/connect/SerializationStrategy.kt

@@ -19,7 +19,7 @@ import kotlin.reflect.KClass
 /**
  * The serialization strategy for completion events from gRPC or Connect.
  *
- * A base data type will need to implement a SerializationStrategy.
+ * A base data type will need to implement a [SerializationStrategy].
  */
 interface SerializationStrategy {
 

library/src/main/kotlin/build/buf/connect/ServerOnlyStreamInterface.kt

@@ -33,11 +33,21 @@ interface ServerOnlyStreamInterface<Input, Output> {
      * Can only be called exactly one time when starting the stream.
      *
      * @param input The request message to send.
+     * @return [Result.success] on send success, [Result.failure] on
+     *         any sends which are not successful.
      */
-    suspend fun send(input: Input)
+    suspend fun send(input: Input): Result<Unit>
 
     /**
-     * Close the stream. No calls to `send()` are valid after calling `close()`.
+     * Close the stream. No calls to [send] are valid after calling [close].
      */
     fun close()
+
+    /**
+     * Determine if the underlying stream is closed.
+     *
+     * @return true if the underlying stream is closed. If the stream is still open,
+     *         this will return false.
+     */
+    fun isClosed(): Boolean
 }

library/src/main/kotlin/build/buf/connect/StreamResult.kt

@@ -17,7 +17,7 @@ package build.buf.connect
 /**
  * Enumeration of result states that can be received over streams.
  *
- * A typical stream receives `Headers > Message > Message > Message ... > Complete`
+ * A typical stream receives [Headers] > [Message] > [Message] > [Message] ... > [Complete]
  */
 sealed class StreamResult<Output> constructor(
     val error: Throwable? = null
@@ -34,7 +34,7 @@ sealed class StreamResult<Output> constructor(
     /**
      * Get the ConnectError from the result.
      *
-     * @return The ConnectError if present, null otherwise.
+     * @return The [ConnectError] if present, null otherwise.
      */
     fun connectError(): ConnectError? {
         if (error is ConnectError) {

library/src/main/kotlin/build/buf/connect/compression/CompressionPool.kt

@@ -20,11 +20,11 @@ import okio.Buffer
  * Conforming types provide the functionality to compress/decompress data using a specific
  * algorithm.
  *
- * `ProtocolClientInterface` implementations are expected to use the first compression pool with
- * a matching `name()` for decompressing inbound responses.
+ * [build.buf.connect.ProtocolClientInterface] implementations are expected to use the first compression pool with
+ * a matching [name] for decompressing inbound responses.
  *
  * Outbound request compression can be specified using additional options that specify a
- * `compressionName` that matches a compression pool's `name()`.
+ * `compressionName` that matches a compression pool's [name].
  */
 interface CompressionPool {
     /**

library/src/main/kotlin/build/buf/connect/http/HTTPClientInterface.kt

@@ -16,6 +16,7 @@ package build.buf.connect.http
 
 import build.buf.connect.StreamResult
 import okio.Buffer
+import java.util.concurrent.atomic.AtomicReference
 
 typealias Cancelable = () -> Unit
 
@@ -49,11 +50,23 @@ class Stream(
     private val onSend: (Buffer) -> Unit,
     private val onClose: () -> Unit
 ) {
-    fun send(buffer: Buffer) {
+    private val isClosed = AtomicReference(false)
+
+    fun send(buffer: Buffer): Result<Unit> {
+        if (isClosed()) {
+            return Result.failure(IllegalStateException("cannot send. underlying stream is closed"))
+        }
         onSend(buffer)
+        return Result.success(Unit)
     }
 
     fun close() {
-        onClose()
+        if (!isClosed.getAndSet(true)) {
+            onClose()
+        }
+    }
+
+    fun isClosed(): Boolean {
+        return isClosed.get()
     }
 }

library/src/main/kotlin/build/buf/connect/http/TracingInfo.kt

@@ -15,7 +15,7 @@
 package build.buf.connect.http
 
 /**
- * Tracing metadata for HTTPClientInterface and Interceptors.
+ * Tracing metadata for [HTTPClientInterface] and [build.buf.connect.Interceptor].
  */
 data class TracingInfo(
     // The underlying http status code

library/src/main/kotlin/build/buf/connect/impl/BidirectionalStream.kt

@@ -20,19 +20,24 @@ import build.buf.connect.StreamResult
 import build.buf.connect.http.Stream
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
+import java.lang.Exception
 
 /**
- * Concrete implementation of `BidirectionalStreamInterface`.
+ * Concrete implementation of [BidirectionalStreamInterface].
  */
 internal class BidirectionalStream<Input, Output>(
     val stream: Stream,
     private val requestCodec: Codec<Input>,
     private val receiveChannel: Channel<StreamResult<Output>>
 ) : BidirectionalStreamInterface<Input, Output> {
 
-    override suspend fun send(input: Input) {
-        val msg = requestCodec.serialize(input)
-        stream.send(msg)
+    override suspend fun send(input: Input): Result<Unit> {
+        val msg = try {
+            requestCodec.serialize(input)
+        } catch (e: Exception) {
+            return Result.failure(e)
+        }
+        return stream.send(msg)
     }
 
     override fun resultChannel(): ReceiveChannel<StreamResult<Output>> {
@@ -42,4 +47,8 @@ internal class BidirectionalStream<Input, Output>(
     override fun close() {
         stream.close()
     }
+
+    override fun isClosed(): Boolean {
+        return stream.isClosed()
+    }
 }