在 Monix 中处理来自任务的错误

Question

我有以下代码片段，它执行简单的 HTTP 查找一组 URL。

  def parse(filter: ParserFilter): Task[Seq[HttpBinaryResponse]] = {
    import scalaj.http._
    // Extracts all HTML body anchor elements into an Option
    val browser = new JsoupBrowser {
      override def requestSettings(conn: Connection): Connection =
        conn.timeout(2000)
    }
    val hrefs =
      browser.get(filter.url) >> elementList("a[href]") >?> attr("href")
    val batched = hrefs.distinct.flatten
      .filter(_.startsWith("http"))
      //.map(toTask)
      .map(href =>
        Task {
          HttpBinaryResponse.asHttpBinaryResponse(href, Http(href).asString)
        })
      .sliding(30, 30)
      .toSeq
      .map(chunk => Task.parSequence(chunk))

    Task.sequence(batched).map(_.flatten)
  }

我有一个 Play 控制器，我在其中调用此函数和运行任务，如下所示：

val batches = appBindings.httpService.parse(parserFilter)
batches.runToFuture.materialize.map {
  case Success(elems) =>
    Ok(Json.prettyPrint(Json.obj(
      "baseURL" -> s"${parserFilter.url}",
      "Total Elements" -> s"${elems.size}",
      "results" -> HttpBinaryResponse.asJson(elems)
    ))).enableCors
  case Failure(err) =>
    Ok(Json.obj("status" -> "error", "message" -> s"${err.getMessage}")).enableCors
}

对于导致 SSLHandshakeException 的 URL 之一，我进入了 Failure(err) 案例块，但我想获得以下内容：

无论错误是什么，我都想进入成功块，我已经在其中捕获了任何失败的错误消息 URL。

如何调整我的任务实现来满足我的需要？有任何想法吗？我尝试了 onErrorRecoverWith 处理程序，但似乎没有任何效果。有什么想法吗？

Answer 1

我设法完成了如下操作：

def parse(filter: ParserFilter): Task[Seq[HttpBinaryResponse]] = {
    import scalaj.http._
    // Extracts all HTML body anchor elements into an Option
    val browser = new JsoupBrowser {
      override def requestSettings(conn: Connection): Connection =
        conn.timeout(2000)
    }
    val hrefs =
      browser.get(filter.url) >> elementList("a[href]") >?> attr("href")
    val batched = hrefs.distinct.flatten
      .filter(_.startsWith("http"))
      //.map(toTask)
      .map(href =>
        Task {
          HttpBinaryResponse.asHttpBinaryResponse(href, Http(href).asString)
        }.onErrorRecoverWith { case ex: Exception =>
          Task.now(
            HttpBinaryResponse(
              origin = href,
              isSuccess = false,
              errorMessage = Some(s"${ex.getMessage}")))
        })
      .sliding(30, 30)
      .toSeq
      .map(chunk => Task.parSequence(chunk))

    Task.sequence(batched).map(_.flatten)
  }

在 Monix 中处理来自任务的错误

Handling Errors From Task in Monix

task

monix