如何在 RxJS 主题中修改地图？

Question

我有一个Map，我想向几个消费者公开。该地图包含Files 的列表及其上传状态：

export interface FileProgress {
    file: File;
    sent: boolean;
}

所以这个Map 存储在BehaviorSubject 中，我将使用.asObservable() 方法公开：

private readonly fileProgressSubject: BehaviorSubject<Map<string, FileProgress>>;

我的问题是，对底层数据结构执行更新的最佳方式是什么？

如果BehaviorSubject 包含一个简单类型，比如mySubject: BehaviorSubject<boolean>，我会简单地调用mySubject.next(true)，瞧，我已经更新了值。

但是在我的fileProgressSubject 中更新Map 更加麻烦。我需要访问地图，然后对其进行更改，所以这可行，但感觉不对：

const temp = fileProgressSubject.getValue();
temp.delete(someFileKey); // EXAMPLE OPERATION - DELETION
fileProgressSubject.next(temp);

这是对我的底层Map 进行更新的“正确”方式吗？我读过在Subject 上调用.getValue() 是强制使用反应式构造，这可能意味着您没有做对。

我的另一个想法是，因为这个 BehaviorSubject 将作为可观察对象公开，即使我可以看到 Typescript 编译器不允许这样做：

const fileObservable = fileProgressSubject.asObservable();

fileObservable.pipe(
    map(fileProgressMap => fileProgressMap.delete(someFileName)), // DELETE A FILE FROM THE MAP
).subscribe(fileProgressSubject);

因此，基本上，将底层数据结构的当前值通过 observable 管道，对其进行更改，然后将新的Map 填充到主题中。但同样，这似乎是不可能的，不知道为什么。

那么，处理这种情况的最佳方法是什么？

Answer 1

如果只是为了状态管理，底层数据结构应该对消费者透明。这样一来，如果您决定有一天将 Map 更改为其他名称，您就不需要找到所有消费者并更新它们。

一种更简洁、更易于维护的方法可能是让文件进度状态的所有者（我假设是 Angular 服务）公开简单的方法，允许消费者将他们的文件添加到进度状态，然后在它更新时更新他们的状态变化。

所以你的状态管理服务看起来像这样：

import { Injectable } from '@angular/core';
import { BehaviorSubject, Observable } from 'rxjs';

export interface FileProgress {
  file: File;
  sent: boolean;
}

Injectable({
  providedIn: 'root'
})
export class FileProgressService {
  private _fileProgressMap = new Map<string, FileProgress>();
  get fileProgressMap(): ReadonlyMap<string, FileProgress> {
    return this._fileProgressMap;
  }

  private _fileProgress$ = new BehaviorSubject<ReadonlyMap<string, FileProgress>>(this.fileProgressMap);
  get fileProgress$(): Observable<ReadonlyMap<string, FileProgress>> {
    return this._fileProgress$.asObservable();
  }

  addFile(fileKey: string, file: File) {
    const initialProgress: FileProgress = {
      file,
      sent: false
    };

    this._fileProgressMap.set(fileKey, initialProgress);
    this.notify();
  }

  updateFileProgress(fileKey: string, isSent: boolean) {
    this._fileProgressMap.get(fileKey).sent = isSent;
    this.notify();
  }

  private notify() {
    this._fileProgress$.next(this.fileProgressMap);
  }
}

您向消费者公开简单的addFile 和updateFileProgress 方法。

该服务恰好在Map 中维护状态，但消费者不需要知道这一点，因此只要addFile 和updateFileProgress 方法签名就可以随时更改它而无需更新消费者不要改变。

您还公开了一个fileProgress$ Observable，消费者可以订阅它以接收状态更新。由于您只想通过公开的方法来控制您的状态，因此 Observable 会发出一个 ReadonlyMap，它是您的状态管理 Map 的只读包装器，因此消费者无法直接更新底层数据结构。

您的消费者都可以订阅fileProgress$ Observable，但您应该考虑他们是否真的需要。您的所有消费者是否都需要知道其他文件的 sent 状态何时发生变化，或者您是否通过 Observable 这样做只是为了让消费者有一种方法来获取状态管理对象并为他们的特定文件更新它？如果是后者，那么可能甚至不需要让他们订阅，因为公开的 add/update 方法应该为他们提供所需的一切。

