Angular subscription triggers only once

Question

I have a material table to show a list of records.




  
     Id 
     {{ deployment.buildId }} 
  

  
     Type 
     {{ deployment.deploymentType }} 
  

  
     Status 
     {{ deployment.status }} 
  

  
     Started At 
    
      {{ deployment.started | date: 'short':locale }}
    
  

  
     Completed At 
    
      {{ deployment.completed | date: 'short':locale }}
    
  

  
     Queued 
     {{ deployment.queuedCount }} 
  

  
     Done 
     {{ deployment.completedCount }} 
  

  
     Error 
     {{ deployment.errorCount }} 
  

  
     In Progress 
    
      {{ deployment.inProgressCount }}
    
  

  
     Job Count 
     {{ deployment.totalJobCount }}

The component subscribes to a service that proxies messages from a websocket connnection

        import { AfterViewInit, ChangeDetectorRef, Component, OnDestroy, OnInit, ViewChild } from '@angular/core';
    import { MatPaginator } from '@angular/material/paginator';
    import { MatSort } from '@angular/material/sort';
    import { MatTableDataSource } from '@angular/material/table';
    import { ActivatedRoute, Router } from '@angular/router';
    import { Deployment } from 'app/shared/models/deployment.model';
    import { DatabaseDeploymentService } from 'app/shared/services/database-deployment.service';
    import { BehaviorSubject, Subject } from 'rxjs';
    import { map } from 'rxjs/operators';

    @Component({
      selector: 'app-list-deployments',
      templateUrl: './list-deployments.component.html',
      styleUrls: ['./list-deployments.component.scss'],
    })
    export class ListDeploymentsComponent implements OnInit, AfterViewInit, OnDestroy {
      @ViewChild(MatPaginator) paginator: MatPaginator;
      @ViewChild(MatSort) sort: MatSort;

      private deploymentsSubject$ = new Subject>();
      private showLoaderSubject$ = new BehaviorSubject(true);
      spinnerMode = 'indeterminate';
      displayedColumns: string[] = [
        'buildId',
        'deploymentType',
        'status',
        'started',
        'completed',
        'queuedCount',
        'inProgressCount',
        'completedCount',
        'errorCount',
        'totalJobCount',
      ];
      deployments: Deployment[];
      showLoader$ = this.showLoaderSubject$.asObservable();
      dataSource$ = this.deploymentsSubject$.asObservable();

      constructor(
        private deploymentService: DatabaseDeploymentService,
        private router: Router,
        private route: ActivatedRoute,
        private cdRef: ChangeDetectorRef
      ) {}

      ngOnDestroy(): void {}

      ngOnInit(): void {}

      ngAfterViewInit(): void {
        this.deploymentService.allDeployments$.subscribe((deployments) => {
          this.deployments = deployments;
          this.publishDataSourceUpdate();
          this.showLoaderSubject$.next(false);
        });
        this.deploymentService.deployment$.subscribe((deployment) => {
          this.addOrUpdateDeployments(deployment);
        });
        this.deploymentService.loadDeployments();
        this.cdRef.detectChanges();
      }

      public viewDeploymentDetails(deployment: Deployment) {
        this.router.navigate([`${deployment.buildId}`], { relativeTo: this.route });
      }

      private addOrUpdateDeployments(deployment: Deployment) {
        const found = this.deployments.find((d) => d.id == deployment.id);
        if (found) {
          this.deployments.map((d, i) => {
            if (d.id == deployment.id) {
              this.deployments[i] = deployment;
            }
          });
        } else {
          this.deployments.push(deployment);
        }
        this.publishDataSourceUpdate();
      }

      private publishDataSourceUpdate() {
        const dataSource = new MatTableDataSource();
        dataSource.paginator = this.paginator;
        dataSource.sort = this.sort;
        dataSource.data = this.deployments;
        this.deploymentsSubject$.next(dataSource);
      }
    }

