Angular2 - Rxjs Subject run twice even after unsubscribing

Question

I'm writing a desktop application using angular2 and electron and there is a download functionality.

My DownloadService is this

import {Injectable} from '@angular/core';
import {Subject} from "rxjs";

interface IQueueItem {
    url: string
}

@Injectable()
export class DownloadService {
    private queue: Array< IQueueItem > = [];

    private downloadSubject: Subject<any>;

    constructor() {
        this.downloadSubject = new Subject();
    }

    addToList(item: IQueueItem) {
        this.queue.unshift(item);

        downloadList();

        return this.downloadSubject;
    }

    downloadList() {
        // pick one item from queue and send it to electron to download and store

        // do this every time a chunk of data received
        this.downloadSubject.next(evt);
        ...
    }

    pauseDownload(item) {
        // send an event to electron to it should stop downloading and therefore no chunk of data will receive
        // also remove item from queue
        ...
    }
}

And my ItemComponent is this:

import {Component} from '@angular/core';
import {DownloadService} from "../services/download.service";

@Component({
    selector: 'app-item',
    template: `
        ...
      `
})
export class ItemComponent {
    constructor(private downloadService: DownloadService) {
        this.addToQueue();
    }

    subscription;

    downloadedBytes = 0;

    fileUrl = '';

    pause() {
        this.downloadService.pauseDownload(this.fileUrl);

        this.subscription.unsubscribe();

        ...
    }

    resume() {
        this.addToQueue();
    }

    private addToQueue() {
        this.subscription = this.downloadService.addToList(this.fileUrl)
            .subscribe(evt => {

                console.log(evt.delta);

                this.downloadedBytes += evt.delta;
            });
    }
}

The problem is when I pause an item I unsubscribe from Subject passed from DownloadService but when I resume again, every console.log() prints twice and add twice data to downloadedBytes. Also if I pause and resume again, It will add more and more bytes and logs!

I searched but I couldn't find any clue to solve.

Answer 1

You can see from the PLUNKER created by yurzui that if you resume and pause everything works as expected, triggered only once.
The issue is when you click resume 2 times in a row as the first subscription will be lost in the memory and only the second one will be stored in this.subscription the access to the first one is lost and you cannot unsubscribe.

This is more like an app issue than rxjs, you should not be able to click resume if the subscription is not paused, so you need to handle the states properly, something like this (edited from @yurzui plunker) :

@Component({
  selector: 'my-app',
  template: `
    <h1>Angular 2 Systemjs start</h1>
    <button *ngIf="started && !paused" (click)="pause()">Pause</button>
    <button *ngIf="started && paused" (click)="resume()">Resume</button>
    <button *ngIf="!started" (click)="start()">Start</button>
  `
})
export class App {
  constructor(private downloadService: DownloadService) {
    this.addToQueue();
  }

  subscription;
  downloadedBytes = 0;
  started = false;
  paused = false;


  pause() {
    this.paused = true;  

    this.subscription.unsubscribe();
  }

  start() {
    this.started = true;  

    this.addToQueue();
  }

  resume() {
    this.paused = false;

    this.addToQueue();
  }

  private addToQueue() {
    this.subscription = this.downloadService.addToList(this)
      .subscribe(evt => {

        console.log(11);

        this.downloadedBytes += evt.delta;
      });
  }
}

You can find a working example in this updated PLUNKER