> For the complete documentation index, see [llms.txt](https://docs.anthriq.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.anthriq.com/bxi-studio/developer/overview/features/parallel.md).

# Run parallel streams

> **Applies to:** [Instinct](/bxi-studio/developer/overview/devices/instinct.md) and [xSys](/bxi-studio/developer/overview/devices/xsys.md). Not available on [xBud](/bxi-studio/developer/overview/devices/xbud.md).

EXG and impedance can run at the same time, which is how you watch electrode contact without interrupting a recording. Several consumers can also read one stream.

## Run EXG and impedance together

Each stream needs its own `stream_id` and port. The two features are independent, so the lifecycles overlap freely.

{% tabs %}
{% tab title="Node.js" %}

```typescript
const streams = [
  { feature: "eeg", streamId: 20, port: 9020, flush: 80 },
  { feature: "impedance", streamId: 21, port: 9021, flush: 1 },
];

const results = await Promise.allSettled(
  streams.map((s) =>
    client.features.executeOperation(s.feature, "add_stream", {
      stream: {
        stream_id: s.streamId,
        protocol: "websocket",
        host: "127.0.0.1",
        port: s.port,
        elements_before_flush: s.flush,
      },
    })
  )
);

const added = results.filter((r) => r.status === "fulfilled" && r.value.success);
console.log(`${added.length}/${streams.length} streams added`);
```

{% endtab %}

{% tab title="Python" %}

```python
import asyncio

streams = [
    {"feature": "eeg", "stream_id": 20, "port": 9020, "flush": 80},
    {"feature": "impedance", "stream_id": 21, "port": 9021, "flush": 1},
]

results = await asyncio.gather(
    *(
        client.execute_operation(s["feature"], "add_stream", {
            "stream": {
                "stream_id": s["stream_id"],
                "protocol": "udp",
                "host": "127.0.0.1",
                "port": s["port"],
                "elements_before_flush": s["flush"],
            },
        })
        for s in streams
    ),
    return_exceptions=True,
)

added = [r for r in results if not isinstance(r, Exception) and r["success"]]
print(f"{len(added)}/{len(streams)} streams added")
```

{% endtab %}
{% endtabs %}

`Promise.allSettled` keeps one failed stream from rejecting the batch. `Promise.all` would reject on the first failure and abandon the others mid-setup, leaving streams registered on the device with no local record of them.

## Watch contact during a recording

The usual arrangement: EXG into storage, impedance into a contact indicator.

{% tabs %}
{% tab title="Node.js" %}

```typescript
await client.features.subscribeToOperation(
  "eeg",
  "stream",
  (update) => {
    if (update.success) writeToDisk(update.data.details.payload.frames);
  },
  { stream_id: 20 }
);

await client.features.subscribeToOperation(
  "impedance",
  "stream",
  (update) => {
    if (!update.success) return;
    for (const frame of update.data.details.payload.frames ?? []) {
      for (const channel of frame.channels ?? []) {
        if (channel.has_error) continue;
        updateContactIndicator(channel.channel_id, channel.adc_data);
      }
    }
  },
  { stream_id: 21 }
);
```

{% endtab %}

{% tab title="Python" %}

```python
await client.subscribe_to_operation(
    "eeg",
    "stream",
    lambda update: write_to_disk(update["data"]["details"]["payload"]["frames"]),
    {"stream_id": 20},
)


def on_impedance(update):
    payload = update["data"]["details"]["payload"]
    for frame in payload.get("frames", []):
        for channel in frame.get("channels", []):
            if channel["has_error"]:
                continue
            update_contact_indicator(channel["channel_id"], channel["adc_data"])


await client.subscribe_to_operation(
    "impedance", "stream", on_impedance, {"stream_id": 21}
)
```

{% endtab %}
{% endtabs %}

The indicator receives magnitudes, not ohms. Turning them into a contact verdict needs a calibration step the SDK does not provide — see [Measure contact impedance](/bxi-studio/developer/overview/features/impedance.md).

## Share one stream between consumers

Subscribing twice to the same stream costs no extra device traffic. The SDK opens one bridge subscription and fans updates out locally.

{% tabs %}
{% tab title="Node.js" %}

```typescript
const plotter = await client.features.subscribeToOperation(
  "eeg", "stream", updatePlot, { stream_id: 20 }
);

const recorder = await client.features.subscribeToOperation(
  "eeg", "stream", writeToDisk, { stream_id: 20 }
);
```

{% endtab %}

{% tab title="Python" %}

```python
plotter = await client.subscribe_to_operation(
    "eeg", "stream", update_plot, {"stream_id": 20}
)

recorder = await client.subscribe_to_operation(
    "eeg", "stream", write_to_disk, {"stream_id": 20}
)
```

{% endtab %}
{% endtabs %}

Each call returns its own subscription id. Unsubscribing the plotter leaves the recorder running; the bridge subscription is released only when the last local callback goes.

Concurrent subscribes are safe. The SDK holds a per-stream lock, so simultaneous calls still produce one bridge subscription rather than two.

{% tabs %}
{% tab title="Node.js" %}

```typescript
// Both resolve against a single bridge subscription.
const [a, b] = await Promise.all([
  client.features.subscribeToOperation("eeg", "stream", first, { stream_id: 20 }),
  client.features.subscribeToOperation("eeg", "stream", second, { stream_id: 20 }),
]);
```

{% endtab %}

{% tab title="Python" %}

```python
import asyncio

# Both resolve against a single bridge subscription.
first_sub, second_sub = await asyncio.gather(
    client.subscribe_to_operation("eeg", "stream", first, {"stream_id": 20}),
    client.subscribe_to_operation("eeg", "stream", second, {"stream_id": 20}),
)
```

{% endtab %}
{% endtabs %}

## Isolate a failing callback

A callback that throws is caught and logged, and the other callbacks on that stream still run. One consumer's bug cannot silence the others or end the subscription.

It does not, however, slow-path safely: a callback that blocks delays every other callback on the stream, because they share the event loop. Keep each one short.

## Tear down in order

Stop every stream before unsubscribing, and remove each one.

{% tabs %}
{% tab title="Node.js" %}

```typescript
try {
  // Acquire.
} finally {
  await Promise.allSettled([
    client.features.executeOperation("eeg", "stop_stream", { stream_id: 20 }),
    client.features.executeOperation("impedance", "stop_stream", { stream_id: 21 }),
  ]);

  // Longest drain window of the streams in play.
  await new Promise((resolve) => setTimeout(resolve, 2000));

  for (const id of subscriptionIds) {
    await client.features.unsubscribeFromOperation("eeg", "stream", { subscriptionId: id });
  }

  await Promise.allSettled([
    client.features.executeOperation("eeg", "remove_stream", { stream_id: 20 }),
    client.features.executeOperation("impedance", "remove_stream", { stream_id: 21 }),
  ]);
}
```

{% endtab %}

{% tab title="Python" %}

```python
import asyncio

try:
    pass  # Acquire.
finally:
    await asyncio.gather(
        client.execute_operation("eeg", "stop_stream", {"stream_id": 20}),
        client.execute_operation("impedance", "stop_stream", {"stream_id": 21}),
        return_exceptions=True,
    )

    # Longest drain window of the streams in play.
    await asyncio.sleep(2)

    for subscription_id in subscription_ids:
        await client.unsubscribe_from_operation(
            "eeg", "stream", {"subscriptionId": subscription_id}
        )

    await asyncio.gather(
        client.execute_operation("eeg", "remove_stream", {"stream_id": 20}),
        client.execute_operation("impedance", "remove_stream", {"stream_id": 21}),
        return_exceptions=True,
    )
```

{% endtab %}
{% endtabs %}

Wait the longest drain window of the streams in play — 2 seconds when impedance is running, rather than EXG's 1.5.

> **Warning:** Use `Promise.allSettled` in teardown, not `Promise.all`. A rejection from one stop would skip the remaining cleanup and leave the other stream transmitting.

## Watch the stream limit

Devices declare a maximum stream count in capabilities. Check it before adding streams in a loop:

{% tabs %}
{% tab title="Node.js" %}

```typescript
const limits = client.getCachedCapabilities().features.eeg?.limits;
console.log(limits?.max_streams);
```

{% endtab %}

{% tab title="Python" %}

```python
limits = client.get_cached_capabilities().features["eeg"].limits
print(limits.get("max_streams"))
```

{% endtab %}
{% endtabs %}

## Next steps

* [Stream EXG](/bxi-studio/developer/overview/features/exg.md)
* [Measure contact impedance](/bxi-studio/developer/overview/features/impedance.md)
* [Run pipelines](/bxi-studio/developer/overview/backend-services/pipelines.md)