How to modify a Map within a RxJS subject?

Question

I have a Map that I'd like to expose to several consumers. The map contains a list of Files and their upload status:

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

So this Map is stored within a BehaviorSubject that I will expose using the .asObservable() method:

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

My question is, what's the best way to perform updates to the underlying data structure?

If the BehaviorSubject contained a simple type, say mySubject: BehaviorSubject<boolean>, I would simply call mySubject.next(true) and voila I've updated the value.

But updating the Map within my fileProgressSubject is more hacky. I need to access the map and then perform changes on it, so this would work but it just feels wrong:

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

Is this the "correct" way of making updates to my underlying Map? I've read that calling .getValue() on a Subject is using reactive constructs imperatively, and probably means that you're not doing something right.

My other thought, since this BehaviorSubject will be exposed as an observable, would be this even though I can see that the Typescript compiler won't allow it:

const fileObservable = fileProgressSubject.asObservable();

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

So, basically, pipe the current value of the underlying data structure through the observable, make a change to it, and then stuff the new Map into the subject. But again, this doesn't seem to be possible, not sure why.

So, what is the best way to handle situations like this?

Answer 1

There's a way to set this up such that the map maintains itself and you don't set or delete values on it manually but through a secondary event stream that transforms your map.

This implementation is straight forward because events are functions over maps. You could, however, design custom events that are data-structure agnostic. Then you're really off to the races in terms of future-proofing and possibilities for optimization/logging/ect.

// 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)
     )
  );
}

What's nice about this design pattern is that you've abstracted away from your underlying datatype entirely. A consumer of this service can watch streams of FileProgress, Map<FileProgress>, Array<FileProgress>, or Array<File> depending on their needs without knowing how it is handled within the service.

What's also nice is that if you alter your _fileProgressUpdate$ stream to accept custom operations (instead of a function that operates on map), then you can start to implement very complex custom logic on arbitrary datatypes or even across different datatypes.

Furthermore, you're not mixing imperative and functional styles as much. Rather than mutating data and then sending an update event, your event encodes transformation/mutation when it's sent and it's the accumulation of these events that create your map.

This means you can cache all your events and replay them to re-create an arbitrary program state.

Finally, since your map becomes a secondary citizen, it becomes trivial to create custom maps/list/ect that reflect some subset of your transformation events. Any number of functions can create any number/order of events and because they don't change data directly, you can hook into their functionality seamlessly.

Answer 2

The underlying data structure should probably be transparent to the consumers if it's simply for state management. That way, if you decide one day to change from a Map to something else, you won't need to find all consumers and update them as well.

A cleaner, more maintainable approach might be to have the owner of the file progress state (I'm assuming an Angular service) expose simple methods that allow consumers to add their files to the progress state and then update their state when it changes.

So your state management service would look something like this:

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);
  }
}

You expose simple addFile and updateFileProgress methods to consumers.

The service happens to maintain the state in a Map but consumers shouldn't need to know that so it can be changed at any time without needing to update consumers as long as the addFile and updateFileProgress method signatures don't change.

You also expose a fileProgress$ Observable that consumers can subscribe to to receive state updates. Since you want to control your state via the exposed methods only, that Observable emits a ReadonlyMap that's a readonly wrapper around your state management Map, so consumers can't directly update the underlying data structure.

Your consumers can all subscribe to the fileProgress$ Observable, but you should consider whether or not they really need to. Do all of your consumers need to know when some other file's sent state has changed, or are you doing this via an Observable just to give consumers a way to get a hold of the state management object and update it for their particular file? If the latter, then it's probably not even necessary to have them subscribe since the exposed add/update methods should give them everything they need.

Answer 3

I believe no way is objectively "correct" unless it doesn't work. Here's a way. However, using map with Map.prototype.delete won't work because it returns a boolean, which would create a new Observable<boolean>. I suggest using tap. Try:

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