Async Request Response using RxJs WebSocketSubject

[abarke]

I'm trying to create a simple promise that resolves after the first response and rejects when an error or timeout occurs.

The use case is a request sent to the server via WebSocketSubject that responds with a single message. I realize this is a typical HTTP style use case, however we supply streams and simple request-response methods via our WebSocket API.

I started using the Observable.toPromise() method thinking this would be a great match. However it only resolves after the Observable completes and then it would take the last value in the stream. However I can't complete the WebSocketSubject as this will kill all subscriptions. Given that I cant really change the inner workings of the Observable to complete after the first response, I just wrapped the subscription in a promise.

I'm not delighted with my current solution so I was wondering what the best way to achieve this is?

Here is my current solution:

ws$: WebSocketSubject<Message>

/**
   * Wrap a subscription in a promise and handle reject, response and timeout.
   * @param channelId
   */
  public request (channelId: string) {
    console.log('Request ' + channelId)

    this.asyncRequest(channelId)
      .then(val => console.log(`${channelId} promise value`, val))
      .catch(err => console.error(err))
  }

  /**
   * Send a request to the server and expect and response. Reject when an error or timeout occurs
   * @param channelId
   * @param requestTimeout
   * @private
   */
  private asyncRequest (channelId: string, requestTimeout = 3000): Promise<Message> {
    const subMsg = {
      event: 'request',
      data: channelId,
      id: Date.now()
    }

    // Filter out response with same message id as the request
    const filter = (msg: Message): boolean => msg.id === subMsg.id

    // Subscribe to subject and filter by message type
    const observable = this.ws$.multiplex(() => subMsg, null, filter)

    // Create a promise that resolves after the first response
    return new Promise((resolve, reject) => {
      const subscription = observable.subscribe({
        next: msg => resolve(msg),
        error: err => reject(err),
        complete: () => reject(new Error(`${channelId} subscription completed before a response`))
      })

      // Handle request timeout and unsubscribe to prevent hanging subscriptions
      setTimeout(() => {
        subscription.unsubscribe()
        reject(new Error(`${channelId} request timeout after ${requestTimeout} milliseconds`))
      }, requestTimeout)
    })
  }

[abarke]

Just as an idea, it would be nice if RxJS supported such a uniplex request that returns a promise, accepts a filter and timeout e.g.

ws$: WebSocketSubject<Message>

const requestPromise = this.ws$.uniplex(reqMsg, filter, timeout)

requestPromise()
      .then(console.log)
      .catch(console.error)

[abarke]

OK seems I found a more elegant solution after some research... Here is the new function:

/**
   * Send a request to the server and expect and response. Reject when an error or timeout occurs.
   * @param channelId
   * @param requestTimeout
   * @private
   */
  private asyncRequest (channelId: string, requestTimeout = 3000): Promise<Message> {
    const request = {
      event: 'request',
      data: channelId,
      id: Date.now()
    }

    // Send request to server
    this.ws$.next(request)

    // Filter out expected response by matching the request and response message id
    const msgFilter = (msg: Message): boolean => msg.id === request.id

    // Create new observable using pipe and convert to a promise
    return this.ws$.pipe(
      first(msgFilter),
      timeout(requestTimeout)
    ).toPromise()
  }

[six-edge]

Here is a new generic function with error handling:

/**
 * Generic asynchronous request that returns a promise
 * @param request A request object that will be sent to the server as the request
 * @param messageFilter A predicate for selecting the appropriate messages from the server for the output stream.
 * @param errorHandler A function that handles an error response from the server.
 * @param requestTimeout Number of milliseconds to wait until the returned promise is resolved.
 */
async request (request: T, messageFilter: (msg: T) => boolean, errorHandler: (msg: T) => any, requestTimeout = 3000) => {
  // instance of WebSocketSubject
  const self = this

  // Send request to server
  self.next(request)

  // Create new observable using pipe and convert to a promise
  return self.pipe(
    first(messageFilter),
    tap(errorHandler),
    timeout(requestTimeout)
  ).toPromise()
}

// Project code
import { webSocket } from 'rxjs/webSocket'

const ws$ = webSocket<Message>('ws://localhost:8000')

const request = { event: 'getServerConfig', id: Date.now() }
const messageFilter = (response: Message): boolean => response.id === request.id
const errorHandler = (msg: Message) => {
  if (msg.event === 'error') {
    const errorMsg = msg as ErrorMessage
    throw new Error(`${errorMsg.reason} (${errorMsg.code})`)
  }
}
try {
  const response = await ws$.request(request, messageFilter, errorHandler)
  console.log('Rx', response)
} catch (err) {
  console.error(err)
}