BREAKING: Refactored SDK I/O

[aajtodd]

An upcoming release of the AWS SDK for Kotlin includes redesigned I/O abstractions (specifically affecting the ByteStream type).

If you do not unwrap a ByteStream into one of the possible variants or supply a custom implementation of ByteStream this change should not affect you.

As an example if you only create and consume instances of ByteStream via convenience functions (e.g. ByteStream.fromFile(...), ByteStream.fromBytes(...), ByteStream.writeToFile(...), ByteStream.toByteArray(), etc) then you should be unaffected.

Release date

This feature will ship with the v0.19.0 release planned for 2022-12-01.

Background

When the AWS SDK for Kotlin project was first started there weren't many options for multi-platform I/O abstractions (there still aren't). There was the kotlinx-io incubator project that looked promising but wasn't ready for consumption yet. kotlinx-io was going to be an attempt to cleanup and standardize the APIs in ktor-io (the underlying I/O abstractions used by the Ktor project). We decided to take a minimal subset of the channel API from Ktor with the expectation that we would later upgrade to kotlinx-io and converge on what we believed would become a community standard.

kotlinx-io has since been deprecated which left us with an I/O API that wasn't necessarily meant to be permanent. This set of breaking changes comes from taking a second look at the underlying I/O abstractions we provide and want to commit to long term.

What's changing?

1. The SdkByteReadChannel interface has been simplified. Previously it required implementing at least 4 non-trivial read functions making it hard to wrap or provide custom implementations of. It also forced at least one additional copy between producer and consumer. The new interface requires implementing only a single read method and minimizes copies.

2. We have introduced a new ByteStream.SourceStream variant that takes an SdkSource type. SdkSource is a new blocking interface. It is simpler to implement and improves performance for file backed I/O by removing a channel middleman.

How is SdkByteReadChannel different from SdkSource?

An SdkByteReadChannel and SdkSource are two different ways of supplying a stream of bytes. SdkSource is a blocking interface whereas SdkByteReadChannel is fully asynchronous. The addition of SdkSource provides additional flexibility to how a stream of bytes can be modeled.

Both interfaces work by requiring data be written to or read from an SdkBuffer. Forcing a concrete type in the interfaces allows for copies to be minimized (because the internal representation of the type is known and can be leveraged when reading and writing data). The inspiration for this (and current backing implementation) comes from Okio. Moving data between buffers is cheap which makes layering these types easy and performant (data is frequently re-assigned between buffers rather than copied).

How to Migrate

NOTE: This would be a good time to review any custom code and update to a convenience function if one already exists that matches your use case.

Consuming ByteStream

Any code that consumes a ByteStream and unwraps it into one of the variants will need to be updated.

// pre-0.19.0 code that unwraps ByteStream
s3.getObject(request) { resp ->
    when(resp.body) { 
       is ByteStream.Buffer -> ...
       is ByteStream.ReplayableStream -> ...
       is ByteStream.OneShotStream -> ...
    }
}

The above code would need to be updated to handle the refactored type hierarchy (and differences in the underlying variants), e.g:

// 0.19.0+ code that unwraps ByteStream
s3.getObject(request) { resp ->
    when(resp.body) { 
       is ByteStream.Buffer -> ...
       is ByteStream.ChannelStream -> ...
       is ByteStream.SourceStream -> ...
    }
}

• ByteStream.ReplayableStream and ByteStream.OneShotStream have been collapsed into ByteStream.ChannelStream and updated to the new SdkByteReadChannel interface.
• Whether a ByteStream is replayable is now indicated by the isOneShot flag shared by all variants.
• ByteStream.SourceStream is a new variant that reads from the SdkSource type

Providing a ByteStream

The new ByteStream.SourceStream variant provides an additional option to consider for how to supply bytes to the SDK. We will focus on how to upgrade the channel based variant here though as that is where the breaking change will be felt most acutely.

The SdkByteChannel type has read and write halves, SdkByteReadChannel and SdkByteWriteChannel respectively. This is a single-reader, single-writer channel that is semantically the same in how it should be used before and after this change (think of it as a pipe connecting an asynchronous producer and consumer together). The only update required should be to re-implement your customized read() logic in terms of the updated interface.

As an example, let's say we implemented a custom channel that wraps an underlying channel and logs something every time data is read from it. This might look something like:

// pre-0.19.0 code that wraps SdkByteReadChannel 
class LoggingChannel(
    private val delegate: SdkByteReadChannel,
    private val onDataRead: (Int) -> Unit
): SdkByteReadChannel by delegate {

    override suspend fun readRemaining(limit: Int): ByteArray = 
        delegate.readRemaining(limit)
            .also { onDataRead(it.size) }
        
    override suspend fun readFully(sink: ByteArray, offset: Int, length: Int) {
        delegate.readFully(sink, offset, length)
        onDataRead(length)
    }

    override suspend fun readAvailable(sink: ByteArray, offset: Int, length: Int): Int =
        delegate.readAvailable(sink, offset, length)
            .also { onDataRead(it) }

    override suspend fun readAvailable(sink: ByteBuffer): Int = 
        delegate.readAvailable(sink)
            .also { onDataRead(it) }
}

