From abada0bf4bc69dd5fc8420355ee79d643566e456 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sun, 26 Dec 2021 22:55:06 +0100 Subject: [PATCH] mucking about trying to get camel-kafka-connect websockets working --- gradle/libs.versions.toml | 6 ++- .../src/main/typescript/package-lock.json | 2 +- .../src/main/typescript/package.json | 12 ++--- .../main/typescript/schema/events-schema.ts | 3 ++ .../dev/adamko/kafkatorio/processor/main.kt | 5 +- .../events/schema/factorioEvents.kt | 27 +++++++---- .../src/docker-compose.yml | 7 ++- modules/infra-kafka-connect/build.gradle.kts | 5 +- .../src/main/docker/docker-compose.yml | 3 +- .../kafkaconnect/websocketClient.kt | 46 +++++++++++++++++++ .../configs/websocket-sink-lua-topics.json5 | 12 ++++- modules/infra-kafka-pipe/build.gradle.kts | 19 ++++++++ .../src/kafka-pipe.Dockerfile | 2 +- modules/infra-kafka-pipe/src/kafka-pipe.sh | 2 +- 14 files changed, 122 insertions(+), 29 deletions(-) create mode 100644 modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/websocketClient.kt diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 4b09960..877b863 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -58,7 +58,8 @@ kotlinx-html = { group = "org.jetbrains.kotlinx", name = "kotlinx- http4k-bom = { group = "org.http4k", name = "http4k-bom", version.ref = "http4k" } http4k-core = { group = "org.http4k", name = "http4k-core" } http4k-contract = { group = "org.http4k", name = "http4k-contract" } -http4k-client = { group = "org.http4k", name = "http4k-client-okhttp" } +http4k-client-okhttp = { group = "org.http4k", name = "http4k-client-okhttp" } +http4k-client-ws = { group = "org.http4k", name = "http4k-client-websocket" } http4k-server = { group = "org.http4k", name = "http4k-server-undertow" } http4k-cloudnative = { group = "org.http4k", name = "http4k-cloudnative" } http4k-format-json = { group = "org.http4k", name = "http4k-format-jackson" } @@ -146,7 +147,8 @@ junit-runtime = [ http4k = [ "http4k-core", "http4k-contract", - "http4k-client", + "http4k-client-okhttp", + "http4k-client-ws", "http4k-server", # "http4k-kotlinx", "http4k-cloudnative", diff --git a/modules/events-mod/src/main/typescript/package-lock.json b/modules/events-mod/src/main/typescript/package-lock.json index fd6ba9b..32fb6ef 100644 --- a/modules/events-mod/src/main/typescript/package-lock.json +++ b/modules/events-mod/src/main/typescript/package-lock.json @@ -1,6 +1,6 @@ { "name": "kafkatorio-events-mod", - "version": "0.1.3", + "version": "0.1.5", "lockfileVersion": 1, "requires": true, "dependencies": { diff --git a/modules/events-mod/src/main/typescript/package.json b/modules/events-mod/src/main/typescript/package.json index 7a9829d..966ac89 100644 --- a/modules/events-mod/src/main/typescript/package.json +++ b/modules/events-mod/src/main/typescript/package.json @@ -9,8 +9,7 @@ "engines": { "node": ">=10.13.0" }, - "devDependencies": { - }, + "devDependencies": {}, "dependencies": { "lua-types": "^2.11.0", "@types/node": "^12.0.8", @@ -19,10 +18,7 @@ "typescript": "4.5.4" }, "license": "UNLICENSED", - "peerDependencies": { - }, - "optionalDependencies": { - }, - "bundledDependencies": [ - ] + "peerDependencies": {}, + "optionalDependencies": {}, + "bundledDependencies": [] } diff --git a/modules/events-mod/src/main/typescript/schema/events-schema.ts b/modules/events-mod/src/main/typescript/schema/events-schema.ts index 27a14dd..fccd740 100644 --- a/modules/events-mod/src/main/typescript/schema/events-schema.ts +++ b/modules/events-mod/src/main/typescript/schema/events-schema.ts @@ -20,6 +20,7 @@ interface PlayerData extends FactorioObjectData { associated_characters_unit_numbers: uint[]; character_unit_number: uint | null; name: string; + object_name: string; position: PositionData; } @@ -28,6 +29,7 @@ interface EntityData extends FactorioObjectData { health: double | null; health_ratio: double; name: string; + object_name: string; player_index: uint | null; position: PositionData; surface_index: int; @@ -39,4 +41,5 @@ interface SurfaceData extends FactorioObjectData { daytime: double; index: uint; name: string; + object_name: string; } \ No newline at end of file diff --git a/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt index a6b5ff1..2f899bf 100644 --- a/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt +++ b/modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt @@ -56,7 +56,10 @@ fun main() { // KXS.asA>(value) } .to( - { key, value, recordContext -> value.data.objectName }, + { key, value, recordContext -> +// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}") + value.data.objectName + }, Produced.with(Serdes.String(), JsonSerdes) ) diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt index 0371d19..f1692b7 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioEvents.kt @@ -1,9 +1,9 @@ package dev.adamko.kafkatorio.events.schema -//import kotlin.reflect.full.starProjectedType import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable +import kotlinx.serialization.Transient import kotlinx.serialization.json.JsonClassDiscriminator @@ -34,10 +34,12 @@ const val FactorioObjectDataDiscriminatorKey: String = "object_name" @JsonClassDiscriminator(FactorioObjectDataDiscriminatorKey) sealed class FactorioObjectData { + abstract val objectName: String + // workaround - https://github.com/Kotlin/kotlinx.serialization/issues/1664 - val objectName: String by lazy { - serializer().descriptor.serialName - } +// val objectName: String by lazy { +// serializer().descriptor.serialName +// } // val objectName: String by lazy { // kotlinx.serialization.serializer(this::class.starProjectedType).descriptor.serialName @@ -45,7 +47,7 @@ sealed class FactorioObjectData { // alternative: find the @SerialName annotation // Again it must be delegated so there's no backing field and kxs ignores it -// val objectNameByAnnotation: String by lazy { +// val objectName: String by lazy { // requireNotNull(this::class.findAnnotation()?.value) { // "Couldn't find @SerialName for ${this::class}!" // } @@ -64,7 +66,10 @@ data class PlayerData( val name: String, @SerialName("position") val position: PositionData, -) : FactorioObjectData() +) : FactorioObjectData() { + @Transient + override val objectName: String= "LuaPlayer" +} @Serializable @SerialName("LuaEntity") @@ -88,7 +93,10 @@ data class EntityData( @SerialName("player_index") val playerIndex: UInt? = null, -) : FactorioObjectData() +) : FactorioObjectData() { + @Transient + override val objectName: String= "LuaEntity" +} @Serializable @SerialName("LuaSurface") @@ -99,7 +107,10 @@ data class SurfaceData( val index: UInt, @SerialName("name") val name: String, -) : FactorioObjectData() +) : FactorioObjectData() { + @Transient + override val objectName: String="LuaSurface" +} @Serializable data class PositionData( diff --git a/modules/infra-kafka-cluster/src/docker-compose.yml b/modules/infra-kafka-cluster/src/docker-compose.yml index 8a4e714..600e071 100644 --- a/modules/infra-kafka-cluster/src/docker-compose.yml +++ b/modules/infra-kafka-cluster/src/docker-compose.yml @@ -28,8 +28,11 @@ services: ports: - "8080:8080" environment: - - KAFKA_CLUSTERS_0_NAME=${COMPOSE_PROJECT_NAME} - - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:29092 + KAFKA_CLUSTERS_0_NAME: "${COMPOSE_PROJECT_NAME}" + KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: "kafka:29092" + + KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: "kafkatorio-connect" + KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: "http://connect:8083" # - KAFKA_CLUSTERS_0_ZOOKEEPER=kafka:9093 # kafka-connect: diff --git a/modules/infra-kafka-connect/build.gradle.kts b/modules/infra-kafka-connect/build.gradle.kts index ff2da04..871f9c1 100644 --- a/modules/infra-kafka-connect/build.gradle.kts +++ b/modules/infra-kafka-connect/build.gradle.kts @@ -24,13 +24,14 @@ fun DependencyHandlerScope.kafkaConnector( } dependencies { + kotlin("reflect") + kafkaConnector("camel-ahc-ws-kafka-connector") implementation("com.github.adamko-dev:json5-kotlin:2.0.3") implementation(platform(libs.http4k.bom)) - implementation("org.http4k:http4k-core") - implementation("org.http4k:http4k-client-okhttp") + implementation(libs.bundles.http4k) implementation(platform(libs.kotlinx.serialization.bom)) implementation("org.jetbrains.kotlinx:kotlinx-serialization-core") diff --git a/modules/infra-kafka-connect/src/main/docker/docker-compose.yml b/modules/infra-kafka-connect/src/main/docker/docker-compose.yml index f44f3af..f152ae6 100644 --- a/modules/infra-kafka-connect/src/main/docker/docker-compose.yml +++ b/modules/infra-kafka-connect/src/main/docker/docker-compose.yml @@ -27,7 +27,8 @@ services: - "kafkatorio-nw" environment: CONNECT_BOOTSTRAP_SERVERS: "kafka:29092" - CONNECT_REST_ADVERTISED_HOST_NAME: ${CONNECTOR_ID} +# CONNECT_REST_ADVERTISED_HOST_NAME: ${CONNECTOR_ID} + CONNECT_REST_ADVERTISED_HOST_NAME: connect # CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://${SCHEMA_REGISTRY}:8081" # CONNECT_REST_PORT: 28083 CONNECT_REST_PORT: 8083 diff --git a/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/websocketClient.kt b/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/websocketClient.kt new file mode 100644 index 0000000..a5f6609 --- /dev/null +++ b/modules/infra-kafka-connect/src/main/kotlin/dev/adamko/kafkatorio/kafkaconnect/websocketClient.kt @@ -0,0 +1,46 @@ +package dev.adamko.kafkatorio.kafkaconnect + +import java.time.Duration +import org.http4k.client.WebsocketClient +import org.http4k.core.Uri +import org.http4k.websocket.WsMessage + +fun main() { + +// val client = WebsocketClient.blocking( +// uri = Uri.of("ws://localhost:8095/ws"), +//// autoReconnection = true, +//// timeout = Duration.ofSeconds(30) +// ) +//// client.send(WsMessage("server sent on connection")) +// +// client.received() +// .take(100) +// .forEachIndexed { index, wsMessage -> +// println("[$index] ${wsMessage.bodyString()}") +// } + + + val nonBlockingClient = WebsocketClient.nonBlocking( + uri = Uri.of("ws://localhost:8095/ws"), + onError = { it.printStackTrace() }, + onConnect = { ws -> + println("onConnect, ws.upgradeRequest ${ws.upgradeRequest}") + ws.send(WsMessage("client sent on connection")) + } + ) + + nonBlockingClient.onMessage { msg -> print(msg.bodyString()) } + + + nonBlockingClient.onClose { + println("non-blocking client closing") + } + + + nonBlockingClient.onError { it.printStackTrace() } + + + Thread.sleep(Duration.ofMinutes(1).toMillis()) + +} diff --git a/modules/infra-kafka-connect/src/main/resources/configs/websocket-sink-lua-topics.json5 b/modules/infra-kafka-connect/src/main/resources/configs/websocket-sink-lua-topics.json5 index 09963ad..d11fdcc 100644 --- a/modules/infra-kafka-connect/src/main/resources/configs/websocket-sink-lua-topics.json5 +++ b/modules/infra-kafka-connect/src/main/resources/configs/websocket-sink-lua-topics.json5 @@ -14,7 +14,7 @@ // mandatory properties (for a complete properties list see the connector documentation): // The URI to use such as http://hostname:port/path - "camel.sink.path.httpUri": "http://localhost:8095/ws", + "camel.sink.path.httpUri": "localhost:8095/ws", // If the option is true, then the Exchange.HTTP_URI header is ignored, and use the endpoint's URI for request. You may also set the throwExceptionOnFailure to be false to let the AhcProducer send all the fault response back. // camel.sink.endpoint.bridgeEndpoint= // The initial in-memory buffer size used when transferring data between Camel and AHC Client. @@ -29,8 +29,16 @@ // camel.sink.endpoint.connectionClose= // Configure a cookie handler to maintain a HTTP session // camel.sink.endpoint.cookieHandler= - // Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing. + + // Whether the producer should be started lazy (on the first message). By starting lazy you can + // use this to allow CamelContext and routes to startup in situations where a producer + // may otherwise fail during starting and cause the route to fail being started. By deferring + // this startup to be lazy then the startup failure can be handled during routing messages via + // Camel's routing error handlers. Beware that when the first message is processed then creating + // and starting the producer may take a little time and prolong the total processing time of the + // processing. // camel.sink.endpoint.lazyStartProducer= + // To enable streaming to send data as multiple text fragments. // camel.sink.endpoint.useStreaming= // To use a custom AhcBinding which allows to control how to bind between AHC and Camel. diff --git a/modules/infra-kafka-pipe/build.gradle.kts b/modules/infra-kafka-pipe/build.gradle.kts index 2991681..d3d3fb0 100644 --- a/modules/infra-kafka-pipe/build.gradle.kts +++ b/modules/infra-kafka-pipe/build.gradle.kts @@ -1,9 +1,28 @@ +import org.jetbrains.kotlin.util.parseSpaceSeparatedArgs + plugins { id("dev.adamko.kafkatorio.infra.docker-compose") } description = "Send events from a Factorio server to a Kafka topic " + +val dockerSrcDir: Directory by extra + tasks.dockerUp { dependsOn(":modules:infra-kafka-cluster:dockerUp") } + +val dockerBuildKafkaPipe by tasks.registering(Exec::class) { + group = "docker-compose" + + logging.captureStandardOutput(LogLevel.LIFECYCLE) + + dependsOn(tasks.dockerEnv) + + inputs.dir(dockerSrcDir) + + workingDir = dockerSrcDir.asFile + commandLine = parseSpaceSeparatedArgs(""" docker-compose build kafka-pipe """) +} +tasks.assemble { dependsOn(dockerBuildKafkaPipe) } diff --git a/modules/infra-kafka-pipe/src/kafka-pipe.Dockerfile b/modules/infra-kafka-pipe/src/kafka-pipe.Dockerfile index edcb96c..02f8881 100644 --- a/modules/infra-kafka-pipe/src/kafka-pipe.Dockerfile +++ b/modules/infra-kafka-pipe/src/kafka-pipe.Dockerfile @@ -8,7 +8,7 @@ ENV KAFKA_HOST="kafka" USER root RUN apk add --update \ curl \ - kafkacat + kcat #USER rootless COPY kafka-pipe.sh kafka-pipe.sh diff --git a/modules/infra-kafka-pipe/src/kafka-pipe.sh b/modules/infra-kafka-pipe/src/kafka-pipe.sh index 1c7aae1..b40a782 100644 --- a/modules/infra-kafka-pipe/src/kafka-pipe.sh +++ b/modules/infra-kafka-pipe/src/kafka-pipe.sh @@ -16,7 +16,7 @@ INSTANCE_ID=$(cat /proc/sys/kernel/random/uuid) while :; do docker logs --tail 0 -f "$FACTORIO_SERVER_CONTAINER_NAME" | sed -n -e 's/^FactorioEvent: //p' | - kafkacat -P -T -b "$KAFKA_HOST" -t factorio-server-log -k "$INSTANCE_ID" + kcat -P -T -b "$KAFKA_HOST" -t factorio-server-log -k "$INSTANCE_ID" echo "Error - retrying in 10 seconds" sleep 10