Streaming API And Pushtopic

Streaming API and PushTopic

This skill activates when an external system needs to receive Salesforce record change notifications or arbitrary event payloads in real time, using the Streaming API over a CometD (Bayeux) connection. It covers PushTopic creation and SOQL constraints, Generic Streaming channels, replay ID handling for durable subscriptions, client configuration, and troubleshooting dropped connections or missed events.

Before Starting

Gather this context before working on anything in this domain:

• Salesforce API version in use: PushTopic requires API v24.0+. Generic Streaming channel replay requires v36.0+. The CometD client version must match or exceed the API version declared in the /cometd/<version> endpoint URL.
• Authentication method: The external client must authenticate first via OAuth (or username-password flow in sandboxes) and obtain a session token before connecting to CometD. Named Credentials cannot be used directly for inbound CometD; the client owns the session.
• Concurrent client limits: Each Streaming API channel supports a maximum of 1,000 concurrent clients per org (across all channels), and a maximum of 100 simultaneous clients per individual channel. Exceeding this causes 402::Unknown client or handshake rejection.
• PushTopic SOQL restrictions: Not all SOQL is valid. Aggregate functions (COUNT, SUM, etc.), relationship traversal in the SELECT clause (e.g., Account.Name from Opportunity), LIMIT, OFFSET, and semi-joins are all prohibited. Only fields on the root object can appear in SELECT.
• Event retention: Streaming API retains events for 24 hours. If a subscriber is offline longer than that, events are permanently lost. For guaranteed delivery, consider Platform Events with a 72-hour window instead.

Core Concepts

PushTopic

A PushTopic is a Salesforce record (PushTopic sObject) that defines a SOQL query against a standard or custom object. Salesforce evaluates each DML operation against the query and fires a streaming notification if the record matches. The notification payload includes changed field values defined in the SELECT clause.

Key fields on the PushTopic record:

NotifyForFields = Referenced is the safe default: it fires when any field used in the SOQL SELECT or WHERE changes. All fires on any field change, increasing noise. Select fires only when SELECT-listed fields change.

Generic Streaming API

Generic Streaming is used when the event payload is arbitrary (not tied to a Salesforce object). A StreamingChannel record is created with a channel name in the form /u/<ChannelName>. External or Apex code publishes payloads using the PushTopic REST endpoint or via Streaming.PushTopic (deprecated). The recommended publisher is the Streaming Channel Push REST endpoint:

POST /services/data/v60.0/sobjects/StreamingChannel/<channelId>/push
Body: { "pushEvents": [{ "payload": "my payload", "userIds": [] }] }

An empty userIds array broadcasts to all subscribers. A non-empty array restricts delivery to those user IDs.

CometD and the Handshake-Subscribe Flow

The Salesforce Streaming API implements the Bayeux protocol via CometD. The client must complete this sequence:

1. POST to /cometd/<version> with handshake message — obtains clientId.
2. POST with connect message — establishes the long-poll connection.
3. POST with subscribe message to /topic/<PushTopicName> or /u/<ChannelName>.
4. Re-issue connect immediately after each server response (long-poll loop).

The Salesforce-provided EMP Connector (Java, open source on GitHub) wraps this lifecycle and handles reconnection, replay, and subscription automatically. For JavaScript, the official cometd npm package or the Lightning empApi LWC wire adapter are the recommended clients.

Replay ID and Durable Subscriptions

Every streaming event carries a monotonically increasing replayId. A client can pass a replayId value in the subscription ext object to replay events stored within the retention window:

{
  "channel": "/topic/MyTopic",
  "ext": { "replay": { "/topic/MyTopic": -1 } }
}

Replay values:

Clients must persist the last-seen replayId to durable storage before processing each event. On reconnect they pass the stored ID. Storing the ID after processing risks duplicate processing if the process crashes between the two steps — weigh idempotency requirements.

Common Patterns

Mode 1: Creating a PushTopic and Subscribing via CometD

When to use: An external service needs to be notified whenever a Salesforce Opportunity is updated (e.g., Stage or Amount changes).

How it works:

1. Create the PushTopic via Apex or REST (one-time setup):

PushTopic pt = new PushTopic();
pt.Name = 'OpportunityUpdates';
pt.Query = 'SELECT Id, Name, StageName, Amount FROM Opportunity WHERE StageName != 'Closed Lost'';
pt.ApiVersion = 60.0;
pt.NotifyForOperationCreate = true;
pt.NotifyForOperationUpdate = true;
pt.NotifyForOperationDelete = false;
pt.NotifyForOperationUndelete = false;
pt.NotifyForFields = 'Referenced';
insert pt;

2. External client authenticates via OAuth 2.0 and obtains access_token and instance_url.
3. Client connects to <instance_url>/cometd/60.0 and performs handshake.
4. Client subscribes to /topic/OpportunityUpdates with a replay extension set to the last stored replayId (or -2 on first run).
5. Client processes each incoming message and persists the replayId before acknowledging.

