@netscript/plugin-streams

Durable Streams development plugin for NetScript: a plugin manifest plus CLI, scaffolding, end-to-end gate, and Aspire integration surfaces for a durable, change-data stream service. This page is written against the plugin public surface reported by deno doc. For the full index of packages and plugins return to the reference overview.

The plugin ships five published entrypoints. The root export (@netscript/plugin-streams) carries the manifest and the typed topic/producer/consumer authoring helpers; four sub-path exports carry the framework integrations:

• @netscript/plugin-streams/cli — plugin CLI command group.
• @netscript/plugin-streams/scaffolding — scaffolder descriptor.
• @netscript/plugin-streams/e2e — E2E gate definitions.
• @netscript/plugin-streams/aspire — Aspire AppHost contribution.

The schema, producer, telemetry, testing, and diagnostics primitives that the plugin builds on live in the separately published @netscript/plugin-streams-core package, which has its own canonical reference page.

Plugin manifest

Topic authoring

Not yet wired. defineStreamProducer and defineStreamConsumer are manifest-layer stubs. A producer's publish() returns a rejected promise and a consumer's subscribe() throws synchronously, both with StreamUnsupportedOperationError. For real producer work use createDurableStream (or the Service-facing createServiceStreamProducer) and defineStreamSchema from @netscript/plugin-streams-core. See the durable streams capability page for the full producer/consumer model.

Browser consumers read over HTTP/SSE. There is no in-process subscribe(); a browser consumes a stream by reading its HTTP/SSE endpoint (for example with EventSource). The local AppHost serves the generated app over http://, and under HTTP/1.1 browsers typically allow only about six concurrent connections per origin — so several long-lived stream subscriptions from one page can starve later requests. Serve over HTTPS (HTTP/2) when a page needs many simultaneous stream consumers.

Re-exported plugin framework types

The root export re-exports the shared NetScript plugin-framework contribution types unchanged from @netscript/plugin. See the @netscript/plugin reference for their full definitions.

Sub-path exports

@netscript/plugin-streams/cli

CLI command group for the plugin, mounted under deno x -A jsr:@netscript/plugin-streams@0.0.6/cli.

The CLI types PluginCli, PluginCliArgs, PluginCliCommand, and PluginCliResult are re-exported from @netscript/plugin and document the base command contract StreamsCli extends.

@netscript/plugin-streams/scaffolding

@netscript/plugin-streams/e2e

@netscript/plugin-streams/aspire

The Aspire entrypoint contributes the Durable Streams development service to an Aspire AppHost.

The remaining symbols on this entrypoint (AspireBuilder, AspireResource, AspireResourceKind, AspireNSPluginContribution, ContributionContext, CacheSpec, ContainerSpec, DatabaseSpec, DenoServiceSpec, DenoBackgroundSpec, HealthCheckSpec, EnvSource) are re-exported unchanged from @netscript/aspire; see that reference page for their definitions.

Core package

The separately published @netscript/plugin-streams-core page is canonical for stream schemas, producers, configuration, telemetry, testing, and diagnostics. This page stays focused on the deployable plugin's manifest and integration entrypoints. The testing example below uses core APIs intentionally; exhaustive core entrypoint and symbol tables live only on its reference page.

Socket-Free Testing and Telemetry

Use MemoryStreamProducer in unit and integration tests to verify change events are published without opening network sockets or starting a real streams server. Use createStreamsInstrumentation (from ./telemetry) to verify context-propagation headers (like OpenTelemetry traceparent) are correctly injected.

import { assertEquals, assertExists } from "@std/assert";
import { MemoryStreamProducer } from "@netscript/plugin-streams-core/testing";
import { createStreamsInstrumentation } from "@netscript/plugin-streams-core/telemetry";

Deno.test("stream socket-free testing with MemoryStreamProducer and telemetry facade", async () => {
  // 1. Socket-free testing using MemoryStreamProducer
  const producer = new MemoryStreamProducer();

  // Publish some mock events
  producer.upsert("orders", { id: "order_123", total: 100 });
  producer.delete("orders", "order_123");
  await producer.flush();

  // Assert events are recorded synchronously in memory without network sockets
  assertEquals(producer.events(), [
    {
      entityType: "orders",
      operation: "upsert",
      key: "order_123",
      value: { id: "order_123", total: 100 },
    },
    {
      entityType: "orders",
      operation: "delete",
      key: "order_123",
    },
  ]);

  await producer.close();

  // 2. Using the Telemetry Facade to publish and inject trace context
  const telemetry = createStreamsInstrumentation();
  let traceHeaders: Record<string, string> = {};

  telemetry.publish({
    streamPath: "/orders/updates",
    collection: "orders",
    operation: "upsert",
    producerId: "orders-service",
    messageId: "order_123",
    emit: (headers) => {
      traceHeaders = headers;
    },
  });

  // Assert traceparent is generated and injected by the telemetry facade
  assertExists(traceHeaders.traceparent);
});

Back to the reference overview.