Remove PickServer()

pull/5144/head
𐲓𐳛𐳪𐳂𐳐 𐲀𐳢𐳦𐳫𐳢 𐲥𐳔𐳛𐳪𐳌𐳑𐳖𐳇 2025-09-14 13:06:42 +08:00 committed by GitHub
parent 746afef8fa
commit 9f5f6b4ee2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 76 additions and 305 deletions

View File

@ -1,89 +0,0 @@
package protocol
import (
"sync"
)
type ServerList struct {
sync.RWMutex
servers []*ServerSpec
}
func NewServerList() *ServerList {
return &ServerList{}
}
func (sl *ServerList) AddServer(server *ServerSpec) {
sl.Lock()
defer sl.Unlock()
sl.servers = append(sl.servers, server)
}
func (sl *ServerList) Size() uint32 {
sl.RLock()
defer sl.RUnlock()
return uint32(len(sl.servers))
}
func (sl *ServerList) GetServer(idx uint32) *ServerSpec {
sl.Lock()
defer sl.Unlock()
for {
if idx >= uint32(len(sl.servers)) {
return nil
}
server := sl.servers[idx]
if !server.IsValid() {
sl.removeServer(idx)
continue
}
return server
}
}
func (sl *ServerList) removeServer(idx uint32) {
n := len(sl.servers)
sl.servers[idx] = sl.servers[n-1]
sl.servers = sl.servers[:n-1]
}
type ServerPicker interface {
PickServer() *ServerSpec
}
type RoundRobinServerPicker struct {
sync.Mutex
serverlist *ServerList
nextIndex uint32
}
func NewRoundRobinServerPicker(serverlist *ServerList) *RoundRobinServerPicker {
return &RoundRobinServerPicker{
serverlist: serverlist,
nextIndex: 0,
}
}
func (p *RoundRobinServerPicker) PickServer() *ServerSpec {
p.Lock()
defer p.Unlock()
next := p.nextIndex
server := p.serverlist.GetServer(next)
if server == nil {
next = 0
server = p.serverlist.GetServer(0)
}
next++
if next >= p.serverlist.Size() {
next = 0
}
p.nextIndex = next
return server
}

View File

@ -1,71 +0,0 @@
package protocol_test
import (
"testing"
"time"
"github.com/xtls/xray-core/common/net"
. "github.com/xtls/xray-core/common/protocol"
)
func TestServerList(t *testing.T) {
list := NewServerList()
list.AddServer(NewServerSpec(net.TCPDestination(net.LocalHostIP, net.Port(1)), AlwaysValid()))
if list.Size() != 1 {
t.Error("list size: ", list.Size())
}
list.AddServer(NewServerSpec(net.TCPDestination(net.LocalHostIP, net.Port(2)), BeforeTime(time.Now().Add(time.Second))))
if list.Size() != 2 {
t.Error("list.size: ", list.Size())
}
server := list.GetServer(1)
if server.Destination().Port != 2 {
t.Error("server: ", server.Destination())
}
time.Sleep(2 * time.Second)
server = list.GetServer(1)
if server != nil {
t.Error("server: ", server)
}
server = list.GetServer(0)
if server.Destination().Port != 1 {
t.Error("server: ", server.Destination())
}
}
func TestServerPicker(t *testing.T) {
list := NewServerList()
list.AddServer(NewServerSpec(net.TCPDestination(net.LocalHostIP, net.Port(1)), AlwaysValid()))
list.AddServer(NewServerSpec(net.TCPDestination(net.LocalHostIP, net.Port(2)), BeforeTime(time.Now().Add(time.Second))))
list.AddServer(NewServerSpec(net.TCPDestination(net.LocalHostIP, net.Port(3)), BeforeTime(time.Now().Add(time.Second))))
picker := NewRoundRobinServerPicker(list)
server := picker.PickServer()
if server.Destination().Port != 1 {
t.Error("server: ", server.Destination())
}
server = picker.PickServer()
if server.Destination().Port != 2 {
t.Error("server: ", server.Destination())
}
server = picker.PickServer()
if server.Destination().Port != 3 {
t.Error("server: ", server.Destination())
}
server = picker.PickServer()
if server.Destination().Port != 1 {
t.Error("server: ", server.Destination())
}
time.Sleep(2 * time.Second)
server = picker.PickServer()
if server.Destination().Port != 1 {
t.Error("server: ", server.Destination())
}
server = picker.PickServer()
if server.Destination().Port != 1 {
t.Error("server: ", server.Destination())
}
}

