feat(rt): event stream framing support #320

aajtodd · 2021-09-24T19:03:27Z

Issue #

N/A

Description of changes

Implements message encode/decode for the vnd.amazon.event-stream content-type.
Some churn from SdkBuffer -> SdkByteBuffer and related API changes

Scope

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

ianbotsf

Looks most good, well done! The biggest question blocking my approval is the interpretation of length fields as signed integers.

ianbotsf · 2021-09-27T16:05:09Z

...time/protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/Header.kt

+         * Read an encoded header from the [buffer]
+         */
+        public fun decode(buffer: Buffer): Header {
+            check(buffer.readRemaining >= MIN_HEADER_LEN.toULong()) { "Invalid frame header; require at least 2 bytes" }


Nit: Reuse the constant in the error message:

check(buffer.readRemaining >= MIN_HEADER_LEN.toULong()) { "Invalid frame header; require at least $MIN_HEADER_LEN bytes" }

ianbotsf · 2021-09-27T16:11:37Z

...protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/HeaderValue.kt

+        fun fromTypeId(value: Byte): HeaderType {
+            return when (value.toInt()) {
+                0 -> TRUE
+                1 -> FALSE
+                2 -> BYTE
+                3 -> INT16
+                4 -> INT32
+                5 -> INT64
+                6 -> BYTE_ARRAY
+                7 -> STRING
+                8 -> TIMESTAMP
+                9 -> UUID
+                else -> throw IllegalArgumentException("Unknown HeaderType: $value")
+            }
+        }


Comment: We can avoid duplication by using values instead:

fun fromTypeId(value: Byte): HeaderType = requireNotNull(values().find { it.value == value }) { "Unknown HeaderType: $value" }

ianbotsf · 2021-09-27T16:19:48Z

...protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/HeaderValue.kt

+        is ByteArray -> {
+            dest.writeHeader(HeaderType.BYTE_ARRAY)
+            check(value.size in Short.MIN_VALUE..Short.MAX_VALUE) { "HeaderValue ByteArray too long" }
+            dest.writeShort(value.size.toShort())
+            dest.writeFully(value)
+        }


Question: Is any recovery possible at this point? If so, we should probably check the array size before writing the header type.

I'm not sure there is any recovery possible here but agree that the check belongs before starting the write.

ianbotsf · 2021-09-27T16:21:42Z

...protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/HeaderValue.kt

+        }
+        is ByteArray -> {
+            dest.writeHeader(HeaderType.BYTE_ARRAY)
+            check(value.size in Short.MIN_VALUE..Short.MAX_VALUE) { "HeaderValue ByteArray too long" }


Nit: The lower bound here should probably be 0 instead of Short.MIN_VALUE.

ianbotsf · 2021-09-27T16:34:04Z

...protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/HeaderValue.kt

+            check(value.size in Short.MIN_VALUE..Short.MAX_VALUE) { "HeaderValue ByteArray too long" }
+            dest.writeShort(value.size.toShort())


Question: It's not clear to me from reading the event streams description that the value byte length is meant to be a signed 16-bit integer. Do we have any other sources that may indicate whether 32KB is the max size for a blob/string? Or are 64K values possible?

I agree. I think the header value lengths and overall lengths in the prelude ought to be unsigned. Good call out, will fix.

ianbotsf · 2021-09-27T16:37:53Z

...protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/HeaderValue.kt

+@Suppress("NOTHING_TO_INLINE")
+private inline fun MutableBuffer.writeHeader(headerType: HeaderType) = writeByte(headerType.value)


Question: Why inline this function if the compiler thinks there's nothing to inline?

I'm smarter of course...will remove inline

ianbotsf · 2021-09-27T16:49:08Z

...ime/protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/Message.kt

+            // read headers
+            var headerBytesConsumed = 0UL
+            while (headerBytesConsumed < prelude.headersLength.toULong()) {
+                val start = messageBuffer.readPosition
+                val header = Header.decode(messageBuffer)
+                headerBytesConsumed += messageBuffer.readPosition - start
+                message.addHeader(header)
+            }


Question: Should we verify when we're done parsing headers that headerBytesConsumed == prelude.headersLength.toULong()? It seems like that could provide an early warning that a message is malformed.

Seems sensible, will update

ianbotsf · 2021-09-27T16:51:46Z

...ime/protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/Message.kt

+     * Encode a message to the [dest] buffer
+     */
+    public fun encode(dest: MutableBuffer) {
+        val encodedHeaders = SdkByteBuffer(16u)


Question: Why 16u?

No reason, had to pick something. Any suggestion on what a sensible default would be?

In the absence of a reason to pick something else, I'd say 0u. In fact, upon further examination of smithy-kotlin#482, I'd suggest adding a default value of 0u to the secondary constructor.

kggilmer

Very clean, nice work.

kggilmer · 2021-09-29T23:17:00Z

aws-runtime/protocols/event-stream/build.gradle.kts

+ */
+
+description = "Support for the vnd.amazon.event-stream content type"
+extra["displayName"] = "AWS :: SDK :: Kotlin :: Event Stream"


nit

seems like it would be more descriptive if the display name has "Protocols" after "Kotlin", also for json and xml.

kggilmer · 2021-09-29T23:28:10Z

...protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/HeaderValue.kt

+        is String -> {
+            dest.writeHeader(HeaderType.STRING)
+            val bytes = value.encodeToByteArray()
+            check(bytes.size in Short.MIN_VALUE..Short.MAX_VALUE) { "HeaderValue String too long" }


Ian's point above applies here as well

Nit: The lower bound here should probably be 0 instead of Short.MIN_VALUE.

kggilmer · 2021-09-29T23:30:48Z

...protocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/HeaderValue.kt

+                HeaderType.BYTE_ARRAY, HeaderType.STRING -> {
+                    val len = buffer.readShort()
+                    if (len < 0 || buffer.readRemaining < len.toULong()) {
+                        throw IllegalStateException("Invalid HeaderValue; type=$type, len=$len")


suggestion

Seems like we may also want to know what buffer.readRemaining was at the time of check if this is thrown

kggilmer · 2021-09-30T00:08:29Z

...rotocols/event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/FrameDecoder.kt

+     * The decoder will consume the prelude when enough data is available. When it is invoked with enough
+     * data it will consume the remaining message bytes.
+     */
+    public fun decodeFrame(buffer: Buffer): DecodedFrame {


question

I gather that the intent here that if not enough bytes are in the buffer to decode the next frame that DecodedFrame.Incomplete is returned. Since there is only one valid type that can be returned Message and no state is captured by Incomplete would it not be simpler to represent this case with null instead?

Ya we could probably do that, I'll take a look and see. I'm not even sure we'll keep FrameDecoder in it's current state depending on how it all gets wired up. I was sort of following along with both the Java and Rust implementations that have similar abstractions.

aajtodd added 8 commits September 21, 2021 17:10

feat(rt): bootstrap event stream frame encode/decode

b7b9091

add todo placeholder

5154cbb

cleanup

f5faf70

more cleanup

48a9761

use markBytesReadable

d61b770

add simple frame decoder

dd60d73

refactor: fallout from SdkBuffer changes

251665d

remove commented out code

0d7fb46

aajtodd mentioned this pull request Sep 24, 2021

feat(rt)!: refactor SdkBuffer to implement immutable and mutable interfaces; use unsigned types for cursors smithy-lang/smithy-kotlin#482

Merged

aajtodd requested review from ianbotsf and kggilmer September 24, 2021 19:09

ianbotsf reviewed Sep 27, 2021

View reviewed changes

kggilmer approved these changes Sep 30, 2021

View reviewed changes

aajtodd added 7 commits October 1, 2021 13:03

Merge remote-tracking branch 'origin/main' into feat-event-stream

0a20d82

signed -> unsigned

ba6a335

get rid of concrete parameter

f5633a5

remove DecodedFrame

af88955

Merge remote-tracking branch 'origin/main' into feat-event-stream

bc46380

SdkByteBuffer updates

e2ec762

force ci

5e8a62e

aajtodd merged commit cf91733 into main Oct 5, 2021

aajtodd deleted the feat-event-stream branch October 5, 2021 16:02

		check(value.size in Short.MIN_VALUE..Short.MAX_VALUE) { "HeaderValue ByteArray too long" }
		dest.writeShort(value.size.toShort())

		@Suppress("NOTHING_TO_INLINE")
		private inline fun MutableBuffer.writeHeader(headerType: HeaderType) = writeByte(headerType.value)

feat(rt): event stream framing support #320

feat(rt): event stream framing support #320

Uh oh!

Conversation

aajtodd commented Sep 24, 2021

Issue #

Description of changes

Scope

Uh oh!

ianbotsf left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kggilmer left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants