|
@@ -5,6 +5,7 @@ package dispatcher
|
|
|
import (
|
|
import (
|
|
|
"context"
|
|
"context"
|
|
|
"strings"
|
|
"strings"
|
|
|
|
|
+ "sync"
|
|
|
"time"
|
|
"time"
|
|
|
|
|
|
|
|
"v2ray.com/core"
|
|
"v2ray.com/core"
|
|
@@ -24,21 +25,27 @@ var (
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
type cachedReader struct {
|
|
type cachedReader struct {
|
|
|
|
|
+ sync.Mutex
|
|
|
reader *pipe.Reader
|
|
reader *pipe.Reader
|
|
|
cache buf.MultiBuffer
|
|
cache buf.MultiBuffer
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
func (r *cachedReader) Cache(b *buf.Buffer) {
|
|
func (r *cachedReader) Cache(b *buf.Buffer) {
|
|
|
mb, _ := r.reader.ReadMultiBufferTimeout(time.Millisecond * 100)
|
|
mb, _ := r.reader.ReadMultiBufferTimeout(time.Millisecond * 100)
|
|
|
|
|
+ r.Lock()
|
|
|
if !mb.IsEmpty() {
|
|
if !mb.IsEmpty() {
|
|
|
common.Must(r.cache.WriteMultiBuffer(mb))
|
|
common.Must(r.cache.WriteMultiBuffer(mb))
|
|
|
}
|
|
}
|
|
|
common.Must(b.Reset(func(x []byte) (int, error) {
|
|
common.Must(b.Reset(func(x []byte) (int, error) {
|
|
|
return r.cache.Copy(x), nil
|
|
return r.cache.Copy(x), nil
|
|
|
}))
|
|
}))
|
|
|
|
|
+ r.Unlock()
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
func (r *cachedReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
|
|
func (r *cachedReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
|
|
|
|
|
+ r.Lock()
|
|
|
|
|
+ defer r.Unlock()
|
|
|
|
|
+
|
|
|
if !r.cache.IsEmpty() {
|
|
if !r.cache.IsEmpty() {
|
|
|
mb := r.cache
|
|
mb := r.cache
|
|
|
r.cache = nil
|
|
r.cache = nil
|
|
@@ -49,6 +56,9 @@ func (r *cachedReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
func (r *cachedReader) ReadMultiBufferTimeout(timeout time.Duration) (buf.MultiBuffer, error) {
|
|
func (r *cachedReader) ReadMultiBufferTimeout(timeout time.Duration) (buf.MultiBuffer, error) {
|
|
|
|
|
+ r.Lock()
|
|
|
|
|
+ defer r.Unlock()
|
|
|
|
|
+
|
|
|
if !r.cache.IsEmpty() {
|
|
if !r.cache.IsEmpty() {
|
|
|
mb := r.cache
|
|
mb := r.cache
|
|
|
r.cache = nil
|
|
r.cache = nil
|
|
@@ -59,7 +69,10 @@ func (r *cachedReader) ReadMultiBufferTimeout(timeout time.Duration) (buf.MultiB
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
func (r *cachedReader) CloseError() {
|
|
func (r *cachedReader) CloseError() {
|
|
|
|
|
+ r.Lock()
|
|
|
r.cache.Release()
|
|
r.cache.Release()
|
|
|
|
|
+ r.cache = nil
|
|
|
|
|
+ r.Unlock()
|
|
|
r.reader.CloseError()
|
|
r.reader.CloseError()
|
|
|
}
|
|
}
|
|
|
|
|
|