Resolve conflicts

common/xray/buf/buffer.go

@@ -0,0 +1,337 @@
+package buf
+
+import (
+	"io"
+
+	"github.com/sagernet/sing-box/common/xray/bytespool"
+	"github.com/sagernet/sing-box/common/xray/net"
+	E "github.com/sagernet/sing/common/exceptions"
+)
+
+const (
+	// Size of a regular buffer.
+	Size = 8192
+)
+
+var zero = [Size * 10]byte{0}
+
+var pool = bytespool.GetPool(Size)
+
+// ownership represents the data owner of the buffer.
+type ownership uint8
+
+const (
+	managed ownership = iota
+	unmanaged
+	bytespools
+)
+
+// Buffer is a recyclable allocation of a byte array. Buffer.Release() recycles
+// the buffer into an internal buffer pool, in order to recreate a buffer more
+// quickly.
+type Buffer struct {
+	v         []byte
+	start     int32
+	end       int32
+	ownership ownership
+	UDP       *net.Destination
+}
+
+// New creates a Buffer with 0 length and 8K capacity, managed.
+func New() *Buffer {
+	buf := pool.Get().([]byte)
+	if cap(buf) >= Size {
+		buf = buf[:Size]
+	} else {
+		buf = make([]byte, Size)
+	}
+
+	return &Buffer{
+		v: buf,
+	}
+}
+
+// NewExisted creates a standard size Buffer with an existed bytearray, managed.
+func NewExisted(b []byte) *Buffer {
+	if cap(b) < Size {
+		panic("Invalid buffer")
+	}
+
+	oLen := len(b)
+	if oLen < Size {
+		b = b[:Size]
+	}
+
+	return &Buffer{
+		v:   b,
+		end: int32(oLen),
+	}
+}
+
+// FromBytes creates a Buffer with an existed bytearray, unmanaged.
+func FromBytes(b []byte) *Buffer {
+	return &Buffer{
+		v:         b,
+		end:       int32(len(b)),
+		ownership: unmanaged,
+	}
+}
+
+// StackNew creates a new Buffer object on stack, managed.
+// This method is for buffers that is released in the same function.
+func StackNew() Buffer {
+	buf := pool.Get().([]byte)
+	if cap(buf) >= Size {
+		buf = buf[:Size]
+	} else {
+		buf = make([]byte, Size)
+	}
+
+	return Buffer{
+		v: buf,
+	}
+}
+
+// NewWithSize creates a Buffer with 0 length and capacity with at least the given size, bytespool's.
+func NewWithSize(size int32) *Buffer {
+	return &Buffer{
+		v:         bytespool.Alloc(size),
+		ownership: bytespools,
+	}
+}
+
+// Release recycles the buffer into an internal buffer pool.
+func (b *Buffer) Release() {
+	if b == nil || b.v == nil || b.ownership == unmanaged {
+		return
+	}
+
+	p := b.v
+	b.v = nil
+	b.Clear()
+
+	switch b.ownership {
+	case managed:
+		if cap(p) == Size {
+			pool.Put(p)
+		}
+	case bytespools:
+		bytespool.Free(p)
+	}
+	b.UDP = nil
+}
+
+// Clear clears the content of the buffer, results an empty buffer with
+// Len() = 0.
+func (b *Buffer) Clear() {
+	b.start = 0
+	b.end = 0
+}
+
+// Byte returns the bytes at index.
+func (b *Buffer) Byte(index int32) byte {
+	return b.v[b.start+index]
+}
+
+// SetByte sets the byte value at index.
+func (b *Buffer) SetByte(index int32, value byte) {
+	b.v[b.start+index] = value
+}
+
+// Bytes returns the content bytes of this Buffer.
+func (b *Buffer) Bytes() []byte {
+	return b.v[b.start:b.end]
+}
+
+// Extend increases the buffer size by n bytes, and returns the extended part.
+// It panics if result size is larger than buf.Size.
+func (b *Buffer) Extend(n int32) []byte {
+	end := b.end + n
+	if end > int32(len(b.v)) {
+		panic("extending out of bound")
+	}
+	ext := b.v[b.end:end]
+	b.end = end
+	copy(ext, zero[:])
+	return ext
+}
+
+// BytesRange returns a slice of this buffer with given from and to boundary.
+func (b *Buffer) BytesRange(from, to int32) []byte {
+	if from < 0 {
+		from += b.Len()
+	}
+	if to < 0 {
+		to += b.Len()
+	}
+	return b.v[b.start+from : b.start+to]
+}
+
+// BytesFrom returns a slice of this Buffer starting from the given position.
+func (b *Buffer) BytesFrom(from int32) []byte {
+	if from < 0 {
+		from += b.Len()
+	}
+	return b.v[b.start+from : b.end]
+}
+
+// BytesTo returns a slice of this Buffer from start to the given position.
+func (b *Buffer) BytesTo(to int32) []byte {
+	if to < 0 {
+		to += b.Len()
+	}
+	if to < 0 {
+		to = 0
+	}
+	return b.v[b.start : b.start+to]
+}
+
+// Check makes sure that 0 <= b.start <= b.end.
+func (b *Buffer) Check() {
+	if b.start < 0 {
+		b.start = 0
+	}
+	if b.end < 0 {
+		b.end = 0
+	}
+	if b.start > b.end {
+		b.start = b.end
+	}
+}
+
+// Resize cuts the buffer at the given position.
+func (b *Buffer) Resize(from, to int32) {
+	oldEnd := b.end
+	if from < 0 {
+		from += b.Len()
+	}
+	if to < 0 {
+		to += b.Len()
+	}
+	if to < from {
+		panic("Invalid slice")
+	}
+	b.end = b.start + to
+	b.start += from
+	b.Check()
+	if b.end > oldEnd {
+		copy(b.v[oldEnd:b.end], zero[:])
+	}
+}
+
+// Advance cuts the buffer at the given position.
+func (b *Buffer) Advance(from int32) {
+	if from < 0 {
+		from += b.Len()
+	}
+	b.start += from
+	b.Check()
+}
+
+// Len returns the length of the buffer content.
+func (b *Buffer) Len() int32 {
+	if b == nil {
+		return 0
+	}
+	return b.end - b.start
+}
+
+// Cap returns the capacity of the buffer content.
+func (b *Buffer) Cap() int32 {
+	if b == nil {
+		return 0
+	}
+	return int32(len(b.v))
+}
+
+// IsEmpty returns true if the buffer is empty.
+func (b *Buffer) IsEmpty() bool {
+	return b.Len() == 0
+}
+
+// IsFull returns true if the buffer has no more room to grow.
+func (b *Buffer) IsFull() bool {
+	return b != nil && b.end == int32(len(b.v))
+}
+
+// Write implements Write method in io.Writer.
+func (b *Buffer) Write(data []byte) (int, error) {
+	nBytes := copy(b.v[b.end:], data)
+	b.end += int32(nBytes)
+	return nBytes, nil
+}
+
+// WriteByte writes a single byte into the buffer.
+func (b *Buffer) WriteByte(v byte) error {
+	if b.IsFull() {
+		return E.New("buffer full")
+	}
+	b.v[b.end] = v
+	b.end++
+	return nil
+}
+
+// WriteString implements io.StringWriter.
+func (b *Buffer) WriteString(s string) (int, error) {
+	return b.Write([]byte(s))
+}
+
+// ReadByte implements io.ByteReader
+func (b *Buffer) ReadByte() (byte, error) {
+	if b.start == b.end {
+		return 0, io.EOF
+	}
+
+	nb := b.v[b.start]
+	b.start++
+	return nb, nil
+}
+
+// ReadBytes implements bufio.Reader.ReadBytes
+func (b *Buffer) ReadBytes(length int32) ([]byte, error) {
+	if b.end-b.start < length {
+		return nil, io.EOF
+	}
+
+	nb := b.v[b.start : b.start+length]
+	b.start += length
+	return nb, nil
+}
+
+// Read implements io.Reader.Read().
+func (b *Buffer) Read(data []byte) (int, error) {
+	if b.Len() == 0 {
+		return 0, io.EOF
+	}
+	nBytes := copy(data, b.v[b.start:b.end])
+	if int32(nBytes) == b.Len() {
+		b.Clear()
+	} else {
+		b.start += int32(nBytes)
+	}
+	return nBytes, nil
+}
+
+// ReadFrom implements io.ReaderFrom.
+func (b *Buffer) ReadFrom(reader io.Reader) (int64, error) {
+	n, err := reader.Read(b.v[b.end:])
+	b.end += int32(n)
+	return int64(n), err
+}
+
+// ReadFullFrom reads exact size of bytes from given reader, or until error occurs.
+func (b *Buffer) ReadFullFrom(reader io.Reader, size int32) (int64, error) {
+	end := b.end + size
+	if end > int32(len(b.v)) {
+		v := end
+		return 0, E.New("out of bound: ", v)
+	}
+	n, err := io.ReadFull(reader, b.v[b.end:end])
+	b.end += int32(n)
+	return int64(n), err
+}
+
+// String returns the string form of this Buffer.
+func (b *Buffer) String() string {
+	return string(b.Bytes())
+}

common/xray/buf/copy.go

