Sivan Israelov + Koral Ben Ami snippet for rxjs post

subscribeToFileRemovals.ts

// Fake remove file 
function remove(filename: string) {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve();
    }, 300);
  });
}

interface FileRemoveInteractorImpl {
  onFileRemoval(): Observable<string>;
  removeFile(): Observable<any>;
  disconnect(): void;
}

class FileRemoveInteractor implements FileRemoveInteractorImpl {
  private subject: Subject<string>;
  private _filename: string;
  constructor(filename: string) {
    this.subject = new Subject();
    this._filename = filename;
  }

  onFileRemoval(): Observable<string> {
    return this.subject.asObservable(); // notify consumers on file removal
  }
  // the side effective act of removing the file 
  removeFile(): Observable<any> {
    return defer(() => remove(this._filename)).pipe(
      tap(() => this.subject.next(this._filename))
    );
  }
  // closing the stream - won't be able to emit anymore 
  disconnect() {
    this.subject.complete();
  }
}

class ConsumerOne {
  constructor(rmInteractor: FileRemoveInteractorImpl) {
    rmInteractor // subscribing to the file removal event
      .onFileRemoval()
      .subscribe((name: string) => this.handler(name));
  }

  handler(name: string) {
    console.log("consumer 1 was notifed: removed", name);
  }
}

class ConsumerTwo {
  constructor(rmInteractor: FileRemoveInteractorImpl) {
    rmInteractor
      .onFileRemoval()
      .subscribe((name: string) => this.handler(name));
  }

  handler(name: string) {
    console.log("consumer 2 was notifed: removed", name);
  }
}

const producer = new FileRemoveInteractor("foo.txt");
const consumerOne = new ConsumerOne(producer);
const consumerTwo = new ConsumerTwo(producer);

producer.removeFile().subscribe(() => {
  producer.disconnect(); // prevent future emissions 
});
// this will still remove the file but from this point the prev consumers won't be notified 
producer.removeFile().subscribe();