Add in-process server transport

Author: gjcairo

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

@@ -16,8 +16,7 @@
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
 extension RPCWriter {
-  @usableFromInline
-  struct Closable: ClosableRPCWriterProtocol {
+  public struct Closable: ClosableRPCWriterProtocol {
     @usableFromInline
     let writer: any ClosableRPCWriterProtocol<Element>
 
@@ -36,7 +35,7 @@ extension RPCWriter {
     ///
     /// - Parameter elements: The elements to write.
     @inlinable
-    func write(contentsOf elements: some Sequence<Element>) async throws {
+    public func write(contentsOf elements: some Sequence<Element>) async throws {
       try await self.writer.write(contentsOf: elements)
     }
 
@@ -45,7 +44,7 @@ extension RPCWriter {
     /// All writes after ``finish()`` has been called should result in an error
     /// being thrown.
     @inlinable
-    func finish() {
+    public func finish() {
       self.writer.finish()
     }
 
@@ -54,7 +53,7 @@ extension RPCWriter {
     /// All writes after ``finish(throwing:)`` has been called should result in an error
     /// being thrown.
     @inlinable
-    func finish(throwing error: Error) {
+    public func finish(throwing error: Error) {
       self.writer.finish(throwing: error)
     }
   }

Sources/GRPCCore/Transport/InProcessServerTransport.swift

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2023, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * 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()
+    }
+}

Tests/GRPCCoreTests/Transport/InProcessServerTransportTest.swift

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2023, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import XCTest
+
+@testable import GRPCCore
+
+final class InProcessServerTransportTest: XCTestCase {
+  func testStartListening() async throws {
+    let transport = InProcessServerTransport()
+    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)))))
+    )
+    
+    let streamSequence = transport.listen()
+    var streamSequenceInterator = streamSequence.makeAsyncIterator()
+    
+    transport.acceptStream(stream)
+    
+    let testStream = try await streamSequenceInterator.next()
+    var inboundIterator = testStream?.inbound.makeAsyncIterator()
+    let rpcRequestPart = try await inboundIterator?.next()
+    XCTAssertEqual(rpcRequestPart, .message([42]))
+  }
+    
+  func testStopListening() async throws {
+    let transport = InProcessServerTransport()
+    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)))))
+    )
+    
+    let streamSequence = transport.listen()
+    var streamSequenceInterator = streamSequence.makeAsyncIterator()
+    
+    transport.acceptStream(firstStream)
+    
+    let firstTestStream = try await streamSequenceInterator.next()
+    var inboundIterator = firstTestStream?.inbound.makeAsyncIterator()
+    let rpcRequestPart = try await inboundIterator?.next()
+    XCTAssertEqual(rpcRequestPart, .message([42]))
+    
+    transport.stopListening()
+    
+    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)))))
+    )
+    
+    transport.acceptStream(secondStream)
+    let secondTestStream = try await streamSequenceInterator.next()
+    XCTAssertNil(secondTestStream)
+  }
+}