将 2 个 RxJs 可观察对象合并为映射的可观察对象

Question

感谢任何帮助 - 我对 RxJs 有点陌生，但我认为我对 Observables 的映射方式存在一些重大误解。

我通过 NGRX 为一个帐户存储了两个 observable，分别是“receivedMessages”和“sentMessages”。每个只是一个包含 toUsername 和 fromUsername 的“MessageModel”数组。

我正在尝试将这两个可观察对象组合成 1 个可观察对象，我可以显示“MessageThreads”以及这两个用户之间的所有对应关系，方法是拥有一个带有“用户名：作为唯一键的对象的数组：就像 HashMap 一样，这看起来像：{username: string, thread: MessageModel[]}

我尝试过使用 forkJoins、Concats、Merge，但遇到了一个奇怪的错误。让我觉得我可能不明白ngrx或rxjs实际上在做什么:(

收件箱容器.component.ts

 import { Component, OnInit } from '@angular/core';
import { AppState } from 'src/app/store/app.reducers';
import { Store } from '@ngrx/store';
import { MessageModel } from '../../shared/models/message.model';
import { Observable, forkJoin, of, from, pipe } from 'rxjs';
import { map, concat, combineLatest, mergeMap, flatMap, switchMap, mapTo, merge } from 'rxjs/operators';
import { stringify } from '@angular/compiler/src/util';

@Component({
  selector: 'app-inbox-container',
  templateUrl: './inbox-container.component.html',
  styleUrls: ['./inbox-container.component.css']
})
export class InboxContainerComponent implements OnInit {

  receivedMessages: Observable<MessageModel[]>;
  sentMessages: Observable<MessageModel[]>;

  uniqueThreads: Observable<{ username: string, thread: MessageModel[] }[]>;

  constructor(private store: Store<AppState>) {
    this.receivedMessages = this.store.select('inbox', 'receivedMessages');
    this.sentMessages = this.store.select('inbox', 'sentMessages');
  }

  ngOnInit() {
    this.uniqueThreads =
      of(
        forkJoin(
          this.receivedMessages,
          this.sentMessages
        ),
        map((messages: MessageModel[]): Observable<{ username: string, thread: MessageModel[] }[]> => {
          let mergedThreads: { username: string, thread: MessageModel[] }[] = [];
          messages.forEach((message) => {
            let activeThreadFrom = mergedThreads.find(any => any.username === message.fromUsername)
            if (activeThreadFrom === null) {
              mergedThreads.push({ username: message.fromUsername, thread: [message] })
            } else {
              activeThreadFrom.thread.push(message);
            }
            mergedThreads.push(activeThreadFrom);

            let activeThreadTo = mergedThreads.find(any => any.username === message.toUsername)
            if (activeThreadTo === null) {
              mergedThreads.push({ username: message.toUsername, thread: [message] })
            } else {
              activeThreadTo.thread.push(message);
            }
            mergedThreads.push(activeThreadTo)

          })
          return of(
            mergedThreads
          )
        }
        )
      )
  }

}

收件箱容器.component.html

<div class="col-xs-12">
  <div class="col-xs-4">
    <div class="list-group">
      <a class="list-group-item" routerLinkActive="active" routerLink="compose">
        <h4 class="list-group-item-heading">
          <div style="margin-top: 10px; margin-right: 10px; padding-right:10px;" class="dropdown" appSearchBarDirective
            [text]=searchText.value>
            <input class="form-control" type="text" placeholder="Username.." aria-label="Search" #searchText>
            <div class="dropdown-menu">
              <app-reactive-accounts-search [searchForm]=searchText.value></app-reactive-accounts-search>
            </div>
          </div>
        </h4>
      </a>
      <div class="list-group">
        <div class="list-group-item" *ngFor="let thread of ( uniqueThreads | async)" appHighlightDirective
          style="cursor: pointer;">
          <div class="list-group-item-heading">
            <h4>{{ thread.username }}
              <div class="badge pull-right">4</div>
            </h4>
          </div>
        </div>
      </div>
    </div>
  </div>
  <div class="col-xs-8">
    <div class="panel panel-default">
      <router-outlet></router-outlet>
    </div>
  </div>
  <button (click) = "wtf()">asdasd</button>
</div>

我在控制台日志中收到错误：

InboxContainerComponent.html:10 ERROR Error: Cannot find a differ supporting object 'function mapOperation(source) {
        if (typeof project !== 'function') {
            throw new TypeError('argument is not a function. Are you looking for `mapTo()`?');
        }
        return source.lift(new MapOperator(project, thisArg));
    }' of type 'mapOperation'. NgFor only supports binding to Iterables such as Arrays.
    at NgForOf.push../node_modules/@angular/common/fesm5/common.js.NgForOf.ngDoCheck (common.js:3184)
    at checkAndUpdateDirectiveInline (core.js:22101)
    at checkAndUpdateNodeInline (core.js:23362)
    at checkAndUpdateNode (core.js:23324)
    at debugCheckAndUpdateNode (core.js:23958)
    at debugCheckDirectivesFn (core.js:23918)
    at Object.eval [as updateDirectives] (InboxContainerComponent.html:16)
    at Object.debugUpdateDirectives [as updateDirectives] (core.js:23910)
    at checkAndUpdateView (core.js:23306)
    at callViewActi

