lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

delete unused code (bye bye http4k!)

Browse source
This commit is contained in:
Adam 2022-07-02 10:21:13 +02:00
parent 79447b5936
commit b14adfb810
8 changed files with 11 additions and 657 deletions
Download patch file Download diff file Expand all files Collapse all files

168
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/SyslogSocketServer.kt
View file

@ -1,168 +0,0 @@
package dev.adamko.kafkatorio.processor.tileserver
//import com.github.palindromicity.syslog.AllowableDeviations
//import com.github.palindromicity.syslog.SyslogParser
//import com.github.palindromicity.syslog.SyslogParserBuilder
//import com.github.palindromicity.syslog.SyslogSpecification
//import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
//import dev.adamko.kafkatorio.server.config.ApplicationProperties
//import dev.adamko.kafkatorio.processor.tileserver.SyslogMsg.Companion.toSyslogMsg
//import io.ktor.network.selector.SelectorManager
//import io.ktor.network.sockets.aSocket
//import io.ktor.network.sockets.awaitClosed
//import io.ktor.network.sockets.openReadChannel
//import io.ktor.utils.io.readUTF8Line
//import java.util.EnumSet
//import kotlinx.coroutines.CoroutineDispatcher
//import kotlinx.coroutines.Dispatchers
//import kotlinx.coroutines.coroutineScope
//import kotlinx.coroutines.flow.MutableSharedFlow
//import kotlinx.coroutines.flow.SharedFlow
//import kotlinx.coroutines.flow.asSharedFlow
//import kotlinx.coroutines.isActive
//import kotlinx.coroutines.launch
//import kotlinx.coroutines.yield
//
//
//class SyslogSocketServer(
// appProps: ApplicationProperties,
// dispatcher: CoroutineDispatcher = Dispatchers.IO,
//) {
//
// private val selectorManager: SelectorManager = SelectorManager(dispatcher)
//
//
// private val serverSocketTcp = aSocket(selectorManager)
// .tcp()
// .bind(
// appProps.socketServerHost,
// appProps.socketServerPort,
// )
//
//
//// private val serverSocketUdp: BoundDatagramSocket = aSocket(selectorManager)
//// .udp()
//// .bind(
//// localAddress = InetSocketAddress(
//// appProps.socketServerHost,
//// appProps.socketServerPortUdp,
//// ),
//// )
//
//
// private val syslogParser: SyslogParser = SyslogParserBuilder()
// .withDeviations(EnumSet.of(AllowableDeviations.PRIORITY, AllowableDeviations.VERSION))
// .forSpecification(SyslogSpecification.RFC_5424)
// .build()
//
//
// private val _messages: MutableSharedFlow<SyslogMsg> = MutableSharedFlow()
// val messages: SharedFlow<SyslogMsg>
// get() = _messages.asSharedFlow()
//
//
// suspend fun start() {
//// udp()
// tcp()
// }
//
//// private suspend fun udp() {
//// println("[SyslogSocketServer] started listening: ${serverSocketUdp.localAddress}")
////
//// serverSocketUdp.incoming
//// .receiveAsFlow()
//// .map {udpPacket: Datagram ->
//// val msg = udpPacket.packet.readText()
//// println("[SyslogSocketServer] received msg: $msg")
//// syslogParser.parseLine(msg).toSyslogMsg()
//// }.collect(_messages)
//// }
//
// private suspend fun tcp() = coroutineScope {
// println("[SyslogSocketServer] started listening: ${serverSocketTcp.localAddress}")
//
// while (isActive) {
// val socket = serverSocketTcp.accept()
//
// launch {
// val channel = socket.openReadChannel()
// try {
// while (socket.socketContext.isActive) {
// val line = channel.readUTF8Line()
// println("[SyslogSocketServer] received: $line")
//
// val syslog = syslogParser.parseLine(line).toSyslogMsg()
//
// _messages.emit(syslog)
//
// yield()
// }
// } catch (e: Exception) {
// socket.awaitClosed()
// }
// }
// }
//
// yield()
// }
//}
//
//
//data class SyslogMsg(
// val src: Map<String, String?>
//) {
//
// val message: String?
// get() = src[SyslogFieldKeys.MESSAGE.field]
//
// val headerAppName: String?
// get() = src[SyslogFieldKeys.HEADER_APPNAME.field]
//
// val headerHostName: String?
// get() = src[SyslogFieldKeys.HEADER_HOSTNAME.field]
//
// val headerPri: String?
// get() = src[SyslogFieldKeys.HEADER_PRI.field]
//
// val headerPriSeverity: String?
// get() = src[SyslogFieldKeys.HEADER_PRI_SEVERITY.field]
//
// val headerPriFacility: String?
// get() = src[SyslogFieldKeys.HEADER_PRI_FACILITY.field]
//
// val headerProcId: String?
// get() = src[SyslogFieldKeys.HEADER_PROCID.field]
//
// val headerTimestamp: String?
// get() = src[SyslogFieldKeys.HEADER_TIMESTAMP.field]
//
// val headerMsgId: String?
// get() = src[SyslogFieldKeys.HEADER_MSGID.field]
//
// val headerVersion: String?
// get() = src[SyslogFieldKeys.HEADER_VERSION.field]
//
// val structuredBase: String?
// get() = src[SyslogFieldKeys.STRUCTURED_BASE.field]
//
// val structuredElementIdFmt: String?
// get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_FMT.field]
//
// val structuredElementIdPnameFmt: String?
// get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_FMT.field]
//
// val structuredElementIdPnamePattern: String?
// get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_PATTERN.field]
//
// companion object {
// fun Map<String, Any?>.toSyslogMsg(): SyslogMsg {
// val map = entries.mapNotNull { (k, v) ->
// when (v) {
// is String -> k to v
// else -> null
// }
// }.toMap()
// return SyslogMsg(map)
// }
// }
//}

