Issue 141 networking basics

game/application/pom.xml

@@ -27,6 +27,10 @@
             <groupId>de.gleex.pltcmd.game</groupId>
             <artifactId>serialization</artifactId>
         </dependency>
+        <dependency>
+            <groupId>de.gleex.pltcmd.game</groupId>
+            <artifactId>networking</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.hexworks.zircon</groupId>
             <artifactId>zircon.jvm.swing</artifactId>

game/application/src/main/kotlin/de/gleex/pltcmd/game/application/main.kt

@@ -2,6 +2,8 @@ package de.gleex.pltcmd.game.application
 
 import de.gleex.pltcmd.game.engine.Game
 import de.gleex.pltcmd.game.engine.entities.types.*
+import de.gleex.pltcmd.game.networking.connect
+import de.gleex.pltcmd.game.networking.createServer
 import de.gleex.pltcmd.game.options.GameOptions
 import de.gleex.pltcmd.game.options.UiOptions
 import de.gleex.pltcmd.game.serialization.StorageId
@@ -21,6 +23,7 @@ import de.gleex.pltcmd.model.world.Sector
 import de.gleex.pltcmd.model.world.WorldMap
 import de.gleex.pltcmd.model.world.coordinate.Coordinate
 import de.gleex.pltcmd.model.world.sectorOrigin
+import io.ktor.server.engine.*
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.runBlocking
 import org.hexworks.amethyst.api.Engine
