In RxJS can I set up a ReplaySubject (or similar) that replays all the most recent messages with a unique name? – Javascript

You can implement your class inherited from Subject. Here is an example code in Typescript (just remove type annotations if you are using JavaScript)

import { PartialObserver, Subject, Subscription } from "rxjs";
import { filter } from "rxjs/operators";

class ReplayLatestPayloadSubject<T> extends Subject<T> {
  // Array of keys used to replay the events for a new subscriber in the order they arrived.
  // If order doesn't matter look on the keys in _cached dictionary instead in replayOldMessages
  private _keys: string[] = [];
  // A dictionary to map a key to the latest payload.
  private _cached: Record<string, T> = {};

  // construtor having a getKey message to get the key for the object (in the example the name property)
  constructor(private getKey: (value: T) => string) {
    super();
  }

  // Override the subscribe method (all overloads)
  public subscribe(observer?: PartialObserver<T>): Subscription;
  public subscribe(
    next?: (value: T) => void,
    error?: (error: any) => void,
    complete?: () => void
  ): Subscription;
  public subscribe(
    observerOrNext?: PartialObserver<T> | ((value: T) => void),
    error?: (error: any) => void,
    complete?: () => void
  ): Subscription {
    if (typeof observerOrNext === "function") {
      // The overload with a next method is used
      this.replayOldMessages(observerOrNext);
    } else if (observerOrNext?.next) {
      // the overload with an observer is used.
      this.replayOldMessages((v) => observerOrNext.next(v));
    }
    // Call base class
    return super.subscribe(observerOrNext as any, error, complete);
  }

  // replay messages when a new subscriber is called
  private replayOldMessages(next: (value: T) => void) {
    for (let key of this._keys) {
      next(this._cached[key]);
    }
  }

  // override the next function to store the values to be replayed later.
  public next(value?: T): void {
    if (value) {
      const key = this.getKey(value);
      if (!this._cached[key]) {
        this._keys.push(key);
      }
      this._cached[key] = value;
    }
    super.next(value);
  }
}

// Example

const replaySource = new ReplayLatestPayloadSubject<{
  name: string;
  payload: number;
}>((v) => v.name);

replaySource.next({ name: "aaa", payload: 1 });
replaySource.next({ name: "bbb", payload: 2 });
replaySource.next({ name: "aaa", payload: 3 });
replaySource.next({ name: "bbb", payload: 4 });
replaySource.next({ name: "bbb", payload: 5 });

replaySource.pipe(filter((x) => x.name === "aaa")).subscribe((x) => {
  console.log(
    "I want this to be logged once ... received the latest aaa " + x.payload
  );
});

replaySource.pipe(filter((x) => x.name === "bbb")).subscribe((x) => {
  console.log(
    "I want this to be logged once ... received the latest bbb " + x.payload
  );
});