Deserializing Blob object from WebSocketSubject

user_2335880 0
Viewed 638

I am trying to use the built-in WebSocketSubject class in rxjs. I have set up most of the base code as an Angular service like so:

import { Injectable } from '@angular/core';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { Subscription, from, Observable } from 'rxjs';

@Injectable({
  providedIn: 'root'
})
export class WebsocketsApiService {
  connection: WebSocketSubject<any>;

  constructor() { 
    this.connection = webSocket({
      url: <websocket URL>,
      closeObserver: {
        next: () => console.log('Sockets disconnected.')
      },
      openObserver: {
        next: () => console.log('Socket connected.')
      }
    });
  }

  connect(observer: (value: any) => void): Subscription {
    return this.connection.subscribe(observer);
  }

  send(data: any) {
    if (this.connection) {
      this.connection.next(data);
    } else {
      console.error('Websockets not connected');
    }
  }

  close() {
    if (this.connection) {
      this.connection.complete();
      this.connection = null;
    }
  }

  ngOnDestroy() {
    this.close();
  }
}

With this base code however, it's failing to deserialize any data that comes in from the websocket connection. Upon debugging I've noticed that it's receiving data as a Blob, which is why the default deserializer function JSON.parse(data) fails.

I've tried rolling a simple custom deserializer function by getting the Blob data as a string, but it turns out Blob.text() is a Promise, and this is the part where I am stuck:

constructor() { 
    this.connection = webSocket({
      /// ...
      deserializer: (e) => e.data.text()
    });
}

So with this function as a deserializer, each component subscribing to this Websocket service is going to get a Promise for a string, which defers the responsibility of resolving the promise to the component.

Questions:

  1. Is this current approach ideal? Having each subscribing component resolve the promise themselves?
  2. Is there a cleaner approach, preferably where websocket subscribers get a JSON object parsed immediately as opposed to a string subscribers have to parse individually?
rxjs angular websocket
1 Answers

You can map to the value of an inner Promise with mergeAll/mergeMap, concatAll/concatMap, switchAll/switchMap. You probably want concatAll/concatMap to emit values in the order they come in.

constructor() { 
  this.connection = webSocket({
    /// ...
    deserializer: (e) => e.data.text()
  });
}

connect(observer: (value: any) => void): Subscription {
  return this.connection.pipe(
    concatAll()
  ).subscribe(observer);
}

Or

constructor() { 
  this.connection = webSocket({
    /// ...
    deserializer: ({data}) => data
  });
}

connect(observer: (value: any) => void): Subscription {
  return this.connection.pipe(
    concatMap(data => data.text())
  ).subscribe(observer);
}
user_94232310
Related