https 、客户端与服务端连接优化

server/base.go

@@ -1,29 +1,24 @@
 package server
 
 import (
+	"errors"
 	"github.com/cnlh/easyProxy/bridge"
 	"github.com/cnlh/easyProxy/utils"
+	"net"
+	"net/http"
 	"sync"
 )
 
 //server base struct
 type server struct {
-	bridge *bridge.Bridge
-	task   *utils.Tunnel
-	config *utils.Config
+	id           int
+	bridge       *bridge.Bridge
+	task         *utils.Tunnel
+	config       *utils.Config
+	errorContent []byte
 	sync.Mutex
 }
 
-func (s *server) GetTunnelAndWriteHost(connType string, clientId int, cnf *utils.Config, addr string) (link *utils.Conn, err error) {
-	if link, err = s.bridge.GetTunnel(clientId, cnf.CompressEncode, cnf.CompressDecode, cnf.Crypt, cnf.Mux); err != nil {
-		return
-	}
-	if _, err = link.WriteHost(connType, addr); err != nil {
-		link.Close()
-	}
-	return
-}
-
 func (s *server) FlowAdd(in, out int64) {
 	s.Lock()
 	defer s.Unlock()
@@ -56,7 +51,6 @@ func (s *server) ResetConfig() bool {
 			s.config.U = client.Cnf.U
 			s.config.P = client.Cnf.P
 			s.config.Compress = client.Cnf.Compress
-			s.config.Mux = client.Cnf.Mux
 			s.config.Crypt = client.Cnf.Crypt
 		}
 	} else {
@@ -64,7 +58,6 @@ func (s *server) ResetConfig() bool {
 			s.config.U = task.Config.U
 			s.config.P = task.Config.P
 			s.config.Compress = task.Config.Compress
-			s.config.Mux = task.Config.Mux
 			s.config.Crypt = task.Config.Crypt
 		}
 	}
@@ -72,3 +65,43 @@ func (s *server) ResetConfig() bool {
 	s.config.CompressDecode, s.config.CompressEncode = utils.GetCompressType(s.config.Compress)
 	return true
 }
+
+func (s *server) linkCopy(link *utils.Link, c *utils.Conn, rb []byte, tunnel *utils.Conn, flow *utils.Flow) {
+	if rb != nil {
+		if _, err := tunnel.SendMsg(rb, link); err != nil {
+			c.Close()
+			return
+		}
+		flow.Add(len(rb), 0)
+	}
+	for {
+		buf := utils.BufPoolCopy.Get().([]byte)
+		if n, err := c.Read(buf); err != nil {
+			tunnel.SendMsg([]byte(utils.IO_EOF), link)
+			break
+		} else {
+			if _, err := tunnel.SendMsg(buf[:n], link); err != nil {
+				utils.PutBufPoolCopy(buf)
+				c.Close()
+				break
+			}
+			utils.PutBufPoolCopy(buf)
+			flow.Add(n, 0)
+		}
+	}
+}
+
+func (s *server) writeConnFail(c net.Conn) {
+	c.Write([]byte(utils.ConnectionFailBytes))
+	c.Write(s.errorContent)
+}
+
+//权限认证
+func (s *server) auth(r *http.Request, c *utils.Conn, u, p string) error {
+	if u != "" && p != "" && !utils.CheckAuth(r, u, p) {
+		c.Write([]byte(utils.UnauthorizedBytes))
+		c.Close()
+		return errors.New("401 Unauthorized")
+	}
+	return nil
+}

server/http.go

