splithttp Read() using blocking mode (#3473)

* blocking splithttp read

* Add testcase

* simplify conditions

---------

Co-authored-by: mmmray <142015632+mmmray@users.noreply.github.com>
This commit is contained in:
maskedeken 2024-06-24 23:09:24 +08:00 committed by GitHub
parent 7acd5a623b
commit e4f9d03bef
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 36 additions and 14 deletions

View file

@ -47,13 +47,19 @@ func (h *UploadQueue) Close() error {
} }
func (h *UploadQueue) Read(b []byte) (int, error) { func (h *UploadQueue) Read(b []byte) (int, error) {
if h.closed && len(h.heap) == 0 && len(h.pushedPackets) == 0 { if h.closed {
return 0, io.EOF return 0, io.EOF
} }
needMorePackets := false if len(h.heap) == 0 {
packet, more := <-h.pushedPackets
if !more {
return 0, io.EOF
}
heap.Push(&h.heap, packet)
}
if len(h.heap) > 0 { for len(h.heap) > 0 {
packet := heap.Pop(&h.heap).(Packet) packet := heap.Pop(&h.heap).(Packet)
n := 0 n := 0
@ -81,18 +87,12 @@ func (h *UploadQueue) Read(b []byte) (int, error) {
return 0, newError("packet queue is too large") return 0, newError("packet queue is too large")
} }
heap.Push(&h.heap, packet) heap.Push(&h.heap, packet)
needMorePackets = true packet2, more := <-h.pushedPackets
}
} else {
needMorePackets = true
}
if needMorePackets {
packet, more := <-h.pushedPackets
if !more { if !more {
return 0, io.EOF return 0, io.EOF
} }
heap.Push(&h.heap, packet) heap.Push(&h.heap, packet2)
}
} }
return 0, nil return 0, nil

View file

@ -0,0 +1,22 @@
package splithttp_test
import (
"testing"
"github.com/xtls/xray-core/common"
. "github.com/xtls/xray-core/transport/internet/splithttp"
)
func Test_regression_readzero(t *testing.T) {
q := NewUploadQueue(10)
q.Push(Packet{
Payload: []byte("x"),
Seq: 0,
})
buf := make([]byte, 20)
n, err := q.Read(buf)
common.Must(err)
if n != 1 {
t.Error("n=", n)
}
}