Skip to content

JSONLineServer.kt

src/jvmMain/kotlin/afield/io/process/JSONLineServer.kt · 429 lines

package and imports — 55 lines, from line 1 
@file:afield.runtime.completion.CompletionCategory("Data")

package afield.io.process

import afield.data.collections.invoke
import afield.data.collections.s
import afield.data.collections.times
import afield.document.graph.minusAssign
import afield.document.graph.plusAssign
import afield.document.properties.compareTo
import afield.document.workspace.file
import afield.graphics.render.rangeTo
import afield.math.autodiff.compareTo
import afield.math.autodiff.div
import afield.math.autodiff.minus
import afield.math.autodiff.plus
import afield.math.autodiff.times
import afield.math.linalg.invoke
import afield.math.linalg.remAssign
import afield.math.linalg.times
import afield.math.units.div
import afield.math.units.minus
import afield.math.units.plus
import afield.math.units.times

import afield.audio.synth.*
import afield.data.collections.get
import afield.document.properties.clear
import afield.math.linalg.set
import afield.media.image.close
import afield.runtime.execution.start
import afield.runtime.lifecycle.cached
import afield.document.graph.Box
import afield.runtime.execution.ReturnValueRegistry
import afield.runtime.lifecycle.RunsAt
import afield.app.diagnostics.logging.lprintln
import afield.document.graph.registerLaunchJob
import afield.runtime.lifecycle.returnValueRegistration
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import org.json.JSONObject
import java.io.Closeable
import java.net.InetSocketAddress
import java.net.ServerSocket
import java.net.Socket
import java.net.SocketException
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.atomic.AtomicReference
import kotlin.concurrent.thread

A TCP listener that turns newline-delimited JSON into a value a box can read each frame. JSONLineServer binds a ServerSocket, accepts any number of clients on daemon threads, and keeps only the most recently received valid JSON line — JSONLineServer.output is a snapshot, not a queue, so lines that arrive between reads are dropped. Non-JSON lines are counted in JSONLineServer.invalidLineCount and logged, never thrown.

jsonl and JSONLineServers hand out one cached, already-started server per (port, bind host, backlog) so that re-running a box does not fight itself for the port; ephemeral port 0 servers are rejected by the cache and must be built with autoStart = false. Returning a JSONLineServer from a box keeps it running for that box's run — registerJSONLineServerReturnValue installs the ReturnValueRegistry handler that does it, and an uncached server is stopped when the run ends.

class JSONLineServer private constructor

Small TCP server for newline-delimited JSON streams.

Example sender:

some_process_that_prints_jsonl | nc field-machine.local 7777
printf '%s\n' '{"value":1}' | nc field-machine.local 7777

Reconnect wrapper:

while true; do some_process_that_prints_jsonl | nc -N field-machine.local 7777; sleep 1; done

Field-side usage:

val stream = JSONLineServer(7777)
loop {
    val latest = stream.output
    println(latest.optDouble("value"))
    frame()
}