@@ -0,0 +1,124 @@
+package buf
+
+import (
+	"io"
+	"time"
+
+	"github.com/sagernet/sing-box/common/xray/errors"
+	"github.com/sagernet/sing-box/common/xray/signal"
+	E "github.com/sagernet/sing/common/exceptions"
+)
+
+type dataHandler func(MultiBuffer)
+
+type copyHandler struct {
+	onData []dataHandler
+}
+
+// SizeCounter is for counting bytes copied by Copy().
+type SizeCounter struct {
+	Size int64
+}
+
+// CopyOption is an option for copying data.
+type CopyOption func(*copyHandler)
+
+// UpdateActivity is a CopyOption to update activity on each data copy operation.
+func UpdateActivity(timer signal.ActivityUpdater) CopyOption {
+	return func(handler *copyHandler) {
+		handler.onData = append(handler.onData, func(MultiBuffer) {
+			timer.Update()
+		})
+	}
+}
+
+// CountSize is a CopyOption that sums the total size of data copied into the given SizeCounter.
+func CountSize(sc *SizeCounter) CopyOption {
+	return func(handler *copyHandler) {
+		handler.onData = append(handler.onData, func(b MultiBuffer) {
+			sc.Size += int64(b.Len())
+		})
+	}
+}
+
+type readError struct {
+	error
+}
+
+func (e readError) Error() string {
+	return e.error.Error()
+}
+
+func (e readError) Unwrap() error {
+	return e.error
+}
+
+// IsReadError returns true if the error in Copy() comes from reading.
+func IsReadError(err error) bool {
+	_, ok := err.(readError)
+	return ok
+}
+
+type writeError struct {
+	error
+}
+
+func (e writeError) Error() string {
+	return e.error.Error()
+}
+
+func (e writeError) Unwrap() error {
+	return e.error
+}
+
+// IsWriteError returns true if the error in Copy() comes from writing.
+func IsWriteError(err error) bool {
+	_, ok := err.(writeError)
+	return ok
+}
+
+func copyInternal(reader Reader, writer Writer, handler *copyHandler) error {
+	for {
+		buffer, err := reader.ReadMultiBuffer()
+		if !buffer.IsEmpty() {
+			for _, handler := range handler.onData {
+				handler(buffer)
+			}
+
+			if werr := writer.WriteMultiBuffer(buffer); werr != nil {
+				return writeError{werr}
+			}
+		}
+
+		if err != nil {
+			return readError{err}
+		}
+	}
+}
+
+// Copy dumps all payload from reader to writer or stops when an error occurs. It returns nil when EOF.
+func Copy(reader Reader, writer Writer, options ...CopyOption) error {
+	var handler copyHandler
+	for _, option := range options {
+		option(&handler)
+	}
+	err := copyInternal(reader, writer, &handler)
+	if err != nil && errors.Cause(err) != io.EOF {
+		return err
+	}
+	return nil
+}
+
+var ErrNotTimeoutReader = E.New("not a TimeoutReader")
+
+func CopyOnceTimeout(reader Reader, writer Writer, timeout time.Duration) error {
+	timeoutReader, ok := reader.(TimeoutReader)
+	if !ok {
+		return ErrNotTimeoutReader
+	}
+	mb, err := timeoutReader.ReadMultiBufferTimeout(timeout)
+	if err != nil {
+		return err
+	}
+	return writer.WriteMultiBuffer(mb)
+}

common/xray/buf/io.go

@@ -0,0 +1,126 @@
+package buf
+
+import (
+	"io"
+	"net"
+	"syscall"
+	"time"
+
+	"github.com/sagernet/sing-box/common/xray/stat"
+	"github.com/sagernet/sing-box/common/xray/stats"
+	E "github.com/sagernet/sing/common/exceptions"
+)
+
+// Reader extends io.Reader with MultiBuffer.
+type Reader interface {
+	// ReadMultiBuffer reads content from underlying reader, and put it into a MultiBuffer.
+	ReadMultiBuffer() (MultiBuffer, error)
+}
+
+// ErrReadTimeout is an error that happens with IO timeout.
+var ErrReadTimeout = E.New("IO timeout")
+
+// TimeoutReader is a reader that returns error if Read() operation takes longer than the given timeout.
+type TimeoutReader interface {
+	ReadMultiBufferTimeout(time.Duration) (MultiBuffer, error)
+}
+
+// Writer extends io.Writer with MultiBuffer.
+type Writer interface {
+	// WriteMultiBuffer writes a MultiBuffer into underlying writer.
+	WriteMultiBuffer(MultiBuffer) error
+}
+
+// WriteAllBytes ensures all bytes are written into the given writer.
+func WriteAllBytes(writer io.Writer, payload []byte, c stats.Counter) error {
+	wc := 0
+	defer func() {
+		if c != nil {
+			c.Add(int64(wc))
+		}
+	}()
+
+	for len(payload) > 0 {
+		n, err := writer.Write(payload)
+		wc += n
+		if err != nil {
+			return err
+		}
+		payload = payload[n:]
+	}
+	return nil
+}
+
+func isPacketReader(reader io.Reader) bool {
+	_, ok := reader.(net.PacketConn)
+	return ok
+}
+
+// NewReader creates a new Reader.
+// The Reader instance doesn't take the ownership of reader.
+func NewReader(reader io.Reader) Reader {
+	if mr, ok := reader.(Reader); ok {
+		return mr
+	}
+
+	if isPacketReader(reader) {
+		return &PacketReader{
+			Reader: reader,
+		}
+	}
+
+	return &SingleReader{
+		Reader: reader,
+	}
+}
+
+// NewPacketReader creates a new PacketReader based on the given reader.
+func NewPacketReader(reader io.Reader) Reader {
+	if mr, ok := reader.(Reader); ok {
+		return mr
+	}
+
+	return &PacketReader{
+		Reader: reader,
+	}
+}
+
+func isPacketWriter(writer io.Writer) bool {
+	if _, ok := writer.(net.PacketConn); ok {
+		return true
+	}
+
+	// If the writer doesn't implement syscall.Conn, it is probably not a TCP connection.
+	if _, ok := writer.(syscall.Conn); !ok {
+		return true
+	}
+	return false
+}
+
+// NewWriter creates a new Writer.
+func NewWriter(writer io.Writer) Writer {
+	if mw, ok := writer.(Writer); ok {
+		return mw
+	}
+
+	iConn := writer
+	if statConn, ok := writer.(*stat.CounterConnection); ok {
+		iConn = statConn.Connection
+	}
+
+	if isPacketWriter(iConn) {
+		return &SequentialWriter{
+			Writer: writer,
+		}
+	}
+
+	var counter stats.Counter
+
+	if statConn, ok := writer.(*stat.CounterConnection); ok {
+		counter = statConn.WriteCounter
+	}
+	return &BufferToBytesWriter{
+		Writer:  iConn,
+		counter: counter,
+	}
+}

common/xray/buf/multi_buffer.go

@@ -0,0 +1,310 @@
+package buf
+
+import (
+	"io"
+
+	"github.com/sagernet/sing-box/common/xray"
+	"github.com/sagernet/sing-box/common/xray/errors"
+	"github.com/sagernet/sing-box/common/xray/serial"
+)
+
+// ReadAllToBytes reads all content from the reader into a byte array, until EOF.
+func ReadAllToBytes(reader io.Reader) ([]byte, error) {
+	mb, err := ReadFrom(reader)
+	if err != nil {
+		return nil, err
+	}
+	if mb.Len() == 0 {
+		return nil, nil
+	}
+	b := make([]byte, mb.Len())
+	mb, _ = SplitBytes(mb, b)
+	ReleaseMulti(mb)
+	return b, nil
+}
+
+// MultiBuffer is a list of Buffers. The order of Buffer matters.
+type MultiBuffer []*Buffer
+
+// MergeMulti merges content from src to dest, and returns the new address of dest and src
+func MergeMulti(dest MultiBuffer, src MultiBuffer) (MultiBuffer, MultiBuffer) {
+	dest = append(dest, src...)
+	for idx := range src {
+		src[idx] = nil
+	}
+	return dest, src[:0]
+}
+
+// MergeBytes merges the given bytes into MultiBuffer and return the new address of the merged MultiBuffer.
+func MergeBytes(dest MultiBuffer, src []byte) MultiBuffer {
+	n := len(dest)
+	if n > 0 && !(dest)[n-1].IsFull() {
+		nBytes, _ := (dest)[n-1].Write(src)
+		src = src[nBytes:]
+	}
+
+	for len(src) > 0 {
+		b := New()
+		nBytes, _ := b.Write(src)
+		src = src[nBytes:]
+		dest = append(dest, b)
+	}
+
+	return dest
+}
+
+// ReleaseMulti releases all content of the MultiBuffer, and returns an empty MultiBuffer.
+func ReleaseMulti(mb MultiBuffer) MultiBuffer {
+	for i := range mb {
+		mb[i].Release()
+		mb[i] = nil
+	}
+	return mb[:0]
+}
+
+// Copy copied the beginning part of the MultiBuffer into the given byte array.
+func (mb MultiBuffer) Copy(b []byte) int {
+	total := 0
+	for _, bb := range mb {
+		nBytes := copy(b[total:], bb.Bytes())
+		total += nBytes
+		if int32(nBytes) < bb.Len() {
+			break
+		}
+	}
+	return total
+}
+
+// ReadFrom reads all content from reader until EOF.
+func ReadFrom(reader io.Reader) (MultiBuffer, error) {
+	mb := make(MultiBuffer, 0, 16)
+	for {
+		b := New()
+		_, err := b.ReadFullFrom(reader, Size)
+		if b.IsEmpty() {
+			b.Release()
+		} else {
+			mb = append(mb, b)
+		}
+		if err != nil {
+			if errors.Cause(err) == io.EOF || errors.Cause(err) == io.ErrUnexpectedEOF {
+				return mb, nil
+			}
+			return mb, err
+		}
+	}
+}
+
+// SplitBytes splits the given amount of bytes from the beginning of the MultiBuffer.
+// It returns the new address of MultiBuffer leftover, and number of bytes written into the input byte slice.
+func SplitBytes(mb MultiBuffer, b []byte) (MultiBuffer, int) {
+	totalBytes := 0
+	endIndex := -1
+	for i := range mb {
+		pBuffer := mb[i]
+		nBytes, _ := pBuffer.Read(b)
+		totalBytes += nBytes
+		b = b[nBytes:]
+		if !pBuffer.IsEmpty() {
+			endIndex = i
+			break
+		}
+		pBuffer.Release()
+		mb[i] = nil
+	}
+
+	if endIndex == -1 {
+		mb = mb[:0]
+	} else {
+		mb = mb[endIndex:]
+	}
+
+	return mb, totalBytes
+}
+
+// SplitFirstBytes splits the first buffer from MultiBuffer, and then copy its content into the given slice.
+func SplitFirstBytes(mb MultiBuffer, p []byte) (MultiBuffer, int) {
+	mb, b := SplitFirst(mb)
+	if b == nil {
+		return mb, 0
+	}
+	n := copy(p, b.Bytes())
+	b.Release()
+	return mb, n
+}
+
+// Compact returns another MultiBuffer by merging all content of the given one together.
+func Compact(mb MultiBuffer) MultiBuffer {
+	if len(mb) == 0 {
+		return mb
+	}
+
+	mb2 := make(MultiBuffer, 0, len(mb))
+	last := mb[0]
+
+	for i := 1; i < len(mb); i++ {
+		curr := mb[i]
+		if last.Len()+curr.Len() > Size {
+			mb2 = append(mb2, last)
+			last = curr
+		} else {
+			common.Must2(last.ReadFrom(curr))
+			curr.Release()
+		}
+	}
+
+	mb2 = append(mb2, last)
+	return mb2
+}
+
+// SplitFirst splits the first Buffer from the beginning of the MultiBuffer.
+func SplitFirst(mb MultiBuffer) (MultiBuffer, *Buffer) {
+	if len(mb) == 0 {
+		return mb, nil
+	}
+
+	b := mb[0]
+	mb[0] = nil
+	mb = mb[1:]
+	return mb, b
+}
+
+// SplitSize splits the beginning of the MultiBuffer into another one, for at most size bytes.
+func SplitSize(mb MultiBuffer, size int32) (MultiBuffer, MultiBuffer) {
+	if len(mb) == 0 {
+		return mb, nil
+	}
+
+	if mb[0].Len() > size {
+		b := New()
+		copy(b.Extend(size), mb[0].BytesTo(size))
+		mb[0].Advance(size)
+		return mb, MultiBuffer{b}
+	}
+
+	totalBytes := int32(0)
+	var r MultiBuffer
+	endIndex := -1
+	for i := range mb {
+		if totalBytes+mb[i].Len() > size {
+			endIndex = i
+			break
+		}
+		totalBytes += mb[i].Len()
+		r = append(r, mb[i])
+		mb[i] = nil
+	}
+	if endIndex == -1 {
+		// To reuse mb array
+		mb = mb[:0]
+	} else {
+		mb = mb[endIndex:]
+	}
+	return mb, r
+}
+
+// SplitMulti splits the beginning of the MultiBuffer into first one, the index i and after into second one
+func SplitMulti(mb MultiBuffer, i int) (MultiBuffer, MultiBuffer) {
+	mb2 := make(MultiBuffer, 0, len(mb))
+	if i < len(mb) && i >= 0 {
+		mb2 = append(mb2, mb[i:]...)
+		for j := i; j < len(mb); j++ {
+			mb[j] = nil
+		}
+		mb = mb[:i]
+	}
+	return mb, mb2
+}
+
+// WriteMultiBuffer writes all buffers from the MultiBuffer to the Writer one by one, and return error if any, with leftover MultiBuffer.
+func WriteMultiBuffer(writer io.Writer, mb MultiBuffer) (MultiBuffer, error) {
+	for {
+		mb2, b := SplitFirst(mb)
+		mb = mb2
+		if b == nil {
+			break
+		}
+
+		_, err := writer.Write(b.Bytes())
+		b.Release()
+		if err != nil {
+			return mb, err
+		}
+	}
+
+	return nil, nil
+}
+
+// Len returns the total number of bytes in the MultiBuffer.
+func (mb MultiBuffer) Len() int32 {
+	if mb == nil {
+		return 0
+	}
+
+	size := int32(0)
+	for _, b := range mb {
+		size += b.Len()
+	}
+	return size
+}
+
+// IsEmpty returns true if the MultiBuffer has no content.
+func (mb MultiBuffer) IsEmpty() bool {
+	for _, b := range mb {
+		if !b.IsEmpty() {
+			return false
+		}
+	}
+	return true
+}
+
+// String returns the content of the MultiBuffer in string.
+func (mb MultiBuffer) String() string {
+	v := make([]interface{}, len(mb))
+	for i, b := range mb {
+		v[i] = b
+	}
+	return serial.Concat(v...)
+}
+
+// MultiBufferContainer is a ReadWriteCloser wrapper over MultiBuffer.
+type MultiBufferContainer struct {
+	MultiBuffer
+}
+
+// Read implements io.Reader.
+func (c *MultiBufferContainer) Read(b []byte) (int, error) {
+	if c.MultiBuffer.IsEmpty() {
+		return 0, io.EOF
+	}
+
+	mb, nBytes := SplitBytes(c.MultiBuffer, b)
+	c.MultiBuffer = mb
+	return nBytes, nil
+}
+
+// ReadMultiBuffer implements Reader.
+func (c *MultiBufferContainer) ReadMultiBuffer() (MultiBuffer, error) {
+	mb := c.MultiBuffer
+	c.MultiBuffer = nil
+	return mb, nil
+}
+
+// Write implements io.Writer.
+func (c *MultiBufferContainer) Write(b []byte) (int, error) {
+	c.MultiBuffer = MergeBytes(c.MultiBuffer, b)
+	return len(b), nil
+}
+
+// WriteMultiBuffer implements Writer.
+func (c *MultiBufferContainer) WriteMultiBuffer(b MultiBuffer) error {
+	mb, _ := MergeMulti(c.MultiBuffer, b)
+	c.MultiBuffer = mb
+	return nil
+}
+
+// Close implements io.Closer.
+func (c *MultiBufferContainer) Close() error {
+	c.MultiBuffer = ReleaseMulti(c.MultiBuffer)
+	return nil
+}

