Callback Patterns

Practical flows built on callback messages: asynchronous responders, error translation, cancellation, and composing responses from several queries with standard RxJS operators. The anatomy of PostboyCallbackMessage<T> and the fireCallback verb are covered in Callbacks.

Asynchronous processing

The responder may take as long as it needs; the sender only sees the Observable<T>.

Promise-based responder:

postboy.sub(LoadUserQuery).subscribe(async (msg) => {
  const user = await api.loadUser(msg.userId);
  msg.finish(user ?? null);
});

Observable-based responder:

postboy.sub(LoadUserQuery).subscribe((msg) => {
  http.get<User>(`/api/users/${msg.userId}`).subscribe({
    next: (user) => msg.finish(user),
    error: () => msg.finish(null),
  });
});

Error handling

Postboy is not an error boundary: an exception thrown in a responder never reaches the sender. Report failures through the result contract — a nullable value, a fallback such as msg.finish([]), or a discriminated result when the sender must tell failures apart. Never throw from a responder and expect the bus to relay it:

type SettingsResult =
  | { status: 'success'; data: Settings }
  | { status: 'error'; code: string };

export class LoadSettingsQuery extends PostboyCallbackMessage<SettingsResult> {
  static readonly ID = 'app.settings.load';
}

// wrong: the throw is lost inside the subscription; the sender still waits
postboy.sub(LoadUserQuery).subscribe((msg) => {
  throw new Error('boom');
});

Cancellation

Unsubscribing the sender's Observable before completion stops the wait and tears the stream down:

const subscription = postboy.fireCallback(new SearchUsersQuery(term, cancel$))
  .subscribe((users) => display(users));

To stop the responder's work as well, carry a cancel signal in the message payload — an application-level convention, not a built-in field — and honor it with takeUntil:

export class SearchUsersQuery extends PostboyCallbackMessage<User[]> {
  static readonly ID = 'search.users';

  constructor(public readonly term: string, public readonly cancel$: Observable<void>) {
    super();
  }
}

postboy.sub(SearchUsersQuery).subscribe((msg) => {
  http.get<User[]>(`/api/users?q=${msg.term}`)
    .pipe(takeUntil(msg.cancel$))
    .subscribe((users) => msg.finish(users));
});

cancel$.next();              // tears the responder's request down
subscription.unsubscribe(); // and the sender's wait

Composing responses

Every fireCallback returns a typed Observable<T>, so queries compose with ordinary operators:

export class GetCustomerQuery extends PostboyCallbackMessage<Customer> {
  static readonly ID = 'crm.customer.get';
  constructor(public readonly id: string) { super(); }
}
export class GetOrdersQuery extends PostboyCallbackMessage<Order[]> {
  static readonly ID = 'crm.orders.get';
  constructor(public readonly customerId: string) { super(); }
}
export class GetUserProfileQuery extends PostboyCallbackMessage<Profile> {
  static readonly ID = 'app.user.profile';
  constructor(public readonly id: string) { super(); }
}
export class GetUserSettingsQuery extends PostboyCallbackMessage<Settings> {
  static readonly ID = 'app.user.settings';
  constructor(public readonly id: string) { super(); }
}

Sequential — the second query needs the first result:

postboy.fireCallback(new GetCustomerQuery(customerId)).pipe(
  mergeMap((c) => c ? postboy.fireCallback(new GetOrdersQuery(c.id)) : of([] as Order[])),
).subscribe((orders) => render(orders));

Parallel — independent queries gathered with forkJoin:

forkJoin({
  profile: postboy.fireCallback(new GetUserProfileQuery(userId)),
  settings: postboy.fireCallback(new GetUserSettingsQuery(userId)),
}).subscribe(({profile, settings}) => buildDashboard(profile, settings));

Pitfalls

A responder that never finishes hangs the whole composition — forkJoin waits forever.
Expecting the sender's stream to error; it only receives what the responder passes to next and finish.
Unsubscribing and assuming the responder's HTTP call stopped — only a cooperative cancel$ does that.
Firing queries whose classes are not registered yet.

Next steps

07 September 2026