stream
class JSONLineServer private constructor(
    localPort: Int,
    val bindHost: String = "0.0.0.0",
    val backlog: Int = 4,
    autoStart: Boolean = true,
    private val cacheKey: JSONLineServerKey? = null
) : Closeable {

val requestedPort: Int = localPort

var port: Int

    @Volatile
    var port: Int = localPort
        private set

    private val latestJsonLineRef = AtomicReference<String?>(null)
    private val latestRawLineRef = AtomicReference<String?>(null)
    private val latestRemoteRef = AtomicReference<String?>(null)
    private val latestConnectionRemoteRef = AtomicReference<String?>(null)
    private val latestDisconnectedRemoteRef = AtomicReference<String?>(null)
    private val lastErrorRef = AtomicReference<String?>(null)
    private val lineCountRef = AtomicLong(0)
    private val invalidLineCountRef = AtomicLong(0)
    private val connectionCountRef = AtomicLong(0)
    private val activeConnectionCountRef = AtomicLong(0)

    private val clients = CopyOnWriteArrayList<Socket>()
    private val clientThreads = CopyOnWriteArrayList<Thread>()

    @Volatile
    private var serverSocket: ServerSocket? = null

    @Volatile
    private var acceptThread: Thread? = null

var running: Boolean

    @Volatile
    var running: Boolean = false
        private set

var latestReceivedAtMillis: Long

    @Volatile
    var latestReceivedAtMillis: Long = 0L
        private set

var latestConnectedAtMillis: Long

    @Volatile
    var latestConnectedAtMillis: Long = 0L
        private set

var latestDisconnectedAtMillis: Long

    @Volatile
    var latestDisconnectedAtMillis: Long = 0L
        private set

val output: JSONObject

    val output: JSONObject
        get() = latestJsonLineRef.get()?.let(::JSONObject) ?: JSONObject()

val latest: JSONObject?

    val latest: JSONObject?
        get() = latestJsonLineRef.get()?.let(::JSONObject)

val outputText: String?

    val outputText: String?
        get() = latestJsonLineRef.get()

val latestLine: String?

    val latestLine: String?
        get() = latestJsonLineRef.get()

val latestRawLine: String?

    val latestRawLine: String?
        get() = latestRawLineRef.get()

val latestRemoteAddress: String?

    val latestRemoteAddress: String?
        get() = latestRemoteRef.get()

val latestConnectionRemoteAddress: String?

    val latestConnectionRemoteAddress: String?
        get() = latestConnectionRemoteRef.get()

val latestDisconnectedRemoteAddress: String?

    val latestDisconnectedRemoteAddress: String?
        get() = latestDisconnectedRemoteRef.get()

val lastError: String?

    val lastError: String?
        get() = lastErrorRef.get()

val lineCount: Long

    val lineCount: Long
        get() = lineCountRef.get()

val invalidLineCount: Long

    val invalidLineCount: Long
        get() = invalidLineCountRef.get()

val connectionCount: Long

    val connectionCount: Long
        get() = connectionCountRef.get()

val activeConnectionCount: Long

    val activeConnectionCount: Long
        get() = activeConnectionCountRef.get()

init

    init {
        if (autoStart) start()
    }

companion object

operator fun invoke

        operator fun invoke(
            localPort: Int,
            bindHost: String = "0.0.0.0",
            backlog: Int = 4,
            autoStart: Boolean = true
        ): JSONLineServer {
            return if (autoStart && localPort != 0) {
                JSONLineServers.get(localPort, bindHost, backlog)
            } else {
                uncached(localPort, bindHost, backlog, autoStart)
            }
        }

internal fun uncached

        internal fun uncached(
            localPort: Int,
            bindHost: String = "0.0.0.0",
            backlog: Int = 4,
            autoStart: Boolean = true
        ): JSONLineServer = JSONLineServer(localPort, bindHost, backlog, autoStart, null)

internal fun cached

        internal fun cached(
            key: JSONLineServerKey
        ): JSONLineServer = JSONLineServer(key.port, key.bindHost, key.backlog, false, key)
    }

fun hasOutput(): Boolean = latestJsonLineRef.get() != null

fun start(): JSONLineServer

    @Synchronized
    fun start(): JSONLineServer {
        if (running) return this

        val socket = try {
            ServerSocket().apply {
                reuseAddress = true
                bind(bindAddress(), backlog)
            }
        } catch (t: Throwable) {
            val message = "Failed to start JSONLineServer on $bindHost:$requestedPort: ${t.message ?: t}"
            lastErrorRef.set(message)
            lprintln("error,jsonlineserver", message)
            return this
        }

        serverSocket = socket
        port = socket.localPort
        running = true
        lastErrorRef.set(null)

        acceptThread = thread(name = "json-line-server-accept-$port", isDaemon = true) {
            acceptLoop(socket)
        }

        return this
    }

fun stop()

    @Synchronized
    fun stop() {
        running = false

        val socket = serverSocket
        serverSocket = null
        runCatching { socket?.close() }

        clients.forEach { client ->
            runCatching { client.close() }
        }
        clients.clear()

        val current = Thread.currentThread()
        acceptThread?.takeIf { it != current }?.join(250)
        acceptThread = null

        clientThreads.removeIf { thread ->
            if (thread != current) thread.join(50)
            !thread.isAlive
        }
    }

override fun close()

    override fun close() {
        JSONLineServers.remove(this)
        stop()
    }

override fun toString(): String

    override fun toString(): String {
        return "JSONLineServer($bindHost:$port, running=$running, lines=$lineCount, connections=$connectionCount)"
    }

internal suspend fun runForBox(box: Box, strand: String, generation: Int)

