How to make a declarative/reactive feed in Angular with Signals and RxJS? (with reusable <feed_type>)

Question

I'm working on an Angular application that uses Signals (part of the new Angular reactive libraries) and RxJS. I have a NotificationService that needs to set up a feed from a third-party service (in this case, stream_client). The feed is specific to the current user and needs to be updated whenever the user changes. Additionally, the <feed_type> should be reusable and accept different string values, such as 'notification', 'general', or 'all'.

Here's the relevant code:

  @Injectable({
      providedIn: 'root',
    })
    export class NotificationService {
      private user_service = inject(UserService);
      private stream_service = inject(StreamService);
    
      current_user$ = this.user_service.current_user$;
      stream_client = this.stream_service.stream_client;
    
      feed = toSignal(
        this.current_user$.pipe(
          map((user) => {
            const stream_client = this.stream_client();
            if (user && stream_client) {
              return stream_client.feed(<feed_type>, user.uid);
            }
            return null;
          }),
        ),
      );
    }

The feed is set up as a Signal that depends on the current_user$ Observable. Whenever the user changes, the map operator creates a new feed based on the user's ID (user.uid), the stream_client, and the <feed_type> string.

While this approach works, it feels a bit imperative and tightly coupled to the specific implementation details of stream_client and UserService.

How can I refactor this code to be more declarative and reactive, following the principles of the new Angular reactive libraries? I'd like the feed to update automatically whenever the user, the stream_client, or the <feed_type> changes, without explicitly mapping the values in the service. The <feed_type> should be reusable and accept different string values as needed.

Any suggestions or alternative approaches would be greatly appreciated!

Answer 1

Imperative code modifies things directly, your code isn't modifying anything, it just provides a stream of values, so it is purely declarative.

I would like to note, that there are no 100% declarative programms - some parts of your code will do imperative things. We should minimize the amount of imperative code.

The code example you provided only reacts to changes of current_user. You are reading a signal not in a reactive context, so it will not be tracked and your code will not be notified when that signal is updated. Right now (v18) Angular has 2 functions with reactive context: computed() and effect(). Templates are also reactive contexts.

With tracking 3 sources of changes, your code could look like this:

export class NotificationService {
  private user_service = inject(UserService);
  private stream_service = inject(StreamService);

  userId$ = this.user_service.current_user$.pipe(map((user) => user?.uid));
  stream_client$ = toObservable(this.stream_service.stream_client);
  feedType$ = this.stream_service.feedType$;

  feed = toSignal(
    combineLatest({
      userId: this.userId$,
      stream_client: this.stream_client$,
      feed_type: this.feedType$,
    }).pipe(
      switchMap(({ userId, stream_client, feed_type }) => {
        if (!userId || !stream_client || !feed_type) {
          return EMPTY;
        }
        return stream_client.feed(feed_type, userId);
      }),
    ),
  );
}

Here I'm making an assumption that stream_client.feed() returns an observable. If it returns a value, replace switchMap() with map() and EMPTY with null or undefined.

Be aware that combineLatest() will only start emitting values when every observable has emitted at least one value.

If stream_client.feed() returns a value, not an observable, you could track changes using computed():

export class NotificationService {
  private user_service = inject(UserService);
  private stream_service = inject(StreamService);

  userId = toSignal(
    this.user_service.current_user$.pipe(
      map((user) => user?.uid)
    ), 
    {initialValue: undefined}
  );
  stream_client = this.stream_service.stream_client;
  feedType = this.stream_service.feedType;

  feed = computed(() => {
    const userId = this.userId();
    const streamClient = this.stream_client();
    const feedType = this.feedType();
    if (!userId || !streamClient || !feedType) {
      return undefined;
    }
    return streamClient.feed(feedType, userId);
  });
}

In this case, computed() will also warn you if you try to do any signal modifications inside (even inside streamClient.feed()).