ddp.md

DDP flow

graph LR
  subgraph Cliente
    DDPConnect["DDP.connect<br/>(namespace.js)"]
    Conn["Connection<br/>livedata_connection.js"]
    StreamH["ConnectionStreamHandlers<br/>connection_stream_handlers.js"]
    MsgP["MessageProcessors<br/>message_processors.js"]
    DocP["DocumentProcessors<br/>document_processors.js"]
    Inv["MethodInvoker<br/>method_invoker.js"]
    MongoMap["MongoIDMap<br/>mongo_id_map.js"]
    HbC["Heartbeat<br/>ddp-common/heartbeat.js"]
  end

  subgraph Servidor
    DDPServer["DDPServer<br/>livedata_server.js"]
    Ses["Session<br/>livedata_server.js"]
    Sub["Subscription<br/>livedata_server.js"]
    CollView["SessionCollectionView<br/>session_collection_view.ts"]
    DocView["SessionDocumentView<br/>session_document_view.ts"]
    DummyView["DummyDocumentView<br/>dummy_document_view.ts"]
    HbS["Heartbeat<br/>ddp-common/heartbeat.js"]
  end

  %% Construção / composição
  DDPConnect --> Conn
  Conn --> StreamH
  Conn --> MsgP
  Conn --> DocP
  Conn --> HbC
  Conn --> Inv
  DocP --> MongoMap

  DDPServer --> Ses
  Ses --> HbS
  Ses --> Sub
  Sub --> CollView
  CollView --> DocView
  CollView --> DummyView

  %% Comunicação
  Conn <--> Ses
  StreamH -- roteia msgs --> MsgP
  MsgP -- dados --> DocP
  MsgP -- lifecycle método --> Inv

Loading

Client (ddp-client)

• DDP.connect Instantiates Connection and exposes it as Meteor.connection. All public methods (subscribe, call, apply, status, etc.) are proxyed there.

• Connection: Central state/manager of the DDP link on the client; manages session, heartbeat, method/sub queues, caches, reconnection, and delegates parsing/routing and data application. - Constructs _stream with ClientStream (SockJS/WebSocket) and injects message/reset/disconnect callbacks.

  • Maintains link state (session, heartbeat, method/sub queues, caches) and delegates message parsing/routing to ConnectionStreamHandlers and DDP protocol logic to MessageProcessors and DocumentProcessors.

  • Creates auxiliary instances:

• ClientStream class is the DDP transport client used in browsers. It extends

• StreamClientCommon to reuse the callback, status, and reconnection logic infrastructure, and adds browser-specific environment details. In summary, it:

  • Initializes a connection to a Meteor server by converting the URL to SockJS(when available) or WebSocket and applies the appropriate transport whitelist to avoid issues in older browsers (e.g., iOS 4/5).

  • Maintains the connection state (currentStatus, retryCount, lastError) and exposes events such as message, reset, and disconnect for the rest of the stack to react.

  • Implements the dual heartbeat flow: the legacy via SockJS (HEARTBEAT_TIMEOUT) and the modern via DDP. It schedules/cancels timers, detects timeouts, and calls _lostConnection to trigger the base class's retry policy.

  • Handles faults and cleanups (_cleanup, clearConnectionAndHeartbeatTimers), including registering window.online listeners to attempt to reconnect when the network returns. It encapsulates helpers such as _changeUrl, send, and _connected, providing the transparent transport layer for the DDP protocol.

• ConnectionStreamHandlers: receives the raw text from WebSocket/SockJS and parses it with DDPCommon.parseDDP, updates the heartbeat with messageReceived(), sets _version upon receiving connected, routes to MessageProcessors's livedata_* and, on onReset, sends a new connect, marks noRetry methods as failed, and resends subs/methods.

• MessageProcessors: Implements the protocol logic by message type _livedata_connected/_data/_result/_nosub/_error; Manages quiescence (_messagesBufferedUntilQuiescence, _methodsBlockingQuiescence, _subsBeingRevived), reconciles sessions (reset stores, revive subs, half-finished methods), and controls write flushing with bufferedWritesInterval/MaxAge. It also dispatches received document messages (mongo) to DocumentProcessors.

• DocumentProcessors: Applies data messages to stores (Minimongo), reconciling stub effects with _serverDocuments/MongoIDMap, generating added/changed/removed/replace and releasing dataVisible.

