HTTPObservable

HTTPObservable.scala

  /**
   * Asynchronously fetch and return a remote HTTP resource.
   *
   * @param request An HTTP request.
   * @param maxBytes The maximum number of bytes to be fetched.
   * @return The fetched byte array on success, or a NetworkError on failure.
   */
  def fetch (request: HttpRequestBase, maxBytes: Long) (implicit xc: ExecutionContext): AsyncResult[NetworkError, HTTPFetchedData] = {
    import ByteObservable._

    this.request(request).flatMap { observableResponse =>
      observableResponse.contentObservable.collectBytes(maxBytes).map { bytes =>
        HTTPFetchedData(observableResponse.response, bytes, observableResponse.contentType)
      }
    }.toAsyncResult.transform(
      /* Validate the HTTP response and map any errors to a NetworkError */
      { fetchedData =>
        val statusLine = fetchedData.httpResponse.getStatusLine
        val statusCode = statusLine.getStatusCode
  
        if (statusCode < 200 || statusCode >= 300) {
          Failure(HttpRequestFailed(s"HTTP request returned a ${statusLine.getStatusCode} error: ${statusLine.getReasonPhrase}", None))
        } else {
          Success(fetchedData)
        }
      },
      /*
       * TODO: Pattern matching of untyped Throwable instances is less than ideal; until such a time as we
       * find a solution to Observable error typing, we'll work with what we've got.
       *
       * See also: https://groups.google.com/forum/#!topic/rxjava/ucstY-JBMpg
       */
      {
        case networkError: NetworkError => Failure(networkError)
        case other => Failure(HttpRequestFailed(s"HTTP request failed", Some(other)))
      }
    )
  }