Real-time streams (SSE)

Three mechanisms deliver "something changed" without a page reload. They are unrelated to each other and to domain events.

1. Server-sent events

Why a polyfill

Native EventSource cannot set request headers, and every Payability backend requires X-Session-Token. SseService therefore uses EventSourcePolyfill from ng-event-source:

@Injectable()
export class SseService {
  private token: string;

  constructor(private zone: NgZone, private store: Store<AppState>) {
    this.store.select('auth').subscribe((authState) => {
      if (authState && authState.user) { this.token = authState.user.jwtToken; }
    });
  }

  getStream(sseEndpointUrl: string): Observable<any> {
    return new Observable<any>((observer) => {
      const eventSource = new EventSourcePolyfill(sseEndpointUrl, {
        headers: { 'X-Session-Token': this.token },
      });

      eventSource.onerror   = (error) => observer.error(error);
      eventSource.onmessage = (event) => {
        try {
          const data = JSON.parse(event.data);
          this.zone.run(() => observer.next(data));     // ← back into Angular's zone
        } catch (error) { Logger.log(error); observer.error(error); }
      };

      return () => { Logger.log('SseService - eventSource.close()'); eventSource.close(); };
    });
  }
}

Points worth knowing:

  • NgZone.run is required — polyfilled EventSource callbacks arrive outside Angular’s zone, so without it change detection never runs and the UI appears frozen.

  • The token is captured from a long-lived store subscription, so a stream opened before login carries undefined.

  • Any onerror terminates the observable. There is no reconnect logic; the browser’s own EventSource retry is bypassed because the error is propagated as an observable error.

  • A JSON parse failure also terminates the stream, not just the message.

  • Teardown closes the connection, so an unsubscribed stream is cleaned up.

A single malformed message or a transient network blip kills the stream permanently. The UI will simply stop updating with no visible error unless the caller handles it. If you depend on a stream staying alive, add your own retry (retryWhen) at the call site.

The one consumer: B2B task monitoring

B2bWorkflowService (suppliers/services/workflow/) wraps the B2B task manager (workflowApiUrl, correlated by workflowGciId):

taskMonitoring(): Observable<any> {
  const endpointURL = `${this.apiUrl}tasksMonitoring`;
  return !isUndefined(this.streamShared$) ? this.streamShared$ : this.initStream(endpointURL);
}

The stream is share()`d so that many subscribers use one connection, and each message is also pushed onto an internal `Subject (handler).

subscribeTask(taskName, callback, searchForInclusion = false) filters that subject by task name — exactly, or by regex search when searchForInclusion is true.

subscribeTask subscribes to taskMonitoring() and then, inside that subscription’s next handler, subscribes to this.handler. Every stream message therefore creates a new inner subscription to handler, and none of them are torn down. Callbacks accumulate and fire repeatedly as the number of received messages grows.

pendingSubs is incremented but never read; two finalize(() ⇒ {}) calls are no-ops; streamShared$ is never reset. Treat subscribeTask as known-broken for long-lived monitoring and prefer subscribing to taskMonitoring() directly with your own filter.

Other methods on the same service:

Method Detail

getLastExecution(name?)

GET getLastExecution?gci=<gci>[&name=…] — the last run of a named job.

getListOfActivities(id)

GET getListOfActivities?id=<encoded>&gci=<gci> — the steps of one run.

executeWorkflow(data)

POST executeWorkflow?gci=<gci> with Content-Type: application/x-www-form-urlencoded, hand-encoding the body with encodeURIComponent. The only form-urlencoded request in the application.

downloadFile(name, url) / downloadPdfFile(name, url)

Accept either a Blob or a URL; fetch with responseType: 'blob' and save via file-saver as .txt or application/pdf.

Who uses it

Feature Use

Load Earnings / Load Remittance

The supplier Actions menu operations return immediately; progress arrives on the stream. See lifecycle actions.

Bulk imports

BatchImportService.getLastExecution / getListOfActivities delegate to this service for job status. See bulk actions.

2. The in-app event bus

src/app/shared/table-events.service.ts is a small Subject-based publish/subscribe bus, used where a store action would be heavier than the problem warrants:

export const ADD_TRANSACTION_EVENT = 'ADD_TRANSACTION_EVENT';
export const CRM_OPEN_ALL_TAB      = 'CRM_OPEN_ALL_TAB';
export const SIDENAV_CLOSED_EVENT  = 'SIDENAV_CLOSED_EVENT';
export const TASK_COUNTERS_UPDATE  = 'TASK_COUNTERS_UPDATE';

@Injectable()
export class TableEventsService {
  private listeners = {};
  private events = new Subject();

  constructor() {
    this.events.subscribe(({ name, args }) => {
      if (this.listeners[name]) { for (const l of this.listeners[name]) { l(...args); } }
    });
  }

  on(name, listener)            { … }   // register
  broadcast(name, ...args)      { … }   // emit
}
Event Broadcast by → consumed by

TASK_COUNTERS_UPDATE

CrmEffects after a task mutation → TaskCountBadgeComponent reloads the sidebar counters.

CRM_OPEN_ALL_TAB

CrmEffects after adding a CRM item → the CRM panel switches to the "All" tab.

ADD_TRANSACTION_EVENT

Transaction dialogs → transaction grids refresh.

SIDENAV_CLOSED_EVENT

Sidebar → layout listeners.

on() has no unsubscribe. Listeners registered by a component are never removed, so a component that registers in ngOnInit leaks a callback on every instantiation. Register once (in a service or a singleton component), or prefer the store.

Also note the constructor subscribes to events and never unsubscribes, and errors thrown by a listener will propagate through the shared Subject and kill it for all listeners.

Other Subject-based buses exist for narrower purposes:

Service Purpose

CancelEmitService

Supplier cancellation / win-back change notifications. See lifecycle actions.

CardFormgroupMessageService

Save/error feedback for the editable-card pattern.

CommunicationService

The iframe postMessage channel — see iframe messaging.

PendingInterceptorService

In-flight request count for the global spinner.

3. Polling

Where neither SSE nor the bus applies, a few features poll.

Feature Mechanism

Task counters

TaskCountBadgeComponent reloads every RELOAD_COUNT_INTERVAL = 60000 ms, and listens to document’s `visibilitychange so it only refreshes while the tab is visible. It also reloads on TASK_COUNTERS_UPDATE. Three triggers for one counter.

Batch status

PaymentBatchRepositoryService.getStatus(batchId) is called on demand by the batch screens rather than streamed. See processes.

Version check

VersionService compares front-end and back-end versions (using semver) for the sidebar footer and the login screen. See versions.

Choosing a mechanism

Need Use

Progress of a long backend job

B2bWorkflowService.taskMonitoring() — subscribe directly and filter yourself.

"Refresh yourself" between sibling components

TableEventsService, registering the listener once.

Shared state more than one screen reads

The NgRx store. See state management.

Cross-application (EVA) notification

The postMessage protocol. See iframe messaging.

Do not add a new polling interval without checking whether the SSE stream already carries the information.

Gotchas summary

Issue Detail

SSE has no reconnect

Any error or parse failure ends the stream silently.

NgZone.run is mandatory

Polyfilled callbacks are outside Angular’s zone.

subscribeTask leaks

Nested subscription per received message.

TableEventsService.on cannot unsubscribe

Listeners accumulate.

A throwing listener kills the bus

One shared Subject, no per-listener isolation.

Token captured by subscription

A stream opened pre-login sends undefined.