PR and format changes

Author: gjcairo

Sources/GRPCCore/Streaming/Internal/RPCWriter+Closable.swift renamed to Sources/GRPCCore/Streaming/RPCWriter+Closable.swift

@@ -14,42 +14,49 @@
  * limitations under the License.
  */
 
-
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
 /// An in-process implementation of a ``ServerTransport``.
 public struct InProcessServerTransport: ServerTransport {
-    public typealias Inbound = RPCAsyncSequence<RPCRequestPart>
-    public typealias Outbound = RPCWriter<RPCResponsePart>.Closable
-    
-    private let newStreams: AsyncStream<RPCStream<Inbound, Outbound>>
-    private let newStreamsContinuation: AsyncStream<RPCStream<Inbound, Outbound>>.Continuation
-    
-    /// Creates a new instance of ``InProcessServerTransport``.
-    public init() {
-        (self.newStreams, self.newStreamsContinuation) = AsyncStream.makeStream()
-    }
-    
-    /// Publish a new ``RPCStream``, which will be returned by the transport's ``RPCAsyncSequence``,
-    /// returned when calling ``listen()``.
-    ///
-    /// - Parameter stream: The new ``RPCStream`` to publish.
-    public func acceptStream(_ stream: RPCStream<Inbound, Outbound>) {
-        self.newStreamsContinuation.yield(stream)
-    }
-    
-    /// Return a new ``RPCAsyncSequence`` that will contain all published ``RPCStream``s published
-    /// to this transport using the ``acceptStream(_:)`` method.
-    ///
-    /// - Returns: An ``RPCAsyncSequence`` of all published ``RPCStream``s.
-    public func listen() -> RPCAsyncSequence<RPCStream<Inbound, Outbound>> {
-        RPCAsyncSequence(wrapping: self.newStreams)
-    }
-    
-    /// Stop listening to any new ``RPCStream`` publications.
-    ///
-    /// All further calls to ``acceptStream(_:)`` will not produce any new elements on the
-    /// ``RPCAsyncSequence`` returned by ``listen()``.
-    public func stopListening() {
-        self.newStreamsContinuation.finish()
+  public typealias Inbound = RPCAsyncSequence<RPCRequestPart>
+  public typealias Outbound = RPCWriter<RPCResponsePart>.Closable
+
+  private let newStreams: AsyncStream<RPCStream<Inbound, Outbound>>
+  private let newStreamsContinuation: AsyncStream<RPCStream<Inbound, Outbound>>.Continuation
+
+  /// Creates a new instance of ``InProcessServerTransport``.
+  public init() {
+    (self.newStreams, self.newStreamsContinuation) = AsyncStream.makeStream()
+  }
+
+  /// Publish a new ``RPCStream``, which will be returned by the transport's ``RPCAsyncSequence``,
+  /// returned when calling ``listen()``.
+  ///
+  /// - Parameter stream: The new ``RPCStream`` to publish.
+  /// - Throws: ``RPCError`` with code ``RPCError/Code-swift.struct/failedPrecondition``
+  /// if the server transport stopped listening to new streams (i.e., if ``stopListening()`` has been called).
+  internal func acceptStream(_ stream: RPCStream<Inbound, Outbound>) throws {
+    let yieldResult = self.newStreamsContinuation.yield(stream)
+    if case .terminated = yieldResult {
+      throw RPCError(
+        code: .failedPrecondition,
+        message: "The server transport is closed."
+      )
     }
+  }
+
+  /// Return a new ``RPCAsyncSequence`` that will contain all published ``RPCStream``s published
+  /// to this transport using the ``acceptStream(_:)`` method.
+  ///
+  /// - Returns: An ``RPCAsyncSequence`` of all published ``RPCStream``s.
+  public func listen() -> RPCAsyncSequence<RPCStream<Inbound, Outbound>> {
+    RPCAsyncSequence(wrapping: self.newStreams)
+  }
+
+  /// Stop listening to any new ``RPCStream`` publications.
+  ///
+  /// All further calls to ``acceptStream(_:)`` will not produce any new elements on the
+  /// ``RPCAsyncSequence`` returned by ``listen()``.
+  public func stopListening() {
+    self.newStreamsContinuation.finish()
+  }
 }

Sources/GRPCCore/Transport/InProcessServerTransport.swift

@@ -24,47 +24,64 @@ final class InProcessServerTransportTest: XCTestCase {
     let stream = RPCStream<RPCAsyncSequence<RPCRequestPart>, RPCWriter<RPCResponsePart>.Closable>(
       descriptor: .init(service: "testService", method: "testMethod"),
       inbound: .elements([.message([42])]),
-      outbound: .init(wrapping: BufferedStream.Source(storage: .init(backPressureStrategy: .watermark(.init(low: 1, high: 1)))))
+      outbound: .init(
+        wrapping: BufferedStream.Source(
+          storage: .init(backPressureStrategy: .watermark(.init(low: 1, high: 1)))
+        )
+      )
     )
-    
+
     let streamSequence = transport.listen()
     var streamSequenceInterator = streamSequence.makeAsyncIterator()
-    
-    transport.acceptStream(stream)
-    
+
+    try transport.acceptStream(stream)
+
     let testStream = try await streamSequenceInterator.next()
-    var inboundIterator = testStream?.inbound.makeAsyncIterator()
-    let rpcRequestPart = try await inboundIterator?.next()
-    XCTAssertEqual(rpcRequestPart, .message([42]))
+    let messages = try await testStream?.inbound.collect()
+    XCTAssertEqual(messages, [.message([42])])
   }
-    
+
   func testStopListening() async throws {
     let transport = InProcessServerTransport()
-    let firstStream = RPCStream<RPCAsyncSequence<RPCRequestPart>, RPCWriter<RPCResponsePart>.Closable>(
+    let firstStream = RPCStream<
+      RPCAsyncSequence<RPCRequestPart>, RPCWriter<RPCResponsePart>.Closable
+    >(
       descriptor: .init(service: "testService1", method: "testMethod1"),
       inbound: .elements([.message([42])]),
-      outbound: .init(wrapping: BufferedStream.Source(storage: .init(backPressureStrategy: .watermark(.init(low: 1, high: 1)))))
+      outbound: .init(
+        wrapping: BufferedStream.Source(
+          storage: .init(backPressureStrategy: .watermark(.init(low: 1, high: 1)))
+        )
+      )
     )
-    
+
     let streamSequence = transport.listen()
     var streamSequenceInterator = streamSequence.makeAsyncIterator()
-    
-    transport.acceptStream(firstStream)
-    
+
+    try transport.acceptStream(firstStream)
+
     let firstTestStream = try await streamSequenceInterator.next()
-    var inboundIterator = firstTestStream?.inbound.makeAsyncIterator()
-    let rpcRequestPart = try await inboundIterator?.next()
-    XCTAssertEqual(rpcRequestPart, .message([42]))
-    
+    let firstStreamMessages = try await firstTestStream?.inbound.collect()
+    XCTAssertEqual(firstStreamMessages, [.message([42])])
+
     transport.stopListening()
-    
-    let secondStream = RPCStream<RPCAsyncSequence<RPCRequestPart>, RPCWriter<RPCResponsePart>.Closable>(
+
+    let secondStream = RPCStream<
+      RPCAsyncSequence<RPCRequestPart>, RPCWriter<RPCResponsePart>.Closable
+    >(
       descriptor: .init(service: "testService1", method: "testMethod1"),
       inbound: .elements([.message([42])]),
-      outbound: .init(wrapping: BufferedStream.Source(storage: .init(backPressureStrategy: .watermark(.init(low: 1, high: 1)))))
+      outbound: .init(
+        wrapping: BufferedStream.Source(
+          storage: .init(backPressureStrategy: .watermark(.init(low: 1, high: 1)))
+        )
+      )
     )
+
+    XCTAssertThrowsRPCError(try transport.acceptStream(secondStream)) { error in
+      XCTAssertEqual(error.code, .failedPrecondition)
+    }
 
-    transport.acceptStream(secondStream)
     let secondTestStream = try await streamSequenceInterator.next()
     XCTAssertNil(secondTestStream)
   }