mucking about trying to get camel-kafka-connect websockets working

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": {

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": []
 }

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;
 }

modules/events-processor/src/main/kotlin/dev/adamko/kafkatorio/processor/main.kt

@@ -56,7 +56,10 @@ fun main() {
 //      KXS.asA<FactorioEvent<FactorioObjectData>>(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)
     )
 

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<SerialName>()?.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(

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:

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")

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

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())
+
+}

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.

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) }

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

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