@livequery/rpc

Small TypeScript utilities for RPC-style communication between a main thread and a SharedWorker or Chrome extension runtime. It uses RxJS for request streams, cancellation, and observable state.

Use it when you want a service-style API across a worker boundary instead of manually passing postMessage objects around.

Install

bun add @livequery/rpc rxjs

npm install @livequery/rpc rxjs

Core Idea

Main thread                          Worker or extension runtime
-----------                          ---------------------------
ServiceLinker -- RpcMessage -------> WorkerManager -> service instance
     ^                                      |
     |----------- response stream ----------|

• ServiceLinker creates a typed client proxy.
• WorkerManager exposes service instances and dispatches incoming calls.
• SharedWorkerChannel transports messages over SharedWorker.
• ExtensionChannel transports messages over chrome.runtime.
• WorkerService<T> maps a worker-side service type into the client-side type.

Error Handling

When a worker-side service method throws (or rejects), WorkerManager converts the thrown value into a structured-clone-safe { code, message, stack? } before it crosses the worker boundary — postMessage cannot transfer a class Error, and a raw object would otherwise surface as "[object Object]". The client rejects with that shape.

The helper is exported for reuse and testing:

import { serializeError } from "@livequery/rpc"

serializeError(new Error("boom"))          // → { code: "Error", message: "boom", stack: "…" }
serializeError({ code: "E_CONN" })          // → { code: "E_CONN", message: '{"code":"E_CONN"}' }
serializeError("nope")                      // → { code: "InternalError", message: "nope" }

It never throws and always returns string code/message. code is taken from err.code ?? err.name ?? "InternalError"; message from err.message, falling back to a JSON of the error's own fields (or code when the value is empty / circular / not serializable).

SharedWorker Example

1. Define a service

import { BehaviorSubject, interval, map } from "rxjs"

export class CounterService {
  value = new BehaviorSubject(0)

  increment(by = 1) {
    const nextValue = this.value.getValue() + by
    this.value.next(nextValue)
    return nextValue
  }

  ticker() {
    return interval(1000).pipe(map(index => `tick-${index}`))
  }

  profile = {
    getName: () => "Ada",
  }
}

2. Expose it in the worker

import { SharedWorkerChannel, WorkerManager } from "@livequery/rpc"
import { CounterService } from "./CounterService"

const channel = new SharedWorkerChannel()
const manager = new WorkerManager(channel)

manager.exposeService("counter", new CounterService())

3. Connect from the main thread

import { ServiceLinker, SharedWorkerChannel, type WorkerService } from "@livequery/rpc"
import type { CounterService } from "./CounterService"

const worker = new SharedWorker(new URL("./worker.ts", import.meta.url), { type: "module" })
const channel = new SharedWorkerChannel(worker)
const linker = new ServiceLinker(channel)

const counter = linker.linkService<WorkerService<CounterService>>("counter")

4. Call methods

Use await for one-shot values:

const nextValue = await counter.increment(2)

Use subscribe() for streamed values:

const subscription = counter.ticker().subscribe(value => {
  console.log(value)
})

subscription.unsubscribe()

Nested methods work through normal property access:

const name = await counter.profile.getName()

Remote State

Expose a BehaviorSubject property on the worker service when the client should observe shared state:

const subscription = counter.value.subscribe(value => {
  console.log("counter value", value)
})

const current = counter.value.getValue()

Notes:

• subscribe, pipe, and getValue are special-cased for remote observable-like properties.
• The first subscription creates and caches the local shared observable for that property path.
• getValue() reads the local cache. Subscribe first if you need the current worker value to be populated.

Chrome Extension Runtime

Use ExtensionChannel inside Chrome extension contexts where chrome.runtime is available.

Background service worker:

import { ExtensionChannel, WorkerManager } from "@livequery/rpc"
import { CounterService } from "./CounterService"

const channel = new ExtensionChannel()
const manager = new WorkerManager(channel)

manager.exposeService("counter", new CounterService())

Popup, options page, or content script:

import { ExtensionChannel, ServiceLinker, type WorkerService } from "@livequery/rpc"
import type { CounterService } from "./CounterService"

const channel = new ExtensionChannel()
const linker = new ServiceLinker(channel)

const counter = linker.linkService<WorkerService<CounterService>>("counter")

ExtensionChannel auto-detects foreground versus background context. If chrome is unavailable, operations silently no-op.

Call Behavior

Every remote method call returns an Observable with a custom then() implementation. That means the same call can often be used as either:

const result = await service.someMethod()

or:

const subscription = service.someMethod().subscribe(value => {
  console.log(value)
})

For clarity, prefer:

• await for plain values and promises
• subscribe() for observable streams

If a client unsubscribes before a request completes, a cancellation message is sent to the worker and WorkerManager unsubscribes from the worker-side stream.

Utility Helpers

LimitConcurrency

Decorator for limiting concurrent method execution:

import { LimitConcurrency } from "@livequery/rpc"

class ApiService {
  @LimitConcurrency(2)
  async fetchItem(id: string) {
    return { id }
  }
}

RxjsQueue

Small concurrency-limited async queue:

import { RxjsQueue } from "@livequery/rpc"

const queue = new RxjsQueue(2)
const result = await queue.run(() => fetchSomething())

The default concurrency is 1.

StorageBehaviorSubject

BehaviorSubject that initializes from storage and persists every next():

import { StorageBehaviorSubject } from "@livequery/rpc"

const theme$ = new StorageBehaviorSubject(storage, "theme", "light")
theme$.next("dark")

The storage adapter shape is:

type IStorage = {
  getItem: <T>(key: string) => Promise<T | undefined> | T | undefined
  setItem: <T>(key: string, value: T) => void
}

Exports

export * from "./RpcChannel.js"
export * from "./ExtensionChannel.js"
export * from "./SharedWorkerChannel.js"
export * from "./ServiceLinker.js"
export * from "./WorkerService.js"
export * from "./WorkerManager.js"
export * from "./LimitConcurrency.js"
export * from "./StorageBehaviorSubject.js"
export * from "./RxjsQueue.js"

Constraints

• Service paths beginning with # are invalid.
• Worker streams are detected by checking for a pipe() method.
• Falsy values such as 0, false, "", and null are valid RPC payloads.
• Errors are propagated to the client as Error(message), with the worker-side stack preserved on the error for debugging.
• There is no built-in readiness API in ServiceLinker.
• When a connection (port) drops, the channel emits { disconnect: true, connection_id } so the WorkerManager releases any streaming subscriptions owned by that connection.

Development

bun run test
bun run build

Build output is written to dist/.