dispose more StableRef

Author: lauzadis

aws-crt-kotlin/native/src/aws/sdk/kotlin/crt/auth/signing/AwsSignerNative.kt

@@ -19,6 +19,7 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.runBlocking
 import libcrt.*
 import platform.posix.UINT64_MAX
+import aws.sdk.kotlin.crt.util.use
 
 /**
  * Static class for a variety of AWS signing APIs.
@@ -36,7 +37,6 @@ public actual object AwsSigner {
         request.toNativeRequest().usePinned { nativeRequest ->
             // Pair of HTTP request and callback channel containing the signature
             val userData = nativeRequest to Channel<ByteArray>(1)
-            val userDataStableRef = StableRef.create(userData)
 
             val signable = checkNotNull(
                 aws_signable_new_http_request(
@@ -47,20 +47,20 @@ public actual object AwsSigner {
 
             val nativeSigningConfig: CPointer<aws_signing_config_base> = config.toNativeSigningConfig().reinterpret()
 
-            awsAssertOpSuccess(
-                aws_sign_request_aws(
-                    allocator = Allocator.Default.allocator,
-                    signable = signable,
-                    base_config = nativeSigningConfig,
-                    on_complete = staticCFunction(::signCallback),
-                    userdata = userDataStableRef.asCPointer(),
-                ),
-            ) { "sign() aws_sign_request_aws" }
+            StableRef.create(userData).use { userDataStableRef ->
+                awsAssertOpSuccess(
+                    aws_sign_request_aws(
+                        allocator = Allocator.Default.allocator,
+                        signable = signable,
+                        base_config = nativeSigningConfig,
+                        on_complete = staticCFunction(::signCallback),
+                        userdata = userDataStableRef.asCPointer(),
+                    ),
+                ) { "sign() aws_sign_request_aws" }
 
-            val callbackChannel = userDataStableRef.get().second
-            val signature = callbackChannel.receive() // wait for async signing to complete....
-            return AwsSigningResult(nativeRequest.get().toHttpRequest(), signature).also {
-                userDataStableRef.dispose()
+                val callbackChannel = userDataStableRef.get().second
+                val signature = callbackChannel.receive() // wait for async signing to complete....
+                return AwsSigningResult(nativeRequest.get().toHttpRequest(), signature)
             }
         }
     }
@@ -126,7 +126,7 @@ public actual object AwsSigner {
 
 private fun signChunkSignable(signable: CPointer<aws_signable>, config: AwsSigningConfig): AwsSigningResult = memScoped {
     val callbackChannel = Channel<ByteArray>(1)
-    val callbackChannelStableRef = StableRef.create(callbackChannel)
+    val callbackChannelStableRef = StableRef.create(callbackChannel) // Done
 
     val nativeConfig: CPointer<aws_signing_config_base> = config.toNativeSigningConfig().reinterpret()
 
@@ -196,7 +196,7 @@ private fun AwsSigningConfig.toNativeSigningConfig(): CPointer<aws_signing_confi
         aws_date_time_init_epoch_millis(date.ptr, this@toNativeSigningConfig.date.toULong())
 
         this@toNativeSigningConfig.shouldSignHeader?.let {
-            val shouldSignHeaderStableRef = StableRef.create(it)
+            val shouldSignHeaderStableRef = StableRef.create(it) // FIXME Dispose this StableRef somehow...
             should_sign_header = staticCFunction(::nativeShouldSignHeaderFn)
             should_sign_header_ud = shouldSignHeaderStableRef.asCPointer()
         }

aws-crt-kotlin/native/src/aws/sdk/kotlin/crt/http/HttpClientConnectionManagerNative.kt

@@ -132,8 +132,9 @@ public actual class HttpClientConnectionManager actual constructor(
     public actual suspend fun acquireConnection(): HttpClientConnection =
         suspendCoroutine { cont ->
             val cb = staticCFunction(::onConnectionAcquired)
-            val userdata = StableRef.create(HttpConnectionAcquisitionRequest(cont, this))
-            aws_http_connection_manager_acquire_connection(manager, cb, userdata.asCPointer())
+            StableRef.create(HttpConnectionAcquisitionRequest(cont, this)).use { userdata ->
+                aws_http_connection_manager_acquire_connection(manager, cb, userdata.asCPointer())
+            }
         }
 
     /**
@@ -153,6 +154,7 @@ public actual class HttpClientConnectionManager actual constructor(
 
     actual override fun close() {
         aws_http_connection_manager_release(manager)
+        shutdownCompleteStableRef.dispose()
     }
 }
 

aws-crt-kotlin/native/src/aws/sdk/kotlin/crt/http/HttpClientConnectionNative.kt

@@ -29,7 +29,7 @@ internal class HttpClientConnectionNative(
     override fun makeRequest(httpReq: HttpRequest, handler: HttpStreamResponseHandler): HttpStream {
         val nativeReq = httpReq.toNativeRequest()
         val cbData = HttpStreamContext(handler, nativeReq)
-        val stableRef = StableRef.create(cbData)
+        val stableRef = StableRef.create(cbData) // FIXME Close this StableRef somehow...
         val reqOptions = cValue<aws_http_make_request_options> {
             self_size = sizeOf<aws_http_make_request_options>().convert()
             request = nativeReq

aws-crt-kotlin/native/src/aws/sdk/kotlin/crt/http/HttpStreamNative.kt

@@ -9,6 +9,7 @@ import aws.sdk.kotlin.crt.CrtRuntimeException
 import aws.sdk.kotlin.crt.NativeHandle
 import aws.sdk.kotlin.crt.awsAssertOpSuccess
 import aws.sdk.kotlin.crt.util.asAwsByteCursor
+import aws.sdk.kotlin.crt.util.use
 import kotlinx.cinterop.*
 import libcrt.*
 import kotlin.coroutines.Continuation
@@ -63,18 +64,19 @@ internal class HttpStreamNative(
                 throw CrtRuntimeException("aws_input_stream_new_from_cursor()")
             }
 
-            val req = StableRef.create(WriteChunkRequest(cont, byteBuf, stream))
-            val chunkOpts = cValue<aws_http1_chunk_options> {
-                chunk_data_size = chunkData.size.convert()
-                chunk_data = stream
-                on_complete = staticCFunction(::onWriteChunkComplete)
-                user_data = req.asCPointer()
-            }
-            awsAssertOpSuccess(
-                aws_http1_stream_write_chunk(ptr, chunkOpts),
-            ) {
-                cleanupWriteChunkCbData(req)
-                "aws_http1_stream_write_chunk()"
+            StableRef.create(WriteChunkRequest(cont, byteBuf, stream)).use { req ->
+                val chunkOpts = cValue<aws_http1_chunk_options> {
+                    chunk_data_size = chunkData.size.convert()
+                    chunk_data = stream
+                    on_complete = staticCFunction(::onWriteChunkComplete)
+                    user_data = req.asCPointer()
+                }
+                awsAssertOpSuccess(
+                    aws_http1_stream_write_chunk(ptr, chunkOpts),
+                ) {
+                    cleanupWriteChunkCbData(req)
+                    "aws_http1_stream_write_chunk()"
+                }
             }
         }
 

aws-crt-kotlin/native/src/aws/sdk/kotlin/crt/http/RequestBodyStream.kt

@@ -112,7 +112,7 @@ internal fun inputStream(khandler: HttpRequestBodyStream): CPointer<aws_input_st
     val stream: aws_input_stream = Allocator.Default.alloc()
     stream.vtable = requestStreamVtable
     val impl = RequestBodyStream(khandler)
-    val stableRef = StableRef.create(impl)
+    val stableRef = StableRef.create(impl) // FIXME Dispose this StableRef somehow...
     stream.impl = stableRef.asCPointer()
     return stream.ptr
 }

aws-crt-kotlin/native/src/aws/sdk/kotlin/crt/io/ClientBootstrapNative.kt

@@ -54,6 +54,7 @@ public actual class ClientBootstrap private constructor(
 
     actual override fun close() {
         aws_client_bootstrap_release(ptr)
+        channelStableRef.dispose()
 
         if (manageHr) hr.close()
         if (manageElg) elg.close()

aws-crt-kotlin/native/src/aws/sdk/kotlin/crt/io/EventLoopGroupNative.kt

@@ -60,6 +60,7 @@ public actual class EventLoopGroup actual constructor(maxThreads: Int) :
 
     actual override fun close() {
         aws_event_loop_group_release(ptr)
+        channelStableRef.dispose()
     }
 }
 

aws-crt-kotlin/native/src/aws/sdk/kotlin/crt/io/HostResolverNative.kt

@@ -57,6 +57,7 @@ public actual class HostResolver private constructor(
 
     actual override fun close() {
         aws_host_resolver_release(ptr)
+        channelStableRef.dispose()
 
         if (manageElg) elg.close()
     }

aws-crt-kotlin/native/src/aws/sdk/kotlin/crt/util/Interop.kt

@@ -5,6 +5,7 @@
 package aws.sdk.kotlin.crt.util
 
 import kotlinx.coroutines.channels.Channel
+import kotlinx.cinterop.StableRef
 
 /**
  * Channel used to signal async shutdown from C callback to a suspend fn
@@ -15,3 +16,14 @@ internal typealias ShutdownChannel = Channel<Unit>
  * Create a new shutdown notification channel
  */
 internal fun shutdownChannel(): ShutdownChannel = Channel(Channel.RENDEZVOUS)
+
+/**
+ * Execute [block] using [StableRef], then dispose it.
+ */
+internal inline fun <T : Any, R> StableRef<T>.use(block: (StableRef<T>) -> R): R {
+    try {
+        return block(this)
+    } finally {
+        dispose()
+    }
+}