还有 IDE 中的警告：

Type 'Observable<Observable<[MessageModel[], MessageModel[]]> | OperatorFunction<MessageModel[], Observable<{ username: string; thread: MessageModel[]; }[]>>>' is not assignable to type 'Observable<{ username: string; thread: MessageModel[]; }[]>'.
  Type 'Observable<[MessageModel[], MessageModel[]]> | OperatorFunction<MessageModel[], Observable<{ username: string; thread: MessageModel[]; }[]>>' is not assignable to type '{ username: string; thread: MessageModel[]; }[]'.
    Type 'Observable<[MessageModel[], MessageModel[]]>' is missing the following properties from type '{ username: string; thread: MessageModel[]; }[]': l

任何帮助将不胜感激！

-- 解决方案 ---

我只是绑定到 NGRX 存储状态，正如答案所说，不需要创建额外的 observables。

 init() {
    this.uniqueThreads = this.store.select('inbox').pipe(
      map((state: fromInbox.State) => {
        return [
          state.receivedMessages,
          state.sentMessages
        ]
      }),
      ).pipe(
        map(([receivedMessages, sentMessages]) => {
          let mergedThreads: {username: string, thread: { message: MessageModel, received: boolean }[]}[] = [];
          const messages: MessageModel[] = [...receivedMessages, ...sentMessages];
          messages.forEach((message) => {
            console.log(message)
            let activeThreadFrom = mergedThreads.find(any => any.username === message.fromUsername)
            if (activeThreadFrom === undefined) {
              mergedThreads.push({ username: message.fromUsername, thread: [{message: message, received: true}] })
            } else {
              activeThreadFrom.thread = [ ...activeThreadFrom.thread, {message: message, received: true}];
            };

            let activeThreadTo = mergedThreads.find(any => any.username === message.toUsername)
            if (activeThreadTo === undefined) {
              mergedThreads.push({ username: message.toUsername, thread: [{message: message, received: false}] })
            } else {
              activeThreadTo.thread = [ ...activeThreadTo.thread, {message: message, received: false}];
            }
          })
          console.log(mergedThreads)
          return mergedThreads;
        })
      );
  }

Answer 1

嗨，我看不出有任何理由在这里使用of，而您在其中使用了forkJoin 和map 错误，当您通过forkJoin 组合两个或多个可观察对象时，它已经返回了一个可观察对象，所以您可以在上面使用.pipe 并在里面做地图，看我的例子

ngOnInit() {
  this.uniqueThreads =
    forkJoin(
      this.receivedMessages,
      this.sentMessages
    ).pipe(
      map(([receivedmessages, sentMessages]): Observable<{ username: string, thread: MessageModel[] }[]> => {
        let mergedThreads: { username: string, thread: MessageModel[] }[] = [];
        const messages: MessageModel[] = [...receivedmessages, ...sentMessages];
        messages.forEach((message) => {
          let activeThreadFrom = mergedThreads.find(any => any.username === message.fromUsername)
          if (activeThreadFrom === null) {
            mergedThreads.push({ username: message.fromUsername, thread: [message] })
          } else {
            activeThreadFrom.thread.push(message);
          }
          mergedThreads.push(activeThreadFrom);

          let activeThreadTo = mergedThreads.find(any => any.username === message.toUsername)
          if (activeThreadTo === null) {
            mergedThreads.push({ username: message.toUsername, thread: [message] })
          } else {
            activeThreadTo.thread.push(message);
          }
          mergedThreads.push(activeThreadTo)
        })
        return mergedThreads;
      })
    );
}