common/xray/buf/override.go

@@ -0,0 +1,38 @@
+package buf
+
+import (
+	"github.com/sagernet/sing-box/common/xray/net"
+)
+
+type EndpointOverrideReader struct {
+	Reader
+	Dest         net.Address
+	OriginalDest net.Address
+}
+
+func (r *EndpointOverrideReader) ReadMultiBuffer() (MultiBuffer, error) {
+	mb, err := r.Reader.ReadMultiBuffer()
+	if err == nil {
+		for _, b := range mb {
+			if b.UDP != nil && b.UDP.Address == r.OriginalDest {
+				b.UDP.Address = r.Dest
+			}
+		}
+	}
+	return mb, err
+}
+
+type EndpointOverrideWriter struct {
+	Writer
+	Dest         net.Address
+	OriginalDest net.Address
+}
+
+func (w *EndpointOverrideWriter) WriteMultiBuffer(mb MultiBuffer) error {
+	for _, b := range mb {
+		if b.UDP != nil && b.UDP.Address == w.Dest {
+			b.UDP.Address = w.OriginalDest
+		}
+	}
+	return w.Writer.WriteMultiBuffer(mb)
+}

common/xray/buf/reader.go

@@ -0,0 +1,175 @@
+package buf
+
+import (
+	"io"
+
+	"github.com/sagernet/sing-box/common/xray"
+	"github.com/sagernet/sing-box/common/xray/errors"
+	E "github.com/sagernet/sing/common/exceptions"
+)
+
+func readOneUDP(r io.Reader) (*Buffer, error) {
+	b := New()
+	for i := 0; i < 64; i++ {
+		_, err := b.ReadFrom(r)
+		if !b.IsEmpty() {
+			return b, nil
+		}
+		if err != nil {
+			b.Release()
+			return nil, err
+		}
+	}
+
+	b.Release()
+	return nil, E.New("Reader returns too many empty payloads.")
+}
+
+// ReadBuffer reads a Buffer from the given reader.
+func ReadBuffer(r io.Reader) (*Buffer, error) {
+	b := New()
+	n, err := b.ReadFrom(r)
+	if n > 0 {
+		return b, err
+	}
+	b.Release()
+	return nil, err
+}
+
+// BufferedReader is a Reader that keeps its internal buffer.
+type BufferedReader struct {
+	// Reader is the underlying reader to be read from
+	Reader Reader
+	// Buffer is the internal buffer to be read from first
+	Buffer MultiBuffer
+	// Splitter is a function to read bytes from MultiBuffer
+	Splitter func(MultiBuffer, []byte) (MultiBuffer, int)
+}
+
+// BufferedBytes returns the number of bytes that is cached in this reader.
+func (r *BufferedReader) BufferedBytes() int32 {
+	return r.Buffer.Len()
+}
+
+// ReadByte implements io.ByteReader.
+func (r *BufferedReader) ReadByte() (byte, error) {
+	var b [1]byte
+	_, err := r.Read(b[:])
+	return b[0], err
+}
+
+// Read implements io.Reader. It reads from internal buffer first (if available) and then reads from the underlying reader.
+func (r *BufferedReader) Read(b []byte) (int, error) {
+	spliter := r.Splitter
+	if spliter == nil {
+		spliter = SplitBytes
+	}
+
+	if !r.Buffer.IsEmpty() {
+		buffer, nBytes := spliter(r.Buffer, b)
+		r.Buffer = buffer
+		if r.Buffer.IsEmpty() {
+			r.Buffer = nil
+		}
+		return nBytes, nil
+	}
+
+	mb, err := r.Reader.ReadMultiBuffer()
+	if err != nil {
+		return 0, err
+	}
+
+	mb, nBytes := spliter(mb, b)
+	if !mb.IsEmpty() {
+		r.Buffer = mb
+	}
+	return nBytes, nil
+}
+
+// ReadMultiBuffer implements Reader.
+func (r *BufferedReader) ReadMultiBuffer() (MultiBuffer, error) {
+	if !r.Buffer.IsEmpty() {
+		mb := r.Buffer
+		r.Buffer = nil
+		return mb, nil
+	}
+
+	return r.Reader.ReadMultiBuffer()
+}
+
+// ReadAtMost returns a MultiBuffer with at most size.
+func (r *BufferedReader) ReadAtMost(size int32) (MultiBuffer, error) {
+	if r.Buffer.IsEmpty() {
+		mb, err := r.Reader.ReadMultiBuffer()
+		if mb.IsEmpty() && err != nil {
+			return nil, err
+		}
+		r.Buffer = mb
+	}
+
+	rb, mb := SplitSize(r.Buffer, size)
+	r.Buffer = rb
+	if r.Buffer.IsEmpty() {
+		r.Buffer = nil
+	}
+	return mb, nil
+}
+
+func (r *BufferedReader) writeToInternal(writer io.Writer) (int64, error) {
+	mbWriter := NewWriter(writer)
+	var sc SizeCounter
+	if r.Buffer != nil {
+		sc.Size = int64(r.Buffer.Len())
+		if err := mbWriter.WriteMultiBuffer(r.Buffer); err != nil {
+			return 0, err
+		}
+		r.Buffer = nil
+	}
+
+	err := Copy(r.Reader, mbWriter, CountSize(&sc))
+	return sc.Size, err
+}
+
+// WriteTo implements io.WriterTo.
+func (r *BufferedReader) WriteTo(writer io.Writer) (int64, error) {
+	nBytes, err := r.writeToInternal(writer)
+	if errors.Cause(err) == io.EOF {
+		return nBytes, nil
+	}
+	return nBytes, err
+}
+
+// Interrupt implements common.Interruptible.
+func (r *BufferedReader) Interrupt() {
+	common.Interrupt(r.Reader)
+}
+
+// Close implements io.Closer.
+func (r *BufferedReader) Close() error {
+	return common.Close(r.Reader)
+}
+
+// SingleReader is a Reader that read one Buffer every time.
+type SingleReader struct {
+	io.Reader
+}
+
+// ReadMultiBuffer implements Reader.
+func (r *SingleReader) ReadMultiBuffer() (MultiBuffer, error) {
+	b, err := ReadBuffer(r.Reader)
+	return MultiBuffer{b}, err
+}
+
+// PacketReader is a Reader that read one Buffer every time.
+type PacketReader struct {
+	io.Reader
+}
+
+// ReadMultiBuffer implements Reader.
+func (r *PacketReader) ReadMultiBuffer() (MultiBuffer, error) {
+	b, err := readOneUDP(r.Reader)
+	if err != nil {
+		return nil, err
+	}
+	return MultiBuffer{b}, nil
+}

common/xray/buf/writer.go

