Bootstrap event streams

[aajtodd]

Issue #

Description of changes

WIP branch for implementing event streams.

This PR mostly covers generating serialization and deserialization for the vnd.amazon.event-stream content type.

• feat: add chunked signing interfaces required to sign each event (could also be used to sign chunked payloads of any type really)
• refactor: update event stream encode/decode to use Flow directly to simplify the generated code

Example Codegen

Example for transcribe streaming StartStreamTranscription operation (which has both an input and output stream)

Serialize:

private suspend fun serializeStartStreamTranscriptionOperationBody(context: ExecutionContext, input: StartStreamTranscriptionRequest): HttpBody {
    val stream = input.audioStream ?: return HttpBody.Empty
    val signingConfig = context.newEventStreamSigningConfig()
    val messages = stream.map {
            encodeStartStreamTranscriptionAudioStreamEventMessage(it)
        }
        .sign(context, signingConfig)
        .encode()

    return messages.asEventStreamHttpBody()
}

private fun encodeStartStreamTranscriptionAudioStreamEventMessage(input: AudioStream): Message = buildMessage {
    addHeader(":message-type", HeaderValue.String("event"))
    when(input) {
        is AudioStream.AudioEvent -> {
            addHeader(":event-type", HeaderValue.String("AudioEvent"))
            addHeader(":content-type", HeaderValue.String("application/octet-stream"))
            payload = input.value.audioChunk
        }
        is AudioStream.SdkUnknown -> error("cannot serialize the unknown event type!")
    }
}

Deserialize:

private suspend fun deserializeStartStreamTranscriptionOperationBody(builder: StartStreamTranscriptionResponse.Builder, body: HttpBody) {
    val chan = body.toSdkByteReadChannel() ?: return
    val events = decodeFrames(chan)
        .map { message ->
            when(val mt = message.type()) {
                is MessageType.Event -> when(mt.shapeType) {
                    "TranscriptEvent" -> {
                        val e = deserializeTranscriptEventPayload(message.payload)
                        TranscriptResultStream.TranscriptEvent(e)
                    }
                    else -> TranscriptResultStream.SdkUnknown
                }
                is MessageType.Exception -> when(mt.shapeType){
                    "BadRequestException" -> {
                        val err = deserializeBadRequestExceptionPayload(message.payload)
                        throw err
                    }
                    "LimitExceededException" -> {
                        val err = deserializeLimitExceededExceptionPayload(message.payload)
                        throw err
                    }
                    "InternalFailureException" -> {
                        val err = deserializeInternalFailureExceptionPayload(message.payload)
                        throw err
                    }
                    "ConflictException" -> {
                        val err = deserializeConflictExceptionPayload(message.payload)
                        throw err
                    }
                    "ServiceUnavailableException" -> {
                        val err = deserializeServiceUnavailableExceptionPayload(message.payload)
                        throw err
                    }
                    else -> throw TranscribeStreamingException("error processing event stream, unrecognized errorType: ${mt.shapeType}")
                }
                is MessageType.Error -> throw TranscribeStreamingException("error processing event stream: errorCode=${mt.errorCode}; message=${mt.message}")
                is MessageType.SdkUnknown -> throw ClientException("unrecognized event stream message `:message-type`: ${mt.messageType}")
            }
        }
    builder.transcriptResultStream = events
}

Current state

I've been able to test deserialization (output event streams) for S3 select object and it works fine. Serialization has been tested as part of some local integration testing but the event stream signing needs a bit more work before I can send a real request with an input event stream (e.g. transcribe streaming).

Left to do (follow on PRs)

• comb through fixme/todo's and triage
• we need to populate quite a bit more into the execution context for signing. In particular signing of the initial request vs the event stream needs looked at.
• I have some generated integration tests that I'm working on that aren't quite ready that I'll PR separately. Similar to pagination and waiter tests.
• more testing of real service calls
• provide new examples and updated developer guide

