🐛 [dweb/helper,dweb/core] 修复一些内存管理问题

🐛 修复 Producer.Consumer 的 close 没有正确移除自身的问题
🐛 修复 IpcBodySender 没有正确销毁 usableByIpcConsumer 的问题

Author: Gaubee

next/kmp/core/src/commonMain/kotlin/org/dweb_browser/core/ipc/helper/IpcFork.kt

@@ -13,7 +13,7 @@ import org.dweb_browser.core.help.types.CommonAppManifest
 class IpcFork(
   val pid: Int,
   val autoStart: Boolean,
-  val startReason: String?,
+  val startReason: String? = null,
   val locale: CommonAppManifest,
   val remote: CommonAppManifest,
 ) : IpcRawMessage, IpcMessage {

next/kmp/core/src/commonTest/kotlin/ReadableStreamTest.kt

@@ -1,17 +1,23 @@
 package info.bagen.dwebbrowser
 
 import io.ktor.utils.io.cancel
+import io.ktor.utils.io.copyAndClose
+import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import org.dweb_browser.core.ipc.helper.ReadableStream
+import org.dweb_browser.helper.ByteReadChannelDelegate
 import org.dweb_browser.helper.SafeInt
 import org.dweb_browser.helper.canReadContent
+import org.dweb_browser.helper.commonConsumeEachArrayRange
 import org.dweb_browser.helper.consumeEachArrayRange
+import org.dweb_browser.helper.createByteChannel
 import org.dweb_browser.helper.readAvailableByteArray
 import org.dweb_browser.test.runCommonTest
 import kotlin.test.Test
+import kotlin.test.assertContentEquals
 import kotlin.test.assertEquals
 import kotlin.test.assertTrue
 
@@ -20,7 +26,7 @@ class ReadableStreamTestTest {
   @Test
   fun readableStreamAvailableTest() = runCommonTest {
     println("start")
-    val stream = ReadableStream(this,onStart = { controller ->
+    val stream = ReadableStream(this, onStart = { controller ->
       GlobalScope.launch {
         var i = 5
         while (i-- > 0) {
@@ -55,10 +61,10 @@ class ReadableStreamTestTest {
 
   @Test
   fun testCancel() = runCommonTest {
-    val readableStream = ReadableStream(this,onStart = { controller ->
+    val readableStream = ReadableStream(this, onStart = { controller ->
       GlobalScope.launch {
         var i = 0
-        while (true)          {
+        while (true) {
           controller.enqueue(byteArrayOf(i++.toByte()))
           delay(400)
         }
@@ -82,4 +88,36 @@ class ReadableStreamTestTest {
     println("readableStream closed")
     assertTrue { true }
   }
+
+  val byteArray = byteArrayOf(elements = (1..640000).distinct().map { it.toByte() }.toByteArray())
+
+  @Test
+  fun testClose() = runCommonTest(1000) { time ->
+    println("test-$time")
+
+    val stream = ReadableStream(this, onStart = { controller ->
+      launch(start = CoroutineStart.UNDISPATCHED) {
+        controller.enqueue(byteArray)
+      }
+      launch {
+        controller.closeWrite()
+      }
+    })
+    val sink = createByteChannel()
+    launch {
+      when (val source = stream.stream.getReader("xix")) {
+        is ByteReadChannelDelegate -> {
+          source.sourceByteReadChannel
+        }
+        else -> source
+      }.copyAndClose(sink)
+    }
+
+    var res = byteArrayOf()
+    sink.commonConsumeEachArrayRange { byteArray, last ->
+      res += byteArray
+    }
+
+    assertContentEquals(res, byteArray)
+  }
 }

toolkit/dweb-core/src/ipc/endpoint/CommonEndpoint.ts

@@ -74,7 +74,7 @@ export abstract class CommonEndpoint extends IpcEndpoint {
           case ENDPOINT_MESSAGE_TYPE.IPC: {
             const producer = this.getIpcMessageProducer(endpointMessage.pid);
             const ipc = await producer.ipcPo.promise;
-            producer.producer.trySend($normalizeIpcMessage(endpointMessage.ipcMessage, ipc));
+            void producer.producer.trySend($normalizeIpcMessage(endpointMessage.ipcMessage, ipc));
             break;
           }
           case ENDPOINT_MESSAGE_TYPE.LIFECYCLE: {

toolkit/dweb-core/src/ipc/ipc-message/stream/IpcBodySender.ts

@@ -296,7 +296,8 @@ export class IpcBodySender extends IpcBody {
       if (usableIpcBodyMapper === undefined) {
         const mapper = new UsableIpcBodyMapper();
         IpcUsableIpcBodyMap.set(ipc, mapper);
-        ipc.onStream("usableByIpc").collect((event) => {
+        const usableByIpcConsumer = ipc.onStream("usableByIpc");
+        usableByIpcConsumer.collect((event) => {
           const message = event.data;
           switch (message.type) {
             case IPC_MESSAGE_TYPE.STREAM_PULLING:
@@ -313,7 +314,10 @@ export class IpcBodySender extends IpcBody {
           }
           event.consume();
         });
-        mapper.onDestroy(() => IpcUsableIpcBodyMap.delete(ipc));
+        mapper.onDestroy(() => {
+          usableByIpcConsumer.close();
+          IpcUsableIpcBodyMap.delete(ipc);
+        });
         usableIpcBodyMapper = mapper;
       }
       if (usableIpcBodyMapper.add(streamId, ipcBody)) {

toolkit/dweb-helper/src/Producer.ts

@@ -80,9 +80,9 @@ export class Producer<T> {
     /**过滤触发 */
     consumeMapNotNull<R>(mapNotNull: (data: T) => R | undefined): R | undefined {
       const result = mapNotNull(this.data);
-      if (result !== null) {
+      if (result !== null && result !== undefined) {
         this.consume();
-        return result as R;
+        return result;
       }
     }
     consumeFilter<R extends T = T>(filter: (data: T) => boolean) {
@@ -148,14 +148,14 @@ export class Producer<T> {
   /**无保证发送 */
   sendBeacon(value: T) {
     const event = this.event(value);
-    this.doEmit(event);
+    return this.doEmit(event);
   }
 
   trySend(value: T) {
     if (this.isClosedForSend) {
-      this.sendBeacon(value);
+      return this.sendBeacon(value);
     } else {
-      this.#doSend(value);
+      return this.#doSend(value);
     }
   }
 
@@ -197,9 +197,7 @@ export class Producer<T> {
   //#region Consumer
   /**消费者 */
   static #Consumer = class Consumer<T> {
-    constructor(readonly name: string, readonly producer: Producer<T>) {
-      producer.consumers.add(this);
-    }
+    constructor(readonly name: string, readonly producer: Producer<T>) {}
 
     toString() {
       return `Consumer<[${this.producer.name}]${this.name}>`;
@@ -279,6 +277,8 @@ export class Producer<T> {
     #destroySignal = createSignal();
     onDestroy = this.#destroySignal.listen;
     close(cause?: string) {
+      this.input.close();
+      this.producer.consumers.delete(this);
       this.#errorCatcher.resolve(cause);
     }
   };