@@ -0,0 +1,169 @@
+package server
+
+import (
+	"bufio"
+	"crypto/tls"
+	"github.com/astaxie/beego"
+	"github.com/cnlh/easyProxy/bridge"
+	"github.com/cnlh/easyProxy/utils"
+	"log"
+	"net/http"
+	"net/http/httputil"
+	"strconv"
+	"sync"
+)
+
+type httpServer struct {
+	server
+	httpPort  int //http端口
+	httpsPort int //https监听端口
+	pemPath   string
+	keyPath   string
+	stop      chan bool
+}
+
+func NewHttp(bridge *bridge.Bridge, c *utils.Tunnel) *httpServer {
+	httpPort, _ := beego.AppConfig.Int("httpProxyPort")
+	httpsPort, _ := beego.AppConfig.Int("httpsProxyPort")
+	pemPath := beego.AppConfig.String("pemPath")
+	keyPath := beego.AppConfig.String("keyPath")
+	return &httpServer{
+		server: server{
+			task:   c,
+			bridge: bridge,
+			Mutex:  sync.Mutex{},
+		},
+		httpPort:  httpPort,
+		httpsPort: httpsPort,
+		pemPath:   pemPath,
+		keyPath:   keyPath,
+		stop:      make(chan bool),
+	}
+}
+
+func (s *httpServer) Start() error {
+	var err error
+	var http, https *http.Server
+	if s.errorContent, err = utils.ReadAllFromFile(beego.AppPath + "/web/static/page/error.html"); err != nil {
+		s.errorContent = []byte("easyProxy 404")
+	}
+
+	if s.httpPort > 0 {
+		http = s.NewServer(s.httpPort)
+		go func() {
+			log.Println("启动http监听,端口为", s.httpPort)
+			err := http.ListenAndServe()
+			if err != nil {
+				log.Fatalln(err)
+			}
+		}()
+	}
+	if s.httpsPort > 0 {
+		https = s.NewServer(s.httpsPort)
+		go func() {
+			log.Println("启动https监听,端口为", s.httpsPort)
+			err := https.ListenAndServeTLS(s.pemPath, s.keyPath)
+			if err != nil {
+				log.Fatalln(err)
+			}
+		}()
+	}
+	select {
+	case <-s.stop:
+		if http != nil {
+			http.Close()
+		}
+		if https != nil {
+			https.Close()
+		}
+	}
+	return nil
+}
+
+func (s *httpServer) Close() {
+	s.stop <- true
+}
+
+func (s *httpServer) handleTunneling(w http.ResponseWriter, r *http.Request) {
+	hijacker, ok := w.(http.Hijacker)
+	if !ok {
+		http.Error(w, "Hijacking not supported", http.StatusInternalServerError)
+		return
+	}
+	conn, _, err := hijacker.Hijack()
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusServiceUnavailable)
+	}
+	s.process(utils.NewConn(conn), r)
+}
+
+func (s *httpServer) process(c *utils.Conn, r *http.Request) {
+	//多客户端域名代理
+	var (
+		isConn = true
+		link   *utils.Link
+		host   *utils.Host
+		tunnel *utils.Conn
+		err    error
+	)
+	for {
+		//首次获取conn
+		if isConn {
+			if host, err = GetInfoByHost(r.Host); err != nil {
+				log.Printf("the host %s is not found !", r.Host)
+				break
+			}
+			//流量限制
+			if host.Client.Flow.FlowLimit > 0 && (host.Client.Flow.FlowLimit<<20) < (host.Client.Flow.ExportFlow+host.Client.Flow.InletFlow) {
+				break
+			}
+			host.Client.Cnf.CompressDecode, host.Client.Cnf.CompressEncode = utils.GetCompressType(host.Client.Cnf.Compress)
+			//权限控制
+			if err = s.auth(r, c, host.Client.Cnf.U, host.Client.Cnf.P); err != nil {
+				break
+			}
+			link = utils.NewLink(host.Client.GetId(), utils.CONN_TCP, host.Target, host.Client.Cnf.CompressEncode, host.Client.Cnf.CompressDecode, host.Client.Cnf.Crypt, c, host.Flow, nil, host.Client.Rate, nil)
+			if tunnel, err = s.bridge.SendLinkInfo(host.Client.Id, link); err != nil {
+				break
+			}
+			isConn = false
+		} else {
+			r, err = http.ReadRequest(bufio.NewReader(c))
+			if err != nil {
+				log.Println(err)
+				break
+			}
+		}
+		//根据设定，修改header和host
+		utils.ChangeHostAndHeader(r, host.HostChange, host.HeaderChange, c.Conn.RemoteAddr().String())
+		b, err := httputil.DumpRequest(r, true)
+		if err != nil {
+			break
+		}
+		host.Flow.Add(len(b), 0)
+		if _, err := tunnel.SendMsg(b, link); err != nil {
+			c.Close()
+			break
+		}
+	}
+
+	if isConn {
+		s.writeConnFail(c.Conn)
+	} else {
+		tunnel.SendMsg([]byte(utils.IO_EOF), link)
+	}
+
+	c.Close()
+
+}
+
+func (s *httpServer) NewServer(port int) *http.Server {
+	return &http.Server{
+		Addr: ":" + strconv.Itoa(port),
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			s.handleTunneling(w, r)
+		}),
+		// Disable HTTP/2.
+		TLSNextProto: make(map[string]func(*http.Server, *tls.Conn, http.Handler)),
+	}
+}

