Publish and subscribe with the JavaScript SDK

• 1: Publish events with the JavaScript SDK
• 2: Receive events with a JavaScript application server
• 3: Receive events with JavaScript streaming subscriptions
• 4: JavaScript pub/sub configuration and usage

npm install @dapr/dapr@3.19.0

1 - Publish events with the JavaScript SDK

Use DaprClient.pubsub to publish events over HTTP or gRPC with @dapr/dapr 3.19.0 or later. Complete the pub/sub prerequisites before following this guide.

Configure a publisher

The examples use this HTTP client, component pubsub, and topic orders:

import { CommunicationProtocolEnum, DaprClient } from "@dapr/dapr";

const client = new DaprClient({
  daprHost: "127.0.0.1",
  daprPort: "3500",
  communicationProtocol: CommunicationProtocolEnum.HTTP,
});

For gRPC, select CommunicationProtocolEnum.GRPC and port "50001". Run statements containing await in an async function or an ES module. Reuse the client and stop it during application shutdown; see configuration and usage.

Single messages and metadata

const response = await client.pubsub.publish(
  "pubsub",
  "orders",
  { orderId: "101", total: 49.99 },
  {
    contentType: "application/json",
    metadata: { ttlInSeconds: "60" },
  },
);

if (response.error) {
  throw response.error;
}

Without a custom serializer or explicit contentType, the SDK infers application/json for objects and arrays, text/plain for strings, and application/cloudevents+json for recognized CloudEvents.

metadata carries Dapr publish options and component-specific settings, such as message TTL or partition keys. It is not a general-purpose mechanism for adding CloudEvent attributes or arbitrary subscriber headers. Supported keys depend on the component.

Publish CloudEvents

Dapr normally wraps published data in a CloudEvent. To supply your own event attributes, publish a CloudEvent explicitly:

const response = await client.pubsub.publish("pubsub", "orders", {
  specversion: "1.0",
  id: "order-101-created",
  source: "/checkout",
  type: "com.example.order.created",
  datacontenttype: "application/json",
  data: { orderId: "101", total: 49.99 },
});

if (response.error) {
  throw response.error;
}

See CloudEvents for envelope attributes and extensions.

Raw payloads

To publish without a CloudEvent envelope:

const response = await client.pubsub.publish("pubsub", "orders-raw", "order-101", {
  contentType: "text/plain",
  metadata: { rawPayload: "true" },
});

if (response.error) {
  throw response.error;
}

Configure the receiving subscription with metadata: { rawPayload: true } for server subscriptions, or metadata: { rawPayload: "true" } for streaming subscriptions. Server subscription metadata values are JSON-stringified before discovery; streaming metadata is sent directly as a string map. This distinction matters for boolean options such as rawPayload.

Raw payloads do not provide the same CloudEvent attributes as enveloped events. See raw payloads.

Bulk publishing

publishBulk(pubSubName, topic, messages, metadata?) accepts strings, objects, or explicit entries with event, optional entryID, and optional contentType. Missing entry IDs are generated as UUIDs:

const response = await client.pubsub.publishBulk(
  "pubsub",
  "orders",
  [
    { entryID: "order-101", event: { orderId: "101" }, contentType: "application/json" },
    { entryID: "order-102", event: { orderId: "102" }, contentType: "application/json" },
  ],
  { ttlInSeconds: "60" },
);

for (const failure of response.failedMessages) {
  console.error("Publish failed:", failure.message.entryID, failure.error);
}

if (response.failedMessages.length > 0) {
  throw new Error(`Failed to publish ${response.failedMessages.length} orders`);
}

Bulk publishing is not transactional. A resolved promise can contain failures; always inspect failedMessages. Each failure contains the original normalized message and an error. Request-level failures are generally represented as failures for every entry, while argument processing, serialization, or client initialization can still reject.

Bulk publishing uses the stable API on Dapr 1.17 or later, with automatic fallback to the deprecated alpha1 API on older runtimes. HTTP selects the API using sidecar metadata; gRPC falls back when the stable RPC is unsupported.

Use the fourth argument for request metadata on either transport. Although the explicit entry type exposes metadata, this SDK version does not forward per-entry metadata in the gRPC bulk request. Also, without a custom serializer, the gRPC bulk path infers content types from each event rather than honoring explicit entry content types. Avoid relying on those fields for cross-protocol behavior.

See bulk publish and subscribe.

Publish error handling

