你在学习Golang相关的知识吗?本文《如何使用 libp2p 在 golang 中处理对等点的缓冲读写流?》,主要介绍的内容就涉及到,如果你想提升自己的开发能力,就不要错过这篇文章,大家要知道编程理论基础和实战操作都是不可或缺的哦!
问题内容我正在关注本教程:
https://github.com/libp2p/go-libp2p-examples/tree/master/chat-with-mdns
简单地说,它:
- 配置 p2p 主机
- 为传入连接设置默认处理函数 (3.非必须)
- 并打开一个到连接对等点的流:
stream, err := host.NewStream(ctx,peer.ID,protocol.ID(cfg.ProtocolID))
之后,创建了一个缓冲流/读写变量:
rw := bufio.NewReadWriter(bufio.NewReader(stream), bufio.NewWriter(stream))
现在该流用于在对等点之间发送和接收数据。这是使用两个以 rw 作为输入的 goroutine 函数完成的:
go writeData(rw)
go readData(rw)
我的问题是:
我想向我的同事发送数据并需要他们的反馈: 例如在 rw 中,有一个问题,他们需要回答是/否。我怎样才能传回这个答案并处理它(启用一些交互)?
我想在 rw 中发送的数据并不总是相同的。有时它是一个仅包含名称的字符串,有时它是一个包含整个块的字符串等。我如何区分?
我考虑过这些解决方案。但我是 golang 新手,所以也许你有更好的:
我是否需要为每个不同的内容创建一个新的流:
stream, err := host.NewStream(ctx,peer.ID,protocol.ID(cfg.ProtocolID))
我是否需要为每个不同的内容打开更多缓冲的读写变量:
rw := bufio.NewReadWriter(bufio.NewReader(stream), bufio.NewWriter(stream))
还有其他解决方案吗?
感谢您帮助解决这个问题!!
解决方案
这就是 readdata
在你的教程中所做的:
func readdata(rw *bufio.readwriter) {
for {
str, err := rw.readstring('\n')
if err != nil {
fmt.println("error reading from buffer")
panic(err)
}
if str == "" {
return
}
if str != "\n" {
// green console colour: \x1b[32m
// reset console colour: \x1b[0m
fmt.printf("\x1b[32m%s\x1b[0m> ", str)
}
}
}
它基本上读取流,直到找到 \n
(这是一个换行符)并将其打印到标准输出。
writedata
:
func writedata(rw *bufio.readwriter) {
stdreader := bufio.newreader(os.stdin)
for {
fmt.print("> ")
senddata, err := stdreader.readstring('\n')
if err != nil {
fmt.println("error reading from stdin")
panic(err)
}
_, err = rw.writestring(fmt.sprintf("%s\n", senddata))
if err != nil {
fmt.println("error writing to buffer")
panic(err)
}
err = rw.flush()
if err != nil {
fmt.println("error flushing buffer")
panic(err)
}
}
}
它从标准输入读取数据,以便您可以键入消息,并将其写入 rw
并刷新它。这种方式可以实现某种 tty 聊天。
如果工作正常,您应该能够启动至少两个对等点并通过标准输入进行通信。
您不应为新内容重新创建新的 rw
。您可以重复使用现有的,直到将其关闭。根据 tuto 的代码,为每个新对等点创建一个新的 rw
。
现在,tcp 流不能作为带有请求和与该请求对应的响应的 http 请求。因此,如果您想发送某些内容并获得对该特定问题的答复,您可以发送以下格式的消息:
[8 bytes unique id][content of the message]\n
当您收到它时,您会解析它,准备响应并以相同的格式发送它,以便您可以匹配消息,创建一种请求/响应通信。
你可以这样做:
func sendMsg(rw *bufio.ReadWriter, id int64, content []byte) error {
// allocate our slice of bytes with the correct size 4 + size of the message + 1
msg := make([]byte, 4 + len(content) + 1)
// write id
binary.LittleEndian.PutUint64(msg, uint64(id))
// add content to msg
copy(msg[13:], content)
// add new line at the end
msg[len(msg)-1] = '\n'
// write msg to stream
_, err = rw.Write(msg)
if err != nil {
fmt.Println("Error writing to buffer")
return err
}
err = rw.Flush()
if err != nil {
fmt.Println("Error flushing buffer")
return err
}
return nil
}
func readMsg(rw *bufio.ReadWriter) {
for {
// read bytes until new line
msg, err := rw.ReadBytes('\n')
if err != nil {
fmt.Println("Error reading from buffer")
continue
}
// get the id
id := int64(binary.LittleEndian.Uint64(msg[0:8]))
// get the content, last index is len(msg)-1 to remove the new line char
content := string(msg[8:len(msg)-1])
if content != "" {
// we print [message ID] content
fmt.Printf("[%d] %s", id, content)
}
// here you could parse your message
// and prepare a response
response, err := prepareResponse(content)
if err != nil {
fmt.Println("Err while preparing response: ", err)
continue
}
if err := s.sendMsg(rw, id, response); err != nil {
fmt.Println("Err while sending response: ", err)
continue
}
}
}
希望这有帮助。
以上就是本文的全部内容了,是否有顺利帮助你解决问题?若是能给你带来学习上的帮助,请大家多多支持编程网!更多关于Golang的相关知识,也可关注编程网公众号。