rxjs объединяет данные http с обновлением websocket

Question

У меня есть пара массивов данных, полученных из вызова API

    protected ngUnsubscribe: Subject<void> = new Subject<void>();

    const users$ = this.userApi
        .find<User>()
        .takeUntil(this.ngUnsubscribe)
        .share();

    this.active$ = users$.map(users =>
        users.filter(user => user.status === 'active')
    );

    this.archived$ = users$.map(users =>
        users.filter(user => user.status === 'inactive')
    );

, затем другой пользователь в другом сеансе обновляет запись пользователя, изменяя ее с активной на неактивную.

Сервер уведомляет мой сеанс об этом обновленном пользователе с помощью веб-сокетов.

Теперь я застрял ... как я могу изменить наблюдаемые $active и $archive, чтобы отразить этоизменение данных?

Answer 1

, поэтому я заставил это работать, используя следующее:

const data = await this.userApi
        .find<User>()
        .takeUntil(this.ngUnsubscribe)
        .share()
        .toPromise();

    const data$ = new BehaviorSubject<User[]>(data);

    this.active$ = data$.map(users => {
        return users.filter(user => user.status === 'active');
    });

    this.archived$ = data$.map(users => {
        return users.filter(user => user.status !== 'active');
    });

Теперь, когда я получаю обновление / удаление / создание из веб-сокета, я соответствующим образом манипулирую массивом данных и выполняю

data$.next(data)

для получения обновленных данных