Feature/pipeline explicit native release

README.md

@@ -810,6 +810,52 @@ setTimeout(async () => {
 }, 5000);
 ```
 
+### Releasing a Pipeline (dispose)
+
+Call `dispose()` when you are permanently done with a pipeline to free its native
+memory right away, instead of waiting for the garbage collector to finalize the
+wrapper.
+
+Why it matters: the native `GstPipeline` allocation (buffers, decoders, GStreamer
+internals) lives outside V8's heap and is invisible to its GC accounting. A
+pipeline you simply drop is only reclaimed when GC happens to collect the small JS
+wrapper — and with a flat JS heap, V8 feels little pressure to do so. A long
+running process that builds a pipeline per recording or transcode can watch RSS
+climb steadily while the JS heap stays flat. `dispose()` closes that gap: it
+drives the pipeline to NULL if it is not already there and drops the native
+reference synchronously, so the memory is released as soon as the last reference
+goes away rather than at some later GC.
+
+```javascript
+import { Pipeline } from "gst-kit";
+
+const pipeline = new Pipeline("videotestsrc ! theoraenc ! oggmux ! filesink location=out.ogv");
+await pipeline.play();
+
+// ... record, then finish cleanly ...
+pipeline.endOfStream();
+while (true) {
+  const msg = await pipeline.busPop(1000);
+  if (msg?.type === "eos") break;
+}
+await pipeline.stop();
+
+// Release the native pipeline now rather than at some later GC.
+pipeline.dispose();
+
+// dispose() is terminal and idempotent. Any method call afterward throws
+// "Pipeline used after dispose()", and a second dispose() is a harmless no-op.
+```
+
+**Key points:**
+
+- **Stop first (recommended)**: prefer `stop()` (or `endOfStream()` + wait for EOS, then `stop()`) before `dispose()`. If the pipeline has not reached the NULL state, `dispose()` drives it there synchronously before releasing it — GStreamer cannot cleanly free a non-NULL pipeline. That transition is normally instant, but on a not-fully-stopped pipeline it runs on the calling thread and is bounded by a 5s wait, so stopping first keeps `dispose()` cheap and non-blocking.
+- **Terminal**: call it once, when the pipeline will not be used again.
+- **Not a graceful stop**: `dispose()` does the hard release, not a flush.
+- **Release elements first**: any element obtained via `getElementByName()`, and any pad-probe or `onSample()` subscription on it, must be released/unsubscribed before `dispose()`. Elements hold their own reference and are not invalidated by disposing the pipeline.
+- **Idempotent**: calling `dispose()` again does nothing.
+- **Use-after-dispose is loud**: subsequent method calls throw rather than touching freed memory.
+
 ### Message Bus Handling
 
 ```javascript
