r - R中的非阻塞式 socket ？

Question

我正在尝试使用套接字在R session 之间进行通信，并且无法使它们像帮助文件所建议的那样运行(?read.socket)。
在一个Rstudio session 中，我运行以下命令

sock_server <- make.socket(port = 22222, server = TRUE)
write.socket(sock_server, "Ahoy-hoy")

然后在另一个我跑

sock <- make.socket(port = 22222, server = FALSE)
read.socket(sock, loop = FALSE)

我确实看到了友好的问候。
但是，如果我重新运行 read.socket行而不发送任何其他信息，则该行将挂起，仅在发送进一步的信息时返回。选项 loop = FALSE应该可以控制此行为，并且如果没有等待读取的内容，则大概返回一个空字符串，但它不会改变任何内容。
我已经在MacOS和Windows上尝试过了，在两种情况下都是一样的。
任何想法如何使套接字表现？

现在，我已经从命令行使用RStudio，RGui和R的各种组合进行了尝试，据我所知，在所有这些工具中它的行为都相同。
我的最终目标是在R中拥有可以处理来自外部主机的FIX连接的功能。除了上述很难解决的问题以外，基本上可以使用这些套接字函数来工作。
我乐于使用人们在这些事情上使用的其他任何东西。

这是第二次尝试!但是不幸的是，仅在Mac上，Windows会做一些奇怪的事情。
在服务器端，将二进制连接与基础R中的一组不同的套接字函数一起使用，

sock_server <- socketConnection(port = 22222, server = TRUE, open = 'a+b')
fix_message <- gsub("\\|","\001","8=FIX.4.4|9=106|35=0|34=1234|49=abc|56=def|10=999|")
writeBin(charToRaw(fix_message), sock_server)

和在客户端上

sock <- socketConnection(port = 22222, server = FALSE, open = 'a+b')
rawToChar(readBin(sock, raw(), n = 32))

请注意，我正在发送的消息中没有换行符(这是FIX消息的功能)。
在Mac上，这可以按预期方式工作-两次运行 readBin行会给您完整的消息，您可以将其粘贴回去，但是在Windows上，消息的第二部分消失了!

# Mac
> rawToChar(readBin(sock, raw(), n = 32))
[1] "8=FIX.4.4\0019=106\00135=0\00134=1234\00149="
> rawToChar(readBin(sock, raw(), n = 32))
[1] "abc\00156=def\00110=999\001"

# Windows
> rawToChar(readBin(sock, raw(), n = 32))
[1] "8=FIX.4.4\0019=106\00135=0\00134=1234\00149="
> rawToChar(readBin(sock, raw(), n = 32))
[1] ""

我在两个平台上都使用R 4.0.2。任何人都有在Windows上如何执行此操作的想法吗？

Answer 1

我还尝试了 future 的包装:

library("future")
plan(multiprocess)


mySocketRead<- function()
{
  sock <- make.socket(port = 22222, server = FALSE)
  msg <-read.socket(sock, loop = FALSE)
  #print(msg)
  print("read done ")
}


# normal non-blocking
res %<-% mySocketRead()

print("here we go")

但是此示例在读取时也会阻塞，因此这似乎是read.socket函数的一个普遍问题。
我还有2个主意:
想法A:也许您互相交换服务器和客户端，因为如果服务器始终阻塞在永远的读取循环中，则可能会更容易接受，但这取决于您的用例。
想法B:使用socketConnection实现并转换字符串
想要发送给问题较少的字符串。
我确实认为，出现了从Mac到Windows的“丢失的字符串”，因为您确实使用了“4\0019”字符串文字，不同的套接字实现可能会“解释”错误。
通常，它对于使用base64encode和base64decode函数很有用
也许您还应该添加一个校验和，以确保传输成功。

require(base64enc)

#send
fix_message <- gsub("\\|","001","8=FIX.4.4|9=106|35=0|34=1234|49=abc|56=def|10=999|")
fix_message_base64 <- base64encode(charToRaw(fix_message))
writeBin(charToRaw(fix_message_base64), sock_server)

#receive
fix_message_input_base64_raw <-readBin(sock, raw(), n = 32)
fix_message_input_raw <-base64decode(fix_message_base64)
fix_message_input <-rawToChar(fix_message_input_raw)

如果这不可行，则可以使用 wireshark调试网络流量。
旧的不起作用，但是是一个示例，如何等待X秒使函数Y完成:
我认为在函数内部使用套接字，然后等待该函数超时。这也将被阻止，但不会永远存在。

require(R.utils)

## function that can take a long time
socketRead<- function()
{
    sock <- make.socket(port = 22222, server = FALSE)
    read.socket(sock, loop = FALSE)
    return("read")
}

# call read function
waitingTimeInSeconds=10
withTimeout(socketRead(), timeout = waitingTimeInSeconds, onTimeout = "error")

使用Timout here查找函数的文档