I will probably leave RPC event streams as a follow up to this work. Basically an event stream can send/receive an initial request/response. That initial request/response can be tied to the HTTP request OR be sent as the first message on the event stream. The former is what we have now and how S3 and transcribe streaming work. The latter will require a bit more investigation and abstraction to figure out how to implement them. To my knowledge QLDB and kinesis leverage RPC event streams

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

[ianbotsf]

Comment: Yes.

[aajtodd]

I'll do it either in one of the follow on PRs or separately after event streams is complete

[ianbotsf]

Question: These attributes are intended to be used by end-users? If so, can we offer better guidance than "This is not a common option"? Maybe something like "Changing this option is only required in advanced use cases".

[aajtodd]

Sure we can change it. I'm not convinced all of these are intended for use by end-users but I also don't see a reason why they couldn't be. We need to spend some time looking at per/operation config and interceptors. Originally the execution context was envisioned to be something that end users may influence directly but currently they have no way of accessing it.

[ianbotsf]

Question: Does AwsSigner.signRequest still need to exist?

[aajtodd]

Are you asking should it still exist in aws-crt-kotlin?

Probably not if the sign() function returning AwsSigningResult supersedes it.

[ianbotsf]

Yeah, asking about the implementation in aws-crt-kotlin specifically. I didn't see any remaining usage after this removal.

[ianbotsf]

Question: Am I reading correctly that this will collect every message (i.e., the entire event stream) and then emit a new stream with signed messages? Isn't that wasteful and unnecessarily greedy? Could/should we map instead?

[aajtodd]

This is inside of a flow collector so collection is expected and how many of the provided transforms are implemented internally.

Note the = flow { on line 31. We are creating a new flow that processes elements from the current one. Collection will not happen until the returned flow is collected

[ianbotsf]

Question: We purposely ignore these values in AwsSigningConfig? Do any services (e.g., S3) need special values here?

[aajtodd]

We purposely ignore these values in AwsSigningConfig?

I don't follow...

Do any services (e.g., S3) need special values here?

Unknown. There is still some work left before I can send an input stream. I've validated most of the message serialization locally with some integration tests (that I will add to future PR after I clean them up). Signing is an area that may change in a future PR once I understand what we need to do better.

[ianbotsf]

We purposely ignore these values in AwsSigningConfig?

I don't follow...

Sorry, disregard. I misread this code.

[ianbotsf]

Question: Why is a Job necessary here? If invokeOnCompletion waits for the job to complete anyway then it seems like this could be replaced with simply invoking collect directly in this method.

[aajtodd]

We don't want to collect directly in this method. We are processing a stream and it needs to be done asynchronously since it is of unknown size. It also needs to happen after we have sent the initial request (which now that I look at the generated code may need tweaked a bit, will find out when I get it actually working on a real service).

We launch a coroutine to do this (which returns Job). invokeOnCompletion does not wait for the Job to complete, it simply wires up a completion handler that will fire when the job transitions to the Complete state (which will happen for either success or failed/cancelled jobs).

[ianbotsf]

Question: Can be removed?

[ianbotsf]

Style: I suggest we sort these lexicographically.

[ianbotsf]

Question: Why do we still need this integration?

[aajtodd]

We may or may not depending on what we do with RPC based event streams. I need to look into how much work it will be to support them (IIRC Kinesis and QLDB are the only ones that use this whereas transcribe streaming and S3 use the HTTP request/response for the initial request/response of the event stream)

[ianbotsf]

Style: This indentation is non-standard and the encoding function doesn't need a full lambda body. I suggest:

writer.withBlock("val messages = stream", "") {
    write(".#T(::#T)", RuntimeTypes.KotlinxCoroutines.Flow.map, encodeFn)
    write(".#T(context, signingConfig)", AwsRuntimeTypes.AwsEventStream.sign)
    write(".#T()", AwsRuntimeTypes.AwsEventStream.encode)
}

Yielding:

val messages = stream
    .map(::encodeFooEventMessage)
    .sign(context, signingConfig)
    .encode()

[sonarqubecloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
20 Code Smells

No Coverage information
0.0% Duplication