lzhr/v2fly
1
0
Fork 0
mirror of https://github.com/v2fly/v2ray-core.git synced 2024-12-30 05:56:54 -05:00
Code Issues Releases Wiki Activity

close method for point server

Browse Source
This commit is contained in:
v2ray 2016-01-04 00:33:25 +01:00
parent 201481a82c
commit 56c5302367
11 changed files with 80 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
proxy/dokodemo/dokodemo.go
View File

@ -13,7 +13,8 @@ import (
)
type DokodemoDoor struct {
sync.Mutex
tcpMutex sync.RWMutex
udpMutex sync.RWMutex
config Config
accepting bool
address v2net.Address
@ -35,16 +36,16 @@ func NewDokodemoDoor(space app.Space, config Config) *DokodemoDoor {
func (this *DokodemoDoor) Close() {
this.accepting = false
if this.tcpListener != nil {
this.Lock()
this.tcpListener.Close()
this.tcpMutex.Lock()
this.tcpListener = nil
this.Unlock()
this.tcpMutex.Unlock()
}
if this.udpConn != nil {
this.Lock()
this.udpConn.Close()
this.udpMutex.Lock()
this.udpConn = nil
this.Unlock()
this.udpMutex.Unlock()
}
}
@ -84,13 +85,12 @@ func (this *DokodemoDoor) ListenUDP(port v2net.Port) error {
func (this *DokodemoDoor) handleUDPPackets() {
for this.accepting {
buffer := alloc.NewBuffer()
var udpConn *net.UDPConn
this.Lock()
if this.udpConn != nil {
udpConn = this.udpConn
this.udpMutex.RLock()
if !this.accepting {
return
}
this.Unlock()
nBytes, addr, err := udpConn.ReadFromUDP(buffer.Value)
nBytes, addr, err := this.udpConn.ReadFromUDP(buffer.Value)
this.udpMutex.RUnlock()
buffer.Slice(0, nBytes)
if err != nil {
buffer.Release()
@ -103,7 +103,13 @@ func (this *DokodemoDoor) handleUDPPackets() {
close(ray.InboundInput())
for payload := range ray.InboundOutput() {
udpConn.WriteToUDP(payload.Value, addr)
this.udpMutex.RLock()
if !this.accepting {
this.udpMutex.RUnlock()
return
}
this.udpConn.WriteToUDP(payload.Value, addr)
this.udpMutex.RUnlock()
}
}
}
@ -126,8 +132,11 @@ func (this *DokodemoDoor) ListenTCP(port v2net.Port) error {
func (this *DokodemoDoor) AcceptTCPConnections() {
for this.accepting {
retry.Timed(100, 100).On(func() error {
this.Lock()
defer this.Unlock()
if !this.accepting {
return nil
}
this.tcpMutex.RLock()
defer this.tcpMutex.RUnlock()
if this.tcpListener != nil {
connection, err := this.tcpListener.AcceptTCP()
if err != nil {

2
proxy/http/http.go
View File

@ -35,8 +35,8 @@ func NewHttpProxyServer(space app.Space, config Config) *HttpProxyServer {
func (this *HttpProxyServer) Close() {
this.accepting = false
if this.tcpListener != nil {
this.Lock()
this.tcpListener.Close()
this.Lock()
this.tcpListener = nil
this.Unlock()
}

31
proxy/socks/socks.go
View File

@ -23,7 +23,8 @@ var (
// SocksServer is a SOCKS 5 proxy server
type SocksServer struct {
sync.RWMutex
tcpMutex sync.RWMutex
udpMutex sync.RWMutex
accepting bool
space app.Space
config Config
@ -42,20 +43,16 @@ func NewSocksServer(space app.Space, config Config) *SocksServer {
func (this *SocksServer) Close() {
this.accepting = false
if this.tcpListener != nil {
this.Lock()
if this.tcpListener != nil {
this.tcpListener.Close()
this.tcpListener = nil
}
this.Unlock()
this.tcpListener.Close()
this.tcpMutex.Lock()
this.tcpListener = nil
this.tcpMutex.Unlock()
}
if this.udpConn != nil {
this.Lock()
if this.udpConn != nil {
this.udpConn.Close()
this.udpConn = nil
}
this.Unlock()
this.udpConn.Close()
this.udpMutex.Lock()
this.udpConn = nil
this.udpMutex.Unlock()
}
}
@ -81,12 +78,16 @@ func (this *SocksServer) Listen(port v2net.Port) error {
func (this *SocksServer) AcceptConnections() {
for this.accepting {
retry.Timed(100 /* times */, 100 /* ms */).On(func() error {
this.RLock()
defer this.RUnlock()
if !this.accepting {
return nil
}
this.tcpMutex.RLock()
if this.tcpListener == nil {
this.tcpMutex.RUnlock()
return nil
}
connection, err := this.tcpListener.AcceptTCP()
this.tcpMutex.RUnlock()
if err != nil {
log.Error("Socks failed to accept new connection %v", err)
return err

12
proxy/socks/udp.go
View File

@ -30,13 +30,13 @@ func (this *SocksServer) ListenUDP(port v2net.Port) error {
func (this *SocksServer) AcceptPackets() error {
for this.accepting {
buffer := alloc.NewBuffer()
this.RLock()
this.udpMutex.RLock()
if !this.accepting {
this.RUnlock()
this.udpMutex.RUnlock()
return nil
}
nBytes, addr, err := this.udpConn.ReadFromUDP(buffer.Value)
this.RUnlock()
this.udpMutex.RUnlock()
if err != nil {
log.Error("Socks failed to read UDP packets: %v", err)
buffer.Release()
@ -82,13 +82,13 @@ func (this *SocksServer) handlePacket(packet v2net.Packet, clientAddr *net.UDPAd
udpMessage := alloc.NewSmallBuffer().Clear()
response.Write(udpMessage)
this.RLock()
this.udpMutex.RLock()
if !this.accepting {
this.RUnlock()
this.udpMutex.RUnlock()
return
}
nBytes, err := this.udpConn.WriteToUDP(udpMessage.Value, clientAddr)
this.RUnlock()
this.udpMutex.RUnlock()
udpMessage.Release()
response.Data.Release()
if err != nil {

6
proxy/vmess/inbound/inbound.go
View File

@ -38,11 +38,9 @@ func NewVMessInboundHandler(space app.Space, clients user.UserSet) *VMessInbound
func (this *VMessInboundHandler) Close() {
this.accepting = false
if this.listener != nil {
this.listener.Close()
this.Lock()
if this.listener != nil {
this.listener.Close()
this.listener = nil
}
this.listener = nil
this.Unlock()
}
}

6
shell/point/inbound_detour.go
View File

@ -39,6 +39,12 @@ func (this *InboundDetourHandler) Initialize() error {
return nil
}
func (this *InboundDetourHandler) Close() {
for _, ich := range this.ich {
ich.handler.Close()
}
}
// Starts the inbound connection handler.
func (this *InboundDetourHandler) Start() error {
for _, ich := range this.ich {

7
shell/point/point.go
View File

@ -113,6 +113,13 @@ func NewPoint(pConfig PointConfig) (*Point, error) {
return vpoint, nil
}
func (this *Point) Close() {
this.ich.Close()
for _, idh := range this.idh {
idh.Close()
}
}
// Start starts the Point server, and return any error during the process.
// In the case of any errors, the state of the server is unpredicatable.
func (this *Point) Start() error {

2
testing/scenarios/dokodemo_test.go
View File

@ -50,4 +50,6 @@ func TestDokodemoTCP(t *testing.T) {
assert.StringLiteral("Processed: " + payload).Equals(string(response[:nBytes]))
conn.Close()
}
CloseAllServers()
}

2
testing/scenarios/router_test.go
View File

@ -76,4 +76,6 @@ func TestRouter(t *testing.T) {
assert.Int(nBytes).Equals(0)
assert.Bool(tcpServer2Accessed).IsFalse()
conn.Close()
CloseAllServers()
}

14
testing/scenarios/server_env.go
View File

@ -27,7 +27,7 @@ import (
)
var (
serverup = make(map[string]bool)
runningServers = make([]*point.Point, 0, 10)
)
func TestFile(filename string) string {
@ -35,9 +35,6 @@ func TestFile(filename string) string {
}
func InitializeServerSetOnce(testcase string) error {
if up, found := serverup[testcase]; found && up {
return nil
}
err := InitializeServer(TestFile(testcase + "_server.json"))
if err != nil {
return err
@ -46,7 +43,6 @@ func InitializeServerSetOnce(testcase string) error {
if err != nil {
return err
}
serverup[testcase] = true
return nil
}
@ -68,6 +64,14 @@ func InitializeServer(configFile string) error {
log.Error("Error starting Point server: %v", err)
return err
}
runningServers = append(runningServers, vPoint)
return nil
}
func CloseAllServers() {
for _, server := range runningServers {
server.Close()
}
runningServers = make([]*point.Point, 0, 10)
}

10
testing/scenarios/socks_end_test.go
View File

@ -12,10 +12,6 @@ import (
"github.com/v2ray/v2ray-core/testing/servers/udp"
)
var (
serverUp = false
)
func TestTCPConnection(t *testing.T) {
v2testing.Current(t)
@ -86,6 +82,8 @@ func TestTCPConnection(t *testing.T) {
conn.Close()
}
CloseAllServers()
}
func TestTCPBind(t *testing.T) {
@ -135,6 +133,8 @@ func TestTCPBind(t *testing.T) {
assert.Bytes(connectResponse[:nBytes]).Equals([]byte{socks5Version, 7, 0, 1, 0, 0, 0, 0, 0, 0})
conn.Close()
CloseAllServers()
}
func TestUDPAssociate(t *testing.T) {
@ -204,4 +204,6 @@ func TestUDPAssociate(t *testing.T) {
udpConn.Close()
conn.Close()
CloseAllServers()
}