[fleet] move http ktor engine related code to rpc.server.ktor

GitOrigin-RevId: 09aee497f8c5f532380918456c3a2a1a0f0e13cb

fleet/rpc.server/src/fleet/rpc/server/DirectRpcClient.kt

@@ -8,24 +8,21 @@ import fleet.rpc.client.rpcClient
 import fleet.rpc.core.Serialization
 import fleet.rpc.core.Transport
 import fleet.rpc.core.TransportMessage
-import fleet.tracing.spannedScope
 import fleet.util.UID
-import fleet.util.async.async
-import fleet.util.async.resource
-import fleet.util.async.use
+import fleet.util.async.*
 import fleet.util.channels.channels
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.launch
 
-suspend fun RequestDispatcher.withDirectRpcClient(serialization: () -> Serialization,
+fun RequestDispatcher.directRpcClient(
+  serialization: () -> Serialization,
   interceptor: RpcInterceptor,
-                                                  body: suspend CoroutineScope.(IRpcClient) -> Unit) {
-  spannedScope("withDirectRpcClient") {
+): Resource<IRpcClient> =
+  resource { cc ->
     val (dispatcherSend, clientReceive) = channels<TransportMessage>(Channel.BUFFERED)
     val (clientSend, dispatcherReceive) = channels<TransportMessage>(Channel.BUFFERED)
     val origin = UID.random()
-    resource<IRpcClient> { continuation ->
     launch {
       handleConnection(route = origin,
                        endpoint = EndpointKind.Client,
@@ -37,11 +34,19 @@ suspend fun RequestDispatcher.withDirectRpcClient(serialization: () -> Serializa
                 serialization = serialization,
                 origin = origin,
                 requestInterceptor = interceptor) { rpcClient ->
-          continuation(rpcClient)
+        cc(rpcClient)
       }
     }
-    }.async().use { rpcClientDeferred ->
+  }.span("directRpcClient")
+
+suspend fun RequestDispatcher.withDirectRpcClient(
+  serialization: () -> Serialization,
+  interceptor: RpcInterceptor,
+  body: suspend CoroutineScope.(IRpcClient) -> Unit,
+) {
+  directRpcClient(serialization, interceptor)
+    .async()
+    .use { rpcClientDeferred ->
       body(promisingRpcClient(rpcClientDeferred))
     }
-  }
 }

fleet/rpc.server/src/fleet/rpc/server/FleetService.kt

@@ -16,7 +16,6 @@ class FleetService private constructor(val serviceId: UID,
     job.join()
   }
 
-  //@fleet.kernel.plugins.InternalInPluginModules(where = ["fleet.app.fleet.tests"])
   suspend fun terminate(cause: String) {
     terminate(CancellationException(cause))
   }
@@ -50,7 +49,7 @@ class FleetService private constructor(val serviceId: UID,
             }
           }
         }.use { serviceJob ->
-          body(FleetService(providerId, serviceJob))
+          body(FleetService(serviceId = providerId, job = serviceJob))
         }
       }
     }

fleet/rpc.server/src/fleet/rpc/server/RpcExecutor.kt

@@ -28,7 +28,6 @@ import java.util.concurrent.ConcurrentHashMap
 import kotlin.coroutines.EmptyCoroutineContext
 import kotlin.coroutines.coroutineContext
 
-//@fleet.kernel.plugins.InternalInPluginModules(where = ["fleet.app.fleet.tests"])
 class RpcExecutor private constructor(private val serialization: () -> Serialization,
                                       private val services: RpcServiceLocator,
                                       private val route: UID,