Asynchronous Processing

When handling a callback message, the work may involve asynchronous operations such as API calls, timers, or complex observable pipelines. The handler must call msg.finish(result) after the async work completes.

Registration pattern (reminder)

Before any callback can be used, the message type must be connected to the bus.

import {AppPostboyService} from '@shared/services/app-postboy.service';
import {ConnectMessage} from '@artstesh/postboy';
import {Subject} from 'rxjs';
import {GetUserQuery} from './messages/queries/get-user.query';

export class UserService {
  constructor(private postboy: AppPostboyService) {
// Register the callback message type
    this.postboy.exec(new ConnectMessage(GetUserQuery, new Subject<GetUserQuery>()));

// Subscribe and handle
    this.postboy.sub(GetUserQuery).subscribe((msg) => {
      const user = this.cache.get(msg.userId);
      msg.finish(user ?? null);
    });
  }

  private cache = new Map<string, User>();
}

Promise-based handler (async / await)

Use async functions for Promise-based work. Call msg.finish(...) when the promise resolves.

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

Observable-based handler (HttpClient / RxJS)

When using Angular’s HttpClient or any Observable, subscribe inside the handler and finish when the value arrives.

import {HttpClient} from '@angular/common/http';

export class UserService {
  constructor(private http: HttpClient, private postboy: AppPostboyService) {
    this.postboy.exec(new ConnectMessage(GetUserQuery, new Subject<GetUserQuery>()));

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

Cancellation-aware handler

If the callback message includes a cancellation signal, use takeUntil to stop the request when the sender cancels.

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

Key points

The handler must always call msg.finish(result) to complete the callback flow.
For HTTP or long-running observables, ensure proper cleanup with takeUntil or manual unsubscription.
The sender simply fires the callback and subscribes to the result — no knowledge of sync or async implementation.

24 апреля 2026