syndicate_utils/src/syndesizer/json_socket_translator.nim

# SPDX-FileCopyrightText: ☭ Emery Hemingway
# SPDX-License-Identifier: Unlicense

import std/[json, options]
import pkg/sys/[ioqueue, sockets]
import preserves, preserves/jsonhooks, syndicate

import ../schema/[config, json_messages]

template translateSocketBody {.dirty.} =
  # Template workaround for CPS and parameterized types.
  var
    guard = initGuard(facet)
    dec = newBufferedDecoder(0)
    buf = new string #TODO: get a pointer into the decoder
    alive = true
  proc kill(turn: Turn) =
    alive = false
  proc setup(turn: Turn) =
    # Closure, not CPS.
    onMessage(turn, ds, ?:SendJson) do (data: JsonNode):
      if alive:
        discard trampoline:
          whelp write(socket[], $data & "\n")
      else:
        stderr.writeLine "dropped send of ", data
    discard publish(turn, ds, initRecord("connected", sa.toPreserves))
  onStop(facet, kill)
  run(facet, setup)
  while alive:
    # TODO: parse buffer
    buf[].setLen(0x4000)
    let n = read(socket[], buf)
    if n < 1:
      stderr.writeLine "socket read returned ", n
    else:
      buf[].setLen(n)
      dec.feed(buf[])
      var data = dec.parse()
      if data.isSome:
        proc send(turn: Turn) =
          # Closure, not CPS.
          message(turn, ds, initRecord("recv", data.get))
        run(facet, send)
  stderr.writeLine "close socket ", sa
  close(socket[])

proc translateSocket(facet: Facet; ds: Cap; sa: TcpAddress) {.asyncio.} =
  var
    socket = new AsyncConn[Protocol.Tcp]
    conn = connectTcpAsync(sa.host, Port sa.port)
  socket[] = conn
  translateSocketBody()

proc translateSocket(facet: Facet; ds: Cap; sa: UnixAddress) {.asyncio.} =
  var
    socket = new AsyncConn[Protocol.Unix]
    conn = connectUnixAsync(sa.path)
  socket[] = conn
  translateSocketBody()

proc spawnJsonSocketTranslator*(turn: Turn; root: Cap): Actor {.discardable.} =
  spawnActor(turn, "json-socket-translator") do (turn: Turn):
    during(turn, root, ?:JsonSocketTranslatorArguments) do (ds: Cap, sa: TcpAddress):
      linkActor(turn, "json-socket-translator") do (turn: Turn):
        discard trampoline:
          whelp translateSocket(turn.facet, ds, sa)
    during(turn, root, ?:JsonSocketTranslatorArguments) do (ds: Cap, sa: UnixAddress):
      linkActor(turn, "json-socket-translator") do (turn: Turn):
        discard trampoline:
          whelp translateSocket(turn.facet, ds, sa)

when isMainModule:
  import syndicate/relays
  runActor("main") do (turn: Turn):
    resolveEnvironment(turn) do (turn: Turn; ds: Cap):
      spawnJsonSocketTranslator(turn, ds)