如何使用 libp2p 在 golang 中处理缓冲的读写流？

Question

我正在学习本教程：

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))

之后，创建了一个缓冲区stream/read-write变量：

rw := bufio.NewReadWriter(bufio.NewReader(stream), bufio.NewWriter(stream))

现在这个流用于在点之间发送和接收数据。这是使用两个以 rw 作为输入的 goroutine 函数完成的：

go writeData(rw) go readData(rw)

我的问题是：

我想向我的同行发送数据并需要他们的反馈：例如在 rw 中有一个问题，他们需要回答 yes/no。我如何传回这个答案并处理它（启用一些交互）？
我想在rw中发送的数据并不总是相同的。有时它是一个仅包含名称的字符串，有时它是一个包含整个块的字符串等。我该如何区分？

我想到了那些解决方案。但我是 golang 的新手，所以也许你有更好的：

我是否需要为每个不同的内容创建一个新流： stream, err := host.NewStream(ctx, peer.ID, protocol.ID(cfg.ProtocolID))
我是否需要为每个不同的内容打开更多缓冲的 rw 变量： rw := bufio.NewReadWriter(bufio.NewReader(stream), bufio.NewWriter(stream))
还有其他解决办法吗？

感谢您帮助解决这个问题！！

Answer 1

这就是 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)
        }
    }
}

它从 stdin 读取数据，因此您可以键入消息，并将其写入 rw 并刷新它。这种启用了一种 tty 聊天。如果它工作正常，你应该能够启动至少两个对等点并通过标准输入进行通信。

您不应为新内容重新创建 rw。您可以重复使用现有的，直到您关闭它。从 tuto 的代码中，为每个新对等点创建一个新的 rw。

现在 tcp 流不能作为具有请求和与该请求对应的响应的 http 请求。所以如果你想发送一些东西，并得到对那个特定问题的回应，你可以发送这种格式的消息：

[8 bytes unique ID][content of the message]\n

当你收到它时，你解析它，准备响应并以相同的格式发送它，这样你就可以匹配消息，创建一种 request/response 通信。

你可以这样做：

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
        }
    }
}

希望这对您有所帮助。

如何使用 libp2p 在 golang 中处理缓冲的读写流？

How to handle buffered Read-Write Stream(s) to peers in golang using libp2p?

p2p

stream

go

libp2p