punchd/examples/app/asyncutils.nim

86 lines
2.6 KiB
Nim

import asyncdispatch
from net import
BufferSize
from os import
osLastError,
newOsError
from posix import
EINTR,
EWOULDBLOCK,
EAGAIN,
IOVec,
Tmsghdr,
Tcmsghdr,
SocketHandle,
SockLen,
CMSG_FIRSTHDR,
CMSG_NXTHDR,
CMSG_DATA,
SOL_SOCKET,
SCM_RIGHTS,
recvmsg
type ControlMessage* = object
level*: int
msgType*: int
data*: string
proc getFd*(cmsg: ControlMessage): AsyncFD =
if cmsg.level != SOL_SOCKET or
cmsg.msgType != SCM_RIGHTS or
cmsg.data.len != sizeof(AsyncFD):
raise(newException(ValueError, "unexpected ancillary data"))
result = cast[ptr AsyncFD](cmsg.data.cstring)[]
proc asyncRecvMsg*(fd: AsyncFD, size: int = BufferSize,
cmsgSize: csize_t = BufferSize.csize_t):
Future[tuple[data: string, cmsgs: seq[ControlMessage]]] =
var retFuture = newFuture[tuple[data: string,
cmsgs: seq[ControlMessage]]]("asyncRecvMsg")
proc cb(sock: AsyncFD): bool =
result = true
var dataBuffer = newString(size)
# TODO: remove the when clause once issue https://github.com/nim-lang/Nim/issues/15197 is solved.
when (defined(linux) and not defined(android)) and defined(amd64):
var iovec = IOVec(iov_base: dataBuffer.cstring,
iov_len: dataBuffer.len.csize_t)
else:
var iovec = IOVec(iov_base: dataBuffer.cstring,
iov_len: dataBuffer.len)
var cmsgBuffer = newString(cmsgSize)
zeroMem(cmsgBuffer.cstring, cmsgBuffer.len)
var msg = Tmsghdr(msg_iov: addr iovec,
msg_iovlen: 1,
msg_control: addr cmsgBuffer[0],
msg_controllen: cmsgBuffer.len.SockLen)
let res = recvmsg(sock.SocketHandle, addr msg, 0)
if res < 0:
let lastError = osLastError()
if lastError.int32 != EINTR and
lastError.int32 != EWOULDBLOCK and
lastError.int32 != EAGAIN:
retFuture.fail(newOSError(lastError))
else:
result = false
return
var cmsgs = newSeq[ControlMessage]()
var cmsgHeader = CMSG_FIRSTHDR(addr msg)
while cmsgHeader != nil:
let dataLen = cmsgHeader.cmsg_len - sizeof(Tcmsghdr).csize_t
var cmsg = ControlMessage(level: cmsgHeader.cmsg_level,
msgType: cmsgHeader.cmsg_type,
data: newString(dataLen))
copyMem(cmsg.data.cstring, CMSG_DATA(cmsgHeader), dataLen)
cmsgs.add(cmsg)
cmsgHeader = CMSG_NXTHDR(addr msg, cmsgHeader)
dataBuffer.setLen(res)
retFuture.complete((dataBuffer, cmsgs))
addRead(fd, cb)
return retFuture