}

	return req, nil
}

type respBodyMonitor struct {
	io.ReadCloser
	expectTimeouts  bool
	errorStatusOnce sync.Once
}

func (r *respBodyMonitor) Read(p []byte) (n int, err error) {
	n, err = r.ReadCloser.Read(p)
	r.errorStatus(err)
	return
}

func (r *respBodyMonitor) Close() (err error) {
	err = r.ReadCloser.Close()
	r.errorStatus(err)
	return
}

	// Close the input.
	err := r.readCloser.Close()
	for range r.valueCh {
		// Drain values so we don't leak a goroutine.
		// Since we have closed the input, it should fail rather quickly.
	}
	return err
}

// NewReader - creates new JSON reader using readCloser.
func NewReader(readCloser io.ReadCloser, args *ReaderArgs) *Reader {
	readCloser = &syncReadCloser{rc: readCloser}

			}
		}()
	}
}

// NewPReader - creates new parallel JSON reader using readCloser.
// Should only be used for LINES types.
func NewPReader(readCloser io.ReadCloser, args *ReaderArgs) *PReader {
	r := &PReader{
		args:       args,
		buf:        bufio.NewReaderSize(readCloser, jsonSplitSize*2),
		readCloser: readCloser,
		close:      make(chan struct{}),
	}
	r.startReaders()
	return r

func newCountUpReader(reader io.Reader) *countUpReader {
	return &countUpReader{
		reader: reader,
	}
}

type progressReader struct {
	rc              io.ReadCloser
	scannedReader   *countUpReader
	processedReader *countUpReader

	closedMu sync.Mutex
	closer   io.ReadCloser
	closed   bool
}

func (pr *progressReader) Read(p []byte) (n int, err error) {
	// This ensures that Close will block until Read has completed.

	return nil
}

// NewReader - creates new CSV reader using readCloser.
func NewReader(readCloser io.ReadCloser, args *ReaderArgs) (*Reader, error) {
	if args == nil || args.IsEmpty() {
		panic(fmt.Errorf("empty args passed %v", args))
	}
	csvIn := io.Reader(readCloser)
	if args.RecordDelimiter != "\n" {
		csvIn = &recordTransform{
			reader:          readCloser,
			recordDelimiter: []byte(args.RecordDelimiter),

type Args struct {
	URL         *xnet.URL             `json:"url"`
	AuthToken   string                `json:"authToken"`
	Transport   http.RoundTripper     `json:"-"`
	CloseRespFn func(r io.ReadCloser) `json:"-"`
}

// Validate - validate opa configuration params.
func (a *Args) Validate() error {
	req, err := http.NewRequest(http.MethodPost, a.URL.String(), bytes.NewReader([]byte("")))
	if err != nil {

	baseOffset int64

	// fileList is a list of files sorted by ename,
	// for use by the Open method.
	fileListOnce sync.Once
	fileList     []fileListEntry
}

// A ReadCloser is a [Reader] that must be closed when no longer needed.
type ReadCloser struct {
	f *os.File
	Reader
}

// A File is a single file in a ZIP archive.
// The file information is in the embedded [FileHeader].

// convenient wrapper to drain any remaining data on response body.
//
// Subsequently this allows golang http RoundTripper
// to reuse the same connection for future requests.
func DrainBody(respBody io.ReadCloser) {
	// Callers should close resp.Body when done reading from it.
	// If resp.Body is not closed, the Client's underlying RoundTripper
	// (typically Transport) may not be able to reuse a persistent TCP

type ObjectSegmentReaderFn func(offset int64) (io.ReadCloser, error)

// ObjectReadSeekCloser implements ReadSeekCloser interface for reading objects.
// It uses a function that returns a io.ReadCloser for the object.
type ObjectReadSeekCloser struct {
	segmentReader ObjectSegmentReaderFn

	size   int64 // actual object size regardless of compression/encryption
	offset int64
	reader io.ReadCloser

	m.wg.Wait()
	close(m.opCh)
	atomic.StoreInt32(&m.closed, 1)

	if len(m.opCh) > 0 {
		healingLogEvent(context.Background(), "Saving MRF healing data (%d entries)", len(m.opCh))
	}

	newReader := func() io.ReadCloser {
		r, w := io.Pipe()
		go func() {
			// Initialize MRF meta header.
			var data [4]byte
			binary.LittleEndian.PutUint16(data[0:2], healMRFMetaFormat)