@@ -0,0 +1,270 @@
+package buf
+
+import (
+	"io"
+	"net"
+	"sync"
+
+	"github.com/sagernet/sing-box/common/xray"
+	"github.com/sagernet/sing-box/common/xray/errors"
+	"github.com/sagernet/sing-box/common/xray/stats"
+)
+
+// BufferToBytesWriter is a Writer that writes alloc.Buffer into underlying writer.
+type BufferToBytesWriter struct {
+	io.Writer
+
+	counter stats.Counter
+	cache   [][]byte
+}
+
+// WriteMultiBuffer implements Writer. This method takes ownership of the given buffer.
+func (w *BufferToBytesWriter) WriteMultiBuffer(mb MultiBuffer) error {
+	defer ReleaseMulti(mb)
+
+	size := mb.Len()
+	if size == 0 {
+		return nil
+	}
+
+	if len(mb) == 1 {
+		return WriteAllBytes(w.Writer, mb[0].Bytes(), w.counter)
+	}
+
+	if cap(w.cache) < len(mb) {
+		w.cache = make([][]byte, 0, len(mb))
+	}
+
+	bs := w.cache
+	for _, b := range mb {
+		bs = append(bs, b.Bytes())
+	}
+
+	defer func() {
+		for idx := range bs {
+			bs[idx] = nil
+		}
+	}()
+
+	nb := net.Buffers(bs)
+	wc := int64(0)
+	defer func() {
+		if w.counter != nil {
+			w.counter.Add(wc)
+		}
+	}()
+	for size > 0 {
+		n, err := nb.WriteTo(w.Writer)
+		wc += n
+		if err != nil {
+			return err
+		}
+		size -= int32(n)
+	}
+
+	return nil
+}
+
+// ReadFrom implements io.ReaderFrom.
+func (w *BufferToBytesWriter) ReadFrom(reader io.Reader) (int64, error) {
+	var sc SizeCounter
+	err := Copy(NewReader(reader), w, CountSize(&sc))
+	return sc.Size, err
+}
+
+// BufferedWriter is a Writer with internal buffer.
+type BufferedWriter struct {
+	sync.Mutex
+	writer   Writer
+	buffer   *Buffer
+	buffered bool
+}
+
+// NewBufferedWriter creates a new BufferedWriter.
+func NewBufferedWriter(writer Writer) *BufferedWriter {
+	return &BufferedWriter{
+		writer:   writer,
+		buffer:   New(),
+		buffered: true,
+	}
+}
+
+// WriteByte implements io.ByteWriter.
+func (w *BufferedWriter) WriteByte(c byte) error {
+	return common.Error2(w.Write([]byte{c}))
+}
+
+// Write implements io.Writer.
+func (w *BufferedWriter) Write(b []byte) (int, error) {
+	if len(b) == 0 {
+		return 0, nil
+	}
+
+	w.Lock()
+	defer w.Unlock()
+
+	if !w.buffered {
+		if writer, ok := w.writer.(io.Writer); ok {
+			return writer.Write(b)
+		}
+	}
+
+	totalBytes := 0
+	for len(b) > 0 {
+		if w.buffer == nil {
+			w.buffer = New()
+		}
+
+		nBytes, err := w.buffer.Write(b)
+		totalBytes += nBytes
+		if err != nil {
+			return totalBytes, err
+		}
+		if !w.buffered || w.buffer.IsFull() {
+			if err := w.flushInternal(); err != nil {
+				return totalBytes, err
+			}
+		}
+		b = b[nBytes:]
+	}
+
+	return totalBytes, nil
+}
+
+// WriteMultiBuffer implements Writer. It takes ownership of the given MultiBuffer.
+func (w *BufferedWriter) WriteMultiBuffer(b MultiBuffer) error {
+	if b.IsEmpty() {
+		return nil
+	}
+
+	w.Lock()
+	defer w.Unlock()
+
+	if !w.buffered {
+		return w.writer.WriteMultiBuffer(b)
+	}
+
+	reader := MultiBufferContainer{
+		MultiBuffer: b,
+	}
+	defer reader.Close()
+
+	for !reader.MultiBuffer.IsEmpty() {
+		if w.buffer == nil {
+			w.buffer = New()
+		}
+		common.Must2(w.buffer.ReadFrom(&reader))
+		if w.buffer.IsFull() {
+			if err := w.flushInternal(); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+// Flush flushes buffered content into underlying writer.
+func (w *BufferedWriter) Flush() error {
+	w.Lock()
+	defer w.Unlock()
+
+	return w.flushInternal()
+}
+
+func (w *BufferedWriter) flushInternal() error {
+	if w.buffer.IsEmpty() {
+		return nil
+	}
+
+	b := w.buffer
+	w.buffer = nil
+
+	if writer, ok := w.writer.(io.Writer); ok {
+		err := WriteAllBytes(writer, b.Bytes(), nil)
+		b.Release()
+		return err
+	}
+
+	return w.writer.WriteMultiBuffer(MultiBuffer{b})
+}
+
+// SetBuffered sets whether the internal buffer is used. If set to false, Flush() will be called to clear the buffer.
+func (w *BufferedWriter) SetBuffered(f bool) error {
+	w.Lock()
+	defer w.Unlock()
+
+	w.buffered = f
+	if !f {
+		return w.flushInternal()
+	}
+	return nil
+}
+
+// ReadFrom implements io.ReaderFrom.
+func (w *BufferedWriter) ReadFrom(reader io.Reader) (int64, error) {
+	if err := w.SetBuffered(false); err != nil {
+		return 0, err
+	}
+
+	var sc SizeCounter
+	err := Copy(NewReader(reader), w, CountSize(&sc))
+	return sc.Size, err
+}
+
+// Close implements io.Closable.
+func (w *BufferedWriter) Close() error {
+	if err := w.Flush(); err != nil {
+		return err
+	}
+	return common.Close(w.writer)
+}
+
+// SequentialWriter is a Writer that writes MultiBuffer sequentially into the underlying io.Writer.
+type SequentialWriter struct {
+	io.Writer
+}
+
+// WriteMultiBuffer implements Writer.
+func (w *SequentialWriter) WriteMultiBuffer(mb MultiBuffer) error {
+	mb, err := WriteMultiBuffer(w.Writer, mb)
+	ReleaseMulti(mb)
+	return err
+}
+
+type noOpWriter byte
+
+func (noOpWriter) WriteMultiBuffer(b MultiBuffer) error {
+	ReleaseMulti(b)
+	return nil
+}
+
+func (noOpWriter) Write(b []byte) (int, error) {
+	return len(b), nil
+}
+
+func (noOpWriter) ReadFrom(reader io.Reader) (int64, error) {
+	b := New()
+	defer b.Release()
+
+	totalBytes := int64(0)
+	for {
+		b.Clear()
+		_, err := b.ReadFrom(reader)
+		totalBytes += int64(b.Len())
+		if err != nil {
+			if errors.Cause(err) == io.EOF {
+				return totalBytes, nil
+			}
+			return totalBytes, err
+		}
+	}
+}
+
+var (
+	// Discard is a Writer that swallows all contents written in.
+	Discard Writer = noOpWriter(0)
+
+	// DiscardBytes is an io.Writer that swallows all contents written in.
+	DiscardBytes io.Writer = noOpWriter(0)
+)

common/xray/bytespool/pool.go

@@ -0,0 +1,72 @@
+package bytespool
+
+import "sync"
+
+func createAllocFunc(size int32) func() interface{} {
+	return func() interface{} {
+		return make([]byte, size)
+	}
+}
+
+// The following parameters controls the size of buffer pools.
+// There are numPools pools. Starting from 2k size, the size of each pool is sizeMulti of the previous one.
+// Package buf is guaranteed to not use buffers larger than the largest pool.
+// Other packets may use larger buffers.
+const (
+	numPools  = 4
+	sizeMulti = 4
+)
+
+var (
+	pool     [numPools]sync.Pool
+	poolSize [numPools]int32
+)
+
+func init() {
+	size := int32(2048)
+	for i := 0; i < numPools; i++ {
+		pool[i] = sync.Pool{
+			New: createAllocFunc(size),
+		}
+		poolSize[i] = size
+		size *= sizeMulti
+	}
+}
+
+// GetPool returns a sync.Pool that generates bytes array with at least the given size.
+// It may return nil if no such pool exists.
+//
+// xray:api:stable
+func GetPool(size int32) *sync.Pool {
+	for idx, ps := range poolSize {
+		if size <= ps {
+			return &pool[idx]
+		}
+	}
+	return nil
+}
+
+// Alloc returns a byte slice with at least the given size. Minimum size of returned slice is 2048.
+//
+// xray:api:stable
+func Alloc(size int32) []byte {
+	pool := GetPool(size)
+	if pool != nil {
+		return pool.Get().([]byte)
+	}
+	return make([]byte, size)
+}
+
+// Free puts a byte slice into the internal pool.
+//
+// xray:api:stable
+func Free(b []byte) {
+	size := int32(cap(b))
+	b = b[0:cap(b)]
+	for i := numPools - 1; i >= 0; i-- {
+		if size >= poolSize[i] {
+			pool[i].Put(b)
+			return
+		}
+	}
+}

common/xray/common.go

@@ -0,0 +1,19 @@
+package common
+
+// Must panics if err is not nil.
+func Must(err error) {
+	if err != nil {
+		panic(err)
+	}
+}
+
+// Must2 panics if the second parameter is not nil, otherwise returns the first parameter.
+func Must2(v interface{}, err error) interface{} {
+	Must(err)
+	return v
+}
+
+// Error2 returns the err from the 2nd parameter.
+func Error2(v interface{}, err error) error {
+	return err
+}

common/xray/crypto/crypto.go

@@ -0,0 +1,14 @@
+package crypto
+
+import (
+	"crypto/rand"
+	"math/big"
+)
+
+func RandBetween(from int64, to int64) int64 {
+	if from == to {
+		return from
+	}
+	bigInt, _ := rand.Int(rand.Reader, big.NewInt(to-from))
+	return from + bigInt.Int64()
+}

common/xray/errors/errors.go

@@ -0,0 +1,25 @@
+package errors
+
+type hasInnerError interface {
+	// Unwrap returns the underlying error of this one.
+	Unwrap() error
+}
+
+func Cause(err error) error {
+	if err == nil {
+		return nil
+	}
+L:
+	for {
+		switch inner := err.(type) {
+		case hasInnerError:
+			if inner.Unwrap() == nil {
+				break L
+			}
+			err = inner.Unwrap()
+		default:
+			break L
+		}
+	}
+	return err
+}

common/xray/interfaces.go

@@ -0,0 +1,52 @@
+package common
+
+// Closable is the interface for objects that can release its resources.
+//
+// xray:api:beta
+type Closable interface {
+	// Close release all resources used by this object, including goroutines.
+	Close() error
+}
+
+// Interruptible is an interface for objects that can be stopped before its completion.
+//
+// xray:api:beta
+type Interruptible interface {
+	Interrupt()
+}
+
+// Close closes the obj if it is a Closable.
+//
+// xray:api:beta
+func Close(obj interface{}) error {
+	if c, ok := obj.(Closable); ok {
+		return c.Close()
+	}
+	return nil
+}
+
+// Interrupt calls Interrupt() if object implements Interruptible interface, or Close() if the object implements Closable interface.
+//
+// xray:api:beta
+func Interrupt(obj interface{}) error {
+	if c, ok := obj.(Interruptible); ok {
+		c.Interrupt()
+		return nil
+	}
+	return Close(obj)
+}
+
+// Runnable is the interface for objects that can start to work and stop on demand.
+type Runnable interface {
+	// Start starts the runnable object. Upon the method returning nil, the object begins to function properly.
+	Start() error
+
+	Closable
+}
+
+// HasType is the interface for objects that knows its type.
+type HasType interface {
+	// Type returns the type of the object.
+	// Usually it returns (*Type)(nil) of the object.
+	Type() interface{}
+}

common/xray/json/badoption/range.go

@@ -0,0 +1,53 @@
+package badoption
+
+import (
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/sagernet/sing-box/common/xray/crypto"
+	E "github.com/sagernet/sing/common/exceptions"
+)
+
+type Range struct {
+	From int32 `json:"from"`
+	To   int32 `json:"to"`
+}
+
+func (c *Range) Build() *Range {
+	return (*Range)(c)
+}
+
+func (c *Range) MarshalJSON() ([]byte, error) {
+	return json.Marshal(fmt.Sprintf("%d-%d", c.From, c.To))
+}
+
+func (c *Range) UnmarshalJSON(content []byte) error {
+	var stringValue string
+	err := json.Unmarshal(content, &stringValue)
+	if err != nil {
+		return err
+	}
+	parts := strings.Split(stringValue, "-")
+	if len(parts) != 2 {
+		return E.New("invalid length of range parts")
+	}
+	from, err := strconv.ParseInt(parts[0], 10, 32)
+	if err != nil {
+		return err
+	}
+	to, err := strconv.ParseInt(parts[1], 10, 32)
+	if err != nil {
+		return err
+	}
+	if from > to {
+		return E.New("invalid range")
+	}
+	*c = Range{int32(from), int32(to)}
+	return nil
+}
+
+func (c Range) Rand() int32 {
+	return int32(crypto.RandBetween(int64(c.From), int64(c.To)))
+}

common/xray/net/address.go

@@ -0,0 +1,181 @@
+package net
+
+import (
+	"bytes"
+	"net"
+	"strings"
+)
+
+var (
+	// LocalHostIP is a constant value for localhost IP in IPv4.
+	LocalHostIP = IPAddress([]byte{127, 0, 0, 1})
+
+	// AnyIP is a constant value for any IP in IPv4.
+	AnyIP = IPAddress([]byte{0, 0, 0, 0})
+
+	// LocalHostDomain is a constant value for localhost domain.
+	LocalHostDomain = DomainAddress("localhost")
+
+	// LocalHostIPv6 is a constant value for localhost IP in IPv6.
+	LocalHostIPv6 = IPAddress([]byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1})
+
+	// AnyIPv6 is a constant value for any IP in IPv6.
+	AnyIPv6 = IPAddress([]byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0})
+)
+
+// AddressFamily is the type of address.
+type AddressFamily byte
+
+const (
+	// AddressFamilyIPv4 represents address as IPv4
+	AddressFamilyIPv4 = AddressFamily(0)
+
+	// AddressFamilyIPv6 represents address as IPv6
+	AddressFamilyIPv6 = AddressFamily(1)
+
+	// AddressFamilyDomain represents address as Domain
+	AddressFamilyDomain = AddressFamily(2)
+)
+
+// IsIPv4 returns true if current AddressFamily is IPv4.
+func (af AddressFamily) IsIPv4() bool {
+	return af == AddressFamilyIPv4
+}
+
+// IsIPv6 returns true if current AddressFamily is IPv6.
+func (af AddressFamily) IsIPv6() bool {
+	return af == AddressFamilyIPv6
+}
+
+// IsIP returns true if current AddressFamily is IPv6 or IPv4.
+func (af AddressFamily) IsIP() bool {
+	return af == AddressFamilyIPv4 || af == AddressFamilyIPv6
+}
+
+// IsDomain returns true if current AddressFamily is Domain.
+func (af AddressFamily) IsDomain() bool {
+	return af == AddressFamilyDomain
+}
+
+// Address represents a network address to be communicated with. It may be an IP address or domain
+// address, not both. This interface doesn't resolve IP address for a given domain.
+type Address interface {
+	IP() net.IP     // IP of this Address
+	Domain() string // Domain of this Address
+	Family() AddressFamily
+
+	String() string // String representation of this Address
+}
+
+func isAlphaNum(c byte) bool {
+	return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z')
+}
+
+// ParseAddress parses a string into an Address. The return value will be an IPAddress when
+// the string is in the form of IPv4 or IPv6 address, or a DomainAddress otherwise.
+func ParseAddress(addr string) Address {
+	// Handle IPv6 address in form as "[2001:4860:0:2001::68]"
+	lenAddr := len(addr)
+	if lenAddr > 0 && addr[0] == '[' && addr[lenAddr-1] == ']' {
+		addr = addr[1 : lenAddr-1]
+		lenAddr -= 2
+	}
+
+	if lenAddr > 0 && (!isAlphaNum(addr[0]) || !isAlphaNum(addr[len(addr)-1])) {
+		addr = strings.TrimSpace(addr)
+	}
+
+	ip := net.ParseIP(addr)
+	if ip != nil {
+		return IPAddress(ip)
+	}
+	return DomainAddress(addr)
+}
+
+var bytes0 = []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
+
+// IPAddress creates an Address with given IP.
+func IPAddress(ip []byte) Address {
+	switch len(ip) {
+	case net.IPv4len:
+		var addr ipv4Address = [4]byte{ip[0], ip[1], ip[2], ip[3]}
+		return addr
+	case net.IPv6len:
+		if bytes.Equal(ip[:10], bytes0) && ip[10] == 0xff && ip[11] == 0xff {
+			return IPAddress(ip[12:16])
+		}
+		var addr ipv6Address = [16]byte{
+			ip[0], ip[1], ip[2], ip[3],
+			ip[4], ip[5], ip[6], ip[7],
+			ip[8], ip[9], ip[10], ip[11],
+			ip[12], ip[13], ip[14], ip[15],
+		}
+		return addr
+	default:
+		return nil
+	}
+}
+
+// DomainAddress creates an Address with given domain.
+// This is an internal function that forcibly converts a string to domain.
+// It's mainly used in test files and mux.
+// Unless you have a specific reason, use net.ParseAddress instead,
+// as this function does not check whether the input is an IP address.
+// Otherwise, you will get strange results like domain: 1.1.1.1
+func DomainAddress(domain string) Address {
+	return domainAddress(domain)
+}
+
+type ipv4Address [4]byte
+
+func (a ipv4Address) IP() net.IP {
+	return net.IP(a[:])
+}
+
+func (ipv4Address) Domain() string {
+	panic("Calling Domain() on an IPv4Address.")
+}
+
+func (ipv4Address) Family() AddressFamily {
+	return AddressFamilyIPv4
+}
+
+func (a ipv4Address) String() string {
+	return a.IP().String()
+}
+
+type ipv6Address [16]byte
+
+func (a ipv6Address) IP() net.IP {
+	return net.IP(a[:])
+}
+
+func (ipv6Address) Domain() string {
+	panic("Calling Domain() on an IPv6Address.")
+}
+
+func (ipv6Address) Family() AddressFamily {
+	return AddressFamilyIPv6
+}
+
+func (a ipv6Address) String() string {
+	return "[" + a.IP().String() + "]"
+}
+
+type domainAddress string
+
+func (domainAddress) IP() net.IP {
+	panic("Calling IP() on a DomainAddress.")
+}
+
+func (a domainAddress) Domain() string {
+	return string(a)
+}
+
+func (domainAddress) Family() AddressFamily {
+	return AddressFamilyDomain
+}
+
+func (a domainAddress) String() string {
+	return a.Domain()
+}

common/xray/net/destination.go

@@ -0,0 +1,146 @@
+package net
+
+import (
+	"net"
+	"strings"
+)
+
+// Destination represents a network destination including address and protocol (tcp / udp).
+type Destination struct {
+	Address Address
+	Port    Port
+	Network Network
+}
+
+// DestinationFromAddr generates a Destination from a net address.
+func DestinationFromAddr(addr net.Addr) Destination {
+	switch addr := addr.(type) {
+	case *net.TCPAddr:
+		return TCPDestination(IPAddress(addr.IP), Port(addr.Port))
+	case *net.UDPAddr:
+		return UDPDestination(IPAddress(addr.IP), Port(addr.Port))
+	case *net.UnixAddr:
+		return UnixDestination(DomainAddress(addr.Name))
+	default:
+		panic("Net: Unknown address type.")
+	}
+}
+
+// ParseDestination converts a destination from its string presentation.
+func ParseDestination(dest string) (Destination, error) {
+	d := Destination{
+		Address: AnyIP,
+		Port:    Port(0),
+	}
+	if strings.HasPrefix(dest, "tcp:") {
+		d.Network = Network_TCP
+		dest = dest[4:]
+	} else if strings.HasPrefix(dest, "udp:") {
+		d.Network = Network_UDP
+		dest = dest[4:]
+	} else if strings.HasPrefix(dest, "unix:") {
+		d = UnixDestination(DomainAddress(dest[5:]))
+		return d, nil
+	}
+
+	hstr, pstr, err := SplitHostPort(dest)
+	if err != nil {
+		return d, err
+	}
+	if len(hstr) > 0 {
+		d.Address = ParseAddress(hstr)
+	}
+	if len(pstr) > 0 {
+		port, err := PortFromString(pstr)
+		if err != nil {
+			return d, err
+		}
+		d.Port = port
+	}
+	return d, nil
+}
+
+// TCPDestination creates a TCP destination with given address
+func TCPDestination(address Address, port Port) Destination {
+	return Destination{
+		Network: Network_TCP,
+		Address: address,
+		Port:    port,
+	}
+}
+
+// UDPDestination creates a UDP destination with given address
+func UDPDestination(address Address, port Port) Destination {
+	return Destination{
+		Network: Network_UDP,
+		Address: address,
+		Port:    port,
+	}
+}
+
+// UnixDestination creates a Unix destination with given address
+func UnixDestination(address Address) Destination {
+	return Destination{
+		Network: Network_UNIX,
+		Address: address,
+	}
+}
+
+// NetAddr returns the network address in this Destination in string form.
+func (d Destination) NetAddr() string {
+	addr := ""
+	if d.Network == Network_TCP || d.Network == Network_UDP {
+		addr = d.Address.String() + ":" + d.Port.String()
+	} else if d.Network == Network_UNIX {
+		addr = d.Address.String()
+	}
+	return addr
+}
+
+// RawNetAddr converts a net.Addr from its Destination presentation.
+func (d Destination) RawNetAddr() net.Addr {
+	var addr net.Addr
+	switch d.Network {
+	case Network_TCP:
+		if d.Address.Family().IsIP() {
+			addr = &net.TCPAddr{
+				IP:   d.Address.IP(),
+				Port: int(d.Port),
+			}
+		}
+	case Network_UDP:
+		if d.Address.Family().IsIP() {
+			addr = &net.UDPAddr{
+				IP:   d.Address.IP(),
+				Port: int(d.Port),
+			}
+		}
+	case Network_UNIX:
+		if d.Address.Family().IsDomain() {
+			addr = &net.UnixAddr{
+				Name: d.Address.String(),
+				Net:  d.Network.SystemString(),
+			}
+		}
+	}
+	return addr
+}
+
+// String returns the strings form of this Destination.
+func (d Destination) String() string {
+	prefix := "unknown:"
+	switch d.Network {
+	case Network_TCP:
+		prefix = "tcp:"
+	case Network_UDP:
+		prefix = "udp:"
+	case Network_UNIX:
+		prefix = "unix:"
+	}
+	return prefix + d.NetAddr()
+}
+
+// IsValid returns true if this Destination is valid.
+func (d Destination) IsValid() bool {
+	return d.Network != Network_Unknown
+}

common/xray/net/net.go

@@ -0,0 +1,13 @@
+package net
+
+import "time"
+
+// defines the maximum time an idle TCP session can survive in the tunnel, so
+// it should be consistent across HTTP versions and with other transports.
+const ConnIdleTimeout = 300 * time.Second
+
+// consistent with quic-go
+const QuicgoH3KeepAlivePeriod = 10 * time.Second
+
+// consistent with chrome
+const ChromeH2KeepAlivePeriod = 45 * time.Second

common/xray/net/network.go

@@ -0,0 +1,33 @@
+package net
+
+type Network int32
+
+const (
+	Network_Unknown Network = 0
+	Network_TCP     Network = 2
+	Network_UDP     Network = 3
+	Network_UNIX    Network = 4
+)
+
+func (n Network) SystemString() string {
+	switch n {
+	case Network_TCP:
+		return "tcp"
+	case Network_UDP:
+		return "udp"
+	case Network_UNIX:
+		return "unix"
+	default:
+		return "unknown"
+	}
+}
+
+// HasNetwork returns true if the network list has a certain network.
+func HasNetwork(list []Network, network Network) bool {
+	for _, value := range list {
+		if value == network {
+			return true
+		}
+	}
+	return false
+}

common/xray/net/port.go

@@ -0,0 +1,55 @@
+package net
+
+import (
+	"encoding/binary"
+	"strconv"
+
+	E "github.com/sagernet/sing/common/exceptions"
+)
+
+// Port represents a network port in TCP and UDP protocol.
+type Port uint16
+
+// PortFromBytes converts a byte array to a Port, assuming bytes are in big endian order.
+// @unsafe Caller must ensure that the byte array has at least 2 elements.
+func PortFromBytes(port []byte) Port {
+	return Port(binary.BigEndian.Uint16(port))
+}
+
+// PortFromInt converts an integer to a Port.
+// @error when the integer is not positive or larger then 65535
+func PortFromInt(val uint32) (Port, error) {
+	if val > 65535 {
+		return Port(0), E.New("invalid port range: ", val)
+	}
+	return Port(val), nil
+}
+
+// PortFromString converts a string to a Port.
+// @error when the string is not an integer or the integral value is a not a valid Port.
+func PortFromString(s string) (Port, error) {
+	val, err := strconv.ParseUint(s, 10, 32)
+	if err != nil {
+		return Port(0), E.New("invalid port range: ", s)
+	}
+	return PortFromInt(uint32(val))
+}
+
+// Value return the corresponding uint16 value of a Port.
+func (p Port) Value() uint16 {
+	return uint16(p)
+}
+
+// String returns the string presentation of a Port.
+func (p Port) String() string {
+	return strconv.Itoa(int(p))
+}
+
+type MemoryPortRange struct {
+	From Port
+	To   Port
+}
+
+func (r MemoryPortRange) Contains(port Port) bool {
+	return r.From <= port && port <= r.To
+}

common/xray/net/system.go

@@ -0,0 +1,84 @@
+package net
+
+import "net"
+
+// DialTCP is an alias of net.DialTCP.
+var (
+	DialTCP  = net.DialTCP
+	DialUDP  = net.DialUDP
+	DialUnix = net.DialUnix
+	Dial     = net.Dial
+)
+
+type ListenConfig = net.ListenConfig
+
+var (
+	Listen     = net.Listen
+	ListenTCP  = net.ListenTCP
+	ListenUDP  = net.ListenUDP
+	ListenUnix = net.ListenUnix
+)
+
+var LookupIP = net.LookupIP
+
+var FileConn = net.FileConn
+
+// ParseIP is an alias of net.ParseIP
+var ParseIP = net.ParseIP
+
+var SplitHostPort = net.SplitHostPort
+
+var CIDRMask = net.CIDRMask
+
+type (
+	Addr       = net.Addr
+	Conn       = net.Conn
+	PacketConn = net.PacketConn
+)
+
+type (
+	TCPAddr = net.TCPAddr
+	TCPConn = net.TCPConn
+)
+
+type (
+	UDPAddr = net.UDPAddr
+	UDPConn = net.UDPConn
+)
+
+type (
+	UnixAddr = net.UnixAddr
+	UnixConn = net.UnixConn
+)
+
+// IP is an alias for net.IP.
+type (
+	IP     = net.IP
+	IPMask = net.IPMask
+	IPNet  = net.IPNet
+)
+
+const (
+	IPv4len = net.IPv4len
+	IPv6len = net.IPv6len
+)
+
+type (
+	Error     = net.Error
+	AddrError = net.AddrError
+)
+
+type (
+	Dialer       = net.Dialer
+	Listener     = net.Listener
+	TCPListener  = net.TCPListener
+	UnixListener = net.UnixListener
+)
+
+var (
+	ResolveTCPAddr  = net.ResolveTCPAddr
+	ResolveUDPAddr  = net.ResolveUDPAddr
+	ResolveUnixAddr = net.ResolveUnixAddr
+)
+
+type Resolver = net.Resolver

common/xray/pipe/impl.go

@@ -0,0 +1,215 @@
+package pipe
+
+import (
+	"errors"
+	"io"
+	"runtime"
+	"sync"
+	"time"
+
+	"github.com/sagernet/sing-box/common/xray"
+	"github.com/sagernet/sing-box/common/xray/buf"
+	"github.com/sagernet/sing-box/common/xray/signal"
+	"github.com/sagernet/sing-box/common/xray/signal/done"
+)
+
+type state byte
+
+const (
+	open state = iota
+	closed
+	errord
+)
+
+type pipeOption struct {
+	limit           int32 // maximum buffer size in bytes
+	discardOverflow bool
+}
+
+func (o *pipeOption) isFull(curSize int32) bool {
+	return o.limit >= 0 && curSize > o.limit
+}
+
+type pipe struct {
+	sync.Mutex
+	data        buf.MultiBuffer
+	readSignal  *signal.Notifier
+	writeSignal *signal.Notifier
+	done        *done.Instance
+	errChan     chan error
+	option      pipeOption
+	state       state
+}
+
+var (
+	errBufferFull = errors.New("buffer full")
+	errSlowDown   = errors.New("slow down")
+)
+
+func (p *pipe) Len() int32 {
+	data := p.data
+	if data == nil {
+		return 0
+	}
+	return data.Len()
+}
+
+func (p *pipe) getState(forRead bool) error {
+	switch p.state {
+	case open:
+		if !forRead && p.option.isFull(p.data.Len()) {
+			return errBufferFull
+		}
+		return nil
+	case closed:
+		if !forRead {
+			return io.ErrClosedPipe
+		}
+		if !p.data.IsEmpty() {
+			return nil
+		}
+		return io.EOF
+	case errord:
+		return io.ErrClosedPipe
+	default:
+		panic("impossible case")
+	}
+}
+
+func (p *pipe) readMultiBufferInternal() (buf.MultiBuffer, error) {
+	p.Lock()
+	defer p.Unlock()
+
+	if err := p.getState(true); err != nil {
+		return nil, err
+	}
+
+	data := p.data
+	p.data = nil
+	return data, nil
+}
+
+func (p *pipe) ReadMultiBuffer() (buf.MultiBuffer, error) {
+	for {
+		data, err := p.readMultiBufferInternal()
+		if data != nil || err != nil {
+			p.writeSignal.Signal()
+			return data, err
+		}
+
+		select {
+		case <-p.readSignal.Wait():
+		case <-p.done.Wait():
+		case err = <-p.errChan:
+			return nil, err
+		}
+	}
+}
+
+func (p *pipe) ReadMultiBufferTimeout(d time.Duration) (buf.MultiBuffer, error) {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+
+	for {
+		data, err := p.readMultiBufferInternal()
+		if data != nil || err != nil {
+			p.writeSignal.Signal()
+			return data, err
+		}
+
+		select {
+		case <-p.readSignal.Wait():
+		case <-p.done.Wait():
+		case <-timer.C:
+			return nil, buf.ErrReadTimeout
+		}
+	}
+}
+
+func (p *pipe) writeMultiBufferInternal(mb buf.MultiBuffer) error {
+	p.Lock()
+	defer p.Unlock()
+
+	if err := p.getState(false); err != nil {
+		return err
+	}
+
+	if p.data == nil {
+		p.data = mb
+		return nil
+	}
+
+	p.data, _ = buf.MergeMulti(p.data, mb)
+	return errSlowDown
+}
+
+func (p *pipe) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	if mb.IsEmpty() {
+		return nil
+	}
+
+	for {
+		err := p.writeMultiBufferInternal(mb)
+		if err == nil {
+			p.readSignal.Signal()
+			return nil
+		}
+
+		if err == errSlowDown {
+			p.readSignal.Signal()
+
+			// Yield current goroutine. Hopefully the reading counterpart can pick up the payload.
+			runtime.Gosched()
+			return nil
+		}
+
+		if err == errBufferFull && p.option.discardOverflow {
+			buf.ReleaseMulti(mb)
+			return nil
+		}
+
+		if err != errBufferFull {
+			buf.ReleaseMulti(mb)
+			p.readSignal.Signal()
+			return err
+		}
+
+		select {
+		case <-p.writeSignal.Wait():
+		case <-p.done.Wait():
+			return io.ErrClosedPipe
+		}
+	}
+}
+
+func (p *pipe) Close() error {
+	p.Lock()
+	defer p.Unlock()
+
+	if p.state == closed || p.state == errord {
+		return nil
+	}
+
+	p.state = closed
+	common.Must(p.done.Close())
+	return nil
+}
+
+// Interrupt implements common.Interruptible.
+func (p *pipe) Interrupt() {
+	p.Lock()
+	defer p.Unlock()
+
+	if p.state == closed || p.state == errord {
+		return
+	}
+
+	p.state = errord
+
+	if !p.data.IsEmpty() {
+		buf.ReleaseMulti(p.data)
+		p.data = nil
+	}
+
+	common.Must(p.done.Close())
+}