server/process.go

@@ -1,111 +0,0 @@
-package server
-
-import (
-	"bufio"
-	"github.com/cnlh/easyProxy/utils"
-	"github.com/pkg/errors"
-	"log"
-	"net/http"
-	"net/http/httputil"
-	"sync"
-)
-
-type process func(c *utils.Conn, s *TunnelModeServer) error
-
-//tcp隧道模式
-func ProcessTunnel(c *utils.Conn, s *TunnelModeServer) error {
-	if !s.ResetConfig() {
-		c.Close()
-		return errors.New("流量超出")
-	}
-	return s.dealClient(c, s.config, s.task.Target, "", nil)
-}
-
-//http代理模式
-func ProcessHttp(c *utils.Conn, s *TunnelModeServer) error {
-	if !s.ResetConfig() {
-		c.Close()
-		return errors.New("流量超出")
-	}
-	method, addr, rb, err, r := c.GetHost()
-	if err != nil {
-		log.Println(err)
-		c.Close()
-		return err
-	}
-	if err := s.auth(r, c, s.config.U, s.config.P); err != nil {
-		return err
-	}
-	return s.dealClient(c, s.config, addr, method, rb)
-}
-
-//多客户端域名代理
-func ProcessHost(c *utils.Conn, s *TunnelModeServer) error {
-	var (
-		isConn = true
-		link   *utils.Conn
-		host   *utils.Host
-		wg     sync.WaitGroup
-	)
-	for {
-		r, err := http.ReadRequest(bufio.NewReader(c))
-		if err != nil {
-			break
-		}
-		//首次获取conn
-		if isConn {
-			if host, err = GetInfoByHost(r.Host); err != nil {
-				log.Printf("the host %s is not found !", r.Host)
-				break
-			}
-			//流量限制
-			if host.Client.Flow.FlowLimit > 0 && (host.Client.Flow.FlowLimit<<20) < (host.Client.Flow.ExportFlow+host.Client.Flow.InletFlow) {
-				break
-			}
-			host.Client.Cnf.CompressDecode, host.Client.Cnf.CompressEncode = utils.GetCompressType(host.Client.Cnf.Compress)
-			//权限控制
-			if err = s.auth(r, c, host.Client.Cnf.U, host.Client.Cnf.P); err != nil {
-				break
-			}
-			if link, err = s.GetTunnelAndWriteHost(utils.CONN_TCP, host.Client.Id, host.Client.Cnf, host.Target); err != nil {
-				log.Println("get bridge tunnel error: ", err)
-				break
-			}
-			if flag, err := link.ReadFlag(); err != nil || flag == utils.CONN_ERROR {
-				log.Printf("the host %s connection to %s error", r.Host, host.Target)
-				break
-			} else {
-				wg.Add(1)
-				go func() {
-					out, _ := utils.Relay(c.Conn, link.Conn, host.Client.Cnf.CompressDecode, host.Client.Cnf.Crypt, host.Client.Cnf.Mux, host.Client.Rate)
-					wg.Done()
-					s.FlowAddHost(host, 0, out)
-				}()
-			}
-			isConn = false
-		}
-		//根据设定，修改header和host
-		utils.ChangeHostAndHeader(r, host.HostChange, host.HeaderChange, c.Conn.RemoteAddr().String())
-		b, err := httputil.DumpRequest(r, true)
-		if err != nil {
-			break
-		}
-		s.FlowAddHost(host, int64(len(b)), 0)
-		if _, err := link.WriteTo(b, host.Client.Cnf.CompressEncode, host.Client.Cnf.Crypt, host.Client.Rate); err != nil {
-			break
-		}
-	}
-	wg.Wait()
-	if host != nil && host.Client.Cnf != nil && host.Client.Cnf.Mux && link != nil {
-		link.WriteTo([]byte(utils.IO_EOF), host.Client.Cnf.CompressEncode, host.Client.Cnf.Crypt, host.Client.Rate)
-		s.bridge.ReturnTunnel(link, host.Client.Id)
-	} else if link != nil {
-		link.Close()
-	}
-
-	if isConn {
-		s.writeConnFail(c.Conn)
-	}
-	c.Close()
-	return nil
-}

