Merge pull request #886 from fatedier/websocket

Connect protocol support websocket

utils/net/conn.go

@@ -96,45 +96,34 @@ func (conn *WrapReadWriteCloserConn) SetWriteDeadline(t time.Time) error {
 	return &net.OpError{Op: "set", Net: "wrap", Source: nil, Addr: nil, Err: errors.New("deadline not supported")}
 }
 
-func ConnectServer(protocol string, addr string) (c Conn, err error) {
-	switch protocol {
-	case "tcp":
-		return ConnectTcpServer(addr)
-	case "kcp":
-		kcpConn, errRet := kcp.DialWithOptions(addr, nil, 10, 3)
-		if errRet != nil {
-			err = errRet
-			return
-		}
-		kcpConn.SetStreamMode(true)
-		kcpConn.SetWriteDelay(true)
-		kcpConn.SetNoDelay(1, 20, 2, 1)
-		kcpConn.SetWindowSize(128, 512)
-		kcpConn.SetMtu(1350)
-		kcpConn.SetACKNoDelay(false)
-		kcpConn.SetReadBuffer(4194304)
-		kcpConn.SetWriteBuffer(4194304)
-		c = WrapConn(kcpConn)
-		return
-	default:
-		return nil, fmt.Errorf("unsupport protocol: %s", protocol)
+type CloseNotifyConn struct {
+	net.Conn
+	log.Logger
+
+	// 1 means closed
+	closeFlag int32
+
+	closeFn func()
+}
+
+// closeFn will be only called once
+func WrapCloseNotifyConn(c net.Conn, closeFn func()) Conn {
+	return &CloseNotifyConn{
+		Conn:    c,
+		Logger:  log.NewPrefixLogger(""),
+		closeFn: closeFn,
 	}
 }
 
-func ConnectServerByProxy(proxyUrl string, protocol string, addr string) (c Conn, err error) {
-	switch protocol {
-	case "tcp":
-		var conn net.Conn
-		if conn, err = gnet.DialTcpByProxy(proxyUrl, addr); err != nil {
-			return
+func (cc *CloseNotifyConn) Close() (err error) {
+	pflag := atomic.SwapInt32(&cc.closeFlag, 1)
+	if pflag == 0 {
+		err = cc.Close()
+		if cc.closeFn != nil {
+			cc.closeFn()
 		}
-		return WrapConn(conn), nil
-	case "kcp":
-		// http proxy is not supported for kcp
-		return ConnectServer(protocol, addr)
-	default:
-		return nil, fmt.Errorf("unsupport protocol: %s", protocol)
 	}
+	return
 }
 
 type StatsConn struct {
@@ -175,3 +164,46 @@ func (statsConn *StatsConn) Close() (err error) {
 	}
 	return
 }
+
+func ConnectServer(protocol string, addr string) (c Conn, err error) {
+	switch protocol {
+	case "tcp":
+		return ConnectTcpServer(addr)
+	case "kcp":
+		kcpConn, errRet := kcp.DialWithOptions(addr, nil, 10, 3)
+		if errRet != nil {
+			err = errRet
+			return
+		}
+		kcpConn.SetStreamMode(true)
+		kcpConn.SetWriteDelay(true)
+		kcpConn.SetNoDelay(1, 20, 2, 1)
+		kcpConn.SetWindowSize(128, 512)
+		kcpConn.SetMtu(1350)
+		kcpConn.SetACKNoDelay(false)
+		kcpConn.SetReadBuffer(4194304)
+		kcpConn.SetWriteBuffer(4194304)
+		c = WrapConn(kcpConn)
+		return
+	default:
+		return nil, fmt.Errorf("unsupport protocol: %s", protocol)
+	}
+}
+
+func ConnectServerByProxy(proxyUrl string, protocol string, addr string) (c Conn, err error) {
+	switch protocol {
+	case "tcp":
+		var conn net.Conn
+		if conn, err = gnet.DialTcpByProxy(proxyUrl, addr); err != nil {
+			return
+		}
+		return WrapConn(conn), nil
+	case "kcp":
+		// http proxy is not supported for kcp
+		return ConnectServer(protocol, addr)
+	case "websocket":
+		return ConnectWebsocketServer(addr)
+	default:
+		return nil, fmt.Errorf("unsupport protocol: %s", protocol)
+	}
+}

utils/net/websocket.go

@@ -0,0 +1,105 @@
+package net
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/fatedier/frp/utils/log"
+
+	"golang.org/x/net/websocket"
+)
+
+var (
+	ErrWebsocketListenerClosed = errors.New("websocket listener closed")
+)
+
+const (
+	FrpWebsocketPath = "/~!frp"
+)
+
+type WebsocketListener struct {
+	net.Addr
+	ln     net.Listener
+	accept chan Conn
+	log.Logger
+
+	server    *http.Server
+	httpMutex *http.ServeMux
+}
+
+// ln: tcp listener for websocket connections
+func NewWebsocketListener(ln net.Listener) (wl *WebsocketListener) {
+	wl = &WebsocketListener{
+		Addr:   ln.Addr(),
+		accept: make(chan Conn),
+		Logger: log.NewPrefixLogger(""),
+	}
+
+	muxer := http.NewServeMux()
+	muxer.Handle(FrpWebsocketPath, websocket.Handler(func(c *websocket.Conn) {
+		notifyCh := make(chan struct{})
+		conn := WrapCloseNotifyConn(c, func() {
+			close(notifyCh)
+		})
+		wl.accept <- conn
+		<-notifyCh
+	}))
+
+	wl.server = &http.Server{
+		Addr:    ln.Addr().String(),
+		Handler: muxer,
+	}
+
+	go wl.server.Serve(ln)
+	return
+}
+
+func ListenWebsocket(bindAddr string, bindPort int) (*WebsocketListener, error) {
+	tcpLn, err := net.Listen("tcp", fmt.Sprintf("%s:%d", bindAddr, bindPort))
+	if err != nil {
+		return nil, err
+	}
+	l := NewWebsocketListener(tcpLn)
+	return l, nil
+}
+
+func (p *WebsocketListener) Accept() (Conn, error) {
+	c, ok := <-p.accept
+	if !ok {
+		return nil, ErrWebsocketListenerClosed
+	}
+	return c, nil
+}
+
+func (p *WebsocketListener) Close() error {
+	return p.server.Close()
+}
+
+// addr: domain:port
+func ConnectWebsocketServer(addr string) (Conn, error) {
+	addr = "ws://" + addr + FrpWebsocketPath
+	uri, err := url.Parse(addr)
+	if err != nil {
+		return nil, err
+	}
+
+	origin := "http://" + uri.Host
+	cfg, err := websocket.NewConfig(addr, origin)
+	if err != nil {
+		return nil, err
+	}
+	cfg.Dialer = &net.Dialer{
+		Timeout: 10 * time.Second,
+	}
+
+	conn, err := websocket.DialConfig(cfg)
+	if err != nil {
+		return nil, err
+	}
+	c := WrapConn(conn)
+	return c, nil
+}