common/xray/pipe/pipe.go

@@ -0,0 +1,53 @@
+package pipe
+
+import (
+	"github.com/sagernet/sing-box/common/xray/signal"
+	"github.com/sagernet/sing-box/common/xray/signal/done"
+)
+
+// Option for creating new Pipes.
+type Option func(*pipeOption)
+
+// WithoutSizeLimit returns an Option for Pipe to have no size limit.
+func WithoutSizeLimit() Option {
+	return func(opt *pipeOption) {
+		opt.limit = -1
+	}
+}
+
+// WithSizeLimit returns an Option for Pipe to have the given size limit.
+func WithSizeLimit(limit int32) Option {
+	return func(opt *pipeOption) {
+		opt.limit = limit
+	}
+}
+
+// DiscardOverflow returns an Option for Pipe to discard writes if full.
+func DiscardOverflow() Option {
+	return func(opt *pipeOption) {
+		opt.discardOverflow = true
+	}
+}
+
+// New creates a new Reader and Writer that connects to each other.
+func New(opts ...Option) (*Reader, *Writer) {
+	p := &pipe{
+		readSignal:  signal.NewNotifier(),
+		writeSignal: signal.NewNotifier(),
+		done:        done.New(),
+		errChan:     make(chan error, 1),
+		option: pipeOption{
+			limit: -1,
+		},
+	}
+
+	for _, opt := range opts {
+		opt(&(p.option))
+	}
+
+	return &Reader{
+			pipe: p,
+		}, &Writer{
+			pipe: p,
+		}
+}

