-
Notifications
You must be signed in to change notification settings - Fork 27
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
add scala3 support for s3 #167
Changes from all commits
91b90bb
a41bec1
f179316
d3c2469
3ef6f4f
7003459
7090dd3
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -29,7 +29,7 @@ import pekko.http.scaladsl.{ ClientTransport, Http } | |
import pekko.stream.connectors.s3.BucketAccess.{ AccessDenied, AccessGranted, NotExists } | ||
import pekko.stream.connectors.s3._ | ||
import pekko.stream.connectors.s3.impl.auth.{ CredentialScope, Signer, SigningKey } | ||
import pekko.stream.scaladsl.{ Flow, Keep, RetryFlow, RunnableGraph, Sink, Source, Tcp } | ||
import pekko.stream.scaladsl.{ Flow, Keep, RetryFlow, RunnableGraph, Sink, Source, SubFlow, Tcp } | ||
import pekko.stream.{ Attributes, Materializer } | ||
import pekko.util.ByteString | ||
import pekko.{ Done, NotUsed } | ||
|
@@ -1177,11 +1177,15 @@ import scala.util.{ Failure, Success, Try } | |
|
||
import conf.multipartUploadSettings.retrySettings._ | ||
|
||
SplitAfterSize(chunkSize, chunkBufferSize)(atLeastOneByteString) | ||
.via(getChunkBuffer(chunkSize, chunkBufferSize, maxRetries)) // creates the chunks | ||
.mergeSubstreamsWithParallelism(parallelism) | ||
val source1: SubFlow[Chunk, NotUsed, Flow[ByteString, ByteString, NotUsed]#Repr, Sink[ByteString, NotUsed]] = | ||
SplitAfterSize(chunkSize, chunkBufferSize)(atLeastOneByteString) | ||
.via(getChunkBuffer(chunkSize, chunkBufferSize, maxRetries)) // creates the chunks | ||
|
||
val source2 = source1.mergeSubstreamsWithParallelism(parallelism) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The source1.mergeSubstreamsWithParallelism(parallelism)
.filter(_.size > 0)
.via(atLeastOne)
.zip(requestInfoOrUploadState(s3Location, contentType, s3Headers, initialUploadState))
// etc etc There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. the original code wouldn't compile in scala 3 and splitting the code up has helped to get it to compile There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. code no longer compiles when I start removing the types on these params There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I can see that for |
||
.filter(_.size > 0) | ||
.via(atLeastOne) | ||
|
||
source2 | ||
.zip(requestInfoOrUploadState(s3Location, contentType, s3Headers, initialUploadState)) | ||
.groupBy(parallelism, { case (_, (_, chunkIndex)) => chunkIndex % parallelism }) | ||
// Allow requests that fail with transient errors to be retried, using the already buffered chunk. | ||
|
@@ -1278,11 +1282,18 @@ import scala.util.{ Failure, Success, Try } | |
Flow[(ByteString, C)].orElse( | ||
Source.single((ByteString.empty, null.asInstanceOf[C]))) | ||
|
||
SplitAfterSizeWithContext(chunkSize)(atLeastOneByteStringAndEmptyContext) | ||
.via(getChunk(chunkBufferSize)) | ||
.mergeSubstreamsWithParallelism(parallelism) | ||
.filter { case (chunk, _) => chunk.size > 0 } | ||
.via(atLeastOne) | ||
val source1: SubFlow[(Chunk, immutable.Iterable[C]), NotUsed, Flow[(ByteString, C), (ByteString, C), | ||
NotUsed]#Repr, Sink[(ByteString, C), NotUsed]] = | ||
SplitAfterSizeWithContext(chunkSize)(atLeastOneByteStringAndEmptyContext) | ||
.via(getChunk(chunkBufferSize)) | ||
|
||
val source2: Flow[(ByteString, C), (Chunk, immutable.Iterable[C]), NotUsed] = | ||
source1 | ||
.mergeSubstreamsWithParallelism(parallelism) | ||
.filter { case (chunk, _) => chunk.size > 0 } | ||
.via(atLeastOne) | ||
|
||
source2 | ||
.zip(requestInfoOrUploadState(s3Location, contentType, s3Headers, initialUploadState)) | ||
.groupBy(parallelism, { case (_, (_, chunkIndex)) => chunkIndex % parallelism }) | ||
.map { | ||
|
@@ -1379,9 +1390,9 @@ import scala.util.{ Failure, Success, Try } | |
import mat.executionContext | ||
Sink | ||
.seq[UploadPartResponse] | ||
.mapMaterializedValue { responseFuture: Future[immutable.Seq[UploadPartResponse]] => | ||
.mapMaterializedValue { (responseFuture: Future[immutable.Seq[UploadPartResponse]]) => | ||
responseFuture | ||
.flatMap { responses: immutable.Seq[UploadPartResponse] => | ||
.flatMap { (responses: immutable.Seq[UploadPartResponse]) => | ||
val successes = responses.collect { case r: SuccessfulUploadPart => r } | ||
val failures = responses.collect { case r: FailedUploadPart => r } | ||
if (responses.isEmpty) { | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,86 @@ | ||
/* | ||
* Copyright 2015 Johan Andrén | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"); | ||
* you may not use this file except in compliance with the License. | ||
* You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.pekko.stream.connectors.s3.impl.retry | ||
|
||
import java.util.concurrent.{ ThreadLocalRandom, TimeUnit } | ||
import scala.concurrent.duration.FiniteDuration | ||
import scala.concurrent.{ ExecutionContext, Future } | ||
import scala.util.Random | ||
|
||
// copied from https://github.com/johanandren/futiles/blob/18868f252bbf5dd71d2cd0fc67e7eb39863b686a/src/main/scala/markatta/futiles/Retry.scala | ||
object Retry { | ||
|
||
private val alwaysRetry: Throwable => Boolean = _ => true | ||
|
||
/** | ||
* Evaluate a block that creates a future up to a specific number of times, if the future fails, decide about | ||
* retrying using a predicate, if it should retry an exponential back off is applied so that the retry waits longer | ||
* and longer for every retry it makes. A jitter is also added so that the exact timing of the retry isn't exactly | ||
* the same for all calls with the same backOffUnit | ||
* | ||
* Any exception in the block creating the future will also be returned as a failed future Default is to retry for | ||
* all throwables. | ||
* | ||
* Based on this wikipedia article: http://en.wikipedia.org/wiki/Truncated_binary_exponential_backoff | ||
*/ | ||
def retryWithBackOff[A]( | ||
times: Int, | ||
backOffUnit: FiniteDuration, | ||
shouldRetry: Throwable => Boolean = alwaysRetry)(fBlock: => Future[A])(implicit ec: ExecutionContext): Future[A] = | ||
try | ||
if (times <= 1) fBlock | ||
else retryWithBackOffLoop(times, 1, backOffUnit, shouldRetry)(fBlock) | ||
catch { | ||
// failure to actually create the future | ||
case x: Throwable => Future.failed(x) | ||
} | ||
|
||
private def retryWithBackOffLoop[A]( | ||
totalTimes: Int, | ||
timesTried: Int, | ||
backOffUnit: FiniteDuration, | ||
shouldRetry: Throwable => Boolean)(fBlock: => Future[A])(implicit ec: ExecutionContext): Future[A] = | ||
if (totalTimes <= timesTried) fBlock | ||
else | ||
fBlock.recoverWith { | ||
case ex: Throwable if shouldRetry(ex) => | ||
val timesTriedNow = timesTried + 1 | ||
val backOff = nextBackOff(timesTriedNow, backOffUnit) | ||
Timeouts | ||
.timeout(backOff)(()) | ||
.flatMap(_ => | ||
retryWithBackOffLoop( | ||
totalTimes, | ||
timesTriedNow, | ||
backOffUnit, | ||
shouldRetry)(fBlock)) | ||
} | ||
|
||
private def nextBackOff( | ||
tries: Int, | ||
backOffUnit: FiniteDuration): FiniteDuration = { | ||
require(tries > 0, "tries should start from 1") | ||
val rng = new Random(ThreadLocalRandom.current()) | ||
// jitter between 0.5 and 1.5 | ||
val jitter = 0.5 + rng.nextDouble() | ||
val factor = math.pow(2, tries) * jitter | ||
FiniteDuration( | ||
(backOffUnit.toMillis * factor).toLong, | ||
TimeUnit.MILLISECONDS) | ||
} | ||
|
||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,45 @@ | ||
/* | ||
* Copyright 2015 Johan Andrén | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"); | ||
* you may not use this file except in compliance with the License. | ||
* You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.pekko.stream.connectors.s3.impl.retry | ||
|
||
import java.util.{ Timer, TimerTask } | ||
import scala.concurrent.duration.FiniteDuration | ||
import scala.concurrent.{ ExecutionContext, Future, Promise } | ||
import scala.util.Try | ||
|
||
// copied from https://github.com/johanandren/futiles/blob/18868f252bbf5dd71d2cd0fc67e7eb39863b686a/src/main/scala/markatta/futiles/Timeouts.scala | ||
object Timeouts { | ||
|
||
private val timer = new Timer() | ||
|
||
/** | ||
* When ```waitFor``` has passed, evaluate ```what``` on the given execution context and complete the future | ||
*/ | ||
def timeout[A](waitFor: FiniteDuration)(what: => A)(implicit ec: ExecutionContext): Future[A] = { | ||
val promise = Promise[A]() | ||
timer.schedule(new TimerTask { | ||
override def run(): Unit = | ||
// make sure we do not block the timer thread | ||
Future { | ||
promise.complete(Try(what)) | ||
} | ||
}, | ||
waitFor.toMillis) | ||
|
||
promise.future | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I presume that this type annotation was auto generated from Intellij and if so it likely can be simplified a bit more. Ill check out this PR and see if I can improve it.