Answer 2

有一种设置方法可以让地图自行维护，您无需手动设置或删除其上的值，而是通过可转换地图的辅助事件流。

这个实现很简单，因为事件是映射上的函数。但是，您可以设计与数据结构无关的自定义事件。那么你就真的在未来的证明和优化/记录/等方面的可能性方面参加比赛。

// Stream that holds the result of applying all the update events 
private _fileProgress$: Observable<Map<string, FileProgress>>;
// Subject that emits update events
private _fileProgressUpdate$: Subject<(input: Map<string, FileProgress>) => any>;

constructor(){
  this._fileProgressUpdate$ = new Subject<(input: Map<string, FileProgress>) => any>();

  // Apply the given function to the most current map
  const accumulator = (accMap, currFn) => {
    currFn(accMap);
  };
  // accumulate and apply all update events to a map
  this._fileProgress$ = _fileProgressUpdate$.pipe(
    scan(accumulator, new Map<string, FileProgress>()),
    shareReplay(1)
  );
  // We want to subscribe right away so later subscribers just
  // start with the most recent shareReplay(1) value.
  this._fileProgress$.subscribe();
}

// You can make any number of custom functions that send operations to
// your map via the _fileProgressUpdate$
newFile(exampleFile: File){
  const fileProgress = {file: exampleFile, sent: false }
  this._fileProgressUpdate$.next(mapO =>
    // At this point we don't know how many maps are 'subscribed' to this update
    // which is part of the beauty of this appraoch 
    mapO.set(exampleFile.name(), fileProgress)

  );
}

// A consumer might only be interested in the progress of a single
// file. That sort of logic is easily implemented here and might mean later
// changes to update your service for performance are non-breaking
watchFileById(id: string): Observable<FileProgress>{
  this._fileProgress$.pipe(
    map(mapO => mapO.get(id)),
    filter(fileProgress => fileProgress != null)
  );
  
}

// Expose your map if you need to
const watchMapOfFileProgress(): Observable<Map<string, FileProgress>> {
  return this._fileProgress$;
} 

// Maybe users don't care about keys and just want a whole list of progress?
watchArrayOfFileProgress(): Observable<Array<FileProgress>>{
  return this._fileProgress$.pipe(
    map(mapO => Array.from(mapO.values()))
  );
}

// Users who don't care about progress and just the most current list of files
// That still need to be sent
watchPendingFiles(): Observable<Array<FileProgress>>{
  return this._fileProgress$.pipe(
    map(mapO => 
      Array.from(mapO.values())
        .filter(prog => !prog.sent)
        .map(prog => prog.file)
     )
  );
}

这种设计模式的好处在于您已经完全抽象出底层数据类型。此服务的使用者可以根据自己的需要观看FileProgress、Map<FileProgress>、Array<FileProgress> 或Array<File> 的流，而无需知道在服务中是如何处理的。

还有一个好处是，如果您更改 _fileProgressUpdate$ 流以接受自定义操作（而不是在 map 上运行的函数），那么您可以开始对任意数据类型甚至跨不同的数据类型实现非常复杂的自定义逻辑数据类型。

此外，您并没有将命令式和函数式风格混为一谈。您的事件不是更改数据然后发送更新事件，而是在发送时对转换/变异进行编码，并且这些事件的累积会创建您的地图。

这意味着您可以缓存所有事件并重放它们以重新创建任意程序状态。

最后，由于您的地图成为次要公民，因此创建自定义地图/列表/ect 以反映您的转换事件的某些子集变得微不足道。任意数量的函数可以创建任意数量/顺序的事件，并且由于它们不直接更改数据，因此您可以无缝地连接到它们的功能中。

Answer 3

我相信没有任何方法是客观上“正确”的，除非它不起作用。这是一个方法。 然而，使用带有Map.prototype.delete 的map 将不起作用，因为它returns a boolean 会创建一个新的Observable。我建议使用tap。 试试：

fileObservable.pipe(
    tap(fileProgressMap => fileProgressMap.delete(someFileName)), // DELETE A FILE FROM THE MAP
).subscribe();