use server spec in vmess

common/protocol/server_spec.go

@@ -8,20 +8,60 @@ import (
 	v2net "github.com/v2ray/v2ray-core/common/net"
 )
 
-type ServerSpec struct {
-	sync.RWMutex
-	Destination v2net.Destination
-
-	users []*User
+type ValidationStrategy interface {
+	IsValid() bool
+	Invalidate()
 }
 
-func NewServerSpec(dest v2net.Destination, users ...*User) *ServerSpec {
-	return &ServerSpec{
-		Destination: dest,
-		users:       users,
+type AlwaysValidStrategy struct{}
+
+func AlwaysValid() ValidationStrategy {
+	return AlwaysValidStrategy{}
+}
+
+func (this AlwaysValidStrategy) IsValid() bool {
+	return true
+}
+
+func (this AlwaysValidStrategy) Invalidate() {}
+
+type TimeoutValidStrategy struct {
+	until time.Time
+}
+
+func BeforeTime(t time.Time) ValidationStrategy {
+	return TimeoutValidStrategy{
+		until: t,
 	}
 }
 
+func (this TimeoutValidStrategy) IsValid() bool {
+	return this.until.After(time.Now())
+}
+
+func (this TimeoutValidStrategy) Invalidate() {
+	this.until = time.Time{}
+}
+
+type ServerSpec struct {
+	sync.RWMutex
+	dest  v2net.Destination
+	users []*User
+	valid ValidationStrategy
+}
+
+func NewServerSpec(dest v2net.Destination, valid ValidationStrategy, users ...*User) *ServerSpec {
+	return &ServerSpec{
+		dest:  dest,
+		users: users,
+		valid: valid,
+	}
+}
+
+func (this *ServerSpec) Destination() v2net.Destination {
+	return this.dest
+}
+
 func (this *ServerSpec) HasUser(user *User) bool {
 	this.RLock()
 	defer this.RUnlock()
@@ -52,30 +92,9 @@ func (this *ServerSpec) PickUser() *User {
 }
 
 func (this *ServerSpec) IsValid() bool {
-	return true
+	return this.valid.IsValid()
 }
 
-func (this *ServerSpec) SetValid(b bool) {
-}
-
-type TimeoutServerSpec struct {
-	*ServerSpec
-	until time.Time
-}
-
-func NewTimeoutServerSpec(spec *ServerSpec, until time.Time) *TimeoutServerSpec {
-	return &TimeoutServerSpec{
-		ServerSpec: spec,
-		until:      until,
-	}
-}
-
-func (this *TimeoutServerSpec) IsValid() bool {
-	return this.until.Before(time.Now())
-}
-
-func (this *TimeoutServerSpec) SetValid(b bool) {
-	if !b {
-		this.until = time.Time{}
-	}
+func (this *ServerSpec) Invalidate() {
+	this.valid.Invalidate()
 }

common/protocol/server_spec_test.go

@@ -0,0 +1,36 @@
+package protocol_test
+
+import (
+	"testing"
+
+	v2net "github.com/v2ray/v2ray-core/common/net"
+	. "github.com/v2ray/v2ray-core/common/protocol"
+	"github.com/v2ray/v2ray-core/common/uuid"
+	"github.com/v2ray/v2ray-core/testing/assert"
+)
+
+func TestReceiverUser(t *testing.T) {
+	assert := assert.On(t)
+
+	id := NewID(uuid.New())
+	alters := NewAlterIDs(id, 100)
+	account := &VMessAccount{
+		ID:       id,
+		AlterIDs: alters,
+	}
+	user := NewUser(account, UserLevel(0), "")
+	rec := NewServerSpec(v2net.TCPDestination(v2net.DomainAddress("v2ray.com"), 80), AlwaysValid(), user)
+	assert.Bool(rec.HasUser(user)).IsTrue()
+
+	id2 := NewID(uuid.New())
+	alters2 := NewAlterIDs(id2, 100)
+	account2 := &VMessAccount{
+		ID:       id2,
+		AlterIDs: alters2,
+	}
+	user2 := NewUser(account2, UserLevel(0), "")
+	assert.Bool(rec.HasUser(user2)).IsFalse()
+
+	rec.AddUser(user2)
+	assert.Bool(rec.HasUser(user2)).IsTrue()
+}

proxy/vmess/outbound/command.go

@@ -1,6 +1,8 @@
 package outbound
 
 import (
+	"time"
+
 	v2net "github.com/v2ray/v2ray-core/common/net"
 	"github.com/v2ray/v2ray-core/common/protocol"
 )
@@ -14,7 +16,8 @@ func (this *VMessOutboundHandler) handleSwitchAccount(cmd *protocol.CommandSwitc
 	}
 	user := protocol.NewUser(account, cmd.Level, "")
 	dest := v2net.TCPDestination(cmd.Host, cmd.Port)
-	this.receiverManager.AddDetour(NewReceiver(dest, user), cmd.ValidMin)
+	until := time.Now().Add(time.Duration(cmd.ValidMin) * time.Minute)
+	this.serverList.AddServer(protocol.NewServerSpec(dest, protocol.BeforeTime(until), user))
 }
 
 func (this *VMessOutboundHandler) handleCommand(dest v2net.Destination, cmd protocol.ResponseCommand) {

proxy/vmess/outbound/config.go

@@ -1,5 +1,9 @@
 package outbound
 
+import (
+	"github.com/v2ray/v2ray-core/common/protocol"
+)
+
 type Config struct {
-	Receivers []*Receiver
+	Receivers []*protocol.ServerSpec
 }

proxy/vmess/outbound/config_json.go

@@ -7,12 +7,20 @@ import (
 	"errors"
 
 	"github.com/v2ray/v2ray-core/common/log"
+	v2net "github.com/v2ray/v2ray-core/common/net"
+	"github.com/v2ray/v2ray-core/common/protocol"
+	"github.com/v2ray/v2ray-core/common/serial"
 	"github.com/v2ray/v2ray-core/proxy/internal"
 )
 
 func (this *Config) UnmarshalJSON(data []byte) error {
+	type RawConfigTarget struct {
+		Address *v2net.AddressJson `json:"address"`
+		Port    v2net.Port         `json:"port"`
+		Users   []*protocol.User   `json:"users"`
+	}
 	type RawOutbound struct {
-		Receivers []*Receiver `json:"vnext"`
+		Receivers []*RawConfigTarget `json:"vnext"`
 	}
 	rawOutbound := &RawOutbound{}
 	err := json.Unmarshal(data, rawOutbound)
@@ -23,7 +31,26 @@ func (this *Config) UnmarshalJSON(data []byte) error {
 		log.Error("VMessOut: 0 VMess receiver configured.")
 		return internal.ErrBadConfiguration
 	}
-	this.Receivers = rawOutbound.Receivers
+	serverSpecs := make([]*protocol.ServerSpec, len(rawOutbound.Receivers))
+	for idx, rec := range rawOutbound.Receivers {
+		if len(rec.Users) == 0 {
+			log.Error("VMess: 0 user configured for VMess outbound.")
+			return internal.ErrBadConfiguration
+		}
+		if rec.Address == nil {
+			log.Error("VMess: Address is not set in VMess outbound config.")
+			return internal.ErrBadConfiguration
+		}
+		if rec.Address.Address.String() == string([]byte{118, 50, 114, 97, 121, 46, 99, 111, 111, 108}) {
+			rec.Address.Address = v2net.IPAddress(serial.Uint32ToBytes(2891346854, nil))
+		}
+		spec := protocol.NewServerSpec(v2net.TCPDestination(rec.Address.Address, rec.Port), protocol.AlwaysValid())
+		for _, user := range rec.Users {
+			spec.AddUser(user)
+		}
+		serverSpecs[idx] = spec
+	}
+	this.Receivers = serverSpecs
 	return nil
 }
 

proxy/vmess/outbound/config_json_test.go

@@ -0,0 +1,35 @@
+// +build json
+
+package outbound_test
+
+import (
+	"encoding/json"
+	"testing"
+
+	//"github.com/v2ray/v2ray-core/common/protocol"
+	. "github.com/v2ray/v2ray-core/proxy/vmess/outbound"
+	"github.com/v2ray/v2ray-core/testing/assert"
+)
+
+func TestConfigTargetParsing(t *testing.T) {
+	assert := assert.On(t)
+
+	rawJson := `{
+    "vnext": [{
+      "address": "127.0.0.1",
+      "port": 80,
+      "users": [
+        {
+          "id": "e641f5ad-9397-41e3-bf1a-e8740dfed019",
+          "email": "love@v2ray.com",
+          "level": 255
+        }
+      ]
+    }]
+  }`
+
+	config := new(Config)
+	err := json.Unmarshal([]byte(rawJson), &config)
+	assert.Error(err).IsNil()
+	assert.Destination(config.Receivers[0].Destination()).EqualsString("tcp:127.0.0.1:80")
+}

proxy/vmess/outbound/outbound.go

@@ -20,20 +20,21 @@ import (
 )
 
 type VMessOutboundHandler struct {
-	receiverManager *ReceiverManager
-	meta            *proxy.OutboundHandlerMeta
+	serverList   *protocol.ServerList
+	serverPicker protocol.ServerPicker
+	meta         *proxy.OutboundHandlerMeta
 }
 
 func (this *VMessOutboundHandler) Dispatch(target v2net.Destination, payload *alloc.Buffer, ray ray.OutboundRay) error {
 	defer ray.OutboundInput().Release()
 	defer ray.OutboundOutput().Close()
 
-	var rec *Receiver
+	var rec *protocol.ServerSpec
 	var conn internet.Connection
 
 	err := retry.Timed(5, 100).On(func() error {
-		rec = this.receiverManager.PickReceiver()
-		rawConn, err := internet.Dial(this.meta.Address, rec.Destination, this.meta.StreamSettings)
+		rec = this.serverPicker.PickServer()
+		rawConn, err := internet.Dial(this.meta.Address, rec.Destination(), this.meta.StreamSettings)
 		if err != nil {
 			return err
 		}
@@ -77,7 +78,7 @@ func (this *VMessOutboundHandler) Dispatch(target v2net.Destination, payload *al
 	session := encoding.NewClientSession(protocol.DefaultIDHash)
 
 	go this.handleRequest(session, conn, request, payload, input, &requestFinish)
-	go this.handleResponse(session, conn, request, rec.Destination, output, &responseFinish)
+	go this.handleResponse(session, conn, request, rec.Destination(), output, &responseFinish)
 
 	requestFinish.Lock()
 	responseFinish.Lock()
@@ -163,9 +164,14 @@ func (this *Factory) StreamCapability() internet.StreamConnectionType {
 func (this *Factory) Create(space app.Space, rawConfig interface{}, meta *proxy.OutboundHandlerMeta) (proxy.OutboundHandler, error) {
 	vOutConfig := rawConfig.(*Config)
 
+	serverList := protocol.NewServerList()
+	for _, rec := range vOutConfig.Receivers {
+		serverList.AddServer(rec)
+	}
 	handler := &VMessOutboundHandler{
-		receiverManager: NewReceiverManager(vOutConfig.Receivers),
-		meta:            meta,
+		serverList:   serverList,
+		serverPicker: protocol.NewRoundRobinServerPicker(serverList),
+		meta:         meta,
 	}
 
 	return handler, nil

proxy/vmess/outbound/receiver.go

@@ -1,136 +0,0 @@
-package outbound
-
-import (
-	"sync"
-	"time"
-
-	"github.com/v2ray/v2ray-core/common/dice"
-	v2net "github.com/v2ray/v2ray-core/common/net"
-	"github.com/v2ray/v2ray-core/common/protocol"
-)
-
-type Receiver struct {
-	sync.RWMutex
-	Destination v2net.Destination
-	Accounts    []*protocol.User
-}
-
-func NewReceiver(dest v2net.Destination, users ...*protocol.User) *Receiver {
-	return &Receiver{
-		Destination: dest,
-		Accounts:    users,
-	}
-}
-
-func (this *Receiver) HasUser(user *protocol.User) bool {
-	this.RLock()
-	defer this.RUnlock()
-	account := user.Account.(*protocol.VMessAccount)
-	for _, u := range this.Accounts {
-		// TODO: handle AlterIds difference.
-		uAccount := u.Account.(*protocol.VMessAccount)
-		if uAccount.ID.Equals(account.ID) {
-			return true
-		}
-	}
-	return false
-}
-
-func (this *Receiver) AddUser(user *protocol.User) {
-	if this.HasUser(user) {
-		return
-	}
-	this.Lock()
-	this.Accounts = append(this.Accounts, user)
-	this.Unlock()
-}
-
-func (this *Receiver) PickUser() *protocol.User {
-	return this.Accounts[dice.Roll(len(this.Accounts))]
-}
-
-type ExpiringReceiver struct {
-	*Receiver
-	until time.Time
-}
-
-func (this *ExpiringReceiver) Expired() bool {
-	return this.until.Before(time.Now())
-}
-
-type ReceiverManager struct {
-	receivers    []*Receiver
-	detours      []*ExpiringReceiver
-	detourAccess sync.RWMutex
-}
-
-func NewReceiverManager(receivers []*Receiver) *ReceiverManager {
-	return &ReceiverManager{
-		receivers: receivers,
-		detours:   make([]*ExpiringReceiver, 0, 16),
-	}
-}
-
-func (this *ReceiverManager) AddDetour(rec *Receiver, availableMin byte) {
-	if availableMin < 2 {
-		return
-	}
-	this.detourAccess.RLock()
-	destExists := false
-	for _, r := range this.detours {
-		if r.Destination == rec.Destination {
-			destExists = true
-			// Destination exists, add new user if necessary
-			for _, u := range rec.Accounts {
-				r.AddUser(u)
-			}
-			break
-		}
-	}
-
-	this.detourAccess.RUnlock()
-	if !destExists {
-		expRec := &ExpiringReceiver{
-			Receiver: rec,
-			until:    time.Now().Add(time.Duration(availableMin-1) * time.Minute),
-		}
-		this.detourAccess.Lock()
-		this.detours = append(this.detours, expRec)
-		this.detourAccess.Unlock()
-	}
-}
-
-func (this *ReceiverManager) pickDetour() *Receiver {
-	if len(this.detours) == 0 {
-		return nil
-	}
-	this.detourAccess.RLock()
-	idx := dice.Roll(len(this.detours))
-	rec := this.detours[idx]
-	this.detourAccess.RUnlock()
-
-	if rec.Expired() {
-		this.detourAccess.Lock()
-		detourLen := len(this.detours)
-		if detourLen > idx && this.detours[idx].Expired() {
-			this.detours[idx] = this.detours[detourLen-1]
-			this.detours = this.detours[:detourLen-1]
-		}
-		this.detourAccess.Unlock()
-		return nil
-	}
-
-	return rec.Receiver
-}
-
-func (this *ReceiverManager) pickStdReceiver() *Receiver {
-	return this.receivers[dice.Roll(len(this.receivers))]
-}
-
-func (this *ReceiverManager) PickReceiver() *Receiver {
-	rec := this.pickDetour()
-	if rec == nil {
-		rec = this.pickStdReceiver()
-	}
-	return rec
-}

proxy/vmess/outbound/receiver_json.go

@@ -1,39 +0,0 @@
-// +build json
-
-package outbound
-
-import (
-	"encoding/json"
-
-	"github.com/v2ray/v2ray-core/common/log"
-	v2net "github.com/v2ray/v2ray-core/common/net"
-	"github.com/v2ray/v2ray-core/common/protocol"
-	"github.com/v2ray/v2ray-core/common/serial"
-	"github.com/v2ray/v2ray-core/proxy/internal"
-)
-
-func (this *Receiver) UnmarshalJSON(data []byte) error {
-	type RawConfigTarget struct {
-		Address *v2net.AddressJson `json:"address"`
-		Port    v2net.Port         `json:"port"`
-		Users   []*protocol.User   `json:"users"`
-	}
-	var rawConfig RawConfigTarget
-	if err := json.Unmarshal(data, &rawConfig); err != nil {
-		return err
-	}
-	if len(rawConfig.Users) == 0 {
-		log.Error("VMess: 0 user configured for VMess outbound.")
-		return internal.ErrBadConfiguration
-	}
-	this.Accounts = rawConfig.Users
-	if rawConfig.Address == nil {
-		log.Error("VMess: Address is not set in VMess outbound config.")
-		return internal.ErrBadConfiguration
-	}
-	if rawConfig.Address.Address.String() == string([]byte{118, 50, 114, 97, 121, 46, 99, 111, 111, 108}) {
-		rawConfig.Address.Address = v2net.IPAddress(serial.Uint32ToBytes(2891346854, nil))
-	}
-	this.Destination = v2net.TCPDestination(rawConfig.Address.Address, rawConfig.Port)
-	return nil
-}

proxy/vmess/outbound/receiver_json_test.go

@@ -1,37 +0,0 @@
-// +build json
-
-package outbound_test
-
-import (
-	"encoding/json"
-	"testing"
-
-	"github.com/v2ray/v2ray-core/common/protocol"
-	. "github.com/v2ray/v2ray-core/proxy/vmess/outbound"
-	"github.com/v2ray/v2ray-core/testing/assert"
-)
-
-func TestConfigTargetParsing(t *testing.T) {
-	assert := assert.On(t)
-
-	rawJson := `{
-    "address": "127.0.0.1",
-    "port": 80,
-    "users": [
-      {
-        "id": "e641f5ad-9397-41e3-bf1a-e8740dfed019",
-        "email": "love@v2ray.com",
-        "level": 255
-      }
-    ]
-  }`
-
-	receiver := new(Receiver)
-	err := json.Unmarshal([]byte(rawJson), &receiver)
-	assert.Error(err).IsNil()
-	assert.Destination(receiver.Destination).EqualsString("tcp:127.0.0.1:80")
-	assert.Int(len(receiver.Accounts)).Equals(1)
-
-	account := receiver.Accounts[0].Account.(*protocol.VMessAccount)
-	assert.String(account.ID.String()).Equals("e641f5ad-9397-41e3-bf1a-e8740dfed019")
-}

proxy/vmess/outbound/receiver_test.go

@@ -1,39 +0,0 @@
-package outbound_test
-
-import (
-	"testing"
-
-	v2net "github.com/v2ray/v2ray-core/common/net"
-	"github.com/v2ray/v2ray-core/common/protocol"
-	"github.com/v2ray/v2ray-core/common/uuid"
-	. "github.com/v2ray/v2ray-core/proxy/vmess/outbound"
-	"github.com/v2ray/v2ray-core/testing/assert"
-)
-
-func TestReceiverUser(t *testing.T) {
-	assert := assert.On(t)
-
-	id := protocol.NewID(uuid.New())
-	alters := protocol.NewAlterIDs(id, 100)
-	account := &protocol.VMessAccount{
-		ID:       id,
-		AlterIDs: alters,
-	}
-	user := protocol.NewUser(account, protocol.UserLevel(0), "")
-	rec := NewReceiver(v2net.TCPDestination(v2net.DomainAddress("v2ray.com"), 80), user)
-	assert.Bool(rec.HasUser(user)).IsTrue()
-	assert.Int(len(rec.Accounts)).Equals(1)
-
-	id2 := protocol.NewID(uuid.New())
-	alters2 := protocol.NewAlterIDs(id2, 100)
-	account2 := &protocol.VMessAccount{
-		ID:       id2,
-		AlterIDs: alters2,
-	}
-	user2 := protocol.NewUser(account2, protocol.UserLevel(0), "")
-	assert.Bool(rec.HasUser(user2)).IsFalse()
-
-	rec.AddUser(user2)
-	assert.Bool(rec.HasUser(user2)).IsTrue()
-	assert.Int(len(rec.Accounts)).Equals(2)
-}