Pipe is an effective way to reuse libp2p streams. While streams are lightweight there are still cases of protocols which needs a lot of messaging between same peers for continuous time. Current libp2p flow suggests to create new stream for every new message or request/response, what could be inefficient in high flood of messages and waste a lot of bandwidth. Pipe's aim is to fix the problems.
Pipe library suggests simple interface for two most common cases of bidirectional stream messaging: simple message without any feedback from remote peer and asynchronous request/response pattern.
Pipe is somewhere similar to request pipelining, but with one key difference - requested host does not have to handle requests in line and can process them for long time, so responses could be sent at any time without any ordering.
Pipe takes full control over stream and handles new stream creation on failures with graceful pipe closing on both sides of the pipe.
$ go get github.com/Wondertan/go-libp2p-pipe
// register Pipe handler on remote Peer
pipe.SetPipeHandler(host, pipeHandler, customProtocol)// create new Pipe on own Peer
pi, err := pipe.NewPipe(ctx, host, remotePeer, protocol)
if err != nil {
return err
}// create simple message with the data
msg := pipe.NewMessage(bytes)
// send message
err := pi.Send(msg)
if err != nil {
return err
}// dequeue received message from pipe
msg := p.Next(context.Background())
// retrieving sent data
bytes := msg.Data()// create new request message
req := pipe.NewRequest(bytes)
// send request
err := pi.Send(req)
if err != nil {
return err
}// dequeue request
req := p.Next(context.Background())
// replying with the same data from the request
req.Reply(req.Data())// getting response with data from original requst
data := req.Response(context.Background())Feel free to dive in! Open an issue or submit PRs.
MIT © Hlib Kanunnikov