package pipeline

import "io"

// ********** The following is common between the request body AND the response body.

// ProgressReceiver defines the signature of a callback function invoked as progress is reported.
type ProgressReceiver func(bytesTransferred int64)

// ********** The following are specific to the request body (a ReadSeekCloser)

// This struct is used when sending a body to the network
type requestBodyProgress struct {
	requestBody io.ReadSeeker // Seeking is required to support retries
	pr          ProgressReceiver
}

// NewRequestBodyProgress adds progress reporting to an HTTP request's body stream.
func NewRequestBodyProgress(requestBody io.ReadSeeker, pr ProgressReceiver) io.ReadSeeker {
	if pr == nil {
		panic("pr must not be nil")
	}
	return &requestBodyProgress{requestBody: requestBody, pr: pr}
}

// Read reads a block of data from an inner stream and reports progress
func (rbp *requestBodyProgress) Read(p []byte) (n int, err error) {
	n, err = rbp.requestBody.Read(p)
	if err != nil {
		return
	}
	// Invokes the user's callback method to report progress
	position, err := rbp.requestBody.Seek(0, io.SeekCurrent)
	if err != nil {
		panic(err)
	}
	rbp.pr(position)
	return
}

func (rbp *requestBodyProgress) Seek(offset int64, whence int) (offsetFromStart int64, err error) {
	return rbp.requestBody.Seek(offset, whence)
}

// requestBodyProgress supports Close but the underlying stream may not; if it does, Close will close it.
func (rbp *requestBodyProgress) Close() error {
	if c, ok := rbp.requestBody.(io.Closer); ok {
		return c.Close()
	}
	return nil
}

// ********** The following are specific to the response body (a ReadCloser)

// This struct is used when sending a body to the network
type responseBodyProgress struct {
	responseBody io.ReadCloser
	pr           ProgressReceiver
	offset       int64
}

// NewResponseBodyProgress adds progress reporting to an HTTP response's body stream.
func NewResponseBodyProgress(responseBody io.ReadCloser, pr ProgressReceiver) io.ReadCloser {
	if pr == nil {
		panic("pr must not be nil")
	}
	return &responseBodyProgress{responseBody: responseBody, pr: pr, offset: 0}
}

// Read reads a block of data from an inner stream and reports progress
func (rbp *responseBodyProgress) Read(p []byte) (n int, err error) {
	n, err = rbp.responseBody.Read(p)
	rbp.offset += int64(n)

	// Invokes the user's callback method to report progress
	rbp.pr(rbp.offset)
	return
}

func (rbp *responseBodyProgress) Close() error {
	return rbp.responseBody.Close()
}