common/xray/pipe/reader.go

@@ -0,0 +1,41 @@
+package pipe
+
+import (
+	"time"
+
+	"github.com/sagernet/sing-box/common/xray/buf"
+)
+
+// Reader is a buf.Reader that reads content from a pipe.
+type Reader struct {
+	pipe *pipe
+}
+
+// ReadMultiBuffer implements buf.Reader.
+func (r *Reader) ReadMultiBuffer() (buf.MultiBuffer, error) {
+	return r.pipe.ReadMultiBuffer()
+}
+
+// ReadMultiBufferTimeout reads content from a pipe within the given duration, or returns buf.ErrTimeout otherwise.
+func (r *Reader) ReadMultiBufferTimeout(d time.Duration) (buf.MultiBuffer, error) {
+	return r.pipe.ReadMultiBufferTimeout(d)
+}
+
+// Interrupt implements common.Interruptible.
+func (r *Reader) Interrupt() {
+	r.pipe.Interrupt()
+}
+
+// ReturnAnError makes ReadMultiBuffer return an error, only once.
+func (r *Reader) ReturnAnError(err error) {
+	r.pipe.errChan <- err
+}
+
+// Recover catches an error set by ReturnAnError, if exists.
+func (r *Reader) Recover() (err error) {
+	select {
+	case err = <-r.pipe.errChan:
+	default:
+	}
+	return
+}