• _stores: createStoreMethods/registerStoreClient|Server sets up wrappers beginUpdate/update/endUpdate/saveOriginals/retrieveOriginals/getDoc and dumps pending updates.

• Heartbeat: - DDPCommon.Heartbeat: ping/pong timer; sends a ping if there was no traffic and triggers a timeout to reconnect/close. When connecting (MessageProcessors._livedata_connected), it creates DDPCommon.Heartbeat (packages/ddp-common/heartbeat.js#L8) with sendPing -> {msg:'ping'} and onTimeout -> _lostConnection(...).

• MethodInvoker: represents a remote call; registers with _methodInvokers, controls resending, noRetry, result callbacks, and only completes when the result + data are visible (dataVisible), releasing the method block in _outstandingMethodBlocks.

• MongoIDMap: IdMap specialized for Mongo/Minimongo IDs (stringify/parse), used to track server documents/stubs.

• DDP.onReconnect injects functions executed before resending pending methods (e.g., redo login/token).

Server (ddp-server)

• DDPServer: instantiates a Session per client and dispatches messages (method, sub, unsub, pings)

• Session: maintains the connection state (user, subs, message queue) and uses SessionCollectionView to maintain the "view" by collection for each client.

• SessionCollectionView composes several SessionDocumentView or DummyDocumentView (for strategies without merge) and generates diffs (added/changed/removed) sent as DDP messages.

• Subscription — is the “server-side” of each Meteor.subscribe that arrives from the client.

  • Stores identity (id sent by the client or undefined for universal publishes) and an internal handle used for precedence in SessionDocumentView.

  • Maintains the publication state (name, parameters, documents that have already been sent, ready/deactivated flag).

  • Exposes the API that the publish code uses (this.added/changed/removed/ready/error/stop, this.userId, this.connection, this.unblock) to send DDP messages and control the subnet lifecycle.

  • Executes the handler within a publish context (DDP._CurrentPublicationInvocation), catching errors, accepting Promises/thenables.

  • When the handler returns cursors, observes them and converts diffs into DDP messages, calling ready() at the end of the initial load.

• SessionCollectionView:

  • Maintains documents: Map<id, DocumentView> and {added,changed,removed} callbacks of the session.

  • Use DiffSequence.diffMaps to issue adds/removes/changes between two views (used in _setUserId).

  • added/changed/removed forwards changes to SessionDocumentView/DummyDocumentView and triggers session callbacks.

• SessionDocumentView:

  • Precedence structure by field (dataByKey: Map<field, [subscriptionHandle, value][]>) and existsIn (subs that see the doc).

  • getFields() returns the dominant value (first in the list) by key; changeField and clearField control precedence and collect changes to send.

• DummyDocumentView: always returns an empty object and logs changes directly (strategies that do not maintain merge).

  • Publishing strategies):

  • SERVER_MERGE: maintains mirroring by collection with full diffing.

  • NO_MERGE_NO_HISTORY: sends everything without saving history (does not send removed).

  • NO_MERGE: remembers IDs to send removed, but does not save fields.

  • NO_MERGE_MULTI: remembers sharing between multiple publications with dummy view for light precedence.

Universal Publications

• Unnamed handlers (Meteor.publish(null, ...)) are stored in server.universal_publish_handlers.

• Each Session.startUniversalSubs() creates an Subscription without an id and places it in [universalSubs]packages/ddp-server/livedata_server.js#L100); reset when userId changes; they do not receive an explicit nosub.

End-to-End Flow

1. Client-side Invocation

• Connection.call/apply/subscribe packages msg ({msg:'method'| 'sub', ...}), creates MethodInvoker if it's a method, and enqueues it in _outstandingMethodBlocks. apply respects wait to serialize blocks.

• _sendQueued/_send (same file) writes to the _stream; if the socket is not yet connected, messages remain in the internal queue of the ClientStream.

2. Server Reception

• Socket delivers text → StreamServer (not listed here) forwards to Session → _processMessage dispatches by type (method/sub/unsub/ping).

• method: creates context DDP._CurrentMethodInvocation, runs handler, captures result or exception, sends result and then updated when data is ready.

• sub: creates Subscription, runs handler (synchronous or thenable), sends ready at the end; data diffs are pushed via added/changed/removed originating from SessionCollectionView.

• Publishing strategy defines whether the session maintains mirrored collectionViews and how to calculate diffs.

3. Data transmission from server to client

