syndicate_utils/src/syndesizer/json_socket_translator.nim

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

import
  std/[json, options],
  pkg/sys/[ioqueue, sockets],
  pkg/preserves, pkg/preserves/jsonhooks,
  pkg/syndicate, pkg/syndicate/protocols/[gatekeeper, sturdy],
   ../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, observer, ResolvedAccepted(responderSession: ds))
      # Resolve the <json-socket-translator { }> step.
  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; sa: TcpAddress; ds, observer: Cap) {.asyncio.} =
  var
    socket = new AsyncConn[Protocol.Tcp]
    conn = connectTcpAsync(sa.host, Port sa.port)
  socket[] = conn
  translateSocketBody()

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

proc spawnJsonSocketTranslator*(turn: Turn; relay: Cap): Actor {.discardable.} =
  let pat = Resolve?:{ 0: JsonSocketTranslatorStep.grabTypeFlat, 1: grab() }
  spawnActor(turn, "json-socket-translator") do (turn: Turn):
    during(turn, relay, pat) do (sa: TcpAddress, observer: Cap):
      linkActor(turn, "json-socket-translator") do (turn: Turn):
        let ds = turn.newDataspace()
        discard trampoline:
          whelp translateSocket(turn.facet, sa, ds, observer)
    during(turn, relay, pat) do (sa: UnixAddress, observer: Cap):
      linkActor(turn, "json-socket-translator") do (turn: Turn):
        let ds = turn.newDataspace()
        discard trampoline:
          whelp translateSocket(turn.facet, sa, ds, observer)

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