server/server.go

@@ -2,27 +2,17 @@ package server
 
 import (
 	"errors"
-	"github.com/astaxie/beego"
 	"github.com/cnlh/easyProxy/bridge"
 	"github.com/cnlh/easyProxy/utils"
 	"log"
 	"reflect"
 	"strings"
-	"sync"
 )
 
-type RunServer struct {
-	flag       int   //标志
-	ExportFlow int64 //出口流量
-	InletFlow  int64 //入口流量
-	service    interface{}
-	sync.Mutex
-}
-
 var (
-	Bridge    *bridge.Bridge
-	RunList   map[int]interface{} //运行中的任务
-	CsvDb     = utils.GetCsvDb()
+	Bridge  *bridge.Bridge
+	RunList map[int]interface{} //运行中的任务
+	CsvDb   = utils.GetCsvDb()
 )
 
 func init() {
@@ -69,9 +59,8 @@ func NewMode(Bridge *bridge.Bridge, c *utils.Tunnel) interface{} {
 		return NewUdpModeServer(Bridge, c)
 	case "webServer":
 		InitFromCsv()
-		p, _ := beego.AppConfig.Int("hostPort")
 		t := &utils.Tunnel{
-			TcpPort: p,
+			TcpPort: 0,
 			Mode:    "httpHostServer",
 			Target:  "",
 			Config:  &utils.Config{},
@@ -82,7 +71,7 @@ func NewMode(Bridge *bridge.Bridge, c *utils.Tunnel) interface{} {
 	case "hostServer":
 		return NewHostServer(c)
 	case "httpHostServer":
-		return NewTunnelModeServer(ProcessHost, Bridge, c)
+		return NewHttp(Bridge, c)
 	}
 	return nil
 }
@@ -161,7 +150,7 @@ func GetTunnel(start, length int, typeVal string, clientId int) ([]*utils.Tunnel
 			continue
 		}
 		cnt++
-		if _, ok := Bridge.SignalList[v.Client.Id]; ok {
+		if _, ok := Bridge.Client[v.Client.Id]; ok {
 			v.Client.IsConnect = true
 		} else {
 			v.Client.IsConnect = false
@@ -189,7 +178,7 @@ func GetClientList(start, length int) (list []*utils.Client, cnt int) {
 
 func dealClientData(list []*utils.Client) {
 	for _, v := range list {
-		if _, ok := Bridge.SignalList[v.Id]; ok {
+		if _, ok := Bridge.Client[v.Id]; ok {
 			v.IsConnect = true
 		} else {
 			v.IsConnect = false
@@ -228,8 +217,7 @@ func DelTunnelAndHostByClientId(clientId int) {
 
 //关闭客户端连接
 func DelClientConnect(clientId int) {
-	Bridge.DelClientTunnel(clientId)
-	Bridge.DelClientSignal(clientId)
+	Bridge.DelClient(clientId)
 }
 
 func GetDashboardData() map[string]int {

server/socks5.go

@@ -106,7 +106,7 @@ func (s *Sock5ModeServer) sendReply(c net.Conn, rep uint8) {
 }
 
 //do conn
-func (s *Sock5ModeServer) doConnect(c net.Conn, command uint8) (proxyConn *utils.Conn, err error) {
+func (s *Sock5ModeServer) doConnect(c net.Conn, command uint8) {
 	addrType := make([]byte, 1)
 	c.Read(addrType)
 	var host string
@@ -127,8 +127,7 @@ func (s *Sock5ModeServer) doConnect(c net.Conn, command uint8) (proxyConn *utils
 		host = string(domain)
 	default:
 		s.sendReply(c, addrTypeNotSupported)
-		err = errors.New("Address type not supported")
-		return nil, err
+		return
 	}
 
 	var port uint16
@@ -141,34 +140,22 @@ func (s *Sock5ModeServer) doConnect(c net.Conn, command uint8) (proxyConn *utils
 	} else {
 		ltype = utils.CONN_TCP
 	}
-	if proxyConn, err = s.GetTunnelAndWriteHost(ltype, s.task.Client.Id, s.config, addr); err != nil {
-		log.Println("get bridge tunnel error: ", err)
+	link := utils.NewLink(s.task.Client.GetId(), ltype, addr, s.config.CompressEncode, s.config.CompressDecode, s.config.Crypt, utils.NewConn(c), s.task.Flow, nil, s.task.Client.Rate, nil)
+
+	if tunnel, err := s.bridge.SendLinkInfo(s.task.Client.Id, link); err != nil {
+		log.Println("error", err, link)
+		c.Close()
 		return
-	}
-	s.sendReply(c, succeeded)
-	var flag string
-	if flag, err = proxyConn.ReadFlag(); err == nil {
-		if flag != utils.CONN_SUCCESS {
-			err = errors.New("conn failed")
-		}
+	} else {
+		s.sendReply(c, succeeded)
+		s.linkCopy(link, utils.NewConn(c), nil, tunnel, s.task.Flow)
 	}
 	return
 }
 
 //conn
 func (s *Sock5ModeServer) handleConnect(c net.Conn) {
-	proxyConn, err := s.doConnect(c, connectMethod)
-	defer func() {
-		if s.config.Mux && proxyConn != nil {
-			s.bridge.ReturnTunnel(proxyConn, s.task.Client.Id)
-		}
-	}()
-	if err != nil {
-		c.Close()
-	} else {
-		out, in := utils.ReplayWaitGroup(proxyConn.Conn, c, s.config.CompressEncode, s.config.CompressDecode, s.config.Crypt, s.config.Mux, s.task.Client.Rate)
-		s.FlowAdd(in, out)
-	}
+	s.doConnect(c, connectMethod)
 }
 
 // passive mode
@@ -195,18 +182,7 @@ func (s *Sock5ModeServer) handleUDP(c net.Conn) {
 		c.Read(dummy)
 	}
 
-	proxyConn, err := s.doConnect(c, associateMethod)
-	defer func() {
-		if s.config.Mux && proxyConn != nil {
-			s.bridge.ReturnTunnel(proxyConn, s.task.Client.Id)
-		}
-	}()
-	if err != nil {
-		c.Close()
-	} else {
-		out, in := utils.ReplayWaitGroup(proxyConn.Conn, c, s.config.CompressEncode, s.config.CompressDecode, s.config.Crypt, s.config.Mux, s.task.Client.Rate)
-		s.FlowAdd(in, out)
-	}
+	s.doConnect(c, associateMethod)
 }
 
 //new conn

server/tcp.go

@@ -2,21 +2,18 @@ package server
 
 import (
 	"errors"
-	"fmt"
 	"github.com/astaxie/beego"
 	"github.com/cnlh/easyProxy/bridge"
 	"github.com/cnlh/easyProxy/utils"
 	"log"
 	"net"
-	"net/http"
 	"strings"
 )
 
 type TunnelModeServer struct {
 	server
-	errorContent []byte
-	process      process
-	listener     *net.TCPListener
+	process  process
+	listener *net.TCPListener
 }
 
 //tcp|http|host
@@ -32,9 +29,6 @@ func NewTunnelModeServer(process process, bridge *bridge.Bridge, task *utils.Tun
 //开始
 func (s *TunnelModeServer) Start() error {
 	var err error
-	if s.errorContent, err = utils.ReadAllFromFile(beego.AppPath + "/web/static/page/error.html"); err != nil {
-		s.errorContent = []byte("easyProxy 404")
-	}
 	s.listener, err = net.ListenTCP("tcp", &net.TCPAddr{net.ParseIP("0.0.0.0"), s.task.TcpPort, ""})
 	if err != nil {
 		return err
@@ -53,44 +47,15 @@ func (s *TunnelModeServer) Start() error {
 	return nil
 }
 
-//权限认证
-func (s *TunnelModeServer) auth(r *http.Request, c *utils.Conn, u, p string) error {
-	if u != "" && p != "" && !utils.CheckAuth(r, u, p) {
-		c.Write([]byte(utils.UnauthorizedBytes))
-		c.Close()
-		return errors.New("401 Unauthorized")
-	}
-	return nil
-}
-
-func (s *TunnelModeServer) writeConnFail(c net.Conn) {
-	c.Write([]byte(utils.ConnectionFailBytes))
-	c.Write(s.errorContent)
-}
-
 //与客户端建立通道
 func (s *TunnelModeServer) dealClient(c *utils.Conn, cnf *utils.Config, addr string, method string, rb []byte) error {
-	var link *utils.Conn
-	var err error
-	defer func() {
-		if cnf.Mux && link != nil {
-			s.bridge.ReturnTunnel(link, s.task.Client.Id)
-		}
-	}()
-	if link, err = s.GetTunnelAndWriteHost(utils.CONN_TCP, s.task.Client.Id, cnf, addr); err != nil {
-		log.Println("get bridge tunnel error: ", err)
+	link := utils.NewLink(s.task.Client.GetId(), utils.CONN_TCP, addr, cnf.CompressEncode, cnf.CompressDecode, cnf.Crypt, c, s.task.Flow, nil, s.task.Client.Rate, nil)
+
+	if tunnel, err := s.bridge.SendLinkInfo(s.task.Client.Id, link); err != nil {
+		c.Close()
 		return err
-	}
-	if flag, err := link.ReadFlag(); err == nil {
-		if flag == utils.CONN_SUCCESS {
-			if method == "CONNECT" {
-				fmt.Fprint(c, "HTTP/1.1 200 Connection established\r\n")
-			} else if rb != nil {
-				link.WriteTo(rb, cnf.CompressEncode, cnf.Crypt, s.task.Client.Rate)
-			}
-			out, in := utils.ReplayWaitGroup(link.Conn, c.Conn, cnf.CompressEncode, cnf.CompressDecode, cnf.Crypt, cnf.Mux, s.task.Client.Rate)
-			s.FlowAdd(in, out)
-		}
+	} else {
+		s.linkCopy(link, c, rb, tunnel, s.task.Flow)
 	}
 	return nil
 }
@@ -142,3 +107,32 @@ func NewHostServer(task *utils.Tunnel) *HostServer {
 func (s *HostServer) Close() error {
 	return nil
 }
+
+type process func(c *utils.Conn, s *TunnelModeServer) error
+
+//tcp隧道模式
+func ProcessTunnel(c *utils.Conn, s *TunnelModeServer) error {
+	if !s.ResetConfig() {
+		c.Close()
+		return errors.New("流量超出")
+	}
+	return s.dealClient(c, s.config, s.task.Target, "", nil)
+}
+
+//http代理模式
+func ProcessHttp(c *utils.Conn, s *TunnelModeServer) error {
+	if !s.ResetConfig() {
+		c.Close()
+		return errors.New("流量超出")
+	}
+	method, addr, rb, err, r := c.GetHost()
+	if err != nil {
+		log.Println(err)
+		c.Close()
+		return err
+	}
+	if err := s.auth(r, c, s.config.U, s.config.P); err != nil {
+		return err
+	}
+	return s.dealClient(c, s.config, addr, method, rb)
+}

server/udp.go

@@ -3,8 +3,6 @@ package server
 import (
 	"github.com/cnlh/easyProxy/bridge"
 	"github.com/cnlh/easyProxy/utils"
-	"io"
-	"log"
 	"net"
 	"strings"
 )
@@ -31,9 +29,9 @@ func (s *UdpModeServer) Start() error {
 	if err != nil {
 		return err
 	}
-	data := make([]byte, 1472) //udp数据包大小
+	buf := utils.BufPoolUdp.Get().([]byte)
 	for {
-		n, addr, err := s.listener.ReadFromUDP(data)
+		n, addr, err := s.listener.ReadFromUDP(buf)
 		if err != nil {
 			if strings.Contains(err.Error(), "use of closed network connection") {
 				break
@@ -43,42 +41,19 @@ func (s *UdpModeServer) Start() error {
 		if !s.ResetConfig() {
 			continue
 		}
-		go s.process(addr, data[:n])
+		go s.process(addr, buf[:n])
 	}
 	return nil
 }
 
-//TODO:效率问题有待解决--->建立稳定通道，重复利用，提高效率，下个版本
 func (s *UdpModeServer) process(addr *net.UDPAddr, data []byte) {
-	conn, err := s.bridge.GetTunnel(s.task.Client.Id, s.config.CompressEncode, s.config.CompressDecode, s.config.Crypt, s.config.Mux)
-	if err != nil {
-		log.Println(err)
+	link := utils.NewLink(s.task.Client.GetId(), utils.CONN_UDP, s.task.Target, s.config.CompressEncode, s.config.CompressDecode, s.config.Crypt, nil, s.task.Flow, s.listener, s.task.Client.Rate, addr)
+
+	if tunnel, err := s.bridge.SendLinkInfo(s.task.Client.Id, link); err != nil {
 		return
-	}
-	if _, err := conn.WriteHost(utils.CONN_UDP, s.task.Target); err != nil {
-		conn.Close()
-		return
-	}
-	if flag, err := conn.ReadFlag(); err == nil {
-		defer func() {
-			if conn != nil && s.config.Mux {
-				conn.WriteTo([]byte(utils.IO_EOF), s.config.CompressEncode, s.config.Crypt, s.task.Client.Rate)
-				s.bridge.ReturnTunnel(conn, s.task.Client.Id)
-			} else {
-				conn.Close()
-			}
-		}()
-		if flag == utils.CONN_SUCCESS {
-			in, _ := conn.WriteTo(data, s.config.CompressEncode, s.config.Crypt, s.task.Client.Rate)
-			buf := utils.BufPoolUdp.Get().([]byte)
-			out, err := conn.ReadFrom(buf, s.config.CompressDecode, s.config.Crypt, s.task.Client.Rate)
-			if err != nil || err == io.EOF {
-				return
-			}
-			s.listener.WriteToUDP(buf[:out], addr)
-			s.FlowAdd(int64(in), int64(out))
-			utils.BufPoolUdp.Put(buf)
-		}
+	} else {
+		s.task.Flow.Add(len(data), 0)
+		tunnel.SendMsg(data, link)
 	}
 }