body — 102 lines, from line 276 
    internal suspend fun runForBox(box: Box, strand: String, generation: Int) {
        start()

        val completionHandle = currentCoroutineContext()[Job]?.invokeOnCompletion {
            stop()
        }

        try {
            while (running && box.runGeneration[strand] == generation) {
                delay(50)
            }
        } finally {
            completionHandle?.dispose()
            if (cacheKey == null) stop()
        }
    }

    private fun bindAddress(): InetSocketAddress {
        return if (bindHost.isBlank() || bindHost == "0.0.0.0" || bindHost == "*") {
            InetSocketAddress(requestedPort)
        } else {
            InetSocketAddress(bindHost, requestedPort)
        }
    }

    private fun acceptLoop(socket: ServerSocket) {
        try {
            while (running && !socket.isClosed) {
                val client = try {
                    socket.accept()
                } catch (t: SocketException) {
                    if (running) recordError("JSONLineServer accept failed: ${t.message ?: t}")
                    break
                } catch (t: Throwable) {
                    if (running) recordError("JSONLineServer accept failed: ${t.message ?: t}")
                    break
                }

                clients += client
                connectionCountRef.incrementAndGet()
                activeConnectionCountRef.incrementAndGet()
                latestConnectedAtMillis = System.currentTimeMillis()
                latestConnectionRemoteRef.set(client.remoteSocketAddress?.toString())

                val readerThread = thread(name = "json-line-server-client-$port-${connectionCountRef.get()}", isDaemon = true) {
                    handleClient(client)
                }
                clientThreads += readerThread
            }
        } finally {
            if (serverSocket === socket) {
                running = false
                serverSocket = null
            }
            runCatching { socket.close() }
        }
    }

    private fun handleClient(client: Socket) {
        val remote = client.remoteSocketAddress?.toString()
        try {
            client.getInputStream().bufferedReader(Charsets.UTF_8).useLines { lines ->
                lines.forEach { line ->
                    handleLine(line, remote)
                }
            }
        } catch (t: Throwable) {
            if (running) recordError("JSONLineServer client read failed: ${t.message ?: t}")
        } finally {
            clients.remove(client)
            activeConnectionCountRef.decrementAndGet()
            latestDisconnectedAtMillis = System.currentTimeMillis()
            latestDisconnectedRemoteRef.set(remote)
            runCatching { client.close() }
            clientThreads.remove(Thread.currentThread())
        }
    }

    private fun handleLine(line: String, remote: String?) {
        latestRawLineRef.set(line)

        val trimmed = line.trim()
        if (trimmed.isEmpty()) return

        try {
            JSONObject(trimmed)
            latestJsonLineRef.set(trimmed)
            latestRemoteRef.set(remote)
            latestReceivedAtMillis = System.currentTimeMillis()
            lineCountRef.incrementAndGet()
            lastErrorRef.set(null)
        } catch (t: Throwable) {
            invalidLineCountRef.incrementAndGet()
            recordError("Ignoring non-JSON line from ${remote ?: "unknown"}: ${t.message ?: trimmed}")
        }
    }

    private fun recordError(message: String) {
        lastErrorRef.set(message)
        lprintln("warn,jsonlineserver", message)
    }
}

internal data class JSONLineServerKey(val port: Int, val bindHost: String, val backlog: Int)

object JSONLineServers

object JSONLineServers {
    private val cache = ConcurrentHashMap<JSONLineServerKey, JSONLineServer>()

fun get

    fun get(
        port: Int = 7777,
        bindHost: String = "0.0.0.0",
        backlog: Int = 4
    ): JSONLineServer {
        require(port != 0) { "Use JSONLineServer(..., autoStart = false) for ephemeral port 0 servers; cached JSONL servers need a fixed port." }
        val key = JSONLineServerKey(port, bindHost, backlog)
        val server = cache.computeIfAbsent(key) {
            JSONLineServer.cached(it)
        }
        server.start()
        return server
    }

fun remove(server: JSONLineServer)

    fun remove(server: JSONLineServer) {
        cache.entries.removeIf { it.value === server }
    }
}

fun jsonl(port: Int = 7777, bindHost: String = "0.0.0.0", backlog: Int = 4): JSONLineServer

fun jsonl(port: Int = 7777, bindHost: String = "0.0.0.0", backlog: Int = 4): JSONLineServer =
    JSONLineServers.get(port, bindHost, backlog)

fun registerJSONLineServerReturnValue()

@RunsAt(returnValueRegistration, priority = 12)
fun registerJSONLineServerReturnValue() {
    ReturnValueRegistry.handlers += handler@{ inside, returnValue, strand, generation, launchEpoch, notify ->
        val server = returnValue as? JSONLineServer ?: return@handler false

        CoroutineScope(Dispatchers.Default + Box.boxContextElement(inside)).launch {
            val job = coroutineContext[Job]!!
            if (!inside.registerLaunchJob(job, launchEpoch)) {
                notify(false, strand, generation)
                return@launch
            }
            notify(true, strand, generation)
            try {
                server.runForBox(inside, strand, generation)
            } finally {
                server.stop()
                inside.jobs.remove(job)
                notify(false, strand, generation)
            }
        }
        true
    }
}