common/xray/pipe/writer.go

@@ -0,0 +1,29 @@
+package pipe
+
+import (
+	"github.com/sagernet/sing-box/common/xray/buf"
+)
+
+// Writer is a buf.Writer that writes data into a pipe.
+type Writer struct {
+	pipe *pipe
+}
+
+// WriteMultiBuffer implements buf.Writer.
+func (w *Writer) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	return w.pipe.WriteMultiBuffer(mb)
+}
+
+// Close implements io.Closer. After the pipe is closed, writing to the pipe will return io.ErrClosedPipe, while reading will return io.EOF.
+func (w *Writer) Close() error {
+	return w.pipe.Close()
+}
+
+func (w *Writer) Len() int32 {
+	return w.pipe.Len()
+}
+
+// Interrupt implements common.Interruptible.
+func (w *Writer) Interrupt() {
+	w.pipe.Interrupt()
+}

common/xray/serial/serial.go

@@ -0,0 +1,29 @@
+package serial
+
+import (
+	"encoding/binary"
+	"io"
+)
+
+// ReadUint16 reads first two bytes from the reader, and then converts them to an uint16 value.
+func ReadUint16(reader io.Reader) (uint16, error) {
+	var b [2]byte
+	if _, err := io.ReadFull(reader, b[:]); err != nil {
+		return 0, err
+	}
+	return binary.BigEndian.Uint16(b[:]), nil
+}
+
+// WriteUint16 writes an uint16 value into writer.
+func WriteUint16(writer io.Writer, value uint16) (int, error) {
+	var b [2]byte
+	binary.BigEndian.PutUint16(b[:], value)
+	return writer.Write(b[:])
+}
+
+// WriteUint64 writes an uint64 value into writer.
+func WriteUint64(writer io.Writer, value uint64) (int, error) {
+	var b [8]byte
+	binary.BigEndian.PutUint64(b[:], value)
+	return writer.Write(b[:])
+}

common/xray/serial/string.go

@@ -0,0 +1,35 @@
+package serial
+
+import (
+	"fmt"
+	"strings"
+)
+
+// ToString serializes an arbitrary value into string.
+func ToString(v interface{}) string {
+	if v == nil {
+		return ""
+	}
+
+	switch value := v.(type) {
+	case string:
+		return value
+	case *string:
+		return *value
+	case fmt.Stringer:
+		return value.String()
+	case error:
+		return value.Error()
+	default:
+		return fmt.Sprintf("%+v", value)
+	}
+}
+
+// Concat concatenates all input into a single string.
+func Concat(v ...interface{}) string {
+	builder := strings.Builder{}
+	for _, value := range v {
+		builder.WriteString(ToString(value))
+	}
+	return builder.String()
+}

common/xray/signal/done/done.go

@@ -0,0 +1,49 @@
+package done
+
+import (
+	"sync"
+)
+
+// Instance is a utility for notifications of something being done.
+type Instance struct {
+	access sync.Mutex
+	c      chan struct{}
+	closed bool
+}
+
+// New returns a new Done.
+func New() *Instance {
+	return &Instance{
+		c: make(chan struct{}),
+	}
+}
+
+// Done returns true if Close() is called.
+func (d *Instance) Done() bool {
+	select {
+	case <-d.Wait():
+		return true
+	default:
+		return false
+	}
+}
+
+// Wait returns a channel for waiting for done.
+func (d *Instance) Wait() <-chan struct{} {
+	return d.c
+}
+
+// Close marks this Done 'done'. This method may be called multiple times. All calls after first call will have no effect on its status.
+func (d *Instance) Close() error {
+	d.access.Lock()
+	defer d.access.Unlock()
+
+	if d.closed {
+		return nil
+	}
+
+	d.closed = true
+	close(d.c)
+
+	return nil
+}

common/xray/signal/notifier.go

@@ -0,0 +1,26 @@
+package signal
+
+// Notifier is a utility for notifying changes. The change producer may notify changes multiple time, and the consumer may get notified asynchronously.
+type Notifier struct {
+	c chan struct{}
+}
+
+// NewNotifier creates a new Notifier.
+func NewNotifier() *Notifier {
+	return &Notifier{
+		c: make(chan struct{}, 1),
+	}
+}
+
+// Signal signals a change, usually by producer. This method never blocks.
+func (n *Notifier) Signal() {
+	select {
+	case n.c <- struct{}{}:
+	default:
+	}
+}
+
+// Wait returns a channel for waiting for changes. The returned channel never gets closed.
+func (n *Notifier) Wait() <-chan struct{} {
+	return n.c
+}

common/xray/signal/pubsub/pubsub.go

@@ -0,0 +1,105 @@
+package pubsub
+
+import (
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/sagernet/sing-box/common/xray"
+	"github.com/sagernet/sing-box/common/xray/signal/done"
+	"github.com/sagernet/sing-box/common/xray/task"
+)
+
+type Subscriber struct {
+	buffer chan interface{}
+	done   *done.Instance
+}
+
+func (s *Subscriber) push(msg interface{}) {
+	select {
+	case s.buffer <- msg:
+	default:
+	}
+}
+
+func (s *Subscriber) Wait() <-chan interface{} {
+	return s.buffer
+}
+
+func (s *Subscriber) Close() error {
+	return s.done.Close()
+}
+
+func (s *Subscriber) IsClosed() bool {
+	return s.done.Done()
+}
+
+type Service struct {
+	sync.RWMutex
+	subs  map[string][]*Subscriber
+	ctask *task.Periodic
+}
+
+func NewService() *Service {
+	s := &Service{
+		subs: make(map[string][]*Subscriber),
+	}
+	s.ctask = &task.Periodic{
+		Execute:  s.Cleanup,
+		Interval: time.Second * 30,
+	}
+	return s
+}
+
+// Cleanup cleans up internal caches of subscribers.
+// Visible for testing only.
+func (s *Service) Cleanup() error {
+	s.Lock()
+	defer s.Unlock()
+
+	if len(s.subs) == 0 {
+		return errors.New("nothing to do")
+	}
+
+	for name, subs := range s.subs {
+		newSub := make([]*Subscriber, 0, len(s.subs))
+		for _, sub := range subs {
+			if !sub.IsClosed() {
+				newSub = append(newSub, sub)
+			}
+		}
+		if len(newSub) == 0 {
+			delete(s.subs, name)
+		} else {
+			s.subs[name] = newSub
+		}
+	}
+
+	if len(s.subs) == 0 {
+		s.subs = make(map[string][]*Subscriber)
+	}
+	return nil
+}
+
+func (s *Service) Subscribe(name string) *Subscriber {
+	sub := &Subscriber{
+		buffer: make(chan interface{}, 16),
+		done:   done.New(),
+	}
+	s.Lock()
+	s.subs[name] = append(s.subs[name], sub)
+	s.Unlock()
+	common.Must(s.ctask.Start())
+	return sub
+}
+
+func (s *Service) Publish(name string, message interface{}) {
+	s.RLock()
+	defer s.RUnlock()
+
+	for _, sub := range s.subs[name] {
+		if !sub.IsClosed() {
+			sub.push(message)
+		}
+	}
+}

