ehmry
/
syndicate-nim
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Use binary Preserves over stdio

This commit is contained in:
Emery Hemingway 2022-10-16 11:42:07 -05:00
parent fa5a4a9cbc
commit 68a742797c
1 changed files with 25 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
src/syndicate/relays.nim
View File

@ -1,7 +1,7 @@
# SPDX-FileCopyrightText: ☭ 2021 Emery Hemingway
# SPDX-License-Identifier: Unlicense
import std/[asyncdispatch, options, tables]
import std/[asyncdispatch, options, streams, tables]
import preserves
import ./actors, ./durings, ./membranes, ./protocols/[protocol, sturdy]
@ -232,7 +232,7 @@ proc dispatch(relay: Relay; v: Preserve[WireRef]) =
of PacketKind.Extension:
discard
else:
stderr.writeLine "discarding unparsed packet ", v
stderr.writeLine "discarding undecoded packet ", v
type
RelayOptions = object of RootObj
@ -350,12 +350,13 @@ proc connectUnix*(turn: var Turn; path: string; cap: SturdyRef; bootProc: Connec
import std/asyncfile
const stdinReadSize = 128
proc connectStdio*(ds: Ref; turn: var Turn) =
## Connect to an external dataspace over stdin and stdout.
proc stdoutWriter(packet: sink Packet): Future[void] {.async.} =
# var buf = encode(packet)
# doAssert writeBytes(stdout, buf, 0, buf.len) == buf.len
write(stdout, packet)
var buf = encode(packet)
doAssert writeBytes(stdout, buf, 0, buf.len) == buf.len
flushFile(stdout)
var opts = RelayActorOptions(
packetWriter: stdoutWriter,
@ -367,15 +368,27 @@ proc connectStdio*(ds: Ref; turn: var Turn) =
asyncStdin = openAsync("/dev/stdin")
facet.actor.atExit do (turn: var Turn):
close(asyncStdin)
var wireBuf = newStringStream()
proc recvCb(pktFut: Future[string]) {.gcsafe.} =
if pktFut.failed: discard
else:
let buf = pktFut.read
if not pktFut.failed:
var buf = pktFut.read
if buf.len == 0:
run(facet) do (turn: var Turn): stopActor(turn)
else:
var v = parsePreserves(buf)
dispatch(relay, cast[Preserve[WireRef]](v))
callSoon: asyncStdin.readLine().addCallback(recvCb)
var decodePos: int
if wireBuf.atEnd:
wireBuf.setPosition(0)
wireBuf.data = move buf
else:
decodePos = wireBuf.getPosition
wireBuf.data.add(buf)
try:
while not wireBuf.atEnd:
decodePos = wireBuf.getPosition
var pr = decodePreserves(wireBuf, WireRef)
dispatch(relay, pr)
except IOError, ValueError:
wireBuf.setPosition(decodePos)
callSoon: asyncStdin.read(stdinReadSize).addCallback(recvCb)
# do not process the next line immediately
asyncStdin.readLine().addCallback(recvCb)
asyncStdin.read(stdinReadSize).addCallback(recvCb)