mirror of
https://github.com/GopeedLab/gopeed.git
synced 2026-11-09 18:10:29 +00:00
feat: HTTP add TLS fingerprint fallback (#1444)
This commit is contained in:
68 files changed
+7243
-1604
No files matched your search
@@ -6,21 +6,11 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/GopeedLab/gopeed/pkg/download/engine/inject/file"
|
||||
"github.com/GopeedLab/gopeed/pkg/download/engine/inject/formdata"
|
||||
"github.com/GopeedLab/gopeed/pkg/download/engine/inject/xhr"
|
||||
"github.com/GopeedLab/gopeed/pkg/download/engine/util"
|
||||
"github.com/dop251/goja"
|
||||
"github.com/dop251/goja_nodejs/eventloop"
|
||||
"github.com/imroc/req/v3"
|
||||
)
|
||||
|
||||
//go:embed stream.js
|
||||
@@ -29,8 +19,6 @@ var script string
|
||||
type Config struct {
|
||||
CreateObjectURL func(opts *ObjectURLOptions, open ObjectURLOpener) (string, error)
|
||||
RevokeObjectURL func(url string) error
|
||||
ProxyHandler func(r *http.Request) (*url.URL, error)
|
||||
RegisterCleanup func(cleanup func())
|
||||
}
|
||||
|
||||
type ObjectURLOptions struct {
|
||||
@@ -96,69 +84,6 @@ func Enable(runtime *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
fetchRegistry := newFetchRegistry()
|
||||
if cfg.RegisterCleanup != nil {
|
||||
cfg.RegisterCleanup(fetchRegistry.CloseAll)
|
||||
}
|
||||
if err := runtime.Set("__gopeed_fetch_open", func(call goja.FunctionCall) goja.Value {
|
||||
reqMeta, err := exportFetchRequest(runtime, call.Argument(0))
|
||||
if err != nil {
|
||||
panic(runtime.NewGoError(err))
|
||||
}
|
||||
fingerprint := util.SafeGet[string](runtime, xhr.FingerprintMagicKey)
|
||||
promise, resolve, reject := runtime.NewPromise()
|
||||
go func() {
|
||||
meta, err := fetchRegistry.Open(fingerprint, cfg.ProxyHandler, reqMeta)
|
||||
ok := loop.RunOnLoop(func(runtime *goja.Runtime) {
|
||||
if err != nil {
|
||||
reject(runtime.NewGoError(err))
|
||||
return
|
||||
}
|
||||
resolve(runtime.ToValue(meta))
|
||||
})
|
||||
if !ok && meta != nil {
|
||||
fetchRegistry.Close(meta.ID)
|
||||
}
|
||||
}()
|
||||
return runtime.ToValue(promise)
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := runtime.Set("__gopeed_fetch_read", func(call goja.FunctionCall) goja.Value {
|
||||
id := call.Argument(0).String()
|
||||
chunkSize := int(call.Argument(1).ToInteger())
|
||||
promise, resolve, reject := runtime.NewPromise()
|
||||
go func() {
|
||||
chunk, done, err := fetchRegistry.Read(id, chunkSize)
|
||||
ok := loop.RunOnLoop(func(runtime *goja.Runtime) {
|
||||
if err != nil {
|
||||
reject(runtime.NewGoError(err))
|
||||
return
|
||||
}
|
||||
if done {
|
||||
resolve(goja.Null())
|
||||
return
|
||||
}
|
||||
resolve(runtime.ToValue(runtime.NewArrayBuffer(chunk)))
|
||||
})
|
||||
if !ok {
|
||||
fetchRegistry.Close(id)
|
||||
}
|
||||
}()
|
||||
return runtime.ToValue(promise)
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := runtime.Set("__gopeed_fetch_close", func(id string) {
|
||||
fetchRegistry.Close(id)
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := runtime.Set("__gopeed_fetch_abort", func(id string, reason string) {
|
||||
fetchRegistry.Abort(id, reason)
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := runtime.Set("__gopeed_blob_pipe_chunk", func(call goja.FunctionCall) goja.Value {
|
||||
id := call.Argument(0).String()
|
||||
ok := blobPipes.Push(id, exportBytes(call.Argument(1)))
|
||||
@@ -550,441 +475,3 @@ func exportJSError(value goja.Value) error {
|
||||
}
|
||||
return errors.New(stack)
|
||||
}
|
||||
|
||||
type fetchRegistry struct {
|
||||
mu sync.Mutex
|
||||
streams map[string]*fetchStream
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
closed bool
|
||||
}
|
||||
|
||||
type fetchStream struct {
|
||||
body io.ReadCloser
|
||||
cancel context.CancelFunc
|
||||
closeOnce sync.Once
|
||||
ctx context.Context
|
||||
ch chan fetchChunk
|
||||
readMu sync.Mutex
|
||||
pending []byte
|
||||
}
|
||||
|
||||
type fetchChunk struct {
|
||||
data []byte
|
||||
err error
|
||||
}
|
||||
|
||||
type fetchRequest struct {
|
||||
URL string
|
||||
Method string
|
||||
Headers [][2]string
|
||||
Body any
|
||||
Redirect string
|
||||
Credentials string
|
||||
}
|
||||
|
||||
type fetchFormDataEntry struct {
|
||||
name string
|
||||
value any
|
||||
}
|
||||
|
||||
type fetchFormDataSnapshot struct {
|
||||
entries []fetchFormDataEntry
|
||||
}
|
||||
|
||||
type fetchOpenMeta struct {
|
||||
ID string `json:"id"`
|
||||
Status int `json:"status"`
|
||||
StatusText string `json:"statusText"`
|
||||
URL string `json:"url"`
|
||||
Headers [][2]string `json:"headers"`
|
||||
}
|
||||
|
||||
func newFetchRegistry() *fetchRegistry {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &fetchRegistry{
|
||||
streams: make(map[string]*fetchStream),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
}
|
||||
|
||||
func exportFetchRequest(runtime *goja.Runtime, value goja.Value) (*fetchRequest, error) {
|
||||
obj := value.ToObject(runtime)
|
||||
if obj == nil {
|
||||
return nil, fmt.Errorf("invalid fetch request")
|
||||
}
|
||||
meta := &fetchRequest{
|
||||
URL: obj.Get("url").String(),
|
||||
Method: obj.Get("method").String(),
|
||||
Redirect: obj.Get("redirect").String(),
|
||||
Credentials: obj.Get("credentials").String(),
|
||||
}
|
||||
if meta.Method == "" {
|
||||
meta.Method = http.MethodGet
|
||||
}
|
||||
if headersVal := obj.Get("headers"); headersVal != nil && !goja.IsUndefined(headersVal) && !goja.IsNull(headersVal) {
|
||||
if exported, ok := headersVal.Export().([]any); ok {
|
||||
for _, item := range exported {
|
||||
pair, ok := item.([]any)
|
||||
if !ok || len(pair) != 2 {
|
||||
continue
|
||||
}
|
||||
meta.Headers = append(meta.Headers, [2]string{fmt.Sprint(pair[0]), fmt.Sprint(pair[1])})
|
||||
}
|
||||
}
|
||||
}
|
||||
bodyVal := obj.Get("body")
|
||||
if bodyVal != nil && !goja.IsUndefined(bodyVal) && !goja.IsNull(bodyVal) {
|
||||
body, err := snapshotFetchBody(bodyVal.Export())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
meta.Body = body
|
||||
}
|
||||
return meta, nil
|
||||
}
|
||||
|
||||
// snapshotFetchBody runs on the JavaScript event loop before fetch work is
|
||||
// handed to a background goroutine. Values exported by goja may still share
|
||||
// mutable storage with JavaScript (notably ArrayBuffer and FormData), so the
|
||||
// background request must only retain immutable Go-owned data.
|
||||
func snapshotFetchBody(body any) (any, error) {
|
||||
switch v := body.(type) {
|
||||
case nil, string:
|
||||
return v, nil
|
||||
case []byte:
|
||||
return append([]byte(nil), v...), nil
|
||||
case goja.ArrayBuffer:
|
||||
return append([]byte(nil), v.Bytes()...), nil
|
||||
case *file.File:
|
||||
return snapshotFetchFile(v), nil
|
||||
case *formdata.FormData:
|
||||
entries := v.Entries()
|
||||
snapshot := &fetchFormDataSnapshot{
|
||||
entries: make([]fetchFormDataEntry, 0, len(entries)),
|
||||
}
|
||||
for _, entry := range entries {
|
||||
pair, ok := entry.([]any)
|
||||
if !ok || len(pair) != 2 {
|
||||
return nil, fmt.Errorf("invalid FormData entry %T", entry)
|
||||
}
|
||||
name, ok := pair[0].(string)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("invalid FormData field name %T", pair[0])
|
||||
}
|
||||
value := pair[1]
|
||||
if formFile, ok := value.(*file.File); ok {
|
||||
value = snapshotFetchFile(formFile)
|
||||
}
|
||||
snapshot.entries = append(snapshot.entries, fetchFormDataEntry{
|
||||
name: name,
|
||||
value: value,
|
||||
})
|
||||
}
|
||||
return snapshot, nil
|
||||
default:
|
||||
if typed, ok := v.(interface{ Bytes() []byte }); ok {
|
||||
return append([]byte(nil), typed.Bytes()...), nil
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
}
|
||||
|
||||
func snapshotFetchFile(src *file.File) *file.File {
|
||||
if src == nil {
|
||||
return nil
|
||||
}
|
||||
return &file.File{
|
||||
Reader: src.Reader,
|
||||
Closer: src.Closer,
|
||||
Name: src.Name,
|
||||
Size: src.Size,
|
||||
}
|
||||
}
|
||||
|
||||
func (r *fetchRegistry) Open(fingerprint string, proxyHandler func(r *http.Request) (*url.URL, error), reqMeta *fetchRequest) (*fetchOpenMeta, error) {
|
||||
r.mu.Lock()
|
||||
if r.closed {
|
||||
r.mu.Unlock()
|
||||
return nil, context.Canceled
|
||||
}
|
||||
registryCtx := r.ctx
|
||||
r.mu.Unlock()
|
||||
|
||||
client := req.C()
|
||||
if proxyHandler != nil {
|
||||
client.SetProxy(proxyHandler)
|
||||
}
|
||||
setFetchFingerprint(client, fingerprint)
|
||||
contentType, body, err := buildFetchBody(reqMeta.Body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, cancel := context.WithCancel(registryCtx)
|
||||
keepCancel := false
|
||||
defer func() {
|
||||
if !keepCancel {
|
||||
cancel()
|
||||
}
|
||||
}()
|
||||
reqBuilder := client.R()
|
||||
reqBuilder.SetContext(ctx)
|
||||
reqBuilder.DisableAutoReadResponse()
|
||||
for _, header := range reqMeta.Headers {
|
||||
reqBuilder.SetHeader(header[0], header[1])
|
||||
}
|
||||
if body != nil && reqMeta.Method != http.MethodGet && reqMeta.Method != http.MethodHead {
|
||||
reqBuilder.SetBody(body)
|
||||
if contentType != "" && !hasHeader(reqMeta.Headers, "Content-Type") {
|
||||
reqBuilder.SetHeader("Content-Type", contentType)
|
||||
}
|
||||
}
|
||||
client.SetRedirectPolicy(func(req *http.Request, via []*http.Request) error {
|
||||
switch reqMeta.Redirect {
|
||||
case "manual":
|
||||
return http.ErrUseLastResponse
|
||||
case "error":
|
||||
return fmt.Errorf("redirect failed")
|
||||
default:
|
||||
if len(via) > 20 {
|
||||
return fmt.Errorf("too many redirects")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
})
|
||||
resp, err := reqBuilder.Send(reqMeta.Method, reqMeta.URL)
|
||||
if err != nil {
|
||||
var ne net.Error
|
||||
if errorsAsTimeout(err, &ne) {
|
||||
return nil, fmt.Errorf("Network request timed out")
|
||||
}
|
||||
return nil, fmt.Errorf("Network request failed: %w", err)
|
||||
}
|
||||
id := fmt.Sprintf("%d", time.Now().UnixNano())
|
||||
meta := &fetchOpenMeta{
|
||||
ID: id,
|
||||
Status: resp.StatusCode,
|
||||
StatusText: resp.Status,
|
||||
URL: reqMeta.URL,
|
||||
}
|
||||
if resp.Response != nil && resp.Response.Request != nil && resp.Response.Request.URL != nil {
|
||||
responseURL := *resp.Response.Request.URL
|
||||
responseURL.Fragment = ""
|
||||
meta.URL = responseURL.String()
|
||||
}
|
||||
for key, values := range resp.Header {
|
||||
meta.Headers = append(meta.Headers, [2]string{key, strings.Join(values, ", ")})
|
||||
}
|
||||
bodyCloser := io.NopCloser(strings.NewReader(""))
|
||||
if resp.Response != nil && resp.Response.Body != nil {
|
||||
bodyCloser = resp.Response.Body
|
||||
}
|
||||
stream := &fetchStream{
|
||||
body: bodyCloser,
|
||||
cancel: cancel,
|
||||
ctx: ctx,
|
||||
ch: make(chan fetchChunk, 8),
|
||||
}
|
||||
r.mu.Lock()
|
||||
if r.closed {
|
||||
r.mu.Unlock()
|
||||
closeFetchStream(stream)
|
||||
return nil, context.Canceled
|
||||
}
|
||||
r.streams[id] = stream
|
||||
r.mu.Unlock()
|
||||
keepCancel = true
|
||||
go stream.readLoop()
|
||||
return meta, nil
|
||||
}
|
||||
|
||||
func (r *fetchRegistry) Read(id string, chunkSize int) ([]byte, bool, error) {
|
||||
stream := r.get(id)
|
||||
if stream == nil {
|
||||
return nil, true, nil
|
||||
}
|
||||
stream.readMu.Lock()
|
||||
defer stream.readMu.Unlock()
|
||||
if len(stream.pending) > 0 {
|
||||
n := len(stream.pending)
|
||||
if chunkSize > 0 && n > chunkSize {
|
||||
n = chunkSize
|
||||
}
|
||||
chunk := append([]byte(nil), stream.pending[:n]...)
|
||||
stream.pending = stream.pending[n:]
|
||||
return chunk, false, nil
|
||||
}
|
||||
select {
|
||||
case item, ok := <-stream.ch:
|
||||
if !ok {
|
||||
r.Close(id)
|
||||
return nil, true, nil
|
||||
}
|
||||
if item.err != nil {
|
||||
r.Close(id)
|
||||
return nil, false, item.err
|
||||
}
|
||||
if chunkSize > 0 && len(item.data) > chunkSize {
|
||||
head := append([]byte(nil), item.data[:chunkSize]...)
|
||||
stream.pending = append(stream.pending[:0], item.data[chunkSize:]...)
|
||||
return head, false, nil
|
||||
}
|
||||
return item.data, false, nil
|
||||
case <-stream.ctx.Done():
|
||||
return nil, false, stream.ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *fetchStream) readLoop() {
|
||||
defer close(s.ch)
|
||||
buf := make([]byte, 64*1024)
|
||||
for {
|
||||
n, err := s.body.Read(buf)
|
||||
if n > 0 {
|
||||
chunk := append([]byte(nil), buf[:n]...)
|
||||
select {
|
||||
case s.ch <- fetchChunk{data: chunk}:
|
||||
case <-s.ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
if err == io.EOF {
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
select {
|
||||
case s.ch <- fetchChunk{err: err}:
|
||||
case <-s.ctx.Done():
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *fetchRegistry) Close(id string) {
|
||||
r.mu.Lock()
|
||||
stream := r.streams[id]
|
||||
delete(r.streams, id)
|
||||
r.mu.Unlock()
|
||||
closeFetchStream(stream)
|
||||
}
|
||||
|
||||
// CloseAll permanently closes the registry and cancels both active and
|
||||
// in-flight fetches. It is safe to call more than once.
|
||||
func (r *fetchRegistry) CloseAll() {
|
||||
if r == nil {
|
||||
return
|
||||
}
|
||||
r.mu.Lock()
|
||||
if r.closed {
|
||||
r.mu.Unlock()
|
||||
return
|
||||
}
|
||||
r.closed = true
|
||||
if r.cancel != nil {
|
||||
r.cancel()
|
||||
}
|
||||
streams := r.streams
|
||||
r.streams = make(map[string]*fetchStream)
|
||||
r.mu.Unlock()
|
||||
|
||||
for _, stream := range streams {
|
||||
closeFetchStream(stream)
|
||||
}
|
||||
}
|
||||
|
||||
func closeFetchStream(stream *fetchStream) {
|
||||
if stream == nil {
|
||||
return
|
||||
}
|
||||
stream.closeOnce.Do(func() {
|
||||
if stream.cancel != nil {
|
||||
stream.cancel()
|
||||
}
|
||||
_ = stream.body.Close()
|
||||
})
|
||||
}
|
||||
|
||||
func (r *fetchRegistry) Abort(id string, _ string) {
|
||||
r.Close(id)
|
||||
}
|
||||
|
||||
func (r *fetchRegistry) get(id string) *fetchStream {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.streams[id]
|
||||
}
|
||||
|
||||
func buildFetchBody(body any) (string, any, error) {
|
||||
switch v := body.(type) {
|
||||
case nil:
|
||||
return "", nil, nil
|
||||
case string:
|
||||
return "text/plain;charset=UTF-8", v, nil
|
||||
case []byte:
|
||||
return "application/octet-stream", v, nil
|
||||
case goja.ArrayBuffer:
|
||||
return "application/octet-stream", v.Bytes(), nil
|
||||
case *file.File:
|
||||
return "application/octet-stream", v.Reader, nil
|
||||
case *fetchFormDataSnapshot:
|
||||
pr, pw := io.Pipe()
|
||||
mw := xhr.NewMultipart(pw)
|
||||
for _, entry := range v.entries {
|
||||
switch vv := entry.value.(type) {
|
||||
case string:
|
||||
if err := mw.WriteField(entry.name, vv); err != nil {
|
||||
_ = pw.CloseWithError(err)
|
||||
return "", nil, err
|
||||
}
|
||||
case *file.File:
|
||||
if err := mw.WriteFile(entry.name, vv); err != nil {
|
||||
_ = pw.CloseWithError(err)
|
||||
return "", nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
go func() {
|
||||
if err := mw.Send(); err != nil {
|
||||
_ = pw.CloseWithError(err)
|
||||
return
|
||||
}
|
||||
if err := mw.Close(); err != nil {
|
||||
_ = pw.CloseWithError(err)
|
||||
return
|
||||
}
|
||||
_ = pw.Close()
|
||||
}()
|
||||
return mw.FormDataContentType(), pr, nil
|
||||
default:
|
||||
if typed, ok := v.(interface{ Bytes() []byte }); ok {
|
||||
return "application/octet-stream", typed.Bytes(), nil
|
||||
}
|
||||
return "", fmt.Sprint(v), nil
|
||||
}
|
||||
}
|
||||
|
||||
func hasHeader(headers [][2]string, key string) bool {
|
||||
for _, header := range headers {
|
||||
if strings.EqualFold(header[0], key) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func setFetchFingerprint(client *req.Client, fingerprint string) {
|
||||
switch fingerprint {
|
||||
case "chrome":
|
||||
client.ImpersonateChrome()
|
||||
case "firefox":
|
||||
client.ImpersonateFirefox()
|
||||
case "safari":
|
||||
client.ImpersonateSafari()
|
||||
}
|
||||
}
|
||||
|
||||
func errorsAsTimeout(err error, target *net.Error) bool {
|
||||
return errors.As(err, target) && (*target).Timeout()
|
||||
}
|
||||
@@ -258,7 +258,6 @@ func waitForJSValue(t *testing.T, engine *engine.Engine, expression, want string
|
||||
func newStreamTestEngine(t *testing.T, created chan<- createdObjectURL) (*engine.Engine, func()) {
|
||||
t.Helper()
|
||||
var next atomic.Int64
|
||||
var cleanup func()
|
||||
e := engine.NewEngine(&engine.Config{
|
||||
StreamConfig: &stream.Config{
|
||||
CreateObjectURL: func(opts *stream.ObjectURLOptions, open stream.ObjectURLOpener) (string, error) {
|
||||
@@ -266,13 +265,7 @@ func newStreamTestEngine(t *testing.T, created chan<- createdObjectURL) (*engine
|
||||
return fmt.Sprintf("blob:test-%d", next.Add(1)), nil
|
||||
},
|
||||
RevokeObjectURL: func(string) error { return nil },
|
||||
RegisterCleanup: func(fn func()) {
|
||||
cleanup = fn
|
||||
},
|
||||
},
|
||||
})
|
||||
if cleanup == nil {
|
||||
t.Fatal("stream cleanup was not registered")
|
||||
}
|
||||
return e, cleanup
|
||||
return e, e.Close
|
||||
}
|
||||
@@ -1,191 +0,0 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"github.com/GopeedLab/gopeed/pkg/download/engine/inject/file"
|
||||
"github.com/GopeedLab/gopeed/pkg/download/engine/inject/formdata"
|
||||
"github.com/dop251/goja"
|
||||
)
|
||||
|
||||
func TestExportFetchRequestSnapshotsArrayBuffer(t *testing.T) {
|
||||
runtime := goja.New()
|
||||
buffer := runtime.NewArrayBuffer([]byte("before"))
|
||||
request := runtime.NewObject()
|
||||
setFetchRequestDefaults(t, request)
|
||||
if err := request.Set("body", buffer); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
meta, err := exportFetchRequest(runtime, request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
buffer.Bytes()[0] = 'X'
|
||||
|
||||
body, ok := meta.Body.([]byte)
|
||||
if !ok {
|
||||
t.Fatalf("expected byte snapshot, got %T", meta.Body)
|
||||
}
|
||||
if got := string(body); got != "before" {
|
||||
t.Fatalf("ArrayBuffer snapshot changed after export: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExportFetchRequestSnapshotsFormData(t *testing.T) {
|
||||
runtime := goja.New()
|
||||
runtime.SetFieldNameMapper(goja.TagFieldNameMapper("json", true))
|
||||
if err := formdata.Enable(runtime); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := runtime.RunString(`
|
||||
globalThis.form = new FormData();
|
||||
form.append("field", "before");
|
||||
form;
|
||||
`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
request := runtime.NewObject()
|
||||
setFetchRequestDefaults(t, request)
|
||||
if err := request.Set("body", value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
meta, err := exportFetchRequest(runtime, request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := runtime.RunString(`form.set("field", "after")`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
body, ok := meta.Body.(*fetchFormDataSnapshot)
|
||||
if !ok {
|
||||
t.Fatalf("expected FormData snapshot, got %T", meta.Body)
|
||||
}
|
||||
if len(body.entries) != 1 {
|
||||
t.Fatalf("expected one FormData entry, got %d", len(body.entries))
|
||||
}
|
||||
if got := body.entries[0]; got.name != "field" || got.value != "before" {
|
||||
t.Fatalf("unexpected FormData snapshot: %#v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func setFetchRequestDefaults(t *testing.T, request *goja.Object) {
|
||||
t.Helper()
|
||||
for key, value := range map[string]any{
|
||||
"url": "https://example.test",
|
||||
"method": "POST",
|
||||
"redirect": "follow",
|
||||
"credentials": "same-origin",
|
||||
} {
|
||||
if err := request.Set(key, value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSnapshotFetchBodyCopiesFileMetadata(t *testing.T) {
|
||||
source := &file.File{
|
||||
Reader: strings.NewReader("payload"),
|
||||
Name: "before.txt",
|
||||
Size: 7,
|
||||
}
|
||||
|
||||
value, err := snapshotFetchBody(source)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
snapshot, ok := value.(*file.File)
|
||||
if !ok {
|
||||
t.Fatalf("expected file snapshot, got %T", value)
|
||||
}
|
||||
source.Name = "after.txt"
|
||||
source.Size = 99
|
||||
|
||||
if snapshot.Name != "before.txt" || snapshot.Size != 7 {
|
||||
t.Fatalf("file metadata was not snapshotted: %#v", snapshot)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFetchRegistryCloseAll(t *testing.T) {
|
||||
registry := newFetchRegistry()
|
||||
streamCtx, cancel := context.WithCancel(context.Background())
|
||||
body := &trackingReadCloser{}
|
||||
registry.streams["active"] = &fetchStream{
|
||||
body: body,
|
||||
cancel: cancel,
|
||||
ctx: streamCtx,
|
||||
ch: make(chan fetchChunk),
|
||||
}
|
||||
|
||||
registry.CloseAll()
|
||||
registry.CloseAll()
|
||||
|
||||
select {
|
||||
case <-streamCtx.Done():
|
||||
default:
|
||||
t.Fatal("active fetch context was not canceled")
|
||||
}
|
||||
if got := body.closeCalls.Load(); got != 1 {
|
||||
t.Fatalf("expected body to close once, got %d", got)
|
||||
}
|
||||
registry.mu.Lock()
|
||||
closed := registry.closed
|
||||
remaining := len(registry.streams)
|
||||
registry.mu.Unlock()
|
||||
if !closed || remaining != 0 {
|
||||
t.Fatalf("registry not fully closed: closed=%v remaining=%d", closed, remaining)
|
||||
}
|
||||
if _, err := registry.Open("", nil, &fetchRequest{URL: "https://example.test"}); err == nil {
|
||||
t.Fatal("expected a closed registry to reject new fetches")
|
||||
}
|
||||
}
|
||||
|
||||
func TestFetchRegistryReadPreservesChunkOrder(t *testing.T) {
|
||||
registry := newFetchRegistry()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
stream := &fetchStream{
|
||||
body: io.NopCloser(strings.NewReader("")),
|
||||
cancel: cancel,
|
||||
ctx: ctx,
|
||||
ch: make(chan fetchChunk, 2),
|
||||
}
|
||||
stream.ch <- fetchChunk{data: []byte("abc")}
|
||||
stream.ch <- fetchChunk{data: []byte("def")}
|
||||
registry.streams["ordered"] = stream
|
||||
defer registry.CloseAll()
|
||||
|
||||
var got strings.Builder
|
||||
for range 4 {
|
||||
chunk, done, err := registry.Read("ordered", 2)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if done {
|
||||
t.Fatal("stream ended before buffered chunks were read")
|
||||
}
|
||||
got.Write(chunk)
|
||||
}
|
||||
if got.String() != "abcdef" {
|
||||
t.Fatalf("fetch chunks were reordered: %q", got.String())
|
||||
}
|
||||
}
|
||||
|
||||
type trackingReadCloser struct {
|
||||
closeCalls atomic.Int32
|
||||
}
|
||||
|
||||
func (r *trackingReadCloser) Read([]byte) (int, error) {
|
||||
return 0, io.EOF
|
||||
}
|
||||
|
||||
func (r *trackingReadCloser) Close() error {
|
||||
r.closeCalls.Add(1)
|
||||
return nil
|
||||
}
|
||||
@@ -1,10 +1,6 @@
|
||||
(function () {
|
||||
const createBlobObjectURL = globalThis.__gopeed_create_blob_object_url;
|
||||
const revokeObjectURL = globalThis.__gopeed_revoke_object_url;
|
||||
const fetchOpen = globalThis.__gopeed_fetch_open;
|
||||
const fetchRead = globalThis.__gopeed_fetch_read;
|
||||
const fetchClose = globalThis.__gopeed_fetch_close;
|
||||
const fetchAbort = globalThis.__gopeed_fetch_abort;
|
||||
|
||||
if (typeof globalThis.ReadableStream === "undefined") {
|
||||
class ReadableStreamDefaultController {
|
||||
@@ -30,7 +26,10 @@
|
||||
this._stream = stream;
|
||||
}
|
||||
|
||||
read() {
|
||||
read(view) {
|
||||
if (view !== undefined) {
|
||||
return this._stream._readInto(view);
|
||||
}
|
||||
return this._stream._read();
|
||||
}
|
||||
|
||||
@@ -59,6 +58,7 @@
|
||||
this._errored = null;
|
||||
this._reader = null;
|
||||
this._pulling = false;
|
||||
this._disturbed = false;
|
||||
this.locked = false;
|
||||
this._controller = new ReadableStreamDefaultController(this);
|
||||
if (typeof source.start === "function") {
|
||||
@@ -96,6 +96,7 @@
|
||||
if (this._errored) {
|
||||
return Promise.reject(this._errored);
|
||||
}
|
||||
this._disturbed = true;
|
||||
this._markBodyUsed();
|
||||
if (this._queue.length > 0) {
|
||||
return Promise.resolve({ done: false, value: this._queue.shift() });
|
||||
@@ -116,6 +117,26 @@
|
||||
});
|
||||
}
|
||||
|
||||
async _readInto(view) {
|
||||
if (!ArrayBuffer.isView(view) || view.byteLength === 0) {
|
||||
throw new TypeError("BYOB reads require a non-empty ArrayBufferView");
|
||||
}
|
||||
const result = await this._read();
|
||||
if (result.done) {
|
||||
return { done: true, value: new Uint8Array(view.buffer, view.byteOffset, 0) };
|
||||
}
|
||||
if (!(result.value instanceof Uint8Array)) {
|
||||
throw new TypeError("Byte streams require Uint8Array chunks");
|
||||
}
|
||||
const length = Math.min(view.byteLength, result.value.byteLength);
|
||||
const target = new Uint8Array(view.buffer, view.byteOffset, length);
|
||||
target.set(result.value.subarray(0, length));
|
||||
if (length < result.value.byteLength) {
|
||||
this._queue.unshift(result.value.subarray(length));
|
||||
}
|
||||
return { done: false, value: target };
|
||||
}
|
||||
|
||||
_maybePull() {
|
||||
if (this._closed || this._errored || this._pulling) {
|
||||
return;
|
||||
@@ -155,6 +176,7 @@
|
||||
}
|
||||
|
||||
cancel(reason) {
|
||||
this._disturbed = true;
|
||||
this._markBodyUsed();
|
||||
if (this._source && typeof this._source.cancel === "function") {
|
||||
return Promise.resolve(this._source.cancel(reason)).then(() => {
|
||||
@@ -185,6 +207,69 @@
|
||||
writer.releaseLock();
|
||||
}
|
||||
}
|
||||
|
||||
tee() {
|
||||
const reader = this.getReader();
|
||||
let reading = false;
|
||||
let currentRead = null;
|
||||
let finished = false;
|
||||
const waiting = [[], []];
|
||||
const queues = [[], []];
|
||||
|
||||
const cloneChunk = (chunk) => {
|
||||
if (chunk instanceof ArrayBuffer) return chunk.slice(0);
|
||||
if (ArrayBuffer.isView(chunk)) {
|
||||
const buffer = chunk.buffer.slice(0);
|
||||
if (chunk instanceof DataView) return new DataView(buffer, chunk.byteOffset, chunk.byteLength);
|
||||
return new chunk.constructor(buffer, chunk.byteOffset, chunk.length);
|
||||
}
|
||||
return chunk;
|
||||
};
|
||||
const flush = (index) => {
|
||||
while (waiting[index].length && queues[index].length) {
|
||||
waiting[index].shift().enqueue(queues[index].shift());
|
||||
}
|
||||
if (finished && queues[index].length === 0) {
|
||||
while (waiting[index].length) waiting[index].shift().close();
|
||||
}
|
||||
};
|
||||
const pump = () => {
|
||||
if (reading) return currentRead;
|
||||
if (finished) return Promise.resolve();
|
||||
reading = true;
|
||||
currentRead = (async () => {
|
||||
try {
|
||||
const result = await reader.read();
|
||||
if (result.done) {
|
||||
finished = true;
|
||||
} else {
|
||||
queues[0].push(result.value);
|
||||
queues[1].push(cloneChunk(result.value));
|
||||
}
|
||||
} catch (error) {
|
||||
finished = true;
|
||||
for (const branches of waiting) {
|
||||
while (branches.length) branches.shift().error(error);
|
||||
}
|
||||
throw error;
|
||||
} finally {
|
||||
reading = false;
|
||||
currentRead = null;
|
||||
flush(0);
|
||||
flush(1);
|
||||
}
|
||||
})();
|
||||
return currentRead;
|
||||
};
|
||||
const branch = (index) => new ReadableStream({
|
||||
pull(controller) {
|
||||
waiting[index].push(controller);
|
||||
flush(index);
|
||||
return pump();
|
||||
},
|
||||
});
|
||||
return [branch(0), branch(1)];
|
||||
}
|
||||
}
|
||||
|
||||
globalThis.ReadableStream = ReadableStream;
|
||||
@@ -351,167 +436,6 @@
|
||||
};
|
||||
}
|
||||
|
||||
function toUint8Array(chunk) {
|
||||
if (chunk == null) {
|
||||
return new Uint8Array(0);
|
||||
}
|
||||
if (chunk instanceof Uint8Array) {
|
||||
return chunk;
|
||||
}
|
||||
if (typeof chunk === "string") {
|
||||
return new TextEncoder().encode(chunk);
|
||||
}
|
||||
if (chunk instanceof ArrayBuffer) {
|
||||
return new Uint8Array(chunk);
|
||||
}
|
||||
if (typeof ArrayBuffer !== "undefined" && ArrayBuffer.isView && ArrayBuffer.isView(chunk)) {
|
||||
return new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength);
|
||||
}
|
||||
if (typeof Blob !== "undefined" && chunk instanceof Blob) {
|
||||
if (chunk._buffer instanceof Uint8Array) {
|
||||
return chunk._buffer;
|
||||
}
|
||||
}
|
||||
return new Uint8Array(0);
|
||||
}
|
||||
|
||||
function createBodyReadableStream(owner) {
|
||||
return new ReadableStream({
|
||||
start(controller) {
|
||||
Promise.resolve().then(async () => {
|
||||
owner.bodyUsed = true;
|
||||
if (owner._noBody) {
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
if (owner._bodyArrayBuffer) {
|
||||
controller.enqueue(toUint8Array(owner._bodyArrayBuffer));
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
if (owner._bodyBlob) {
|
||||
const data = await owner._bodyBlob.arrayBuffer();
|
||||
controller.enqueue(new Uint8Array(data));
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
if (owner._bodyText != null) {
|
||||
controller.enqueue(toUint8Array(owner._bodyText));
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
if (owner._bodyInit != null) {
|
||||
controller.enqueue(toUint8Array(owner._bodyInit));
|
||||
}
|
||||
controller.close();
|
||||
}).catch((err) => {
|
||||
controller.error(err);
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async function readAllFromStream(stream, asText) {
|
||||
if (!stream) {
|
||||
return asText ? "" : new Uint8Array(0);
|
||||
}
|
||||
const reader = stream.getReader();
|
||||
const chunks = [];
|
||||
let total = 0;
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) {
|
||||
break;
|
||||
}
|
||||
const chunk = toUint8Array(value);
|
||||
chunks.push(chunk);
|
||||
total += chunk.byteLength;
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
const merged = new Uint8Array(total);
|
||||
let offset = 0;
|
||||
for (const chunk of chunks) {
|
||||
merged.set(chunk, offset);
|
||||
offset += chunk.byteLength;
|
||||
}
|
||||
if (asText) {
|
||||
return new TextDecoder().decode(merged);
|
||||
}
|
||||
return merged;
|
||||
}
|
||||
|
||||
function attachResponseStreaming(response, stream) {
|
||||
response.__gopeedBodyStream = stream;
|
||||
response.__gopeedBodyConsumed = false;
|
||||
const ensureUnused = () => {
|
||||
if (response.__gopeedBodyConsumed) {
|
||||
throw new TypeError("Already read");
|
||||
}
|
||||
};
|
||||
const markBodyUsed = () => {
|
||||
ensureUnused();
|
||||
response.__gopeedBodyConsumed = true;
|
||||
response.bodyUsed = true;
|
||||
};
|
||||
if (stream) {
|
||||
stream.__gopeedMarkBodyUsed = () => {
|
||||
if (!response.__gopeedBodyConsumed) {
|
||||
response.__gopeedBodyConsumed = true;
|
||||
response.bodyUsed = true;
|
||||
}
|
||||
};
|
||||
}
|
||||
response.text = async function () {
|
||||
ensureUnused();
|
||||
markBodyUsed();
|
||||
return readAllFromStream(stream, true);
|
||||
};
|
||||
response.arrayBuffer = async function () {
|
||||
ensureUnused();
|
||||
markBodyUsed();
|
||||
const bytes = await readAllFromStream(stream, false);
|
||||
return bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength);
|
||||
};
|
||||
response.blob = async function () {
|
||||
ensureUnused();
|
||||
markBodyUsed();
|
||||
const contentType = this.headers && this.headers.get ? (this.headers.get("content-type") || "") : "";
|
||||
// Response.blob() is a materializing body consumer. Keeping the live
|
||||
// response stream behind an empty one-shot Blob makes retry, range and a
|
||||
// second object-URL reader silently produce an empty file.
|
||||
const bytes = await readAllFromStream(stream, false);
|
||||
return new Blob([bytes], { type: contentType });
|
||||
};
|
||||
response.json = async function () {
|
||||
const text = await this.text();
|
||||
return JSON.parse(text);
|
||||
};
|
||||
return response;
|
||||
}
|
||||
|
||||
if (typeof globalThis.Response === "function") {
|
||||
const responseProto = globalThis.Response.prototype;
|
||||
const bodyDescriptor = Object.getOwnPropertyDescriptor(responseProto, "body");
|
||||
if (!bodyDescriptor || typeof bodyDescriptor.get !== "function") {
|
||||
Object.defineProperty(responseProto, "body", {
|
||||
configurable: true,
|
||||
enumerable: true,
|
||||
get() {
|
||||
if (this.__gopeedBodyStream) {
|
||||
return this.__gopeedBodyStream;
|
||||
}
|
||||
if (!this.__gopeedBodyStream) {
|
||||
this.__gopeedBodyStream = createBodyReadableStream(this);
|
||||
}
|
||||
return this.__gopeedBodyStream;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const blobReaders = new Map();
|
||||
|
||||
function getValueTypeName(value) {
|
||||
@@ -883,82 +807,4 @@
|
||||
}
|
||||
};
|
||||
|
||||
const originalFetch = typeof globalThis.fetch === "function"
|
||||
? globalThis.fetch.bind(globalThis)
|
||||
: null;
|
||||
|
||||
if (typeof fetchOpen === "function") {
|
||||
globalThis.fetch = async function (input, init) {
|
||||
const request = new Request(input, init);
|
||||
if (originalFetch && (request.method === "HEAD" || request.redirect === "manual" || request.redirect === "error")) {
|
||||
return originalFetch(input, init);
|
||||
}
|
||||
let body = null;
|
||||
if (request._bodyFormData) {
|
||||
body = request._bodyFormData;
|
||||
} else if (request._bodyArrayBuffer) {
|
||||
body = request._bodyArrayBuffer;
|
||||
} else if (request._bodyBlob) {
|
||||
body = await request._bodyBlob.arrayBuffer();
|
||||
} else if (request._bodyInit != null && typeof request._bodyInit === "object") {
|
||||
body = request._bodyInit;
|
||||
} else if (request._bodyText != null) {
|
||||
body = request._bodyText;
|
||||
} else if (request._bodyInit != null) {
|
||||
body = request._bodyInit;
|
||||
}
|
||||
const headers = [];
|
||||
request.headers.forEach((value, key) => {
|
||||
headers.push([key, value]);
|
||||
});
|
||||
let meta;
|
||||
try {
|
||||
meta = await fetchOpen({
|
||||
url: request.url,
|
||||
method: request.method,
|
||||
headers,
|
||||
body,
|
||||
redirect: request.redirect,
|
||||
credentials: request.credentials
|
||||
});
|
||||
} catch (error) {
|
||||
throw error instanceof Error ? error : new TypeError(String(error));
|
||||
}
|
||||
const stream = new ReadableStream({
|
||||
async pull(controller) {
|
||||
let chunk;
|
||||
try {
|
||||
chunk = await fetchRead(meta.id, 64 * 1024);
|
||||
} catch (error) {
|
||||
fetchClose(meta.id);
|
||||
controller.error(error instanceof Error ? error : new TypeError(String(error)));
|
||||
return;
|
||||
}
|
||||
if (chunk == null) {
|
||||
fetchClose(meta.id);
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
const bytes = chunk instanceof Uint8Array ? chunk : new Uint8Array(chunk);
|
||||
if (bytes.byteLength === 0) {
|
||||
fetchClose(meta.id);
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
controller.enqueue(bytes);
|
||||
},
|
||||
cancel(reason) {
|
||||
fetchAbort(meta.id, reason == null ? "" : String(reason));
|
||||
}
|
||||
});
|
||||
const response = new Response(null, {
|
||||
status: meta.status,
|
||||
statusText: meta.statusText,
|
||||
headers: meta.headers,
|
||||
url: meta.url
|
||||
});
|
||||
return attachResponseStreaming(response, stream);
|
||||
};
|
||||
globalThis.fetch.__gopeedOriginalFetch = originalFetch;
|
||||
}
|
||||
})();
|
||||
Reference in new issue
Block a user