func (d *Decoder) drainOutput() {
	if d.current.cancel != nil {
		println("cancelling current")
		close(d.current.cancel)
		d.current.cancel = nil
	}
	if d.current.d != nil {
		if debug {
			printf("re-adding current decoder %p, decoders: %d", d.current.d, len(d.decoders))
		}
		d.decoders <- d.current.d
		d.current.d = nil
		d.current.b = nil
	}
	if d.current.output == nil || d.current.flushed {
		println("current already flushed")
		return
	}
	for {
		select {
		case v := <-d.current.output:
			if v.d != nil {
				if debug {
					printf("re-adding decoder %p", v.d)
				}
				d.decoders <- v.d
			}
			if v.err == errEndOfStream {
				println("current flushed")
				d.current.flushed = true
				return
			}
		}
	}
}
drainOutput will drain the output until errEndOfStream is sent.
package zstd
import (
	"bytes"
	"errors"
	"io"
	"sync"
)
type Decoder struct {
	o decoderOptions
	decoders chan *blockDec
	frames chan *frameDec
	stream chan decodeStream
	current decoderState
	dicts map[uint32]struct{}
	streamWg sync.WaitGroup
}
type decoderState struct {
	decodeOutput
	output chan decodeOutput
	cancel chan struct{}
	flushed bool
}
var (
	_ = io.WriterTo(&Decoder{})
	_ = io.Reader(&Decoder{})
)
func NewReader(r io.Reader, opts ...DOption) (*Decoder, error) {
	initPredefined()
	var d Decoder
	d.o.setDefault()
	for _, o := range opts {
		err := o(&d.o)
		if err != nil {
			return nil, err
		}
	}
	d.current.output = make(chan decodeOutput, d.o.concurrent)
	d.current.flushed = true
	d.decoders = make(chan *blockDec, d.o.concurrent)
	d.frames = make(chan *frameDec, d.o.concurrent)
	for i := 0; i < d.o.concurrent; i++ {
		d.frames <- newFrameDec(d.o)
		d.decoders <- newBlockDec(d.o.lowMem)
	}
	if r == nil {
		return &d, nil
	}
	return &d, d.Reset(r)
}
func (d *Decoder) Read(p []byte) (int, error) {
	if d.stream == nil {
		return 0, errors.New("no input has been initialized")
	}
	var n int
	for {
		if len(d.current.b) > 0 {
			filled := copy(p, d.current.b)
			p = p[filled:]
			d.current.b = d.current.b[filled:]
			n += filled
		}
		if len(p) == 0 {
			break
		}
		if len(d.current.b) == 0 {
			if d.current.err != nil {
				break
			}
			if !d.nextBlock(n == 0) {
				return n, nil
			}
		}
	}
	if len(d.current.b) > 0 {
		if debug {
			println("returning", n, "still bytes left:", len(d.current.b))
		}
		return n, nil
	}
	if d.current.err != nil {
		d.drainOutput()
	}
	if debug {
		println("returning", n, d.current.err, len(d.decoders))
	}
	return n, d.current.err
}
func (d *Decoder) Reset(r io.Reader) error {
	if d.current.err == ErrDecoderClosed {
		return d.current.err
	}
	if r == nil {
		return errors.New("nil Reader sent as input")
	}
	if d.stream == nil {
		d.stream = make(chan decodeStream, 1)
		d.streamWg.Add(1)
		go d.startStreamDecoder(d.stream)
	}
	d.drainOutput()
	if bb, ok := r.(*bytes.Buffer); ok && bb.Len() < 1<<20 {
		if debug {
			println("*bytes.Buffer detected, doing sync decode, len:", bb.Len())
		}
		b := bb.Bytes()
		dst, err := d.DecodeAll(b, nil)
		if err == nil {
			err = io.EOF
		}
		d.current.b = dst
		d.current.err = err
		d.current.flushed = true
		if debug {
			println("sync decode to ", len(dst), "bytes, err:", err)
		}
		return nil
	}
	d.current.decodeOutput = decodeOutput{}
	d.current.err = nil
	d.current.cancel = make(chan struct{})
	d.current.flushed = false
	d.current.d = nil
	d.stream <- decodeStream{
		r:      r,
		output: d.current.output,
		cancel: d.current.cancel,
	}
	return nil
}