summaryrefslogtreecommitdiff
path: root/pkg/server/queue
diff options
context:
space:
mode:
Diffstat (limited to 'pkg/server/queue')
-rw-r--r--pkg/server/queue/queue.go60
1 files changed, 60 insertions, 0 deletions
diff --git a/pkg/server/queue/queue.go b/pkg/server/queue/queue.go
new file mode 100644
index 0000000..8d0f395
--- /dev/null
+++ b/pkg/server/queue/queue.go
@@ -0,0 +1,60 @@
+package queue
+
+import (
+ "io"
+)
+
+type Q chan []byte
+
+type reader struct {
+ b []byte
+ q Q
+}
+
+type writer struct {
+ q Q
+}
+
+func New() Q {
+ return make(Q)
+}
+
+func (q Q) Reader() io.Reader {
+ return &reader{q: q}
+}
+
+func (r *reader) Read(p []byte) (int, error) {
+ if len(r.b) == 0 {
+ r.b = <-r.q
+ if r.b == nil {
+ return 0, io.EOF
+ }
+ }
+
+ n := copy(p, r.b)
+ r.b = r.b[n:]
+
+ return n, nil
+}
+
+func (q Q) Writer() io.Writer {
+ return &writer{q: q}
+}
+
+func (w *writer) Write(p []byte) (int, error) {
+ buf := make([]byte, len(p))
+ copy(buf, p)
+ w.q <- buf
+
+ return len(p), nil
+}
+
+func IoCopy(r io.Reader, w io.Writer) error {
+ if _, err := io.Copy(w, r); err != nil {
+ if err != io.EOF {
+ return err
+ }
+ }
+
+ return nil
+}