如何过滤和转换 Observable 中的数据 - 重复数据答案

【问题标题】：How to filter and transform data in Observable - duplicate data如何过滤和转换 Observable 中的数据 - 重复数据
【发布时间】：2019-01-19 01:54:40
【问题描述】：

我有一个主题事件模型 - 定义了它应该具有的属性：

export class EventSchema {
    title: string;
    start: string;
    end: string;
    price: number;

    constructor(title: string, start: string, end: string, price: number){
        this.title=title;
        this.start=start;
        this.end=end;
        this.price=price;
    }
}

我有另一个封装了上述模型的模型 - 有一个主题名称和一个数组中的事件属性，因为单个主题可以有多个事件：

import { EventSchema } from './eventSchema.model';

export class SubjectEvents {
    subjectName: string;
    eventData: EventSchema[];

    constructor(subjectName: string, eventData: EventSchema[]){
        this.subjectName=subjectName;
        this.eventData=eventData;
    }
}

我在 Firebase 中有一个与这些模型匹配的集合。到目前为止，只有两份文件，一份是数学科目，一份是物理科目，每份都有 2 个事件。示例数据为：

我有这项服务，它可以获取数据并对其进行转换。在这种情况下，我想将每个事件作为一个主题（文档）并将其放入一个大数组中。也许我需要合并每个文档的输出......不知道如何。我能够做到这一点 - 但结果是重复的，我不明白为什么。请参阅下面的最终输出屏幕截图。

import { Injectable } from '@angular/core';
import { AngularFirestore, AngularFirestoreCollection, AngularFirestoreDocument, QuerySnapshot, QueryDocumentSnapshot } from 'angularfire2/firestore';
import { Observable, of, Subscription } from 'rxjs';
import { map, filter } from 'rxjs/operators';
import 'rxjs/add/operator/mergeMap';
import { SubjectEvents } from '../_models/subjects/subjectEvents.model';
import { EventSchema } from '../_models/subjects/eventSchema.model';

@Injectable()
export class SubjectEventsService {

    subjectEventCollection : AngularFirestoreCollection<SubjectEvents>;
    subjectEventDocument : AngularFirestoreDocument<SubjectEvents>;
    subjectEvents : Observable<SubjectEvents[]>;
    subjectEvent : Observable<SubjectEvents>;
    filteredArr : Observable<any[]>;

    constructor(private afs : AngularFirestore) { 
      //Get user collection on initialise
      this.subjectEventCollection = this.afs.collection('subjectEvents');
    }
    getSubjectEvents(subjectFilter: string): Observable<SubjectEvents[]> {
      this.filteredArr = this.subjectEventCollection.snapshotChanges()
      .pipe(map(changes => {
        let filteredArr : any = [];
        console.log(filteredArr);
        return changes
            .map(action => {
              console.log(action);
              const data = action.payload.doc.data() as SubjectEvents;
              data.subjectName = action.payload.doc.data().subjectName;
              data.eventData = action.payload.doc.data().eventData;
              console.log(data);

              data.eventData.forEach(result => {filteredArr.push(result); console.log(filteredArr)});

              return filteredArr;
        });
      }));
      return this.filteredArr;
    }
}

【问题讨论】：

我会说你需要 .pipe(mergeMap(changes... 或 return changes.pipe(mergeMap(action... 因为更改是 Observable<SubjectEvents[]> 并且你想将该数组展开到 Observable<SubjectEvents>。如果您键入参数，它有助于推理。（顺便说一句，混合 pipable 运算符和 non-pipable 运算符可能是不好的形式）。

标签： angular rxjs observable google-cloud-firestore angular6

【解决方案1】：

Live example

我刚刚提出了一种更好的方法，我创建了一个名为 hasBeenEmitted 的自定义运算符，您只需提供要过滤的属性即可。

const duplicateData = from(data);
const hasBeenEmitted = n => source =>
  new Observable(observer => {
    const emmited = [];
    return source.subscribe({
      next(item) {
        if (emmited.filter(e => e.title !== item.title).length === 0) {
          emmited.push(item);
          observer.next(item);
        }
      },
      error(err) { observer.error(err); },
      complete() { observer.complete(); }
    })
  });



const source = duplicateData.pipe(
  hasBeenEmitted(item => item.title),
  map(e => ({ value: e.price, start: e.start }))
);

source.subscribe(console.log);

【讨论】：

【解决方案2】：

Live working example

希望能帮上忙，这是我的建议：使用filter 和map，但在filter 中它使用自定义函数来实现您的预期结果：

const duplicateData = from(data);
const emmited = [];
const hasBeenEmitted = (item, emmited) => {
  if (emmited.filter(e => e.title !== item.title).length === 0) {
    emmited.push(item);
    return item;
  } else {
    return null;
  }
};

const source = duplicateData.pipe(
  filter(item => hasBeenEmitted(item, emmited)),
  map(e => ({ value: e.price, start: e.start }))
);

source.subscribe(console.log);

【讨论】：