diff options
Diffstat (limited to 'streamio/stream.go')
-rw-r--r-- | streamio/stream.go | 44 |
1 files changed, 44 insertions, 0 deletions
diff --git a/streamio/stream.go b/streamio/stream.go new file mode 100644 index 000000000..dafade8fb --- /dev/null +++ b/streamio/stream.go @@ -0,0 +1,44 @@ +package streamio + +import ( + "io" +) + +// NewReader turns receiver into an io.Reader. Errors from the receiver +// function are passed on unmodified. This means receiver should emit +// io.EOF when done. +func NewReader(receiver func() ([]byte, error)) io.Reader { + return &receiveReader{receiver: receiver} +} + +type receiveReader struct { + receiver func() ([]byte, error) + data []byte + err error +} + +func (rr *receiveReader) Read(p []byte) (int, error) { + if len(rr.data) == 0 { + rr.data, rr.err = rr.receiver() + } + n := copy(p, rr.data) + rr.data = rr.data[n:] + if len(rr.data) == 0 { + return n, rr.err + } + return n, nil +} + +// NewWriter turns sender into an io.Writer. The number of 'bytes +// written' reported back is always len(p). +func NewWriter(sender func(p []byte) error) io.Writer { + return &sendWriter{sender: sender} +} + +type sendWriter struct { + sender func([]byte) error +} + +func (sw *sendWriter) Write(p []byte) (int, error) { + return len(p), sw.sender(p) +} |