Returning Observable Array from Rxjs WebSocket Subject call

Question

Trying to return an Observable from a rxjs6 Websocket Subject call. Having difficulty understanding the asynchronous nature for the subscribe. Code also filters into a returned envelope for extract the Product[].

I understand making a websocket call when the subscriber retrieves the result and routes for a new angular page.

getAllProducts() : Observable<Product[]> {
  let document: GetAllDocument = new GetAllDocument().init();
  let envelope = 
    new AuctionEnvelope().init(AuctionMethods.getAll,document);
  this.ws.subject.next(envelope); // call server
  let result__: Product[] = [];
  this.ws.subject.asObservable().subscribe( (resp: AuctionEnvelope) => {
    pipe(
      filter((resp)=>resp.method===AuctionMethods.getAllResponse,
      map((resp:AuctionEnvelope) =>resp.getAllResponseDocument.result),
      map((products: Product[]) => this.result__ = products) 
    );
  }
  );
  return from(this.result__);
}

I would like to return the result in an asynchronous way so the application can get the result when it's ready with an Observable.

Answer 1

getAllProducts() : Observable<Product[]> {
        let document: GetAllDocument_V1 = new GetAllDocument_V1().init();
        let envelope = new AuctionEnvelope_V1().init(AuctionMethods.getAll_1, document);
        this.ws.subject.next(envelope); // call server
        return this.ws.subject.asObservable().pipe(
          filter( (resp: AuctionEnvelope_V1) => resp.method === AuctionMethods.getAllResponse_1 ), // mine
          map((resp: AuctionEnvelope_V1) => resp.getAllResponseDocument_V1.result) // get result from envelope
        );
    }

I figured out that I could return the result of the pipe() call. This allowed me to retrieve the product[] from inside the envelope->document to get the result.