141 lines
3.6 KiB
Go
141 lines
3.6 KiB
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/quic-go/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 func() { _ = 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 func() { _ = 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)
|
|
}
|