Why not polling: REST polling on Opportunity every N seconds burns API request quota and has latency proportional to the polling interval. Streaming delivers sub-second notifications with a persistent connection.

Mode 2: Generic Streaming for Custom Payloads

When to use: A middleware system needs to push arbitrary JSON payloads to subscribed Salesforce users or external listeners — for example, triggering a UI refresh on a custom Lightning page.

How it works:

1. Create a StreamingChannel record:

StreamingChannel sc = new StreamingChannel();
sc.Name = '/u/WorkflowUpdates';
insert sc;

2. Grant permission: Users must have the "Streaming API" permission (included in most standard profiles).
3. Publisher pushes events via the Streaming Channel Push REST endpoint:

POST /services/data/v60.0/sobjects/StreamingChannel/<id>/push
Content-Type: application/json
{ "pushEvents": [{ "payload": "{\"status\":\"ready\"}", "userIds": [] }] }

4. Subscribers follow the same CometD handshake-subscribe flow, subscribing to /u/WorkflowUpdates.

Mode 3: Troubleshooting Connection Drops and Missed Events

When to use: The external client was offline or disconnected and suspects events were missed.

How it works:

1. Check the last persisted replayId in client-side storage.
2. On reconnect, set the replay extension to that replayId. The server replays all events with a higher ID within the 24-hour retention window.
3. If no replayId was persisted, set replay to -1 to receive all retained events (risk: replay storm if the backlog is large).
4. Verify the client is re-issuing connect immediately after every server response. A gap in the long-poll loop causes the server to expire the client session (402::Unknown client), requiring a full re-handshake.
5. Check for rate limit errors: 401::Authentication required after a session token expires. The client must re-authenticate and reconnect, not just re-subscribe.

Decision Guidance

Recommended Workflow

Step-by-step instructions for an AI agent or practitioner activating this skill:

1. Gather context — confirm the org edition, relevant objects, and current configuration state
2. Review official sources — check the references in this skill's well-architected.md before making changes
3. Implement or advise — apply the patterns from Core Concepts and Common Patterns sections above
4. Validate — run the skill's checker script and verify against the Review Checklist below
5. Document — record any deviations from standard patterns and update the template if needed

Review Checklist

Run through these before marking work in this area complete:

• PushTopic SOQL does not use aggregate functions, LIMIT, OFFSET, relationship fields in SELECT, or semi-joins.
• ApiVersion on the PushTopic is 24.0 or higher; CometD URL version matches.
• NotifyForFields is set deliberately (Referenced for most cases; All only if justified).
• CometD client persists replayId to durable storage before processing each event.
• Client re-issues connect immediately after each server response (long-poll loop is unbroken).
• Concurrent subscriber count per channel has been estimated and is below 100 per channel / 1,000 org-wide.
• Authentication token refresh is handled — CometD sessions expire when the session token expires.
• For Generic Streaming, the StreamingChannel record exists and users have the Streaming API permission.
• Event retention requirement has been confirmed to be within 24 hours; if not, Platform Events is the right choice.

Salesforce-Specific Gotchas

Non-obvious platform behaviors that cause real production problems:

1. PushTopic SOQL bans aggregate functions and relationship traversal — SELECT COUNT(Id), Account.Name FROM Opportunity is invalid for two separate reasons: COUNT() is an aggregate and Account.Name is a parent relationship field. Only root-object fields in the SELECT list are allowed. The insert of the PushTopic will fail with a SOQL validation error if violations exist.
2. 24-hour event retention — not 72 — Unlike Platform Events, Streaming API retains events for only 24 hours. If a subscriber is offline for more than 24 hours, those events are gone. This is a frequent surprise for teams that assume Streaming API and Platform Events behave identically.
3. CometD client session expiry after Salesforce session timeout — The CometD connection is bound to the OAuth session. When the session token expires (default 2 hours for connected apps without refreshTokenIssuanceTime), the server returns 401::Authentication required. The client must re-authenticate (get a fresh token) and then perform a full CometD handshake before re-subscribing. Simply re-subscribing on the old clientId will fail.

Output Artifacts

Related Skills

• platform-events-apex — Use when you need Apex-triggered event publishing, higher throughput, or 72-hour retention instead of 24-hour Streaming API retention.
• platform-events-integration — Use when an external system publishes or subscribes to Platform Events via CometD or Pub/Sub API with guaranteed delivery semantics.
• named-credentials-setup — Use when the external CometD client needs to make additional Salesforce REST callouts (separate from the streaming connection itself) and needs secure credential storage.
• oauth-flows-and-connected-apps — Use when configuring the OAuth Connected App that issues the access token the CometD client uses for authentication.