Single-message publishing has different failure behavior by transport in this version:

Await calls inside your application’s error boundary and inspect their response fields. Do not assume that resolution means publication succeeded, or that every rejection is an Error instance. Publisher success means Dapr accepted the publish request, not that all subscribers processed the event.

2 - Receive events with a JavaScript application server

Use DaprServer.pubsub to receive events through HTTP or gRPC callbacks with @dapr/dapr 3.19.0 or later. Complete the pub/sub prerequisites before following this guide. For workers without an application callback listener, use streaming subscriptions instead.

Register subscriptions before startup

Save the following as subscriber.js in an ES module application:

import { CommunicationProtocolEnum, DaprPubSubStatusEnum, DaprServer } from "@dapr/dapr";

const server = new DaprServer({
  serverHost: "127.0.0.1",
  serverPort: "50002",
  communicationProtocol: CommunicationProtocolEnum.HTTP,
  clientOptions: { daprHost: "127.0.0.1", daprPort: "3500" },
});

await server.pubsub.subscribe("pubsub", "orders", async (data, headers) => {
  console.log("Order received:", data, headers);
  return DaprPubSubStatusEnum.SUCCESS;
});

await server.start();

Run the subscriber with an application port matching serverPort:

dapr run --app-id orders-subscriber --app-port 50002 --app-protocol http --dapr-http-port 3500 --resources-path ./components -- node subscriber.js

For a gRPC server, select CommunicationProtocolEnum.GRPC, set the client port to "50001", and run Dapr with --app-protocol grpc --dapr-grpc-port 50001.

The remaining examples assume this server and enum import. Add their subscription registrations before server.start().

HTTP subscriptions are exposed through /dapr/subscribe; gRPC subscriptions use ListTopicSubscriptions. Configure each component/topic pair once. Repeating subscribe() or subscribeWithOptions() for an already registered pair throws. Use subscribeToRoute() to add another handler to an existing route instead.

Handlers can be attached to existing routes after startup, but adding a handler does not cause Dapr to rediscover new topics or routes. There is no server-side unsubscribe method in this interface. Stop the application listener with await server.stop() during shutdown. See configuration and usage for client lifetime and serialization.

Callback data and headers

Callbacks receive (data, headers) and should be async. data is normally the deserialized event payload, not the full CloudEvent envelope.

Do not assume that server callbacks expose CloudEvent attributes in headers. If your application requires these attributes, the streaming callback explicitly provides them.

To consume raw payloads, use subscribeWithOptions() with metadata: { rawPayload: true }. Server subscription metadata values are JSON-stringified before discovery. See publishing raw payloads for publisher settings and the different streaming metadata representation.

Acknowledge, retry, or drop

Return the enum value, not an object such as { status: "RETRY" }. When multiple server handlers share a route, all run sequentially and their combined status has priority RETRY > DROP > SUCCESS.

Retries and dead-letter forwarding are controlled by Dapr resiliency policies and the pub/sub component, not by an SDK retry loop. Configure an inbound retry policy appropriate to your broker. For example, save the following in components/resiliency.yaml, targeting the component name pubsub:

apiVersion: dapr.io/v1alpha1
kind: Resiliency
metadata:
  name: pubsub-retries
spec:
  policies:
    retries:
      processing:
        policy: constant
        duration: 500ms
        maxRetries: 10
  targets:
    components:
      pubsub:
        inbound:
          retry: processing

Make handlers idempotent: Dapr pub/sub is at-least-once, and a message can be delivered again after processing.

Content-based routing

Use subscribeWithOptions() to declare the routes, then attach handlers with subscribeToRoute(). Dapr evaluates the CEL expressions against the CloudEvent:

await server.pubsub.subscribeWithOptions("pubsub", "routed-orders", {
  route: {
    default: "standard",
    rules: [{ match: 'event.type == "com.example.order.priority"', path: "priority" }],
  },
});

server.pubsub.subscribeToRoute("pubsub", "routed-orders", "standard", async (data) => {
  console.log("Standard order:", data);
});

server.pubsub.subscribeToRoute("pubsub", "routed-orders", "priority", async (data) => {
  console.log("Priority order:", data);
});

Route names are logical identifiers: the SDK generates unique callback paths from the component, topic, and route. A callback supplied alongside a route object is attached to all its declared routes, not just its default route.

See message routing.

Wildcard topics

