-
Notifications
You must be signed in to change notification settings - Fork 4.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Bulk load cdk: split the destination process stuff into multiple file…
…s in their own package (#46362)
- Loading branch information
Showing
8 changed files
with
227 additions
and
213 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
209 changes: 0 additions & 209 deletions
209
...cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/test/util/DestinationProcess.kt
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
51 changes: 51 additions & 0 deletions
51
...rc/testFixtures/kotlin/io/airbyte/cdk/test/util/destination_process/DestinationProcess.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| /* | ||
| * Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.cdk.test.util.destination_process | ||
|
|
||
| import io.airbyte.cdk.command.ConfigurationSpecification | ||
| import io.airbyte.protocol.models.v0.AirbyteMessage | ||
| import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog | ||
|
|
||
| /** | ||
| * Represents a destination process, whether running in-JVM via micronaut, or as a separate Docker | ||
| * container. The general lifecycle is: | ||
| * 1. `val dest = DestinationProcessFactory.createDestinationProcess(...)` | ||
| * 2. `launch { dest.run() }` | ||
| * 3. [sendMessage] as many times as you want | ||
| * 4. [readMessages] as needed (e.g. to check that state messages are emitted during the sync) | ||
| * 5. [shutdown] once you have no more messages to send to the destination | ||
| */ | ||
| interface DestinationProcess { | ||
| /** | ||
| * Run the destination process. Callers who want to interact with the destination should | ||
| * `launch` this method. | ||
| */ | ||
| fun run() | ||
|
|
||
| fun sendMessage(message: AirbyteMessage) | ||
|
|
||
| /** Return all messages the destination emitted since the last call to [readMessages]. */ | ||
| fun readMessages(): List<AirbyteMessage> | ||
|
|
||
| /** | ||
| * Wait for the destination to terminate, then return all messages it emitted since the last | ||
| * call to [readMessages]. | ||
| */ | ||
| fun shutdown() | ||
| } | ||
|
|
||
| enum class TestDeploymentMode { | ||
| CLOUD, | ||
| OSS | ||
| } | ||
|
|
||
| interface DestinationProcessFactory { | ||
| fun createDestinationProcess( | ||
| command: String, | ||
| config: ConfigurationSpecification? = null, | ||
| catalog: ConfiguredAirbyteCatalog? = null, | ||
| deploymentMode: TestDeploymentMode = TestDeploymentMode.OSS, | ||
| ): DestinationProcess | ||
| } |
88 changes: 88 additions & 0 deletions
88
...testFixtures/kotlin/io/airbyte/cdk/test/util/destination_process/DockerizedDestination.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| /* | ||
| * Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.cdk.test.util.destination_process | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode | ||
| import io.airbyte.protocol.models.v0.AirbyteMessage | ||
| import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog | ||
| import java.io.ByteArrayOutputStream | ||
| import java.io.InputStream | ||
|
|
||
| // TODO define a factory for this class + @Require(env = CI_master_merge) | ||
| class DockerizedDestination( | ||
| val command: String, | ||
| val config: JsonNode?, | ||
| val catalog: ConfiguredAirbyteCatalog?, | ||
| ) : DestinationProcess { | ||
| override fun run() { | ||
| TODO("launch a docker container") | ||
| } | ||
|
|
||
| override fun sendMessage(message: AirbyteMessage) { | ||
| // push a message to the docker process' stdin | ||
| TODO("Not yet implemented") | ||
| } | ||
|
|
||
| override fun readMessages(): List<AirbyteMessage> { | ||
| // read everything from the process' stdout | ||
| TODO("Not yet implemented") | ||
| } | ||
|
|
||
| override fun shutdown() { | ||
| // close stdin, wait until process exits | ||
| TODO("Not yet implemented") | ||
| } | ||
| } | ||
|
|
||
| // This is currently unused, but we'll need it for the Docker version. | ||
| // it exists right now b/c I wrote it prior to the CliRunner retooling. | ||
| /** | ||
| * There doesn't seem to be a built-in equivalent to this? Scanner and BufferedReader both have | ||
| * `hasNextLine` methods which block until the stream has data to read, which we don't want to do. | ||
| * | ||
| * This class simply buffers the next line in-memory until it reaches a newline or EOF. | ||
| */ | ||
| private class LazyInputStreamReader(private val input: InputStream) { | ||
| private val buffer: ByteArrayOutputStream = ByteArrayOutputStream() | ||
| private var eof = false | ||
|
|
||
| /** | ||
| * Returns the next line of data, or null if no line is available. Doesn't block if the | ||
| * inputstream has no data. | ||
| */ | ||
| fun nextLine(): MaybeLine { | ||
| if (eof) { | ||
| return NoLine.EOF | ||
| } | ||
| while (input.available() != 0) { | ||
| when (val read = input.read()) { | ||
| -1 -> { | ||
| eof = true | ||
| val line = Line(buffer.toByteArray().toString(Charsets.UTF_8)) | ||
| buffer.reset() | ||
| return line | ||
| } | ||
| '\n'.code -> { | ||
| val bytes = buffer.toByteArray() | ||
| buffer.reset() | ||
| return Line(bytes.toString(Charsets.UTF_8)) | ||
| } | ||
| else -> { | ||
| buffer.write(read) | ||
| } | ||
| } | ||
| } | ||
| return NoLine.NOT_YET_AVAILABLE | ||
| } | ||
|
|
||
| companion object { | ||
| interface MaybeLine | ||
| enum class NoLine : MaybeLine { | ||
| EOF, | ||
| NOT_YET_AVAILABLE | ||
| } | ||
| data class Line(val line: String) : MaybeLine | ||
| } | ||
| } |
Oops, something went wrong.