10990: Use AsyncStream to minimize store-and-forward latency

[arvados.git] / sdk / go / streamer / streamer.go

diff --git a/sdk/go/streamer/streamer.go b/sdk/go/streamer/streamer.go
index 0c4d20834852ab9c6b8fe7170c6a39a18f71158d..c8d012f76591470551e51dcb8faea87eea2aecea 100644 (file)
--- a/sdk/go/streamer/streamer.go
+++ b/sdk/go/streamer/streamer.go
@@ -36,8 +36,9 @@ Alternately, if you already have a filled buffer and just want to read out from
 package streamer
 
 import (
-       "io"
        "errors"
+       "fmt"
+       "io"
 )
 
 var ErrAlreadyClosed = errors.New("cannot close a stream twice")
@@ -48,6 +49,7 @@ type AsyncStream struct {
        add_reader        chan bool
        subtract_reader   chan bool
        wait_zero_readers chan bool
+       closed            bool
 }
 
 // Reads from the buffer managed by the Transfer()
@@ -58,7 +60,13 @@ type StreamReader struct {
 }
 
 func AsyncStreamFromReader(buffersize int, source io.Reader) *AsyncStream {
-       t := &AsyncStream{make([]byte, buffersize), make(chan sliceRequest), make(chan bool), make(chan bool), make(chan bool)}
+       t := &AsyncStream{
+               buffer:            make([]byte, buffersize),
+               requests:          make(chan sliceRequest),
+               add_reader:        make(chan bool),
+               subtract_reader:   make(chan bool),
+               wait_zero_readers: make(chan bool),
+       }
 
        go t.transfer(source)
        go t.readersMonitor()
@@ -67,7 +75,13 @@ func AsyncStreamFromReader(buffersize int, source io.Reader) *AsyncStream {
 }
 
 func AsyncStreamFromSlice(buf []byte) *AsyncStream {
-       t := &AsyncStream{buf, make(chan sliceRequest), make(chan bool), make(chan bool), make(chan bool)}
+       t := &AsyncStream{
+               buffer:            buf,
+               requests:          make(chan sliceRequest),
+               add_reader:        make(chan bool),
+               subtract_reader:   make(chan bool),
+               wait_zero_readers: make(chan bool),
+       }
 
        go t.transfer(nil)
        go t.readersMonitor()
@@ -127,10 +141,41 @@ func (this *StreamReader) Close() error {
        return nil
 }
 
-func (this *AsyncStream) Close() {
+func (this *AsyncStream) Close() error {
+       if this.closed {
+               return ErrAlreadyClosed
+       }
+       this.closed = true
        this.wait_zero_readers <- true
        close(this.requests)
        close(this.add_reader)
        close(this.subtract_reader)
        close(this.wait_zero_readers)
+       return nil
+}
+
+func (this *StreamReader) Seek(offset int64, whence int) (int64, error) {
+       var want int64
+       switch whence {
+       case io.SeekStart:
+               want = offset
+       case io.SeekCurrent:
+               want = int64(this.offset) + offset
+       case io.SeekEnd:
+               want = int64(this.Len()) + offset
+       default:
+               return int64(this.offset), fmt.Errorf("invalid whence %d", whence)
+       }
+       if want < 0 {
+               return int64(this.offset), fmt.Errorf("attempted seek to %d", want)
+       }
+       if want > int64(this.Len()) {
+               want = int64(this.Len())
+       }
+       this.offset = int(want)
+       return want, nil
+}
+
+func (this *StreamReader) Len() uint64 {
+       return uint64(len(this.stream.buffer))
 }