86
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebMapTileServer.kt
View file

@ -1,86 +0,0 @@
package dev.adamko.kafkatorio.processor.tileserver
//import dev.adamko.kafkatorio.processor.config.ApplicationProperties
//import java.security.MessageDigest
//import org.http4k.core.Filter
//import org.http4k.core.HttpHandler
//import org.http4k.core.MemoryBody
//import org.http4k.core.Method.GET
//import org.http4k.core.Request
//import org.http4k.core.Response
//import org.http4k.core.Status
//import org.http4k.core.then
//import org.http4k.routing.ResourceLoader
//import org.http4k.routing.RoutingHttpHandler
//import org.http4k.routing.bind
//import org.http4k.routing.routes
//import org.http4k.routing.static
//
//internal class WebMapTileServer(
// appProps: ApplicationProperties
//) {
//
// private val tilesLoader = ResourceLoader.Directory(appProps.serverDataDir.toFile().canonicalPath)
//
//
// private val filters = listOf(
// ETagChecker(),
// ).reduce(Filter::then)
//
//
// private val routes: RoutingHttpHandler = routes(
// "/tiles" bind GET to static(tilesLoader).withFilter(filters)
// )
//
//
// fun build(): HttpHandler {
// return routes
// }
//
//
// private class ETagChecker : Filter {
//
// override fun invoke(next: HttpHandler): HttpHandler = { request: Request ->
// val requestedETag = request.header("If-None-Match")
//
// var response = next(request)
// val responseBytes = response.body.stream.use { it.readAllBytes() }
// val currentETag = responseBytes.md5()
//
// response = response.body(MemoryBody(responseBytes))
//
// when {
// currentETag == null -> {
//// println("[${request.etagKey}] currentETag is null (requested: $requestedETag), executing request")
// response
// .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
// }
// requestedETag == currentETag -> {
//// println("[${request.etagKey}] currentETag == requested, returning 304 $requestedETag")
// Response(Status.NOT_MODIFIED)
// .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
// }
// else -> {
//// println("[${request.etagKey}] currentETag != requested, executing request and adding currentETag:$currentETag (requested:$requestedETag)")
// response
// .header("ETag", currentETag)
//// .header("Cache-Control", "no-store")
// .header("Cache-Control", "no-cache, max-age=0, must-revalidate")
// }
// }
// }
// }
//
//
// companion object {
// private fun ByteArray.md5(): String? {
// return if (isEmpty()) {
// null
// } else {
// MessageDigest.getInstance("MD5")
// .digest(this)
// .joinToString("") { "%02x".format(it) }
// }
// }
// }
//}

16
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebServer.kt
View file

@ -1,16 +0,0 @@
package dev.adamko.kafkatorio.processor.tileserver
//
//import dev.adamko.kafkatorio.processor.config.ApplicationProperties
//import org.http4k.server.Http4kServer
//import org.http4k.server.PolyHandler
//import org.http4k.server.Undertow
//import org.http4k.server.asServer
//
//internal fun webServer(
// tileServer: WebMapTileServer,
// wsServer: WebsocketServer,
// appProps: ApplicationProperties,
//): Http4kServer = PolyHandler(
// tileServer.build(),
// wsServer.build(),
//).asServer(Undertow(appProps.webPort()))

