diff --git a/buffer.go b/buffer.go index c4aa556..53d6dae 100644 --- a/buffer.go +++ b/buffer.go @@ -70,8 +70,8 @@ func (mb *MemoryBuffer) Append(l Line) { } func (mb *MemoryBuffer) Size() int { - mb.mutex.Lock() - defer mb.mutex.Unlock() + mb.mutex.RLock() + defer mb.mutex.RUnlock() return len(mb.lines) } @@ -83,8 +83,8 @@ func (mb *MemoryBuffer) Reset() { } func (mb *MemoryBuffer) Done() <-chan struct{} { - mb.mutex.Lock() - defer mb.mutex.Unlock() + mb.mutex.RLock() + defer mb.mutex.RUnlock() return mb.done } @@ -121,8 +121,8 @@ func (mb *MemoryBuffer) Accept(ctx context.Context, p pipeline.Producer) { } func (mb *MemoryBuffer) LineAt(n int) (Line, error) { - mb.mutex.Lock() - defer mb.mutex.Unlock() + mb.mutex.RLock() + defer mb.mutex.RUnlock() if s := len(mb.lines); s <= 0 || n >= s { return nil, errors.New("empty buffer") @@ -139,6 +139,7 @@ func NewSource(in io.Reader, enableSep bool) *Source { enableSep: enableSep, done: make(chan struct{}), ready: make(chan struct{}), + start: make(chan struct{}, 1), setupOnce: sync.Once{}, OutputChannel: pipeline.OutputChannel(make(chan interface{})), } diff --git a/interface.go b/interface.go index d36b7da..e515b74 100644 --- a/interface.go +++ b/interface.go @@ -480,7 +480,7 @@ type Buffer interface { type MemoryBuffer struct { done chan struct{} lines []Line - mutex sync.Mutex + mutex sync.RWMutex } type ActionMap interface {