common/xray/signal/semaphore/semaphore.go

@@ -0,0 +1,27 @@
+package semaphore
+
+// Instance is an implementation of semaphore.
+type Instance struct {
+	token chan struct{}
+}
+
+// New create a new Semaphore with n permits.
+func New(n int) *Instance {
+	s := &Instance{
+		token: make(chan struct{}, n),
+	}
+	for i := 0; i < n; i++ {
+		s.token <- struct{}{}
+	}
+	return s
+}
+
+// Wait returns a channel for acquiring a permit.
+func (s *Instance) Wait() <-chan struct{} {
+	return s.token
+}
+
+// Signal releases a permit into the semaphore.
+func (s *Instance) Signal() {
+	s.token <- struct{}{}
+}

common/xray/signal/timer.go

@@ -0,0 +1,82 @@
+package signal
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/sagernet/sing-box/common/xray"
+	"github.com/sagernet/sing-box/common/xray/task"
+)
+
+type ActivityUpdater interface {
+	Update()
+}
+
+type ActivityTimer struct {
+	sync.RWMutex
+	updated   chan struct{}
+	checkTask *task.Periodic
+	onTimeout func()
+}
+
+func (t *ActivityTimer) Update() {
+	select {
+	case t.updated <- struct{}{}:
+	default:
+	}
+}
+
+func (t *ActivityTimer) check() error {
+	select {
+	case <-t.updated:
+	default:
+		t.finish()
+	}
+	return nil
+}
+
+func (t *ActivityTimer) finish() {
+	t.Lock()
+	defer t.Unlock()
+
+	if t.onTimeout != nil {
+		t.onTimeout()
+		t.onTimeout = nil
+	}
+	if t.checkTask != nil {
+		t.checkTask.Close()
+		t.checkTask = nil
+	}
+}
+
+func (t *ActivityTimer) SetTimeout(timeout time.Duration) {
+	if timeout == 0 {
+		t.finish()
+		return
+	}
+
+	checkTask := &task.Periodic{
+		Interval: timeout,
+		Execute:  t.check,
+	}
+
+	t.Lock()
+
+	if t.checkTask != nil {
+		t.checkTask.Close()
+	}
+	t.checkTask = checkTask
+	t.Unlock()
+	t.Update()
+	common.Must(checkTask.Start())
+}
+
+func CancelAfterInactivity(ctx context.Context, cancel context.CancelFunc, timeout time.Duration) *ActivityTimer {
+	timer := &ActivityTimer{
+		updated:   make(chan struct{}, 1),
+		onTimeout: cancel,
+	}
+	timer.SetTimeout(timeout)
+	return timer
+}

common/xray/stat/connection.go

@@ -0,0 +1,34 @@
+package stat
+
+import (
+	"net"
+
+	"github.com/sagernet/sing-box/common/xray/stats"
+)
+
+type Connection interface {
+	net.Conn
+}
+
+type CounterConnection struct {
+	Connection
+	ReadCounter  stats.Counter
+	WriteCounter stats.Counter
+}
+
+func (c *CounterConnection) Read(b []byte) (int, error) {
+	nBytes, err := c.Connection.Read(b)
+	if c.ReadCounter != nil {
+		c.ReadCounter.Add(int64(nBytes))
+	}
+
+	return nBytes, err
+}
+
+func (c *CounterConnection) Write(b []byte) (int, error) {
+	nBytes, err := c.Connection.Write(b)
+	if c.WriteCounter != nil {
+		c.WriteCounter.Add(int64(nBytes))
+	}
+	return nBytes, err
+}

common/xray/stats/stats.go

@@ -0,0 +1,13 @@
+package stats
+
+// Counter is the interface for stats counters.
+//
+// xray:api:stable
+type Counter interface {
+	// Value is the current value of the counter.
+	Value() int64
+	// Set sets a new value to the counter, and returns the previous one.
+	Set(int64) int64
+	// Add adds a value to the current counter value, and returns the previous value.
+	Add(int64) int64
+}

common/xray/task/common.go

@@ -0,0 +1,10 @@
+package task
+
+import "github.com/sagernet/sing-box/common/xray"
+
+// Close returns a func() that closes v.
+func Close(v interface{}) func() error {
+	return func() error {
+		return common.Close(v)
+	}
+}

common/xray/task/periodic.go

@@ -0,0 +1,85 @@
+package task
+
+import (
+	"sync"
+	"time"
+)
+
+// Periodic is a task that runs periodically.
+type Periodic struct {
+	// Interval of the task being run
+	Interval time.Duration
+	// Execute is the task function
+	Execute func() error
+
+	access  sync.Mutex
+	timer   *time.Timer
+	running bool
+}
+
+func (t *Periodic) hasClosed() bool {
+	t.access.Lock()
+	defer t.access.Unlock()
+
+	return !t.running
+}
+
+func (t *Periodic) checkedExecute() error {
+	if t.hasClosed() {
+		return nil
+	}
+
+	if err := t.Execute(); err != nil {
+		t.access.Lock()
+		t.running = false
+		t.access.Unlock()
+		return err
+	}
+
+	t.access.Lock()
+	defer t.access.Unlock()
+
+	if !t.running {
+		return nil
+	}
+
+	t.timer = time.AfterFunc(t.Interval, func() {
+		t.checkedExecute()
+	})
+
+	return nil
+}
+
+// Start implements common.Runnable.
+func (t *Periodic) Start() error {
+	t.access.Lock()
+	if t.running {
+		t.access.Unlock()
+		return nil
+	}
+	t.running = true
+	t.access.Unlock()
+
+	if err := t.checkedExecute(); err != nil {
+		t.access.Lock()
+		t.running = false
+		t.access.Unlock()
+		return err
+	}
+
+	return nil
+}
+
+// Close implements common.Closable.
+func (t *Periodic) Close() error {
+	t.access.Lock()
+	defer t.access.Unlock()
+
+	t.running = false
+	if t.timer != nil {
+		t.timer.Stop()
+		t.timer = nil
+	}
+
+	return nil
+}

common/xray/task/task.go

@@ -0,0 +1,64 @@
+package task
+
+import (
+	"context"
+
+	"github.com/sagernet/sing-box/common/xray/signal/semaphore"
+)
+
+// OnSuccess executes g() after f() returns nil.
+func OnSuccess(f func() error, g func() error) func() error {
+	return func() error {
+		if err := f(); err != nil {
+			return err
+		}
+		return g()
+	}
+}
+
+// Run executes a list of tasks in parallel, returns the first error encountered or nil if all tasks pass.
+func Run(ctx context.Context, tasks ...func() error) error {
+	n := len(tasks)
+	s := semaphore.New(n)
+	done := make(chan error, 1)
+
+	for _, task := range tasks {
+		<-s.Wait()
+		go func(f func() error) {
+			err := f()
+			if err == nil {
+				s.Signal()
+				return
+			}
+
+			select {
+			case done <- err:
+			default:
+			}
+		}(task)
+	}
+
+	/*
+		if altctx := ctx.Value("altctx"); altctx != nil {
+			ctx = altctx.(context.Context)
+		}
+	*/
+
+	for i := 0; i < n; i++ {
+		select {
+		case err := <-done:
+			return err
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-s.Wait():
+		}
+	}
+
+	/*
+		if cancel := ctx.Value("cancel"); cancel != nil {
+			cancel.(context.CancelFunc)()
+		}
+	*/
+
+	return nil
+}

common/xray/uuid/uuid.go

@@ -0,0 +1,101 @@
+package uuid
+
+import (
+	"bytes"
+	"crypto/rand"
+	"crypto/sha1"
+	"encoding/hex"
+
+	"github.com/sagernet/sing-box/common/xray"
+	E "github.com/sagernet/sing/common/exceptions"
+)
+
+var byteGroups = []int{8, 4, 4, 4, 12}
+
+type UUID [16]byte
+
+// String returns the string representation of this UUID.
+func (u *UUID) String() string {
+	bytes := u.Bytes()
+	result := hex.EncodeToString(bytes[0 : byteGroups[0]/2])
+	start := byteGroups[0] / 2
+	for i := 1; i < len(byteGroups); i++ {
+		nBytes := byteGroups[i] / 2
+		result += "-"
+		result += hex.EncodeToString(bytes[start : start+nBytes])
+		start += nBytes
+	}
+	return result
+}
+
+// Bytes returns the bytes representation of this UUID.
+func (u *UUID) Bytes() []byte {
+	return u[:]
+}
+
+// Equals returns true if this UUID equals another UUID by value.
+func (u *UUID) Equals(another *UUID) bool {
+	if u == nil && another == nil {
+		return true
+	}
+	if u == nil || another == nil {
+		return false
+	}
+	return bytes.Equal(u.Bytes(), another.Bytes())
+}
+
+// New creates a UUID with random value.
+func New() UUID {
+	var uuid UUID
+	common.Must2(rand.Read(uuid.Bytes()))
+	uuid[6] = (uuid[6] & 0x0f) | (4 << 4)
+	uuid[8] = (uuid[8]&(0xff>>2) | (0x02 << 6))
+	return uuid
+}
+
+// ParseBytes converts a UUID in byte form to object.
+func ParseBytes(b []byte) (UUID, error) {
+	var uuid UUID
+	if len(b) != 16 {
+		return uuid, E.New("invalid UUID: ", b)
+	}
+	copy(uuid[:], b)
+	return uuid, nil
+}
+
+// ParseString converts a UUID in string form to object.
+func ParseString(str string) (UUID, error) {
+	var uuid UUID
+
+	text := []byte(str)
+	if l := len(text); l < 32 || l > 36 {
+		if l == 0 || l > 30 {
+			return uuid, E.New("invalid UUID: ", str)
+		}
+		h := sha1.New()
+		h.Write(uuid[:])
+		h.Write(text)
+		u := h.Sum(nil)[:16]
+		u[6] = (u[6] & 0x0f) | (5 << 4)
+		u[8] = (u[8]&(0xff>>2) | (0x02 << 6))
+		copy(uuid[:], u)
+		return uuid, nil
+	}
+
+	b := uuid.Bytes()
+
+	for _, byteGroup := range byteGroups {
+		if text[0] == '-' {
+			text = text[1:]
+		}
+
+		if _, err := hex.Decode(b[:byteGroup/2], text[:byteGroup]); err != nil {
+			return uuid, err
+		}
+
+		text = text[byteGroup:]
+		b = b[byteGroup/2:]
+	}
+
+	return uuid, nil
+}