141
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/tileserver/WebsocketServer.kt
View file

@ -1,141 +0,0 @@
package dev.adamko.kafkatorio.processor.tileserver
//
//import kotlin.coroutines.resume
//import kotlin.coroutines.resumeWithException
//import kotlin.time.Duration.Companion.seconds
//import kotlin.time.TimeSource
//import kotlinx.coroutines.CoroutineExceptionHandler
//import kotlinx.coroutines.CoroutineName
//import kotlinx.coroutines.CoroutineScope
//import kotlinx.coroutines.Dispatchers
//import kotlinx.coroutines.SupervisorJob
//import kotlinx.coroutines.channels.BufferOverflow
//import kotlinx.coroutines.delay
//import kotlinx.coroutines.flow.MutableSharedFlow
//import kotlinx.coroutines.flow.launchIn
//import kotlinx.coroutines.flow.onEach
//import kotlinx.coroutines.flow.withIndex
//import kotlinx.coroutines.isActive
//import kotlinx.coroutines.job
//import kotlinx.coroutines.launch
//import kotlinx.coroutines.supervisorScope
//import kotlinx.coroutines.suspendCancellableCoroutine
//import org.http4k.routing.bind
//import org.http4k.routing.websockets
//import org.http4k.websocket.Websocket
//import org.http4k.websocket.WsHandler
//import org.http4k.websocket.WsMessage
//import org.http4k.websocket.WsStatus
//
//
//class WebsocketServer {
//
// private val serverScope = CoroutineScope(
// CoroutineName("WsServer") +
// Dispatchers.IO +
// SupervisorJob() +
// CoroutineExceptionHandler { _, e ->
// println("[ws.WsServer] handling exception ${e.message}")
// }
// )
//
//
// private val clients: MutableSharedFlow<Websocket> = MutableSharedFlow(
// replay = 100,
// extraBufferCapacity = 100,
// BufferOverflow.DROP_OLDEST,
// )
//
//
// private val messages: MutableSharedFlow<String> = MutableSharedFlow(
// replay = 100,
// extraBufferCapacity = 100,
// BufferOverflow.DROP_OLDEST,
// )
//
//
// private fun bindClient(ws: Websocket) = serverScope.launch {
// println("[ws.bindClient] ${ws.upgradeRequest.toMessage()}")
// clients.emit(ws)
// }
//
//
// fun sendMessage(msg: String) = serverScope.launch {
// println("[ws.sendMessage] $msg")
// messages.emit(msg)
// }
//
//
// fun build(): WsHandler {
// val startTime = TimeSource.Monotonic.markNow()
//
// serverScope.launch {
// supervisorScope {
// clients
// .withIndex()
// .onEach { (id, ws) ->
// createClient(id, ws)
// }.launchIn(serverScope)
// }
// }
//
// serverScope.launch {
// while (isActive) {
// delay(10.seconds)
// sendMessage("elapsed ${startTime.elapsedNow()}")
// }
// }
//
// return websockets(
// "/ws/foo" bind { ws: Websocket ->
// serverScope.launch { bindClient(ws) }
// }
// )
// }
//
//
// private fun createClient(id: Int, ws: Websocket) {
// val clientJob = SupervisorJob(serverScope.coroutineContext.job)
//
// serverScope.launch(CoroutineName("ws-client-$id") + clientJob) {
// launch {
// suspendCancellableCoroutine<WsStatus> { cont ->
// println("[ws-client] entering suspendCancellableCoroutine")
//
// cont.invokeOnCancellation { e: Throwable? ->
// launch { sendMessage("[ws-client.invokeOnCompletion] closing client ws ${e?.message}") }
// when (e) {
// null -> ws.close(WsStatus.NORMAL)
// else -> ws.close(WsStatus.ABNORMAL_CLOSE)
// }
// }
//
// ws.onError { e: Throwable ->
// launch { sendMessage("[ws-client.onError] client $id error [${e.message}]") }
// if (!cont.isCompleted) {
// cont.resumeWithException(e)
// }
// }
//
// ws.onClose { status: WsStatus ->
// launch { sendMessage("[ws-client.onClose] client $id disconnected [${status}]") }
// if (!cont.isCompleted) {
// cont.resume(status)
// }
// }
//
// ws.onMessage { wsMsg ->
// launch { sendMessage("$id says: ${wsMsg.bodyString()}") }
// }
// }
// }
//
// println("[ws-client] $id listening to messages")
// messages
// .onEach { msg ->
// println("[ws-client] sending message to $id")
// ws.send(WsMessage(msg))
// }.launchIn(this)
// }
// }
//}