Function ReadByteStream

core/transfer/streaming/reader.go:39–74

(ctx context.Context, stream streaming.Stream)

}

func ReadByteStream(ctx context.Context, stream streaming.Stream) io.ReadCloser {
	rbs := &readByteStream{
		ctx:     ctx,
		stream:  stream,
		window:  0,
		errCh:   make(chan error),
		updated: make(chan struct{}, 1),
	}
	go func() {
		for {
			if rbs.window >= windowSize {
				select {
				case <-ctx.Done():
					return
				case <-rbs.updated:
					continue
				}
			}
			update := &transferapi.WindowUpdate{
				Update: windowSize,
			}
			anyType, err := typeurl.MarshalAny(update)
			if err != nil {
				rbs.errCh <- err
				return
			}
			if err := stream.Send(anyType); err == nil {
				rbs.window += windowSize
			} else if !errors.Is(err, io.EOF) {
				rbs.errCh <- err
			}
		}

	}()
	return rbs
}

func (r *readByteStream) Read(p []byte) (n int, err error) {
	plen := len(p)