Add new Server object

Sources/GRPCCore/Call/Server/RPCRouter.swift

@@ -133,3 +133,27 @@ public struct RPCRouter: Sendable {
     return self.handlers.removeValue(forKey: descriptor) != nil
   }
 }
+
+@available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+extension RPCRouter {
+  internal func handle(
+    stream: RPCStream<RPCAsyncSequence<RPCRequestPart>, RPCWriter<RPCResponsePart>.Closable>,
+    interceptors: [any ServerInterceptor]
+  ) async {
+    if let handler = self.handlers[stream.descriptor] {
+      await handler.handle(stream: stream, interceptors: interceptors)
+    } else {
+      // If this throws then the stream must be closed which we can't do anything about, so ignore
+      // any error.
+      try? await stream.outbound.write(.status(.rpcNotImplemented, [:]))
+      stream.outbound.finish()
+    }
+  }
+}
+
+extension Status {
+  fileprivate static let rpcNotImplemented = Status(
+    code: .unimplemented,
+    message: "Requested RPC isn't implemented by this server."
+  )
+}

Sources/GRPCCore/Call/Server/ServerRequest.swift

@@ -72,8 +72,24 @@ extension ServerRequest {
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
 extension ServerRequest.Stream {
-  @_spi(Testing)
   public init(single request: ServerRequest.Single<Message>) {
     self.init(metadata: request.metadata, messages: .one(request.message))
   }
 }
+
+@available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
+extension ServerRequest.Single {
+  public init(stream request: ServerRequest.Stream<Message>) async throws {
+    var iterator = request.messages.makeAsyncIterator()
+
+    guard let message = try await iterator.next() else {
+      throw RPCError(code: .internalError, message: "Empty stream.")
+    }
+
+    guard try await iterator.next() == nil else {
+      throw RPCError(code: .internalError, message: "Too many messages.")
+    }
+
+    self = ServerRequest.Single(metadata: request.metadata, message: message)
+  }
+}

Sources/GRPCCore/Call/Server/ServerResponse.swift

@@ -327,7 +327,6 @@ extension ServerResponse.Stream {
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
 extension ServerResponse.Stream {
-  @_spi(Testing)
   public init(single response: ServerResponse.Single<Message>) {
     switch response.accepted {
     case .success(let contents):