move transport methods from net to io

common/alloc/buffer.go

@@ -11,6 +11,13 @@ func Release(buffer *Buffer) {
 	}
 }
 
+func Len(buffer *Buffer) int {
+	if buffer == nil {
+		return 0
+	}
+	return buffer.Len()
+}
+
 // 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.

common/crypto/authenticator.go

@@ -1,6 +1,6 @@
 package crypto
 
 type Authenticator interface {
-	AuthBytes() int
+	AuthSize() int
 	Authenticate(auth []byte, data []byte) []byte
 }

common/io/reader.go

@@ -0,0 +1,122 @@
+package io // import "github.com/v2ray/v2ray-core/common/io"
+
+import (
+	"io"
+
+	"github.com/v2ray/v2ray-core/common/alloc"
+	"github.com/v2ray/v2ray-core/common/crypto"
+	"github.com/v2ray/v2ray-core/common/serial"
+	"github.com/v2ray/v2ray-core/transport"
+)
+
+// ReadFrom reads from a reader and put all content to a buffer.
+// If buffer is nil, ReadFrom creates a new normal buffer.
+func ReadFrom(reader io.Reader, buffer *alloc.Buffer) (*alloc.Buffer, error) {
+	if buffer == nil {
+		buffer = alloc.NewBuffer()
+	}
+	nBytes, err := reader.Read(buffer.Value)
+	buffer.Slice(0, nBytes)
+	return buffer, err
+}
+
+type Reader interface {
+	Read() (*alloc.Buffer, error)
+}
+
+type AdaptiveReader struct {
+	reader   io.Reader
+	allocate func() *alloc.Buffer
+	isLarge  bool
+}
+
+func NewAdaptiveReader(reader io.Reader) *AdaptiveReader {
+	return &AdaptiveReader{
+		reader:   reader,
+		allocate: alloc.NewBuffer,
+		isLarge:  false,
+	}
+}
+
+func (this *AdaptiveReader) Read() (*alloc.Buffer, error) {
+	buffer, err := ReadFrom(this.reader, this.allocate())
+
+	if buffer.IsFull() && !this.isLarge {
+		this.allocate = alloc.NewLargeBuffer
+		this.isLarge = true
+	} else if !buffer.IsFull() {
+		this.allocate = alloc.NewBuffer
+		this.isLarge = false
+	}
+
+	if err != nil {
+		alloc.Release(buffer)
+		return nil, err
+	}
+	return buffer, nil
+}
+
+type ChunkReader struct {
+	reader io.Reader
+}
+
+func NewChunkReader(reader io.Reader) *ChunkReader {
+	return &ChunkReader{
+		reader: reader,
+	}
+}
+
+func (this *ChunkReader) Read() (*alloc.Buffer, error) {
+	buffer := alloc.NewLargeBuffer()
+	if _, err := io.ReadFull(this.reader, buffer.Value[:2]); err != nil {
+		alloc.Release(buffer)
+		return nil, err
+	}
+	length := serial.BytesLiteral(buffer.Value[:2]).Uint16Value()
+	if _, err := io.ReadFull(this.reader, buffer.Value[:length]); err != nil {
+		alloc.Release(buffer)
+		return nil, err
+	}
+	buffer.Slice(0, int(length))
+	return buffer, nil
+}
+
+type AuthenticationReader struct {
+	reader            Reader
+	authenticator     crypto.Authenticator
+	authBeforePayload bool
+}
+
+func NewAuthenticationReader(reader io.Reader, auth crypto.Authenticator, authBeforePayload bool) *AuthenticationReader {
+	return &AuthenticationReader{
+		reader:            NewChunkReader(reader),
+		authenticator:     auth,
+		authBeforePayload: authBeforePayload,
+	}
+}
+
+func (this *AuthenticationReader) Read() (*alloc.Buffer, error) {
+	buffer, err := this.reader.Read()
+	if err != nil {
+		alloc.Release(buffer)
+		return nil, err
+	}
+
+	authSize := this.authenticator.AuthSize()
+	var authBytes, payloadBytes []byte
+	if this.authBeforePayload {
+		authBytes = buffer.Value[:authSize]
+		payloadBytes = buffer.Value[authSize:]
+	} else {
+		payloadBytes = buffer.Value[:authSize]
+		authBytes = buffer.Value[authSize:]
+	}
+
+	actualAuthBytes := this.authenticator.Authenticate(nil, payloadBytes)
+	if !serial.BytesLiteral(authBytes).Equals(serial.BytesLiteral(actualAuthBytes)) {
+		alloc.Release(buffer)
+		return nil, transport.CorruptedPacket
+	}
+	buffer.Value = payloadBytes
+	return buffer, nil
+}

common/io/transport.go

@@ -0,0 +1,42 @@
+package io
+
+import (
+	"io"
+
+	"github.com/v2ray/v2ray-core/common/alloc"
+)
+
+func RawReaderToChan(stream chan<- *alloc.Buffer, reader io.Reader) error {
+	return ReaderToChan(stream, NewAdaptiveReader(reader))
+}
+
+// ReaderToChan dumps all content from a given reader to a chan by constantly reading it until EOF.
+func ReaderToChan(stream chan<- *alloc.Buffer, reader Reader) error {
+	for {
+		buffer, err := reader.Read()
+		if alloc.Len(buffer) > 0 {
+			stream <- buffer
+		} else {
+			alloc.Release(buffer)
+		}
+
+		if err != nil {
+			return err
+		}
+	}
+}
+
+// ChanToWriter dumps all content from a given chan to a writer until the chan is closed.
+func ChanToWriter(writer io.Writer, stream <-chan *alloc.Buffer) error {
+	for buffer := range stream {
+		nBytes, err := writer.Write(buffer.Value)
+		if nBytes < buffer.Len() {
+			_, err = writer.Write(buffer.Value[nBytes:])
+		}
+		buffer.Release()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

common/io/transport_test.go

@@ -0,0 +1,37 @@
+package io_test
+
+import (
+	"bytes"
+	"crypto/rand"
+	"io"
+	"testing"
+
+	"github.com/v2ray/v2ray-core/common/alloc"
+	. "github.com/v2ray/v2ray-core/common/io"
+	v2testing "github.com/v2ray/v2ray-core/testing"
+	"github.com/v2ray/v2ray-core/testing/assert"
+)
+
+func TestReaderAndWrite(t *testing.T) {
+	v2testing.Current(t)
+
+	size := 1024 * 1024
+	buffer := make([]byte, size)
+	nBytes, err := rand.Read(buffer)
+	assert.Int(nBytes).Equals(len(buffer))
+	assert.Error(err).IsNil()
+
+	readerBuffer := bytes.NewReader(buffer)
+	writerBuffer := bytes.NewBuffer(make([]byte, 0, size))
+
+	transportChan := make(chan *alloc.Buffer, 1024)
+
+	err = ReaderToChan(transportChan, NewAdaptiveReader(readerBuffer))
+	assert.Error(err).Equals(io.EOF)
+	close(transportChan)
+
+	err = ChanToWriter(writerBuffer, transportChan)
+	assert.Error(err).IsNil()
+
+	assert.Bytes(buffer).Equals(writerBuffer.Bytes())
+}