README
¶
go-libp2p-pipe
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. Pipe suggests simple interface for two most common cases of stream messaging: simple message without any feedback 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.
Documentation
¶
Overview ¶
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 and bandwidth wasteful in high flood of messages. Pipe suggests simple interface for two most common cases of messaging protocols over streams: simple message without any feedback and 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 some 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.
Index ¶
- Variables
- func MarshalMessage(msg *Message, buf []byte) (int, error)
- func ReadMessage(r io.Reader, msg *Message) error
- func RemovePipeHandler(host core.Host, proto core.ProtocolID)
- func SetPipeHandler(host core.Host, h Handler, proto core.ProtocolID)
- func UnmarshalMessage(msg *Message, b []byte) error
- func WriteMessage(w io.Writer, msg *Message) error
- type Handler
- type Message
- type Pipe
- type RepOpt
Constants ¶
This section is empty.
Variables ¶
var ( // MaxWriteAttempts specifies amount of retries to write on failure MaxWriteAttempts = 3 // MessageBuffer specifies the size of buffer for incoming messages // If buffer is full, new messages will be dropped MessageBuffer = 8 )
var ( ErrClosed = errors.New("pipe: closed pipe") ErrReset = errors.New("pipe: reset pipe") )
var Protocol protocol.ID = "/pipe/1.0.0"
Functions ¶
func MarshalMessage ¶ added in v1.1.2
MarshalMessage fills given byte slice with the Message
func ReadMessage ¶ added in v1.1.2
ReadMessage fills given Message from Reader
func RemovePipeHandler ¶
func RemovePipeHandler(host core.Host, proto core.ProtocolID)
RemovePipeHandler removes pipe handler from host
func SetPipeHandler ¶
func SetPipeHandler(host core.Host, h Handler, proto core.ProtocolID)
SetPipeHandler sets new stream handler which wraps stream into the pipe
func UnmarshalMessage ¶ added in v1.1.2
UnmarshalMessage fills given Message from byte slice
Types ¶
type Message ¶
type Message struct {
// contains filtered or unexported fields
}
Message is the central object in Pipe's communication. It is responsible for transferring Pipe's user data. Also, it has 3 forms: simple message, request, response.
func NewMessage ¶
NewMessage creates new simple pipe Message. Pipe just them pass through without additional handling.
func NewRequest ¶
NewRequest creates new Message with request form. Requests allow users to send any data through the Pipe and asynchronously receive responses at any time after other end calculates them.
type Pipe ¶
type Pipe interface {
// Closes pipe for writing
io.Closer
// Send puts message in the pipe which after are transported to other pipe's end
Send(*Message) error
// Next iteratively reads new messages from pipe
Next(context.Context) (*Message, error)
// Protocol returns protocol identifier defined in pipe
Protocol() protocol.ID
// Conn returns underlying connection used by pipe
Conn() network.Conn
// Reset closes the pipe for reading and writing on both sides
Reset() error
}
Source Files
Directories