The service implementation:

    import { Injectable } from '@angular/core';
    import { BehaviorSubject, Subject } from 'rxjs';
    import { takeUntil } from 'rxjs/operators';
    import { ApiMessage } from '../models/apiMessage.model';
    import { Deployment } from '../models/deployment.model';
    import { DeploymentDetails } from '../models/deploymentDetails.model';
    import { Job } from '../models/job.model';
    import { RequestMessage } from '../models/requestMessage.model';
    import { WebsocketConnectionService } from './websocket-connection.service';

    @Injectable({
      providedIn: 'root',
    })
    export class DatabaseDeploymentService {
      private messageTypes: string[] = [
        'list-deployments-result',
        'deployment-created',
        'deployment-updated',
        'get-deployment-details-result',
        'job-created',
        'job-updated',
      ];
      private unsubscribe = new Subject();
      private deploymentDetailsSubject$ = new Subject();
      private deploymentSubject$ = new Subject();
      private allDeploymentsSubject$ = new Subject();
      private jobSubject$ = new Subject();

      
      public deployment$ = this.deploymentSubject$.asObservable();
      public allDeployments$ = this.allDeploymentsSubject$.asObservable();
      public deploymentDetails$ = this.deploymentDetailsSubject$.asObservable();
      public job$ = this.jobSubject$.asObservable();


      constructor(private websocketConnectionService: WebsocketConnectionService) {
        this.websocketConnectionService.messages$
          .pipe(takeUntil(this.unsubscribe))
          .subscribe((apiMessage: ApiMessage) => this.handleMessage(apiMessage));
      }
      ngOnDestroy(): void {
        this.unsubscribe.next();
        this.unsubscribe.complete();
      }

      public loadDeployments() {
        const requestMessage: RequestMessage = {
          action: 'list-deployments',
        };
        this.websocketConnectionService.sendMessage(requestMessage);
      }

      public getDeploymentById(deploymentId: string) {
        const requestMessage: RequestMessage = {
          action: 'get-deployment-details',
          parameter: {
            deploymentId: deploymentId,
          },
        };
        this.websocketConnectionService.sendMessage(requestMessage);
      }

      public getDeploymentByBuild(buildId: string) {
        const requestMessage: RequestMessage = {
          action: 'get-deployment-details',
          parameter: {
            buildId: buildId,
          },
        };
        this.websocketConnectionService.sendMessage(requestMessage);
      }

      private handleMessage(apiMessage: ApiMessage) {
        if (this.messageTypes.includes(apiMessage.type)) {
          switch (apiMessage.type) {
            case 'list-deployments-result':
              this.allDeploymentsSubject$.next(apiMessage.data);
              break;
            case 'deployment-created' || 'deployment-updated':
              this.deploymentSubject$.next(apiMessage.data);
              break;
            case 'get-deployment-details-result':
              this.deploymentDetailsSubject$.next(apiMessage.data);
              break;
            case 'job-created' || 'job-updated':
              this.jobSubject$.next(apiMessage.data);
              break;
            default:
              break;
          }
        }
      }
    }

The issue is my subscription is only being triggered once.

The "deployment$" observable should be triggered multiple times as the updates are occurring but it is not.

I know the messages are coming to the app as I am logging them in the WebSocket.Connection service.

UPDATED.

I added logging statements the switch statement as suggested - the messages are getting published to the subject.

I also added in error handling on the subscription.

The subscription only fires once (I see next in the log). The error handler is never invoked.

For completeness - here is the websocket connection that the deployment service subscribes to:

        import { Injectable } from '@angular/core';
    import { EMPTY, Subject } from 'rxjs';
    import { WebSocketSubject, webSocket } from 'rxjs/webSocket';
    import { catchError, takeUntil, tap } from 'rxjs/operators';
    import { environment } from 'environments/environment';
    import { ApiMessage } from '../models/apiMessage.model';
    import { RequestMessage } from '../models/requestMessage.model';

    const apiUrl = environment.apiURL;

    @Injectable({
      providedIn: 'root',
    })
    export class WebsocketConnectionService {
      private unsubscribe = new Subject();
      private socket$!: WebSocketSubject;
      private messagesSubject$ = new Subject();
      public messages$ = this.messagesSubject$.asObservable();

      constructor() {
        this.connect();
      }

      ngOnDestroy(): void {
        this.unsubscribe.next();
        this.unsubscribe.complete();
      }

      public sendMessage(requestMessage: RequestMessage) {
        if (!this.socket$) {
          this.connect();
        }
        const request = JSON.stringify(requestMessage);
        console.log(`Sending ${request}`);
        this.socket$.next(requestMessage);
      }

      private connect() {
        if (!this.socket$ || this.socket$.closed) {
          this.socket$ = this.getNewSocket();
          this.socket$
            .pipe(
              tap({
                error: (error) => console.log(error),
              }),
              catchError((_) => EMPTY),
              takeUntil(this.unsubscribe)
            )
            .subscribe((message) => this.handleMessage(message));
        }
      }

      private getNewSocket() {
        return webSocket({
          url: apiUrl,
          closeObserver: {
            next: () => {
              console.log(`Connection closed`);
            },
          },
        });
      }

      private handleMessage(apiMessage: ApiMessage) {
        if (apiMessage.type) {
          this.messagesSubject$.next(apiMessage);
        }
      }
    }

The updated subscription....

    ngAfterViewInit(): void {
        this.deploymentService.allDeployments$.subscribe((deployments) => {
          this.deployments = deployments;
          this.publishDataSourceUpdate();
          this.showLoaderSubject$.next(false);
        });
        this.deploymentService.deployment$
          .pipe(
            map((deployment) => this.addOrUpdateDeployments(deployment))
          )
          .subscribe(
            (d) => console.log(`next`),  <-- Shows up once
            (e) => console.log(`Error ${e}`, <---never
            () => console.log('complete'))
          );
        this.deploymentService.loadDeployments();
        this.cdRef.detectChanges();
      }

Vasileios Kagklis · Answer

The deployment$ observable emits every time the next method is called on the deploymentSubject. This is done in handleMessage inside your service, which in turn is called whenever this.websocketConnectionService.messages$ emits.

So I would start by checking how many times this one emits. Also, inside handleMessage the deploymentSubject is used only in the deployment-created case. If messages$ emits, I would make sure that it has the correct case value.

Angular subscription triggers only once

1 Answers