XHTTP: Add "stream-up" mode for client & server (#3994)

transport/internet/splithttp/upload_queue.go

@@ -6,17 +6,20 @@ package splithttp
 import (
 	"container/heap"
 	"io"
+	"runtime"
 	"sync"
 
 	"github.com/xtls/xray-core/common/errors"
 )
 
 type Packet struct {
+	Reader  io.ReadCloser
 	Payload []byte
 	Seq     uint64
 }
 
 type uploadQueue struct {
+	reader          io.ReadCloser
 	pushedPackets   chan Packet
 	writeCloseMutex sync.Mutex
 	heap            uploadHeap
@@ -39,7 +42,16 @@ func (h *uploadQueue) Push(p Packet) error {
 	h.writeCloseMutex.Lock()
 	defer h.writeCloseMutex.Unlock()
 
+	runtime.Gosched()
+	if h.reader != nil && p.Reader != nil {
+		p.Reader.Close()
+		return errors.New("h.reader already exists")
+	}
+
 	if h.closed {
+		if p.Reader != nil {
+			p.Reader.Close()
+		}
 		return errors.New("splithttp packet queue closed")
 	}
 
@@ -55,10 +67,18 @@ func (h *uploadQueue) Close() error {
 		h.closed = true
 		close(h.pushedPackets)
 	}
+	runtime.Gosched()
+	if h.reader != nil {
+		return h.reader.Close()
+	}
 	return nil
 }
 
 func (h *uploadQueue) Read(b []byte) (int, error) {
+	if h.reader != nil {
+		return h.reader.Read(b)
+	}
+
 	if h.closed {
 		return 0, io.EOF
 	}
@@ -68,6 +88,10 @@ func (h *uploadQueue) Read(b []byte) (int, error) {
 		if !more {
 			return 0, io.EOF
 		}
+		if packet.Reader != nil {
+			h.reader = packet.Reader
+			return h.reader.Read(b)
+		}
 		heap.Push(&h.heap, packet)
 	}