• SessionCollectionView/SessionDocumentView (or Dummy) produce mutations; Session decides whether to send immediately or just register (when _isSending is false in _setUserId).

DDP messages generated: added, changed, removed, ready, updated, nosub, ping/pong.

4. Arrival at the client (stream layer)

• Raw text → ConnectionStreamHandlers.onMessage: parseDDP, heartbeat.messageReceived(), sets _version if connected, and routes.

• In reset, mount connect again, mark noRetry methods with errors, and resend subs/methods.

5. Protocol Processing

• MessageProcessors decides whether to buffer during quiescence: reviving subs (_subsBeingRevived) or wait/active methods (_methodsBlockingQuiescence). When free, drains the data buffer and, if _resetStores, calls beginUpdate/endUpdate on all stores to clear and reapply.

• result: ensures flush of writes before delivering callback; looks for the corresponding MethodInvoker in the first block; writes the result and attempts completion.

• updated: forwards to DocumentProcessors to resolve stubs and then calls dataVisible from each MethodInvoker.

6. Local Cached Application

• DocumentProcessors for added/changed/removed: if _serverDoc exists due to a stub, mutates the snapshot and/or throws a coherent error; otherwise, _pushUpdate accumulates in updates by collection. ready/updated uses _runWhenAllServerDocsAreFlushed to ensure order.

• Store (Minimongo) consumes updates via beginUpdate/update/endUpdate (calls triggered by _flushBufferedWrites).

7. Method Finalization and Reactivity

• When MethodInvoker has a result + dataVisible, it removes itself from _methodInvokers and frees _outstandingMethodFinished() to send the next block.

• Subs: ready marks subRecord.ready, triggers readyCallback and readyDeps.changed() (Tracker reactivity).

8. Heartbeats and Reconnection

• Client: Heartbeat sends ping if it hasn't seen traffic in the interval; timeout triggers _lostConnection → automatic reconnection of the ClientStream, resending of pending subs and methods (_callOnReconnectAndSendAppropriateOutstandingMethods).

• Server: Heartbeat sends ping; timeout calls Session.close() and clears subs/collectionViews.

Quick Distinctions

• _streamHandlers: transport + initial routing.

• _messageProcessors: DDP protocol rules and local consistency.

• DocumentProcessors: applies data to local collections, reconciling stubs.

• _universalSubs: unnamed subs that run automatically throughout the session.

Mermaid Diagram — chronological order of the flow (starting from DDP.connect)

sequenceDiagram
  participant DDPc as DDP.connect<br/>(namespace.js)
  participant UI as UI/Tracker
  participant Conn as Connection<br/>livedata_connection.js
  participant Inv as MethodInvoker<br/>method_invoker.js
  participant SH as StreamHandlers<br/>connection_stream_handlers.js
  participant Net as WebSocket/SockJS
  participant Ses as Session<br/>livedata_server.js
  participant Sub as Subscription<br/>livedata_server.js
  participant CV as SessionCollectionView<br/>session_collection_view.ts
  participant DV as SessionDocumentView/Dummy<br/>session_document_view.ts
  participant MP as MessageProcessors<br/>message_processors.js
  participant DP as DocumentProcessors<br/>document_processors.js
  participant Store as Store/Minimongo

  DDPc->>Conn: cria Connection (ClientStream)
  Conn->>SH: registra handlers message/reset/disconnect
  Conn->>Net: msg connect (DDP)
  Net-->>Ses: connect
  Ses-->>Net: connected (sessionId) + heartbeat iniciado
  Net-->>SH: connected
  SH->>Conn: seta version/heartbeat

  UI->>Conn: Meteor.call / subscribe
  Conn->>Inv: cria invoker (methods) / registra sub
  Conn->>SH: _sendQueued(msg)
  SH->>Net: envia DDP (method/sub)
  Net-->>Ses: texto DDP
  Ses->>Sub: cria/reencontra Subscription
  Sub->>CV: add/changed/removed
  CV->>DV: aplica precedência por sub
  Ses-->>Net: DDP added/changed/removed/ready/updated/result
  Net-->>SH: entrega raw msg
  SH->>Conn: parseDDP + heartbeat.messageReceived()
  SH->>MP: route por tipo
  MP->>DP: dados (added/changed/removed/ready/updated)
  DP->>Store: beginUpdate/update/endUpdate
  MP->>Inv: result / updated → dataVisible
  Inv-->>UI: callback/Promise resolvida

Loading