Wildcard support depends on the broker and component. For MQTT-style subscriptions, the SDK recognizes # for multiple levels and + for one level when dispatching gRPC callbacks:

await server.pubsub.subscribe("pubsub", "sensors/#", async (data) => {
  console.log("Sensor event:", data);
});

await server.pubsub.subscribe("pubsub", "devices/+/temperature", async (data) => {
  console.log("Temperature:", data);
});

Use a component that supports these topic patterns; they are not a general wildcard facility for every pub/sub broker.

Bulk subscriptions

Bulk delivery batches messages between Dapr and the application. The SDK still invokes your callback for one entry at a time, not for an array. It constructs the per-entry acknowledgement response:

await server.pubsub.subscribeBulk(
  "pubsub",
  "bulk-orders",
  async (data) => {
    console.log("Order from batch:", data);
    return DaprPubSubStatusEnum.SUCCESS;
  },
  { maxMessagesCount: 100, maxAwaitDurationMs: 40 },
);

The options also accept route and metadata. Omitted batching limits use runtime/component defaults. For routing or dead-letter configuration combined with bulk delivery, use subscribeWithOptions() with bulkSubscribe: { enabled: true, maxMessagesCount: 100, maxAwaitDurationMs: 40 }.

In this version, the HTTP bulk handler without a custom serializer extracts CloudEvent data and decodes application/octet-stream entries, but does not populate callback data for direct application/json or text/plain entries. Prefer the normal Dapr CloudEvent envelope for HTTP bulk delivery, or use a custom serializer or gRPC when consuming those direct entry formats.

See bulk publish and subscribe.

Dead-letter topics

Set deadLetterTopic (case-sensitive) on the source subscription and register a separate subscription to consume that topic:

await server.pubsub.subscribeWithOptions("pubsub", "validated-orders", {
  deadLetterTopic: "orders-deadletter",
  callback: async (data) => {
    if (!data.orderId) {
      return DaprPubSubStatusEnum.RETRY;
    }
    console.log("Validated order:", data);
    return DaprPubSubStatusEnum.SUCCESS;
  },
});

await server.pubsub.subscribe("pubsub", "orders-deadletter", async (data) => {
  console.error("Order requires investigation:", data);
  return DaprPubSubStatusEnum.SUCCESS;
});

Dapr controls forwarding after processing fails and the configured retry policy is exhausted.

Although the SDK exposes deadLetterCallback, it only registers an additional local route on the source subscription. It does not create a subscription to the dead-letter topic, and specifying the callback alone does not configure runtime dead-letter forwarding. Do not use it as a substitute for the separate subscription above.

See dead-letter topics.

3 - Receive events with JavaScript streaming subscriptions

Use DaprClient.pubsub.subscribeStream() to open a bidirectional gRPC stream from your application to the sidecar with @dapr/dapr 3.19.0 or later. It does not require DaprServer, an application callback port, or subscription discovery.

Complete the pub/sub prerequisites first. Streaming uses SubscribeTopicEventsAlpha1, an alpha gRPC API; your Dapr runtime must implement it. HTTP clients reject this operation with HTTPNotSupportedError. Streaming subscriptions do not support routing rules or bulk subscription options.

Run a streaming worker

Save the following as worker.js in an ES module application. The worker runs until SIGINT or SIGTERM, or until the stream fails:

import { CommunicationProtocolEnum, DaprClient, DaprPubSubStatusEnum } from "@dapr/dapr";

const client = new DaprClient({
  daprHost: "127.0.0.1",
  daprPort: "50001",
  communicationProtocol: CommunicationProtocolEnum.GRPC,
});

const controller = new AbortController();
const stop = () => controller.abort();
process.once("SIGINT", stop);
process.once("SIGTERM", stop);

try {
  const subscription = await client.pubsub.subscribeStream(
    "pubsub",
    "orders",
    async (data, headers) => {
      console.log("Order received:", data, "CloudEvent:", headers);
      return DaprPubSubStatusEnum.SUCCESS;
    },
    { signal: controller.signal, deadLetterTopic: "orders-deadletter" },
  );

  try {
    await subscription.closed;
  } finally {
    await subscription.close();
  }
} finally {
  process.removeListener("SIGINT", stop);
  process.removeListener("SIGTERM", stop);
  await client.stop();
}

Run it without --app-port:

dapr run --app-id orders-worker --dapr-grpc-port 50001 --resources-path ./components -- node worker.js

