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

【问题标题】：Handling Errors From Task in Monix在 Monix 中处理来自任务的错误
【发布时间】：2021-03-21 17:49:45
【问题描述】：

我有以下代码 sn-p，它对一组 URL 进行简单的 HTTP 查找。

  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)
  }

我有一个播放控制器，我在其中调用此函数并运行如下任务：

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) 案例块，但我想获得以下信息：

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

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

【问题讨论】：

标签： task monix

【解决方案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)
  }

【讨论】：