@@ -916,6 +962,9 @@ class Pipeline {
 
   // Message handling
   busPop(timeoutMs?: number): Promise<GstMessage | null>;
+
+  // Lifecycle — release the native pipeline and free its memory immediately
+  dispose(): void;
 }
 ```
 
@@ -1072,6 +1121,7 @@ gst-kit/
 │   ├── appsrc.mjs            # AppSrc usage
 │   ├── appsrc-eos.mjs        # AppSrc with end-of-stream
 │   ├── pipeline-eos.mjs      # Pipeline-level end-of-stream
+│   ├── dispose.mjs           # Releasing a pipeline's native memory
 │   ├── record-to-file.mjs    # Recording to file example
 │   ├── rtp-timestamp.mjs     # RTP handling
 │   ├── bus.mjs               # Message bus handling

examples/dispose.mjs

@@ -0,0 +1,60 @@
+#!/usr/bin/env node
+/**
+ * Pipeline Disposal Example
+ *
+ * Demonstrates dispose() — releasing a pipeline's native memory eagerly instead
+ * of waiting for garbage collection to finalize the wrapper.
+ *
+ * The native GstPipeline allocation is invisible to V8's heap accounting, so a
+ * pipeline you simply drop is not reclaimed until GC happens to collect the small
+ * JS wrapper. A process that builds a pipeline per unit of work (recording,
+ * transcode, feed) can see RSS climb steadily while the JS heap stays flat.
+ * dispose() closes that gap: it drives the pipeline to NULL if it is not already
+ * there and drops the native reference synchronously, so the memory is released
+ * without waiting for GC. Stopping first is recommended so dispose() stays cheap
+ * and non-blocking, but dispose() will force the NULL transition itself if
+ * needed.
+ *
+ * This example builds, plays, stops, and disposes many short-lived pipelines in a
+ * loop — the exact pattern where relying on GC timing leaks native memory. RSS is
+ * always reported; run with `node --expose-gc examples/dispose.mjs` to also force
+ * a GC at the end so the "after" figure excludes collectable JS wrappers.
+ */
+import { Pipeline } from "../dist/esm/index.mjs";
+
+const ITERATIONS = 25;
+
+function rssMb() {
+  return (process.memoryUsage().rss / 1024 / 1024).toFixed(1);
+}
+
+console.log(`📊 RSS before: ${rssMb()} MB`);
+
+for (let i = 1; i <= ITERATIONS; i++) {
+  const pipeline = new Pipeline("videotestsrc ! fakesink");
+
+  await pipeline.play();
+  await pipeline.stop();
+
+  // Release the native pipeline now. Without this, each iteration's native
+  // allocation would linger until GC decided to collect the wrapper.
+  pipeline.dispose();
+
+  // dispose() is idempotent and terminal: a second call is a no-op, and any
+  // other method call after dispose() throws "Pipeline used after dispose()".
+  pipeline.dispose();
+  try {
+    pipeline.playing();
+  } catch (err) {
+    if (i === 1) console.log(`🔒 use-after-dispose is guarded: ${err.message}`);
+  }
+
+  if (i % 5 === 0) console.log(`♻️  disposed ${i}/${ITERATIONS} — RSS: ${rssMb()} MB`);
+}
+
+if (typeof globalThis.gc === "function") {
+  globalThis.gc();
+}
+
+console.log(`📊 RSS after: ${rssMb()} MB`);
+console.log("✅ Done — native pipelines were released as each one finished.");

src/cpp/async-workers.cpp

@@ -1,4 +1,5 @@
 #include "async-workers.hpp"
+#include "pipeline.hpp"
 #include "type-conversion.hpp"
 #include <gst/gst.h>
 
@@ -195,11 +196,12 @@ void PullSampleWorker::cleanup() {
 
 // StateChangeWorker implementation
 StateChangeWorker::StateChangeWorker(
-  const Napi::Env &env, GstPipeline *pipeline, GstState target_state, GstClockTime timeout
+  const Napi::Env &env, Pipeline *owner, GstPipeline *pipeline, GstState target_state,
+  GstClockTime timeout
 ) :
-    Napi::AsyncWorker(env), pipeline(pipeline), target_state(target_state), timeout(timeout),
-    state_change_result(GST_STATE_CHANGE_FAILURE), final_state(GST_STATE_VOID_PENDING),
-    deferred(env) {
+    Napi::AsyncWorker(env), owner(owner), pipeline(pipeline), target_state(target_state),
+    timeout(timeout), state_change_result(GST_STATE_CHANGE_FAILURE),
+    final_state(GST_STATE_VOID_PENDING), deferred(env), owner_notified(false) {
   // Increase reference count since we'll be using this in another thread
   gst_object_ref(pipeline);
 }
@@ -257,11 +259,17 @@ void StateChangeWorker::OnOK() {
   result.Set("finalState", Napi::Number::New(Env(), final_state));
   result.Set("targetState", Napi::Number::New(Env(), target_state));
 
+  // Notify the owner before resolving so any deferred dispose() teardown runs
+  // while this worker's own reference is still held — the release then sees the
+  // last reference drop and finalizes a truly-NULL pipeline.
+  notify_owner_finished();
+
   deferred.Resolve(result);
 }
 
 void StateChangeWorker::OnError(const Napi::Error &error) {
   Napi::HandleScope scope(Env());
+  notify_owner_finished();
   deferred.Reject(error.Value());
 }
 
@@ -271,3 +279,12 @@ void StateChangeWorker::cleanup() {
     pipeline = nullptr;
   }
 }
+
+void StateChangeWorker::notify_owner_finished() {
+  // Runs on the JS thread (OnOK/OnError). Idempotent: OnOK and OnError are
+  // mutually exclusive, but guard anyway so the owner count can never be
+  // decremented twice for one worker.
+  if (owner_notified) return;
+  owner_notified = true;
+  owner->state_worker_finished();
+}

src/cpp/async-workers.hpp

@@ -53,11 +53,14 @@ class PullSampleWorker : public Napi::AsyncWorker {
   Napi::Promise::Deferred deferred;
 };
 
+class Pipeline;
+
 // AsyncWorker for pipeline state changes with timeout
 class StateChangeWorker : public Napi::AsyncWorker {
 public:
   StateChangeWorker(
-    const Napi::Env &env, GstPipeline *pipeline, GstState target_state, GstClockTime timeout
+    const Napi::Env &env, Pipeline *owner, GstPipeline *pipeline, GstState target_state,
+    GstClockTime timeout
   );
   ~StateChangeWorker();
 
@@ -69,11 +72,14 @@ class StateChangeWorker : public Napi::AsyncWorker {
 
 private:
   void cleanup();
+  void notify_owner_finished();
 
+  Pipeline *owner;
   GstPipeline *pipeline;
   GstState target_state;
   GstClockTime timeout;
   GstStateChangeReturn state_change_result;
   GstState final_state;
   Napi::Promise::Deferred deferred;
+  bool owner_notified;
 };