Using it may look like:

...

val wrappedChannel = LoggingChannel(channel) { size -> 
    logger.trace("read $size bytes")
}

...

This can now be written as:

// 0.19.0+ code that wraps SdkByteReadChannel 
class LoggingChannel(
    private val delegate: SdkByteReadChannel,
    private val onDataRead: (Long) -> Unit
): SdkByteReadChannel by delegate {

    override suspend fun read(sink: SdkBuffer, limit: Long): Long = 
        delegate.read(sink, limit)
            .also { onDataRead(it) }
}

Additional resources

If you have any questions concerning this change, please feel free to engage with us in this discussion. If you've encountered a bug with these changes, please file an issue.

• AWS SDK Kotlin Streaming Operations Guide
• Smithy Kotlin Binary Streaming Design
• A closer look at the Okio Library

[teimatinimarin]

Hi guys!

I have a method that download an S3 Objects in bunches of 1 Mib and sent it to a channel for processing...

Something like this

           s3.getObject(it) { resp ->
                val reader = (resp.body as ByteStream.OneShotStream).readFrom()

                val buffer = ByteArray(MIB_1)
                var bytesRead = reader.readAvailable(buffer)
                while (bytesRead != -1) {
                    log.debug("Buffer was read, bytes: $bytesRead")
                    send(buffer.copyOfRange(0, bytesRead))
                    bytesRead = reader.readAvailable(buffer)
                }
            }

I tried to rewrite it, but I'm getting a NotFound for SdkBuffer class.
This is where I got stuck... Am I going the right direction? How do I create a SdkBuffer? Do I need any other dependency for my JVM project?

             s3.getObject(it) { resp ->
                val reader = (resp.body as ByteStream.ChannelStream).readFrom()

                val buffer = SkdBuffer()
                var bytesRead = reader.read(buffer, MIB_1)
                while (bytesRead >= 0) {
                    log.debug("Buffer was read, bytes: $bytesRead")
                    send(buffer.copyOfRange(0, bytesRead))
                    bytesRead = reader.read(buffer, MIB_1)
                }
            }

Thanks in advance!

[aajtodd]

You should just be able to instantiate an SdkBuffer (import aws.smithy.kotlin.runtime.io.SdkBuffer). If not please file a bug, it's possible our dependencies need looked at. If it's not working you can add the dependency manually as a workaround to your build.gradle.kts file (or the equivalent for whatever build system you use):

implementation("aws.smithy.kotlin:io:0.14.3")

NOTE: 0.14.3 is latest, you'll want to keep this in sync with the SDK version expected though.

---

You don't want to rely on the body always being a ChannelStream. You'll need to handle each ByteStream variant. I've demonstrated the two most interesting ones below. The SDK provides convenience functions to handle the most common cases (e.g. writing to a file, decoding to string, reading into ByteArray, etc). We don't currently have any reactive streams/Flow integration which would make this use case simpler I think. Feel free to upvote and comment on your use case in #612.

        // assumes myChannel is type `Channel<ByteArray>`, adapt as needed
        s3.getObject(req) { resp ->
            when(val body = resp.body) {
                // `SdkSource` is a blocking interface, should always be consumed from `Dispatchers.IO`
                is ByteStream.SourceStream -> withContext(Dispatchers.IO){
                    val source = body.readFrom()
                    val buffer = SdkBuffer()
                    while(true) {
                        val rc = source.read(buffer, MIB_1)
                        if (rc < 0) break
                        val bytes = buffer.readByteArray()
                        myChannel.send(bytes)
                    }
                }
                is ByteStream.ChannelStream -> {
                    val bodyChan = body.readFrom()
                    val buffer = SdkBuffer()
                    while(true) {
                        val rc = bodyChan.read(buffer, MIB_1)
                        if (rc < 0) break
                        val bytes = buffer.readByteArray()
                        myChannel.send(bytes)
                    }
                }
                is ByteStream.Buffer -> { ... }
                else -> error("unexpected ByteStream variant")
            }
        }

This hopefully demonstrates the meat and potatoes, you can of course abstract this into some utilities to make it cleaner.

[teimatinimarin]

Thank you so much for the answer @aajtodd and sorry for the delay in the response...
Looks like I had some mistakes in my gradle config, once I fixed that I was able to intanciate a SdkBuffer.
One question I have now is, How/When different ByteStream variants are used? Is there any documentation somewhere to read about when one variant is used over another?
Thanks in advance!

[aajtodd]

Different HTTP engines may use one vs the other depending on their I/O model and what is most efficient to map the response to. The SDK provides convenience wrappers for the most common usage such as in-memory and file I/O. We may add additional ones in the future if there are common use cases/patterns that emerge.