TRKJ/EdgeNode
2
0
Fork 0
mirror of https://github.com/TeaOSLab/EdgeNode.git synced 2025-11-18 03:35:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

缓存支持Range分片查询

Browse Source
This commit is contained in:
GoEdgeLab
2021-01-13 12:52:38 +08:00
parent ba6fa92dc9
commit 15e4045ec9
3 changed files with 181 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
internal/nodes/http_request_cache.go
View File

@@ -5,6 +5,7 @@ import (
"errors"
"github.com/TeaOSLab/EdgeNode/internal/caches"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/iwind/TeaGo/logs"
"net/http"
"strconv"
)
@@ -115,10 +116,142 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
}
this.processResponseHeaders(reader.Status())
this.writer.WriteHeader(reader.Status())
// 输出Body
if this.RawReq.Method != http.MethodHead {
if this.RawReq.Method == http.MethodHead {
this.writer.WriteHeader(reader.Status())
} else {
ifRangeHeaders, ok := this.RawReq.Header["If-Range"]
supportRange := true
if ok {
supportRange = false
for _, v := range ifRangeHeaders {
if v == this.writer.Header().Get("ETag") || v == this.writer.Header().Get("Last-Modified") {
supportRange = true
}
}
}
// 支持Range
rangeSet := [][]int64{}
if supportRange {
fileSize := reader.BodySize()
contentRange := this.RawReq.Header.Get("Range")
if len(contentRange) > 0 {
if fileSize == 0 {
this.processResponseHeaders(http.StatusRequestedRangeNotSatisfiable)
this.writer.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
return true
}
set, ok := httpRequestParseContentRange(contentRange)
if !ok {
this.processResponseHeaders(http.StatusRequestedRangeNotSatisfiable)
this.writer.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
return true
}
if len(set) > 0 {
rangeSet = set
for _, arr := range rangeSet {
if arr[0] == -1 {
arr[0] = fileSize + arr[1]
arr[1] = fileSize - 1
if arr[0] < 0 {
this.processResponseHeaders(http.StatusRequestedRangeNotSatisfiable)
this.writer.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
return true
}
}
if arr[1] < 0 {
arr[1] = fileSize - 1
}
if arr[1] >= fileSize {
arr[1] = fileSize - 1
}
if arr[0] > arr[1] {
this.processResponseHeaders(http.StatusRequestedRangeNotSatisfiable)
this.writer.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
return true
}
}
}
}
}
respHeader := this.writer.Header()
if len(rangeSet) == 1 {
respHeader.Set("Content-Range", "bytes "+strconv.FormatInt(rangeSet[0][0], 10)+"-"+strconv.FormatInt(rangeSet[0][1], 10)+"/"+strconv.FormatInt(reader.BodySize(), 10))
respHeader.Set("Content-Length", strconv.FormatInt(rangeSet[0][1]-rangeSet[0][0]+1, 10))
this.writer.WriteHeader(http.StatusPartialContent)
err = reader.ReadBodyRange(buf, rangeSet[0][0], rangeSet[0][1], func(n int) (goNext bool, err error) {
_, err = this.writer.Write(buf[:n])
if err != nil {
return false, err
}
return true, nil
})
if err != nil {
if err == caches.ErrInvalidRange {
this.processResponseHeaders(http.StatusRequestedRangeNotSatisfiable)
this.writer.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
return true
}
remotelogs.Error("REQUEST_CACHE", "read from cache failed: "+err.Error())
return
}
} else if len(rangeSet) > 1 {
boundary := httpRequestGenBoundary()
respHeader.Set("Content-Type", "multipart/byteranges; boundary="+boundary)
respHeader.Del("Content-Length")
contentType := respHeader.Get("Content-Type")
this.writer.WriteHeader(http.StatusPartialContent)
for index, set := range rangeSet {
if index == 0 {
_, err = this.writer.WriteString("--" + boundary + "\r\n")
} else {
_, err = this.writer.WriteString("\r\n--" + boundary + "\r\n")
}
if err != nil {
logs.Error(err)
return true
}
_, err = this.writer.WriteString("Content-Range: " + "bytes " + strconv.FormatInt(set[0], 10) + "-" + strconv.FormatInt(set[1], 10) + "/" + strconv.FormatInt(reader.BodySize(), 10) + "\r\n")
if err != nil {
logs.Error(err)
return true
}
if len(contentType) > 0 {
_, err = this.writer.WriteString("Content-Type: " + contentType + "\r\n\r\n")
if err != nil {
logs.Error(err)
return true
}
}
err := reader.ReadBodyRange(buf, set[0], set[1], func(n int) (goNext bool, err error) {
_, err = this.writer.Write(buf[:n])
return true, err
})
if err != nil {
remotelogs.Error("REQUEST_CACHE", "read from cache failed: "+err.Error())
return true
}
}
_, err = this.writer.WriteString("\r\n--" + boundary + "--\r\n")
if err != nil {
logs.Error(err)
return true
}
} else {
this.writer.WriteHeader(reader.Status())
err = reader.ReadBody(buf, func(n int) (goNext bool, err error) {
_, err = this.writer.Write(buf[:n])
if err != nil {
@@ -131,6 +264,7 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
return
}
}
}
this.cacheRef = nil // 终止读取不再往下传递
return true

30
internal/nodes/http_request_root.go
View File

@@ -224,13 +224,22 @@ func (this *HTTPRequest) doRoot() (isBreak bool) {
supportRange = true
}
}
if !supportRange {
respHeader.Del("Accept-Ranges")
}
}
// 支持Range
rangeSet := [][]int{}
rangeSet := [][]int64{}
if supportRange {
contentRange := this.RawReq.Header.Get("Range")
if len(contentRange) > 0 {
if fileSize == 0 {
this.processResponseHeaders(http.StatusRequestedRangeNotSatisfiable)
this.writer.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
return true
}
set, ok := httpRequestParseContentRange(contentRange)
if !ok {
this.processResponseHeaders(http.StatusRequestedRangeNotSatisfiable)
@@ -241,8 +250,8 @@ func (this *HTTPRequest) doRoot() (isBreak bool) {
rangeSet = set
for _, arr := range rangeSet {
if arr[0] == -1 {
arr[0] = int(fileSize) + arr[1]
arr[1] = int(fileSize) - 1
arr[0] = fileSize + arr[1]
arr[1] = fileSize - 1
if arr[0] < 0 {
this.processResponseHeaders(http.StatusRequestedRangeNotSatisfiable)
@@ -250,6 +259,17 @@ func (this *HTTPRequest) doRoot() (isBreak bool) {
return true
}
}
if arr[1] > 0 {
arr[1] = fileSize - 1
}
if arr[1] < 0 {
arr[1] = fileSize - 1
}
if arr[0] > arr[1] {
this.processResponseHeaders(http.StatusRequestedRangeNotSatisfiable)
this.writer.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
return true
}
}
}
} else {
@@ -284,7 +304,7 @@ func (this *HTTPRequest) doRoot() (isBreak bool) {
}()
if len(rangeSet) == 1 {
respHeader.Set("Content-Range", "bytes "+strconv.Itoa(rangeSet[0][0])+"-"+strconv.Itoa(rangeSet[0][1])+"/"+strconv.FormatInt(fileSize, 10))
respHeader.Set("Content-Range", "bytes "+strconv.FormatInt(rangeSet[0][0], 10)+"-"+strconv.FormatInt(rangeSet[0][1], 10)+"/"+strconv.FormatInt(fileSize, 10))
this.writer.WriteHeader(http.StatusPartialContent)
ok, err := httpRequestReadRange(reader, buf, rangeSet[0][0], rangeSet[0][1], func(buf []byte, n int) error {
@@ -317,7 +337,7 @@ func (this *HTTPRequest) doRoot() (isBreak bool) {
return true
}
_, err = this.writer.WriteString("Content-Range: " + "bytes " + strconv.Itoa(set[0]) + "-" + strconv.Itoa(set[1]) + "/" + strconv.FormatInt(fileSize, 10) + "\r\n")
_, err = this.writer.WriteString("Content-Range: " + "bytes " + strconv.FormatInt(set[0], 10) + "-" + strconv.FormatInt(set[1], 10) + "/" + strconv.FormatInt(fileSize, 10) + "\r\n")
if err != nil {
logs.Error(err)
return true

22
internal/nodes/http_request_utils.go
View File

@@ -9,7 +9,7 @@ import (
)
// 分解Range
func httpRequestParseContentRange(rangeValue string) (result [][]int, ok bool) {
func httpRequestParseContentRange(rangeValue string) (result [][]int64, ok bool) {
// 参考RFC：https://tools.ietf.org/html/rfc7233
index := strings.Index(rangeValue, "=")
if index == -1 {
@@ -33,20 +33,20 @@ func httpRequestParseContentRange(rangeValue string) (result [][]int, ok bool) {
return
}
first := piece[:index]
firstInt := -1
firstInt := int64(-1)
var err error
last := piece[index+1:]
var lastInt = -1
var lastInt = int64(-1)
if len(first) > 0 {
firstInt, err = strconv.Atoi(first)
firstInt, err = strconv.ParseInt(first, 10, 64)
if err != nil {
return
}
if len(last) > 0 {
lastInt, err = strconv.Atoi(last)
lastInt, err = strconv.ParseInt(last, 10, 64)
if err != nil {
return
}
@@ -59,14 +59,14 @@ func httpRequestParseContentRange(rangeValue string) (result [][]int, ok bool) {
return
}
lastInt, err = strconv.Atoi(last)
lastInt, err = strconv.ParseInt(last, 10, 64)
if err != nil {
return
}
lastInt = -lastInt
}
result = append(result, []int{firstInt, lastInt})
result = append(result, []int64{firstInt, lastInt})
}
ok = true
@@ -74,7 +74,7 @@ func httpRequestParseContentRange(rangeValue string) (result [][]int, ok bool) {
}
// 读取内容Range
func httpRequestReadRange(reader io.Reader, buf []byte, start int, end int, callback func(buf []byte, n int) error) (ok bool, err error) {
func httpRequestReadRange(reader io.Reader, buf []byte, start int64, end int64, callback func(buf []byte, n int) error) (ok bool, err error) {
if start < 0 || end < 0 {
return
}
@@ -82,7 +82,7 @@ func httpRequestReadRange(reader io.Reader, buf []byte, start int, end int, call
if !ok {
return
}
_, err = seeker.Seek(int64(start), io.SeekStart)
_, err = seeker.Seek(start, io.SeekStart)
if err != nil {
return false, nil
}
@@ -91,9 +91,9 @@ func httpRequestReadRange(reader io.Reader, buf []byte, start int, end int, call
for {
n, err := reader.Read(buf)
if n > 0 {
offset += n
offset += int64(n)
if end < offset {
err = callback(buf, n-(offset-end-1))
err = callback(buf, n-int(offset-end-1))
if err != nil {
return false, err
}