使用 Angular 11 和 RxJS 的类型上不存在 combineLatest

Question

晚上好， 我正在尝试使用 combineLatest 运算符在 Angular 中组合两个流，但出现错误“combineLatest 在类型上不存在”。 我试图将代码放入 .pipe() 中，但这不起作用。 我需要管道地图然后结合最新吗？ 感谢您的帮助！

import { Injectable } from '@angular/core';
import {Subject, BehaviorSubject, Observable, combineLatest} from 'rxjs';
import { filter, map, scan } from 'rxjs/operators';

// Model and Service imports
import {Thread} from './thread.model';
import {Message} from '../message/message.model';
import {MessagesService} from '../message/messages.service';

import * as _ from 'lodash';



@Injectable({
  providedIn: 'root'
})
export class ThreadsService {
threads: Observable<{[key:string]: Thread}>; // (a)
orderedThreads: Observable<Thread[]>; // (a1)
currentThread: Subject<Thread> = new BehaviorSubject<Thread>(new Thread()); // (a2)
currentThreadMessages: Observable<Message[]>; // (a3)

  constructor(public messagesService: MessagesService) { 
    
    this.threads = messagesService.messages // (b)
      .pipe( map((messages: Message[]) => {
        const threads: { [key: string]: Thread } = {}; //(b1)

        messages.map((message: Message) => { // (b2)
          this.threads[message.thread.id] =
            this.threads[message.thread.id] ||
            message.thread;

          const messagesThread: Thread = // (c)
            threads[message.thread.id];
          if (!messagesThread.lastMessage || messagesThread.lastMessage.sentAt < message.sentAt) {
            messagesThread.lastMessage = message;
          }
        });
        return threads;
      }))
      
    this.orderedThreads = this.threads // (d)
      .pipe(map((threadGroups: { [key: string]: Thread }) => {
        const threads: Thread[] = _.values(threadGroups);
        return _.sortBy(threads, (t: Thread) => t.lastMessage.sentAt).reverse();
      }));  

      this.currentThread.subscribe(this.messagesService.markThreadAsRead); // (e1)

    this.currentThreadMessages = combineLatest([this.currentThread, messagesService.messages]).pipe(
      map(currentThread: Thread, messages: Message[]) => {
        if(currentThread && messages.length > 0) {
      return _.chain(messages)
        .filter((message: Message) => // (f1)
          (message.thread.id === currentThread.id))
        .map((message: Message) => {
          message.isRead = true;
          return message;
        })
        .value();
    } else {
      return [];
    }
  }
);
    }

  setCurrentThread(newThread: Thread): void { // (e)
    this.currentThread.next(newThread);
  }
}

Answer 1

问题是您在Observable 上调用.combineLatest。使用 combine latest 的语法是

newObservable$ = combineLatest([observable1$, observable2$, ...]).pipe(...)

在你的情况下，你可以拥有

 this.currentThreadMessages = combineLatest([this.currentThread, messagesService.messages]).pipe(
  map(currentThread: Thread, messages: Message[]) => {
     if (currentThread && messages.length > 0) {
          return _.chain(messages)
              .filter((message: Message) => // (f1)
                (message.thread.id === currentThread.id))
              .map((message: Message) => {
                message.isRead = true;
                return message;
              })
              .value();
        } else {
            return [];
        }
  }
)