View File

@ -4,7 +4,6 @@ import (
"sync" "sync"
"time" "time"
"github.com/xtls/xray-core/common/dice"
"github.com/xtls/xray-core/common/net" "github.com/xtls/xray-core/common/net"
) )
@ -98,6 +97,8 @@ func (s *ServerSpec) AddUser(user *MemoryUser) {
s.users = append(s.users, user) s.users = append(s.users, user)
} }
// Locking it only using the first user when user(s) exists.
// Should change after refactor
func (s *ServerSpec) PickUser() *MemoryUser { func (s *ServerSpec) PickUser() *MemoryUser {
s.RLock() s.RLock()
defer s.RUnlock() defer s.RUnlock()
@ -106,10 +107,8 @@ func (s *ServerSpec) PickUser() *MemoryUser {
switch userCount { switch userCount {
case 0: case 0:
return nil return nil
case 1:
return s.users[0]
default: default:
return s.users[dice.Roll(userCount)] return s.users[0]
} }
} }

View File

@ -31,7 +31,7 @@ import (
) )
type Client struct { type Client struct {
serverPicker protocol.ServerPicker server *protocol.ServerSpec
policyManager policy.Manager policyManager policy.Manager
header []*Header header []*Header
} }
@ -48,21 +48,19 @@ var (
// NewClient create a new http client based on the given config. // NewClient create a new http client based on the given config.
func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) { func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) {
serverList := protocol.NewServerList() if len(config.Server) != 1 {
for _, rec := range config.Server { return nil, errors.New(`only one target server allowed`)
s, err := protocol.NewServerSpecFromPB(rec) }
// Harcoded [0] for processing compatibility.
// Should change after refactor.
server, err := protocol.NewServerSpecFromPB(config.Server[0])
if err != nil { if err != nil {
return nil, errors.New("failed to get server spec").Base(err) return nil, errors.New("failed to get server spec").Base(err)
} }
serverList.AddServer(s)
}
if serverList.Size() == 0 {
return nil, errors.New("0 target server")
}
v := core.MustFromContext(ctx) v := core.MustFromContext(ctx)
return &Client{ return &Client{
serverPicker: protocol.NewRoundRobinServerPicker(serverList), server: server,
policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager), policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
header: config.Header, header: config.Header,
}, nil }, nil
@ -102,9 +100,8 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
} }
if err := retry.ExponentialBackoff(5, 100).On(func() error { if err := retry.ExponentialBackoff(5, 100).On(func() error {
server := c.serverPicker.PickServer() dest := c.server.Destination()
dest := server.Destination() user = c.server.PickUser()
user = server.PickUser()
netConn, err := setUpHTTPTunnel(ctx, dest, targetAddr, user, dialer, header, firstPayload) netConn, err := setUpHTTPTunnel(ctx, dest, targetAddr, user, dialer, header, firstPayload)
if netConn != nil { if netConn != nil {

View File

@ -22,27 +22,25 @@ import (
// Client is a inbound handler for Shadowsocks protocol // Client is a inbound handler for Shadowsocks protocol
type Client struct { type Client struct {
serverPicker protocol.ServerPicker server *protocol.ServerSpec
policyManager policy.Manager policyManager policy.Manager
} }
// NewClient create a new Shadowsocks client. // NewClient create a new Shadowsocks client.
func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) { func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) {
serverList := protocol.NewServerList() if len(config.Server) != 1 {
for _, rec := range config.Server { return nil, errors.New(`only one target server allowed`)
s, err := protocol.NewServerSpecFromPB(rec) }
// Harcoded [0] for processing compatibility.
// Should change after refactor.
server, err := protocol.NewServerSpecFromPB(config.Server[0])
if err != nil { if err != nil {
return nil, errors.New("failed to parse server spec").Base(err) return nil, errors.New("failed to get server spec").Base(err)
}
serverList.AddServer(s)
}
if serverList.Size() == 0 {
return nil, errors.New("0 server")
} }
v := core.MustFromContext(ctx) v := core.MustFromContext(ctx)
client := &Client{ client := &Client{
serverPicker: protocol.NewRoundRobinServerPicker(serverList), server: server,
policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager), policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
} }
return client, nil return client, nil
@ -60,12 +58,10 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
destination := ob.Target destination := ob.Target
network := destination.Network network := destination.Network
var server *protocol.ServerSpec
var conn stat.Connection var conn stat.Connection
err := retry.ExponentialBackoff(5, 100).On(func() error { err := retry.ExponentialBackoff(5, 100).On(func() error {
server = c.serverPicker.PickServer() dest := c.server.Destination()
dest := server.Destination()
dest.Network = network dest.Network = network
rawConn, err := dialer.Dial(ctx, dest) rawConn, err := dialer.Dial(ctx, dest)
if err != nil { if err != nil {
@ -78,7 +74,7 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
if err != nil { if err != nil {
return errors.New("failed to find an available destination").AtWarning().Base(err) return errors.New("failed to find an available destination").AtWarning().Base(err)
} }
errors.LogInfo(ctx, "tunneling request to ", destination, " via ", network, ":", server.Destination().NetAddr()) errors.LogInfo(ctx, "tunneling request to ", destination, " via ", network, ":", c.server.Destination().NetAddr())
defer conn.Close() defer conn.Close()
@ -93,7 +89,7 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
request.Command = protocol.RequestCommandUDP request.Command = protocol.RequestCommandUDP
} }
user := server.PickUser() user := c.server.PickUser()
_, ok := user.Account.(*MemoryAccount) _, ok := user.Account.(*MemoryAccount)
if !ok { if !ok {
return errors.New("user account is not valid") return errors.New("user account is not valid")

View File

@ -22,27 +22,25 @@ import (
// Client is a Socks5 client. // Client is a Socks5 client.
type Client struct { type Client struct {
serverPicker protocol.ServerPicker server *protocol.ServerSpec
policyManager policy.Manager policyManager policy.Manager
} }
// NewClient create a new Socks5 client based on the given config. // NewClient create a new Socks5 client based on the given config.
func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) { func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) {
serverList := protocol.NewServerList() if len(config.Server) != 1 {
for _, rec := range config.Server { return nil, errors.New(`only one target server allowed`)
s, err := protocol.NewServerSpecFromPB(rec) }
// Harcoded [0] for processing compatibility.
// Should change after refactor.
server, err := protocol.NewServerSpecFromPB(config.Server[0])
if err != nil { if err != nil {
return nil, errors.New("failed to get server spec").Base(err) return nil, errors.New("failed to get server spec").Base(err)
} }
serverList.AddServer(s)
}
if serverList.Size() == 0 {
return nil, errors.New("0 target server")
}
v := core.MustFromContext(ctx) v := core.MustFromContext(ctx)
c := &Client{ c := &Client{
serverPicker: protocol.NewRoundRobinServerPicker(serverList), server: server,
policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager), policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
} }
@ -61,16 +59,13 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
// Destination of the inner request. // Destination of the inner request.
destination := ob.Target destination := ob.Target
// Outbound server.
var server *protocol.ServerSpec
// Outbound server's destination. // Outbound server's destination.
var dest net.Destination var dest net.Destination
// Connection to the outbound server. // Connection to the outbound server.
var conn stat.Connection var conn stat.Connection
if err := retry.ExponentialBackoff(5, 100).On(func() error { if err := retry.ExponentialBackoff(5, 100).On(func() error {
server = c.serverPicker.PickServer() dest = c.server.Destination()
dest = server.Destination()
rawConn, err := dialer.Dial(ctx, dest) rawConn, err := dialer.Dial(ctx, dest)
if err != nil { if err != nil {
return err return err
@ -101,7 +96,7 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
request.Command = protocol.RequestCommandUDP request.Command = protocol.RequestCommandUDP
} }
user := server.PickUser() user := c.server.PickUser()
if user != nil { if user != nil {
request.User = user request.User = user
p = c.policyManager.ForLevel(user.Level) p = c.policyManager.ForLevel(user.Level)

View File

@ -22,27 +22,25 @@ import (
// Client is a inbound handler for trojan protocol // Client is a inbound handler for trojan protocol
type Client struct { type Client struct {
serverPicker protocol.ServerPicker server *protocol.ServerSpec
policyManager policy.Manager policyManager policy.Manager
} }
// NewClient create a new trojan client. // NewClient create a new trojan client.
func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) { func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) {
serverList := protocol.NewServerList() if len(config.Server) != 1 {
for _, rec := range config.Server { return nil, errors.New(`only one target server allowed`)
s, err := protocol.NewServerSpecFromPB(rec) }
// Harcoded [0] for processing compatibility.
// Should change after refactor.
server, err := protocol.NewServerSpecFromPB(config.Server[0])
if err != nil { if err != nil {
return nil, errors.New("failed to parse server spec").Base(err) return nil, errors.New("failed to get server spec").Base(err)
}
serverList.AddServer(s)
}
if serverList.Size() == 0 {
return nil, errors.New("0 server")
} }
v := core.MustFromContext(ctx) v := core.MustFromContext(ctx)
client := &Client{ client := &Client{
serverPicker: protocol.NewRoundRobinServerPicker(serverList), server: server,
policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager), policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
} }
return client, nil return client, nil
@ -60,12 +58,10 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
destination := ob.Target destination := ob.Target
network := destination.Network network := destination.Network
var server *protocol.ServerSpec
var conn stat.Connection var conn stat.Connection
err := retry.ExponentialBackoff(5, 100).On(func() error { err := retry.ExponentialBackoff(5, 100).On(func() error {
server = c.serverPicker.PickServer() rawConn, err := dialer.Dial(ctx, c.server.Destination())
rawConn, err := dialer.Dial(ctx, server.Destination())
if err != nil { if err != nil {
return err return err
} }
@ -76,11 +72,11 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
if err != nil { if err != nil {
return errors.New("failed to find an available destination").AtWarning().Base(err) return errors.New("failed to find an available destination").AtWarning().Base(err)
} }
errors.LogInfo(ctx, "tunneling request to ", destination, " via ", server.Destination().NetAddr()) errors.LogInfo(ctx, "tunneling request to ", destination, " via ", c.server.Destination().NetAddr())
defer conn.Close() defer conn.Close()
user := server.PickUser() user := c.server.PickUser()
account, ok := user.Account.(*MemoryAccount) account, ok := user.Account.(*MemoryAccount)
if !ok { if !ok {
return errors.New("user account is not valid") return errors.New("user account is not valid")

View File

@ -47,8 +47,7 @@ func init() {
// Handler is an outbound connection handler for VLess protocol. // Handler is an outbound connection handler for VLess protocol.
type Handler struct { type Handler struct {
serverList *protocol.ServerList server *protocol.ServerSpec
serverPicker protocol.ServerPicker
policyManager policy.Manager policyManager policy.Manager
cone bool cone bool
encryption *encryption.ClientInstance encryption *encryption.ClientInstance
@ -57,24 +56,24 @@ type Handler struct {
// New creates a new VLess outbound handler. // New creates a new VLess outbound handler.
func New(ctx context.Context, config *Config) (*Handler, error) { func New(ctx context.Context, config *Config) (*Handler, error) {
serverList := protocol.NewServerList() if len(config.Vnext) != 1 {
for _, rec := range config.Vnext { return nil, errors.New(`only one vnext allowed`)
s, err := protocol.NewServerSpecFromPB(rec)
if err != nil {
return nil, errors.New("failed to parse server spec").Base(err).AtError()
} }
serverList.AddServer(s) // Harcoded [0] for processing compatibility.
// Should change after refactor.
server, err := protocol.NewServerSpecFromPB(config.Vnext[0])
if err != nil {
return nil, errors.New("failed to get server spec").Base(err).AtError()
} }
v := core.MustFromContext(ctx) v := core.MustFromContext(ctx)
handler := &Handler{ handler := &Handler{
serverList: serverList, server: server,
serverPicker: protocol.NewRoundRobinServerPicker(serverList),
policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager), policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
cone: ctx.Value("cone").(bool), cone: ctx.Value("cone").(bool),
} }
a := handler.serverPicker.PickServer().PickUser().Account.(*vless.MemoryAccount) a := handler.server.PickUser().Account.(*vless.MemoryAccount)
if a.Encryption != "" && a.Encryption != "none" { if a.Encryption != "" && a.Encryption != "none" {
s := strings.Split(a.Encryption, ".") s := strings.Split(a.Encryption, ".")
var nfsPKeysBytes [][]byte var nfsPKeysBytes [][]byte
@ -125,12 +124,10 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
} }
ob.Name = "vless" ob.Name = "vless"
var rec *protocol.ServerSpec
var conn stat.Connection var conn stat.Connection
if err := retry.ExponentialBackoff(5, 200).On(func() error { if err := retry.ExponentialBackoff(5, 200).On(func() error {
rec = h.serverPicker.PickServer()
var err error var err error
conn, err = dialer.Dial(ctx, rec.Destination()) conn, err = dialer.Dial(ctx, h.server.Destination())
if err != nil { if err != nil {
return err return err
} }
@ -145,7 +142,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
iConn = statConn.Connection iConn = statConn.Connection
} }
target := ob.Target target := ob.Target
errors.LogInfo(ctx, "tunneling request to ", target, " via ", rec.Destination().NetAddr()) errors.LogInfo(ctx, "tunneling request to ", target, " via ", h.server.Destination().NetAddr())
if h.encryption != nil { if h.encryption != nil {
var err error var err error
@ -172,7 +169,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
request := &protocol.RequestHeader{ request := &protocol.RequestHeader{
Version: encoding.Version, Version: encoding.Version,
User: rec.PickUser(), User: h.server.PickUser(),
Command: command, Command: command,
Address: target.Address, Address: target.Address,
Port: target.Port, Port: target.Port,

View File

@ -1,41 +0,0 @@
package outbound
import (
"time"
"github.com/xtls/xray-core/common"
"github.com/xtls/xray-core/common/net"
"github.com/xtls/xray-core/common/protocol"
"github.com/xtls/xray-core/proxy/vmess"
)
func (h *Handler) handleSwitchAccount(cmd *protocol.CommandSwitchAccount) {
rawAccount := &vmess.Account{
Id: cmd.ID.String(),
SecuritySettings: &protocol.SecurityConfig{
Type: protocol.SecurityType_AUTO,
},
}
account, err := rawAccount.AsAccount()
common.Must(err)
user := &protocol.MemoryUser{
Email: "",
Level: cmd.Level,
Account: account,
}
dest := net.TCPDestination(cmd.Host, cmd.Port)
until := time.Now().Add(time.Duration(cmd.ValidMin) * time.Minute)
h.serverList.AddServer(protocol.NewServerSpec(dest, protocol.BeforeTime(until), user))
}
func (h *Handler) handleCommand(dest net.Destination, cmd protocol.ResponseCommand) {
switch typedCommand := cmd.(type) {
case *protocol.CommandSwitchAccount:
if typedCommand.Host == nil {
typedCommand.Host = dest.Address
}
h.handleSwitchAccount(typedCommand)
default:
}
}

View File

@ -29,27 +29,26 @@ import (
// Handler is an outbound connection handler for VMess protocol. // Handler is an outbound connection handler for VMess protocol.
type Handler struct { type Handler struct {
serverList *protocol.ServerList server *protocol.ServerSpec
serverPicker protocol.ServerPicker
policyManager policy.Manager policyManager policy.Manager
cone bool cone bool
} }
// New creates a new VMess outbound handler. // New creates a new VMess outbound handler.
func New(ctx context.Context, config *Config) (*Handler, error) { func New(ctx context.Context, config *Config) (*Handler, error) {
serverList := protocol.NewServerList() if len(config.Receiver) != 1 {
for _, rec := range config.Receiver { return nil, errors.New(`only one vnext allowed`)
s, err := protocol.NewServerSpecFromPB(rec)
if err != nil {
return nil, errors.New("failed to parse server spec").Base(err)
} }
serverList.AddServer(s) // Harcoded [0] for processing compatibility.
// Should change after refactor.
server, err := protocol.NewServerSpecFromPB(config.Receiver[0])
if err != nil {
return nil, errors.New("failed to get server spec").Base(err)
} }
v := core.MustFromContext(ctx) v := core.MustFromContext(ctx)
handler := &Handler{ handler := &Handler{
serverList: serverList, server: server,
serverPicker: protocol.NewRoundRobinServerPicker(serverList),
policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager), policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
cone: ctx.Value("cone").(bool), cone: ctx.Value("cone").(bool),
} }
@ -67,11 +66,9 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
ob.Name = "vmess" ob.Name = "vmess"
ob.CanSpliceCopy = 3 ob.CanSpliceCopy = 3
var rec *protocol.ServerSpec
var conn stat.Connection var conn stat.Connection
err := retry.ExponentialBackoff(5, 200).On(func() error { err := retry.ExponentialBackoff(5, 200).On(func() error {
rec = h.serverPicker.PickServer() rawConn, err := dialer.Dial(ctx, h.server.Destination())
rawConn, err := dialer.Dial(ctx, rec.Destination())
if err != nil { if err != nil {
return err return err
} }
@ -85,7 +82,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
defer conn.Close() defer conn.Close()
target := ob.Target target := ob.Target
errors.LogInfo(ctx, "tunneling request to ", target, " via ", rec.Destination().NetAddr()) errors.LogInfo(ctx, "tunneling request to ", target, " via ", h.server.Destination().NetAddr())
command := protocol.RequestCommandTCP command := protocol.RequestCommandTCP
if target.Network == net.Network_UDP { if target.Network == net.Network_UDP {
@ -95,7 +92,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
command = protocol.RequestCommandMux command = protocol.RequestCommandMux
} }
user := rec.PickUser() user := h.server.PickUser()
request := &protocol.RequestHeader{ request := &protocol.RequestHeader{
Version: encoding.Version, Version: encoding.Version,
User: user, User: user,
@ -198,11 +195,6 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
defer timer.SetTimeout(sessionPolicy.Timeouts.UplinkOnly) defer timer.SetTimeout(sessionPolicy.Timeouts.UplinkOnly)
reader := &buf.BufferedReader{Reader: buf.NewReader(conn)} reader := &buf.BufferedReader{Reader: buf.NewReader(conn)}
header, err := session.DecodeResponseHeader(reader)
if err != nil {
return errors.New("failed to read header").Base(err)
}
h.handleCommand(rec.Destination(), header.Command)
bodyReader, err := session.DecodeResponseBody(request, reader) bodyReader, err := session.DecodeResponseBody(request, reader)
if err != nil { if err != nil {