@@ -112,11 +115,25 @@ open class Main {
 
         val (elementsToCommand, hq) = prepareGame(game, gameWorld)
 
+        // networking
+        log.debug("starting network server")
+        val serverEngine = createServer(hq)
+        serverEngine.start(wait = false)
+
+        log.debug("starting network client thread")
+        val clientThread = Thread { connect() }
+        clientThread.start()
+
         screen.dock(GameView(gameWorld, tileGrid, game, hq, elementsToCommand))
 
         Ticker.start()
         // cleanup
-        screen.onShutdown { Ticker.stop() }
+        screen.onShutdown {
+            log.debug("shutdown game")
+            Ticker.stop()
+            clientThread.stop()
+            serverEngine.stop(300, 500, TimeUnit.MILLISECONDS)
+        }
     }
 
     /**

game/engine/src/main/kotlin/de/gleex/pltcmd/game/engine/entities/types/Communicating.kt

@@ -27,7 +27,8 @@ typealias CommunicatingEntity = GameEntity<Communicating>
 
 private val log = LoggerFactory.getLogger(Communicating::class)
 
-internal val CommunicatingEntity.communicator: RadioCommunicator
+// FIXME only public for testing! Should be internal and use extension functions to access transmissions.
+val CommunicatingEntity.communicator: RadioCommunicator
     get() = getAttribute(RadioAttribute::class).communicator
 
 /**

game/networking/pom.xml

@@ -0,0 +1,62 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>game</artifactId>
+        <groupId>de.gleex.pltcmd.game</groupId>
+        <version>0.2.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>networking</artifactId>
+
+    <properties>
+        <ktor.version>1.6.3</ktor.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>de.gleex.pltcmd.game</groupId>
+            <artifactId>ui</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>de.gleex.pltcmd.game</groupId>
+            <artifactId>engine</artifactId>
+        </dependency>
+        <!-- serialization -->
+        <dependency>
+            <groupId>org.jetbrains.kotlinx</groupId>
+            <artifactId>kotlinx-serialization-protobuf</artifactId>
+            <version>1.1.0</version>
+        </dependency>
+        <dependency>
+            <groupId>io.ktor</groupId>
+            <artifactId>ktor-serialization</artifactId>
+            <version>${ktor.version}</version>
+        </dependency>
+        <!-- ktor server -->
+        <dependency>
+            <groupId>io.ktor</groupId>
+            <artifactId>ktor-server-netty</artifactId>
+            <version>${ktor.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.ktor</groupId>
+            <artifactId>ktor-websockets</artifactId>
+            <version>${ktor.version}</version>
+        </dependency>
+        <!-- ktor client -->
+        <dependency>
+            <groupId>io.ktor</groupId>
+            <artifactId>ktor-client-cio-jvm</artifactId>
+            <version>${ktor.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.ktor</groupId>
+            <artifactId>ktor-client-websockets</artifactId>
+            <version>${ktor.version}</version>
+        </dependency>
+    </dependencies>
+
+</project>

game/networking/src/main/kotlin/de/gleex/pltcmd/game/networking/Client.kt

@@ -0,0 +1,44 @@
+package de.gleex.pltcmd.game.networking
+
+import de.gleex.pltcmd.model.radio.UiBroadcastEvent
+import de.gleex.pltcmd.model.radio.UiBroadcasts
+import de.gleex.pltcmd.util.events.uiEventBus
+import io.ktor.client.*
+import io.ktor.client.features.websocket.*
+import io.ktor.http.*
+import io.ktor.http.cio.websocket.*
+import kotlinx.coroutines.runBlocking
+import kotlinx.serialization.decodeFromByteArray
+import kotlinx.serialization.protobuf.ProtoBuf
+import org.hexworks.cobalt.events.api.simpleSubscribeTo
+import org.hexworks.cobalt.logging.api.LoggerFactory
+
+private val log = LoggerFactory.getLogger(::connect::class)
+
+fun connect(host: String = "127.0.0.1", port: Int = defaultPort) {
+    val client = HttpClient {
+        install(WebSockets)
+    }
+    runBlocking {
+        log.info("Connecting to server $host:$port")
+        client.webSocket(method = HttpMethod.Get, host = host, port = port, path = pathBroadcastEvents) {
+            for (frame in incoming) {
+                val bytes = frame.readBytes()
+                val broadcastEvent = ProtoBuf.decodeFromByteArray<UiBroadcastEvent>(bytes)
+                log.trace{"received event $broadcastEvent"}
+                uiEventBus.publish(broadcastEvent, UiBroadcasts)
+            }
+        }
+    }
+    client.close()
+    log.info("Connection to server closed.")
+}
+
+fun main() {
+    uiEventBus.simpleSubscribeTo(UiBroadcasts) { event: UiBroadcastEvent ->
+        log.info("received UI event: $event")
+        val message = event.message
+        println(message)
+    }
+    connect()
+}

game/networking/src/main/kotlin/de/gleex/pltcmd/game/networking/Server.kt

@@ -0,0 +1,143 @@
+package de.gleex.pltcmd.game.networking
+
+import de.gleex.pltcmd.game.engine.entities.EntityFactory
+import de.gleex.pltcmd.game.engine.entities.types.CommunicatingEntity
+import de.gleex.pltcmd.game.engine.entities.types.communicator
+import de.gleex.pltcmd.game.engine.entities.types.onReceivedTransmission
+import de.gleex.pltcmd.game.engine.entities.types.onSendTransmission
+import de.gleex.pltcmd.game.ticks.Ticker
+import de.gleex.pltcmd.model.elements.CallSign
+import de.gleex.pltcmd.model.faction.Faction
+import de.gleex.pltcmd.model.radio.UiBroadcastEvent
+import de.gleex.pltcmd.model.radio.communication.building.ConversationBuilder
+import de.gleex.pltcmd.model.radio.communication.transmissions.Transmission
+import de.gleex.pltcmd.model.radio.communication.transmissions.decoding.isOpening
+import de.gleex.pltcmd.model.radio.communication.transmissions.decoding.sender
+import de.gleex.pltcmd.model.radio.receivedTransmission
+import de.gleex.pltcmd.model.world.Sector
+import de.gleex.pltcmd.model.world.WorldMap
+import de.gleex.pltcmd.model.world.WorldTile
+import de.gleex.pltcmd.model.world.coordinate.Coordinate
+import de.gleex.pltcmd.model.world.coordinate.CoordinateRectangle
+import de.gleex.pltcmd.model.world.terrain.Terrain
+import de.gleex.pltcmd.model.world.terrain.TerrainHeight
+import de.gleex.pltcmd.model.world.terrain.TerrainType
+import de.gleex.pltcmd.util.events.globalEventBus
+import io.ktor.application.*
+import io.ktor.features.*
+import io.ktor.http.cio.websocket.*
+import io.ktor.routing.*
+import io.ktor.server.engine.*
+import io.ktor.server.netty.*
+import io.ktor.websocket.*
+import kotlinx.coroutines.channels.Channel
+import kotlinx.serialization.encodeToByteArray
+import kotlinx.serialization.protobuf.ProtoBuf
+import org.hexworks.cobalt.logging.api.LoggerFactory
+import java.util.concurrent.TimeUnit
+
+// there is also an `Application.log` provided by Ktor!
+private val logger = LoggerFactory.getLogger(::createServer::class)
+
+internal const val defaultPort = 9170
+internal val pathBroadcastEvents = "/broadcasts"
+
+// TODO encapsulate return type. Providing the Ktor implementation class to the caller couples the code to this implementation.
+fun createServer(hq: CommunicatingEntity): ApplicationEngine {
+    return embeddedServer(Netty, port = defaultPort) {
+        install(WebSockets)
+        routing {
+            broadcastsRoute(hq)
+        }
+    }
+}
+
+private fun Routing.broadcastsRoute(hq: CommunicatingEntity) {
+    webSocket(pathBroadcastEvents) {
+        logger.info("sending broadcasts to $logId")
+
+        val eventChannel = Channel<UiBroadcastEvent>(Channel.BUFFERED)
+        // listen to local events
+        val subscriptionReceived = hq.onReceivedTransmission(eventChannel::trySendLogging)
+        val subscriptionSend = hq.onSendTransmission(eventChannel::trySendLogging)
+        // clean up on disconnect
+        closeReason.invokeOnCompletion {
+            logger.debug("client closed connection $logId")
+            subscriptionReceived.dispose()
+            subscriptionSend.dispose()
+            eventChannel.close()
+        }
+        // send events to client
+        for (event in eventChannel) {
+            // second send over network
+            val bytes = ProtoBuf.encodeToByteArray(event)
+            send(bytes)
+        }
+
+        logger.info("finished sending broadcasts to $logId")
+    }
+}
+
+/** sends the given Event to the given channel and logs the result. */
+private fun Channel<UiBroadcastEvent>.trySendLogging(event: Transmission) {
+    // first convert
+    val uiEvent = event.uiEvent
+    // and queue event
+    val sendResult = trySend(uiEvent)
+    if (sendResult.isFailure) {
+        logger.error("failed to queue event $uiEvent for network transmission due to ${sendResult.exceptionOrNull()}")
+    } else if (sendResult.isSuccess) {
+        logger.trace("successfully queued event $uiEvent for network transmission")
+    }
+}
+
+// TODO move somewhere else
+val Transmission.uiEvent: UiBroadcastEvent
+    get() {
+        val message = "${Ticker.currentTimeString.value}: ${message}"
+        val senderName = sender.name
+        return UiBroadcastEvent(message, isOpening, senderName)
+    }
+
+internal val DefaultWebSocketServerSession.logId: String
+    get() {
+        val origin = call.request.origin
+        return origin.remoteHost
+    }
+
+fun main(args: Array<String>) {
+    // setup communication model
+    logger.info("creating HQ...")
+    val sender = CallSign("sender")
+    val testTransmission = ConversationBuilder(
+        sender,
+        CallSign("receiver")
+    ).terminatingResponse("The test finished successfully :)")
+    val origin = Coordinate(0, 0)
+    val map = dummyMapAt(origin)
+    val hq: CommunicatingEntity = EntityFactory.newBaseAt(origin, map, Faction("example"), sender)
+    logger.info("done!")
+
+    // start networking
+    val serverThread = createServer(hq)
+    serverThread.start(wait = false)
+
+    // transfer data
+    repeat(3) {
+        logger.info("sending test")
+        globalEventBus.receivedTransmission(hq.communicator, testTransmission)
+        Thread.sleep(3000)
+    }
+
+    // done
+    logger.info("Stopping server...")
+    serverThread.stop(200, 500, TimeUnit.MILLISECONDS)
+    logger.info("Stopped")
+}
+
+private fun dummyMapAt(origin: Coordinate) = WorldMap.create(setOf(Sector(
+    origin,
+    CoordinateRectangle(origin, Sector.TILE_COUNT, Sector.TILE_COUNT)
+        .map { coordinate -> WorldTile(coordinate, Terrain.of(TerrainType.FOREST, TerrainHeight.FIVE)) }
+        .toSortedSet()
+)))

game/networking/src/main/resources/logback.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<configuration>
+
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="info">
+        <appender-ref ref="STDOUT"/>
+    </root>
+
+    <logger name="de.gleex.pltcmd" level="debug"/>
+
+</configuration>

game/options/pom.xml

@@ -17,6 +17,10 @@
             <groupId>de.gleex.pltcmd.model</groupId>
             <artifactId>world</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.hexworks.zircon</groupId>
+            <artifactId>zircon.core-jvm</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

game/pom.xml

@@ -22,13 +22,10 @@
         <module>engine</module>
         <module>ui-strings</module>
         <module>serialization</module>
+        <module>networking</module>
     </modules>
 
     <dependencies>
-        <dependency>
-            <groupId>org.hexworks.zircon</groupId>
-            <artifactId>zircon.core-jvm</artifactId>
-        </dependency>
         <!-- tests -->
         <dependency>
             <groupId>de.gleex.pltcmd.util</groupId>
@@ -62,6 +59,11 @@
                 <artifactId>engine</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>de.gleex.pltcmd.game</groupId>
+                <artifactId>networking</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>de.gleex.pltcmd.game</groupId>
                 <artifactId>serialization</artifactId>