You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by md...@apache.org on 2024/09/13 09:07:12 UTC
(pekko-connectors) branch main updated: Improve message in case of gcs response failure
This is an automated email from the ASF dual-hosted git repository.
mdedetrich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 20bcdd5f1 Improve message in case of gcs response failure
20bcdd5f1 is described below
commit 20bcdd5f102c1953c01d9f7cff6c801d60e28bb8
Author: Matthew de Detrich <md...@gmail.com>
AuthorDate: Thu Sep 12 17:15:04 2024 +0200
Improve message in case of gcs response failure
---
.../pekko/stream/connectors/google/ResumableUpload.scala | 15 +++++++++++----
1 file changed, 11 insertions(+), 4 deletions(-)
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
index c6a78b9cb..a043f9fa5 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
@@ -16,6 +16,7 @@ package org.apache.pekko.stream.connectors.google
import org.apache.pekko
import pekko.NotUsed
import pekko.annotation.InternalApi
+import pekko.dispatch.ExecutionContexts
import pekko.http.scaladsl.model.HttpMethods.{ POST, PUT }
import pekko.http.scaladsl.model.StatusCodes.{ Created, OK, PermanentRedirect }
import pekko.http.scaladsl.model._
@@ -93,10 +94,16 @@ private[connectors] object ResumableUpload {
import implicits._
implicit val um: FromResponseUnmarshaller[Uri] =
- Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
- response.discardEntityBytes().future.map { _ =>
- response.header[Location].fold(throw InvalidResponseException(ErrorInfo("No Location header")))(_.uri)
- }
+ Unmarshaller.withMaterializer { _ => implicit mat => (response: HttpResponse) =>
+ if (response.status.isSuccess())
+ response.discardEntityBytes().future.map { _ =>
+ response.header[Location].fold(throw InvalidResponseException(ErrorInfo("No Location header")))(_.uri)
+ }(ExecutionContexts.parasitic)
+ else
+ Unmarshal(response.entity).to[String].flatMap { errorString =>
+ Future.failed(InvalidResponseException(
+ ErrorInfo(s"Resumable upload failed with status: ${response.status}", s"body: $errorString")))
+ }(ExecutionContexts.parasitic)
}.withDefaultRetry
GoogleHttp(mat.system).singleAuthenticatedRequest[Uri](request)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org