nanonext (version 0.9.2)

recv_aio: Receive Async

Description

Receive data asynchronously over a connection (Socket, Context or Stream).

A signalling version of the function takes a 'conditionVariable' as an additional argument and signals it when the async receive is complete.

Usage

recv_aio(
  con,
  mode = c("serial", "character", "complex", "double", "integer", "logical", "numeric",
    "raw"),
  timeout = NULL,
  keep.raw = FALSE,
  n = 65536L
)
recv_aio_signal(
  con,
  mode = c("serial", "character", "complex", "double", "integer", "logical", "numeric",
    "raw"),
  timeout = NULL,
  keep.raw = FALSE,
  n = 65536L,
  cv
)

Value

A 'recvAio' (object of class 'recvAio') (invisibly).

Arguments

con

a Socket, Context or Stream.

mode

[default 'serial'] mode of vector to be received - one of 'serial', 'character', 'complex', 'double', 'integer', 'logical', 'numeric', or 'raw'. The default 'serial' means a serialised R object, for the other modes, the raw vector received will be converted into the respective mode. For Streams, 'serial' is not an option and the default is 'character'. Alternatively, for performance, specify an integer position in the vector of choices e.g. 1L for 'serial', 2L for 'character' etc.

timeout

[default NULL] integer value in milliseconds or NULL, which applies a socket-specific default, usually the same as no timeout.

keep.raw

[default FALSE] [DEPRECATED - this argument will be removed in a future package version] logical flag whether to keep and return the received raw vector along with the converted data. Supplying a non-logical value will error.

n

[default 65536L] applicable to Streams only, the maximum number of bytes to receive. Can be an over-estimate, but note that a buffer of this size is reserved.

cv

For the signalling version: a 'conditionVariable' that should be signalled when the async receive is complete.

Details

Async receive is always non-blocking and returns a 'recvAio' immediately.

For a 'recvAio', the received message is available at $data, and the raw message at $raw (if kept). An 'unresolved' logical NA is returned if the async operation is yet to complete.

To wait for the async operation to complete and retrieve the received message, use call_aio on the returned 'recvAio' object.

Alternatively, to stop the async operation, use stop_aio.

In case of an error, an integer 'errorValue' is returned (to be distiguishable from an integer message value). This can be verified using is_error_value.

For mode = "serial", attempting to unserialise a non-serialised message will result in the error 'unknown input format'.

For all other modes, if an error occurred in conversion of the data to the specified mode, a raw vector will be returned at $data instead to allow for the data to be recovered.

For the signalling version: when the receive is complete, the supplied 'conditionVariable' is signalled by incrementing its value by 1. This happens asynchronously and independently of the R execution thread.

Examples

s1 <- socket("pair", listen = "inproc://nanonext")
s2 <- socket("pair", dial = "inproc://nanonext")

res <- send_aio(s1, data.frame(a = 1, b = 2), timeout = 100)
msg <- recv_aio(s2, timeout = 100)
msg
msg$data

res <- send_aio(s1, c(1.1, 2.2, 3.3), mode = "raw", timeout = 100)
msg <- recv_aio(s2, mode = "double", timeout = 100)
msg
msg$raw
msg$data

res <- send_aio(s1, "example message", mode = "raw", timeout = 100)
msg <- recv_aio(s2, mode = "character", timeout = 100)
call_aio(msg)
msg$raw
msg$data

close(s1)
close(s2)

# Signalling a condition variable

s1 <- socket("pair", listen = "tcp://127.0.0.1:6546")
cv <- cv()
msg <- recv_aio_signal(s1, timeout = 100, cv = cv)
until(cv, 10L)
msg$data
close(s1)

# in another process in parallel
s2 <- socket("pair", dial = "tcp://127.0.0.1:6546")
res <- send_aio(s2, c(1.1, 2.2, 3.3), mode = "raw", timeout = 100)
close(s2)