The subscription promise resolves only after Dapr confirms setup. Setup failures reject. See configuration and usage for connection settings, authentication, and custom serialization.

Callback data and acknowledgements

Callbacks receive deserialized payload data and CloudEvent attributes in headers. Attributes include id, source, type, specversion, datacontenttype, topic, pubsubname, and path, together with flattened extensions. Unlike server callback headers, these explicitly expose CloudEvent attributes.

Callbacks run sequentially within a stream and use the client’s serializer. Return DaprPubSubStatusEnum.SUCCESS, RETRY, DROP, or no value; completing without a value means success. Callback exceptions, deserialization errors, and invalid return values are logged and acknowledged as RETRY.

Retry and dead-letter behavior depends on the runtime resiliency policy and component. Make handlers idempotent because a message can be delivered more than once.

Streaming options and lifecycle

For raw payloads, pass metadata: { rawPayload: "true" } in the subscription options. See publishing raw payloads. To consume failed messages, create a separate subscription to deadLetterTopic; setting the option does not register a consumer for that topic.

TypeScript applications can import PubSubStreamingSubscription and PubSubStreamingSubscriptionOptions as types from @dapr/dapr.

Supervise failures and shut down

Observe closed throughout the subscription lifetime. The SDK logs stream failures but does not reconnect automatically. If reconnecting is appropriate, implement a supervised retry policy that creates a new subscription and respects application shutdown.

client.stop() does not close these subscriptions. Explicitly close every subscription and await its closed promise before stopping the client. Cancellation during a callback leaves that event unacknowledged, so it may be redelivered. The API does not expose .NET-style queue capacity, per-message timeout, or automatic reconnect options.

4 - JavaScript pub/sub configuration and usage

This guide covers shared configuration and operational guidance for pub/sub with @dapr/dapr 3.19.0 or later. See the overview to choose between server callbacks and streaming subscriptions.

Client lifetime

Reuse a long-lived DaprClient rather than creating one per message. Await outstanding operations before calling await client.stop() at shutdown.

For server subscriptions, stop the application listener with await server.stop(). For streaming subscriptions, close each subscription and await its closed promise before stopping the client; client.stop() does not close streams. See streaming lifecycle guidance.

Endpoints and authentication

The default protocol is HTTP. To explicitly configure a gRPC connection:

import { CommunicationProtocolEnum, DaprClient } from "@dapr/dapr";

const client = new DaprClient({
  daprHost: "127.0.0.1",
  daprPort: "50001",
  communicationProtocol: CommunicationProtocolEnum.GRPC,
  daprApiToken: process.env.DAPR_API_TOKEN,
});

For HTTP, select CommunicationProtocolEnum.HTTP and port "3500". Port options are strings. The application callback protocol and port are configured separately on DaprServer; see server subscriptions.

When explicit connection settings are omitted, the SDK uses the environment settings for the selected protocol:

Custom serialization

Configure an IDaprSerializer using DaprClient’s serializer option. It converts application payloads to and from strings and can supply a contentType. It applies to single and bulk publishing and streaming callbacks. For server callbacks, configure it through DaprServer’s clientOptions.serializer.

import { DaprClient, DaprServer } from "@dapr/dapr";

const serializer = {
  contentType: "application/json",
  serialize(value) {
    return JSON.stringify({ payload: value });
  },
  deserialize(data) {
    return JSON.parse(data).payload;
  },
};

const customClient = new DaprClient({ serializer });
const customServer = new DaprServer({
  serverHost: "127.0.0.1",
  serverPort: "50003",
  clientOptions: { serializer },
});

Use compatible serializers on both ends. A custom serializer replaces the default payload encoding; it does not automatically preserve a supplied CloudEvent envelope. TypeScript type parameters do not provide runtime validation or recreate class instances. TypeScript applications can import IDaprSerializer as a type from @dapr/dapr.

Transport-specific serialization limitations are documented beside bulk publishing and bulk subscriptions.

Reliable message processing

Await SDK operations inside your application’s error boundary. Publication failure behavior differs by transport, and bulk publishing can partially fail even when its promise resolves; see publish error handling.

Make subscribers idempotent and await all processing before returning an acknowledgement. Dapr pub/sub is at-least-once, so successfully processed messages can still be redelivered. Configure retries through Dapr resiliency policies and component settings, not by assuming that the SDK retries every failure.

See server acknowledgements and retry configuration and streaming failure supervision for delivery-mode-specific behavior.