Error using reactive streams toStreamBuffered with newer versions of the AWS SDK #3359
Comments
|
Thanks for reporting!
Sure, are you on JDK 9+? In that case, try using the converters in |
|
Thanks for your reply. Sorry about this but I've got my wires crossed in the original report. The download method with the publisher created by def upload(bucket: String, key: String, contentLength: Long, publisher: Publisher[ByteBuffer]): IO[CompletedUpload] = {
val putObjectRequest = PutObjectRequest.builder
// usual put object config
.build
val requestBody = AsyncRequestBody.fromPublisher(publisher)
val response: Upload = transferManager.upload(
UploadRequest.builder().requestBody(requestBody).putObjectRequest(putObjectRequest).build()
)
IO.fromCompletableFuture(IO(response.completionFuture()))
}I'm then calling it like: Stream.eval(IO(ByteBuffer.wrap("test".getBytes)))
.toUnicastPublisher
.use { pub =>
s3.upload("uploadBucket", "key", 4, pub)
}Then I'm getting the exception reported above. I've tried to use the def streamToPublisher(stream: Stream[IO, ByteBuffer]): Resource[IO, Publisher[ByteBuffer]] =
fs2.interop.flow.toPublisher(stream).map(pub => FlowAdapters.toPublisher[ByteBuffer](pub))I'm still getting the same exception when I try to upload with a |
Great, thanks for trying that! Can you please share the stack trace? |
|
Can you try removing FS2 completely, and using this custom import java.nio.ByteBuffer
import java.util.concurrent.Flow.Publisher
import java.util.concurrent.Flow.Subscriber
import java.util.concurrent.Flow.Subscription
val publisher = new Publisher[ByteBuffer] {
def subscribe(subscriber: Subscriber[? >: ByteBuffer]): Unit = {
subscriber.onSubscribe {
new Subscription {
def request(n: Long): Unit = {
subscriber.onNext(ByteBuffer.wrap("test".getBytes))
subscriber.onComplete()
}
def cancel(): Unit = ()
}
}
}
} |
|
If I create the publisher like that and then use the I don't know if this helps but it does work using this https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/utils/async/SimplePublisher.html val publisher = new SimplePublisher[ByteBuffer]
publisher.send(ByteBuffer.wrap("test".getBytes))
val requestBody = AsyncRequestBody.fromPublisher(publisher)
transferManager.upload(UploadRequest.builder.requestBody(requestBody).build()).completionFuture().get
publisher.complete() |
Ah, that is very helpful ... so it still suggests there's something wrong with how we are doing interop with FS2 🤔 Ok, I think we will need to dig into this more to understand what the problem is. Is there a possibility that you can make a self-contained example? i.e. something that can run against localstack, works with the SimplePublisher, but doesn't work with FS2. That would be really helpful. |
|
@MancunianSam I discussed this issue with @BalmungSan and we think we know what the problem is. You can hold on the reproducer for now, we'll put up our fix and see if it works for you. Thanks! |
|
@MancunianSam can you try resolvers += "s01-oss-sonatype-org-snapshots" at "https://s01.oss.sonatype.org/content/repositories/snapshots" |
|
@armanbilge I've tried converting the stream to a publisher using |
I have a method which uses the S3 transfer manager to return a publisher:
I'm then calling this and calling
toStreamBuffereds3.download(downloadBucket, downloadBucketKey) .flatMap( _.toStreamBuffered[IO](10 * 1024)With recent (I'm not 100% sure which one) versions of the AWS SDK, I'm now getting the following error:
SDK version 2.20.69 is working but 2.21.42 is not. The only recent change in the SDK I can see is aws/aws-sdk-java-v2#4308
If I revert this change locally and run my code again, it's working so it looks like that change to the AWS SDK is at least partly responsible.
Do you know what is causing this error or if there are any workarounds I can use?
Thanks
The text was updated successfully, but these errors were encountered: