reflector.go/peer/http3/client.go

package http3

import (
	"bytes"
	"encoding/hex"
	"fmt"
	"io"
	"net/http"
	"sync"
	"time"

	"github.com/lbryio/reflector.go/internal/metrics"
	"github.com/lbryio/reflector.go/shared"
	"github.com/lbryio/reflector.go/store"

	"github.com/lbryio/lbry.go/v2/extras/errors"
	"github.com/lbryio/lbry.go/v2/stream"

	"github.com/lucas-clemente/quic-go/http3"
	log "github.com/sirupsen/logrus"
)

// Client is an instance of a client connected to a server.
type Client struct {
	Timeout      time.Duration
	conn         *http.Client
	roundTripper *http3.RoundTripper
	ServerAddr   string
}

// Close closes the connection with the client.
func (c *Client) Close() error {
	c.conn.CloseIdleConnections()
	return c.roundTripper.Close()
}

// GetStream gets a stream
func (c *Client) GetStream(sdHash string, blobCache store.BlobStore) (stream.Stream, error) {
	var sd stream.SDBlob

	b, _, err := c.GetBlob(sdHash)
	if err != nil {
		return nil, err
	}

	err = sd.FromBlob(b)
	if err != nil {
		return nil, err
	}

	s := make(stream.Stream, len(sd.BlobInfos)+1-1) // +1 for sd blob, -1 for last null blob
	s[0] = b

	for i := 0; i < len(sd.BlobInfos)-1; i++ {
		var trace shared.BlobTrace
		s[i+1], trace, err = c.GetBlob(hex.EncodeToString(sd.BlobInfos[i].BlobHash))
		if err != nil {
			return nil, err
		}
		log.Debug(trace.String())
	}

	return s, nil
}

// HasBlob checks if the blob is available
func (c *Client) HasBlob(hash string) (bool, error) {
	resp, err := c.conn.Get(fmt.Sprintf("https://%s/has/%s", c.ServerAddr, hash))
	if err != nil {
		return false, errors.Err(err)
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusOK {
		return true, nil
	}
	if resp.StatusCode == http.StatusNotFound {
		return false, nil
	}
	return false, errors.Err("non 200 status code returned: %d", resp.StatusCode)
}

// GetBlob gets a blob
func (c *Client) GetBlob(hash string) (stream.Blob, shared.BlobTrace, error) {
	start := time.Now()
	resp, err := c.conn.Get(fmt.Sprintf("https://%s/get/%s?trace=true", c.ServerAddr, hash))
	if err != nil {
		return nil, shared.NewBlobTrace(time.Since(start), "http3"), errors.Err(err)
	}
	defer resp.Body.Close()

	if resp.StatusCode == http.StatusNotFound {
		fmt.Printf("%s blob not found %d\n", hash, resp.StatusCode)
		return nil, shared.NewBlobTrace(time.Since(start), "http3"), errors.Err(store.ErrBlobNotFound)
	} else if resp.StatusCode != http.StatusOK {
		return nil, shared.NewBlobTrace(time.Since(start), "http3"), errors.Err("non 200 status code returned: %d", resp.StatusCode)
	}

	tmp := getBuffer()
	defer putBuffer(tmp)
	serialized := resp.Header.Get("Via")
	trace := shared.NewBlobTrace(time.Since(start), "http3")
	if serialized != "" {
		parsedTrace, err := shared.Deserialize(serialized)
		if err != nil {
			return nil, shared.NewBlobTrace(time.Since(start), "http3"), err
		}
		trace = *parsedTrace
	}
	written, err := io.Copy(tmp, resp.Body)
	if err != nil {
		return nil, trace.Stack(time.Since(start), "http3"), errors.Err(err)
	}

	blob := make([]byte, written)
	copy(blob, tmp.Bytes())

	metrics.MtrInBytesUdp.Add(float64(len(blob)))

	return blob, trace.Stack(time.Since(start), "http3"), nil
}

// buffer pool to reduce GC
// https://www.captaincodeman.com/2017/06/02/golang-buffer-pool-gotcha
var buffers = sync.Pool{
	// New is called when a new instance is needed
	New: func() interface{} {
		buf := make([]byte, 0, stream.MaxBlobSize)
		return bytes.NewBuffer(buf)
	},
}

// getBuffer fetches a buffer from the pool
func getBuffer() *bytes.Buffer {
	return buffers.Get().(*bytes.Buffer)
}

// putBuffer returns a buffer to the pool
func putBuffer(buf *bytes.Buffer) {
	buf.Reset()
	buffers.Put(buf)
}