From cb1356be757c4504b51a7499473428eb1872b7b3 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:32:47 +0800 Subject: [PATCH 01/47] Register ws network --- common/net/network.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/common/net/network.go b/common/net/network.go index 3fa9509e..1244fe98 100644 --- a/common/net/network.go +++ b/common/net/network.go @@ -15,6 +15,9 @@ const ( // KCPNetwork represents the KCP network. KCPNetwork = Network("kcp") + + // WSNetwork represents the Websocket over HTTP network. + WSNetwork = Network("ws") ) // Network represents a communication network on internet. From 508f79f95f1cd90ce037132febdee3d41dfd2396 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:33:34 +0800 Subject: [PATCH 02/47] Advertise WS cap --- proxy/vmess/inbound/inbound.go | 2 +- proxy/vmess/outbound/outbound.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/proxy/vmess/inbound/inbound.go b/proxy/vmess/inbound/inbound.go index 14117cf0..2be40b33 100644 --- a/proxy/vmess/inbound/inbound.go +++ b/proxy/vmess/inbound/inbound.go @@ -242,7 +242,7 @@ func (this *VMessInboundHandler) HandleConnection(connection internet.Connection type Factory struct{} func (this *Factory) StreamCapability() internet.StreamConnectionType { - return internet.StreamConnectionTypeRawTCP | internet.StreamConnectionTypeTCP | internet.StreamConnectionTypeKCP + return internet.StreamConnectionTypeRawTCP | internet.StreamConnectionTypeTCP | internet.StreamConnectionTypeKCP | internet.StreamConnectionTypeWebSocket } func (this *Factory) Create(space app.Space, rawConfig interface{}, meta *proxy.InboundHandlerMeta) (proxy.InboundHandler, error) { diff --git a/proxy/vmess/outbound/outbound.go b/proxy/vmess/outbound/outbound.go index bcdc6b12..c4f95745 100644 --- a/proxy/vmess/outbound/outbound.go +++ b/proxy/vmess/outbound/outbound.go @@ -158,7 +158,7 @@ func (this *VMessOutboundHandler) handleResponse(session *encoding.ClientSession type Factory struct{} func (this *Factory) StreamCapability() internet.StreamConnectionType { - return internet.StreamConnectionTypeRawTCP | internet.StreamConnectionTypeTCP | internet.StreamConnectionTypeKCP + return internet.StreamConnectionTypeRawTCP | internet.StreamConnectionTypeTCP | internet.StreamConnectionTypeKCP | internet.StreamConnectionTypeWebSocket } func (this *Factory) Create(space app.Space, rawConfig interface{}, meta *proxy.OutboundHandlerMeta) (proxy.OutboundHandler, error) { From 8c96b1961b0ae7881efe527be72b6455b518647b Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:35:31 +0800 Subject: [PATCH 03/47] load ws transport on init --- shell/point/main/main.go | 1 + 1 file changed, 1 insertion(+) diff --git a/shell/point/main/main.go b/shell/point/main/main.go index eaf5160f..889336d2 100644 --- a/shell/point/main/main.go +++ b/shell/point/main/main.go @@ -25,6 +25,7 @@ import ( _ "github.com/v2ray/v2ray-core/transport/internet/kcp" _ "github.com/v2ray/v2ray-core/transport/internet/tcp" _ "github.com/v2ray/v2ray-core/transport/internet/udp" + _ "github.com/v2ray/v2ray-core/transport/internet/ws" _ "github.com/v2ray/v2ray-core/transport/internet/authenticators/noop" _ "github.com/v2ray/v2ray-core/transport/internet/authenticators/srtp" From 8c4a6d94f9e0973bbefe0ed19982461e285bc041 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:37:17 +0800 Subject: [PATCH 04/47] WS Conf structure --- transport/config.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/transport/config.go b/transport/config.go index efe221fe..55c7ccc1 100644 --- a/transport/config.go +++ b/transport/config.go @@ -3,12 +3,14 @@ package transport import ( "github.com/v2ray/v2ray-core/transport/internet/kcp" "github.com/v2ray/v2ray-core/transport/internet/tcp" + "github.com/v2ray/v2ray-core/transport/internet/ws" ) // Config for V2Ray transport layer. type Config struct { tcpConfig *tcp.Config kcpConfig kcp.Config + wsConfig *ws.Config } // Apply applies this Config. @@ -17,5 +19,8 @@ func (this *Config) Apply() error { this.tcpConfig.Apply() } this.kcpConfig.Apply() + if this.wsConfig != nil { + this.wsConfig.Apply() + } return nil } From 2eced2bd6c228d032ff4cc01f4725689971b368d Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:39:34 +0800 Subject: [PATCH 05/47] Load Configure from json --- transport/config_json.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/transport/config_json.go b/transport/config_json.go index c7cb2d63..6dbd9c8f 100644 --- a/transport/config_json.go +++ b/transport/config_json.go @@ -7,12 +7,14 @@ import ( "github.com/v2ray/v2ray-core/transport/internet/kcp" "github.com/v2ray/v2ray-core/transport/internet/tcp" + "github.com/v2ray/v2ray-core/transport/internet/ws" ) func (this *Config) UnmarshalJSON(data []byte) error { type JsonConfig struct { TCPConfig *tcp.Config `json:"tcpSettings"` KCPConfig kcp.Config `json:"kcpSettings"` + WSConfig *ws.Config `json:"wsSettings"` } jsonConfig := &JsonConfig{ KCPConfig: kcp.DefaultConfig(), @@ -22,6 +24,6 @@ func (this *Config) UnmarshalJSON(data []byte) error { } this.tcpConfig = jsonConfig.TCPConfig this.kcpConfig = jsonConfig.KCPConfig - + this.wsConfig = jsonConfig.WSConfig return nil } From 0d60d2449af39bb0e992c507a5d1807641afae8e Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:41:13 +0800 Subject: [PATCH 06/47] Add Websocket Stream type --- transport/internet/connection.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/transport/internet/connection.go b/transport/internet/connection.go index 61435cbd..a5ce7c2d 100644 --- a/transport/internet/connection.go +++ b/transport/internet/connection.go @@ -15,9 +15,10 @@ type Reusable interface { type StreamConnectionType int const ( - StreamConnectionTypeRawTCP StreamConnectionType = 1 - StreamConnectionTypeTCP StreamConnectionType = 2 - StreamConnectionTypeKCP StreamConnectionType = 4 + StreamConnectionTypeRawTCP StreamConnectionType = 1 + StreamConnectionTypeTCP StreamConnectionType = 2 + StreamConnectionTypeKCP StreamConnectionType = 4 + StreamConnectionTypeWebSocket StreamConnectionType = 8 ) type StreamSecurityType int From 20f16309e6ed6845dab315b60bda09643ab90e1c Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:42:18 +0800 Subject: [PATCH 07/47] Load Websocket Streamtype from json --- transport/internet/connection_json.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/transport/internet/connection_json.go b/transport/internet/connection_json.go index eba3b0e8..aff84812 100644 --- a/transport/internet/connection_json.go +++ b/transport/internet/connection_json.go @@ -50,6 +50,9 @@ func (this *StreamSettings) UnmarshalJSON(data []byte) error { if jsonConfig.Network.HasNetwork(v2net.KCPNetwork) { this.Type |= StreamConnectionTypeKCP } + if jsonConfig.Network.HasNetwork(v2net.WSNetwork) { + this.Type |= StreamConnectionTypeWebSocket + } if jsonConfig.Network.HasNetwork(v2net.TCPNetwork) { this.Type |= StreamConnectionTypeTCP } From 29f69d63c41ea3ad13c5bbcab681964842c78793 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:42:56 +0800 Subject: [PATCH 08/47] Support Dial WebSocket Connection --- transport/internet/dialer.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/transport/internet/dialer.go b/transport/internet/dialer.go index ad4e4b24..c7ce5fee 100644 --- a/transport/internet/dialer.go +++ b/transport/internet/dialer.go @@ -20,9 +20,11 @@ var ( KCPDialer Dialer RawTCPDialer Dialer UDPDialer Dialer + WSDialer Dialer ) func Dial(src v2net.Address, dest v2net.Destination, settings *StreamSettings) (Connection, error) { + var connection Connection var err error if dest.IsTCP() { @@ -31,6 +33,15 @@ func Dial(src v2net.Address, dest v2net.Destination, settings *StreamSettings) ( connection, err = TCPDialer(src, dest) case settings.IsCapableOf(StreamConnectionTypeKCP): connection, err = KCPDialer(src, dest) + case settings.IsCapableOf(StreamConnectionTypeWebSocket): + connection, err = WSDialer(src, dest) + + /*Warning: Hours wasted: the following item must be last one + + internet.StreamConnectionType have a default value of 1, + so the following attempt will catch all. + */ + case settings.IsCapableOf(StreamConnectionTypeRawTCP): connection, err = RawTCPDialer(src, dest) default: @@ -39,6 +50,7 @@ func Dial(src v2net.Address, dest v2net.Destination, settings *StreamSettings) ( if err != nil { return nil, err } + if settings.Security == StreamSecurityTypeNone { return connection, nil } From 7ef66c57dc12b45ba3e0e8edcecc060cfd2c126f Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:43:46 +0800 Subject: [PATCH 09/47] Support Listen WebSocket Connection --- transport/internet/tcp_hub.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/transport/internet/tcp_hub.go b/transport/internet/tcp_hub.go index d0642de7..82daa82d 100644 --- a/transport/internet/tcp_hub.go +++ b/transport/internet/tcp_hub.go @@ -17,6 +17,7 @@ var ( KCPListenFunc ListenFunc TCPListenFunc ListenFunc RawTCPListenFunc ListenFunc + WSListenFunc ListenFunc ) type ListenFunc func(address v2net.Address, port v2net.Port) (Listener, error) @@ -42,6 +43,8 @@ func ListenTCP(address v2net.Address, port v2net.Port, callback ConnectionHandle listener, err = TCPListenFunc(address, port) case settings.IsCapableOf(StreamConnectionTypeKCP): listener, err = KCPListenFunc(address, port) + case settings.IsCapableOf(StreamConnectionTypeWebSocket): + listener, err = WSListenFunc(address, port) case settings.IsCapableOf(StreamConnectionTypeRawTCP): listener, err = RawTCPListenFunc(address, port) default: From fecf9cc5b89a23c9909becee781220a9a23db4a9 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:44:36 +0800 Subject: [PATCH 10/47] Added WebSocket transport --- transport/internet/ws/config.go | 21 +++ transport/internet/ws/config_json.go | 25 +++ transport/internet/ws/connection.go | 110 ++++++++++++ transport/internet/ws/connection_cache.go | 112 ++++++++++++ transport/internet/ws/connection_test.go | 19 +++ transport/internet/ws/dialer.go | 119 +++++++++++++ transport/internet/ws/hub.go | 162 ++++++++++++++++++ transport/internet/ws/wsconn.go | 199 ++++++++++++++++++++++ 8 files changed, 767 insertions(+) create mode 100644 transport/internet/ws/config.go create mode 100644 transport/internet/ws/config_json.go create mode 100644 transport/internet/ws/connection.go create mode 100644 transport/internet/ws/connection_cache.go create mode 100644 transport/internet/ws/connection_test.go create mode 100644 transport/internet/ws/dialer.go create mode 100644 transport/internet/ws/hub.go create mode 100644 transport/internet/ws/wsconn.go diff --git a/transport/internet/ws/config.go b/transport/internet/ws/config.go new file mode 100644 index 00000000..e078d607 --- /dev/null +++ b/transport/internet/ws/config.go @@ -0,0 +1,21 @@ +package ws + +type Config struct { + ConnectionReuse bool + Path string + Pto string + Cert string + PrivKey string +} + +func (this *Config) Apply() { + effectiveConfig = this +} + +var ( + effectiveConfig = &Config{ + ConnectionReuse: true, + Path: "", + Pto: "", + } +) diff --git a/transport/internet/ws/config_json.go b/transport/internet/ws/config_json.go new file mode 100644 index 00000000..843e549b --- /dev/null +++ b/transport/internet/ws/config_json.go @@ -0,0 +1,25 @@ +package ws + +import ( + "encoding/json" +) + +func (this *Config) UnmarshalJSON(data []byte) error { + type JsonConfig struct { + ConnectionReuse bool `json:"connectionReuse"` + Path string `json:"Path"` + Pto string `json:"Pto"` + } + jsonConfig := &JsonConfig{ + ConnectionReuse: true, + Path: "", + Pto: "", + } + if err := json.Unmarshal(data, jsonConfig); err != nil { + return err + } + this.ConnectionReuse = jsonConfig.ConnectionReuse + this.Path = jsonConfig.Path + this.Pto = jsonConfig.Pto + return nil +} diff --git a/transport/internet/ws/connection.go b/transport/internet/ws/connection.go new file mode 100644 index 00000000..141fdb32 --- /dev/null +++ b/transport/internet/ws/connection.go @@ -0,0 +1,110 @@ +package ws + +import ( + "errors" + "io" + "net" + "reflect" + "time" +) + +var ( + ErrInvalidConn = errors.New("Invalid Connection.") +) + +type ConnectionManager interface { + Recycle(string, *wsconn) +} + +type Connection struct { + dest string + conn *wsconn + listener ConnectionManager + reusable bool +} + +func NewConnection(dest string, conn *wsconn, manager ConnectionManager) *Connection { + return &Connection{ + dest: dest, + conn: conn, + listener: manager, + reusable: effectiveConfig.ConnectionReuse, + } +} + +func (this *Connection) Read(b []byte) (int, error) { + if this == nil || this.conn == nil { + return 0, io.EOF + } + + return this.conn.Read(b) +} + +func (this *Connection) Write(b []byte) (int, error) { + if this == nil || this.conn == nil { + return 0, io.ErrClosedPipe + } + return this.conn.Write(b) +} + +func (this *Connection) Close() error { + if this == nil || this.conn == nil { + return io.ErrClosedPipe + } + if this.Reusable() { + this.listener.Recycle(this.dest, this.conn) + return nil + } + err := this.conn.Close() + this.conn = nil + return err +} + +func (this *Connection) LocalAddr() net.Addr { + return this.conn.LocalAddr() +} + +func (this *Connection) RemoteAddr() net.Addr { + return this.conn.RemoteAddr() +} + +func (this *Connection) SetDeadline(t time.Time) error { + return this.conn.SetDeadline(t) +} + +func (this *Connection) SetReadDeadline(t time.Time) error { + return this.conn.SetReadDeadline(t) +} + +func (this *Connection) SetWriteDeadline(t time.Time) error { + return this.conn.SetWriteDeadline(t) +} + +func (this *Connection) SetReusable(reusable bool) { + if !effectiveConfig.ConnectionReuse { + return + } + this.reusable = reusable +} + +func (this *Connection) Reusable() bool { + return this.reusable +} + +func (this *Connection) SysFd() (int, error) { + return getSysFd(this.conn) +} + +func getSysFd(conn net.Conn) (int, error) { + cv := reflect.ValueOf(conn) + switch ce := cv.Elem(); ce.Kind() { + case reflect.Struct: + netfd := ce.FieldByName("conn").FieldByName("fd") + switch fe := netfd.Elem(); fe.Kind() { + case reflect.Struct: + fd := fe.FieldByName("sysfd") + return int(fd.Int()), nil + } + } + return 0, ErrInvalidConn +} diff --git a/transport/internet/ws/connection_cache.go b/transport/internet/ws/connection_cache.go new file mode 100644 index 00000000..e0743d75 --- /dev/null +++ b/transport/internet/ws/connection_cache.go @@ -0,0 +1,112 @@ +package ws + +import ( + "net" + "sync" + "time" + + "github.com/v2ray/v2ray-core/common/signal" +) + +type AwaitingConnection struct { + conn *wsconn + expire time.Time +} + +func (this *AwaitingConnection) Expired() bool { + return this.expire.Before(time.Now()) +} + +type ConnectionCache struct { + sync.Mutex + cache map[string][]*AwaitingConnection + cleanupOnce signal.Once +} + +func NewConnectionCache() *ConnectionCache { + return &ConnectionCache{ + cache: make(map[string][]*AwaitingConnection), + } +} + +func (this *ConnectionCache) Cleanup() { + defer this.cleanupOnce.Reset() + + for len(this.cache) > 0 { + time.Sleep(time.Second * 4) + this.Lock() + for key, value := range this.cache { + size := len(value) + changed := false + for i := 0; i < size; { + if value[i].Expired() { + value[i].conn.Close() + value[i] = value[size-1] + size-- + changed = true + } else { + i++ + } + } + if changed { + for i := size; i < len(value); i++ { + value[i] = nil + } + value = value[:size] + this.cache[key] = value + } + } + this.Unlock() + } +} + +func (this *ConnectionCache) Recycle(dest string, conn *wsconn) { + this.Lock() + defer this.Unlock() + + aconn := &AwaitingConnection{ + conn: conn, + expire: time.Now().Add(time.Second * 4), + } + + var list []*AwaitingConnection + if v, found := this.cache[dest]; found { + v = append(v, aconn) + list = v + } else { + list = []*AwaitingConnection{aconn} + } + this.cache[dest] = list + + go this.cleanupOnce.Do(this.Cleanup) +} + +func FindFirstValid(list []*AwaitingConnection) int { + for idx, conn := range list { + if !conn.Expired() && !conn.conn.connClosing { + return idx + } + go conn.conn.Close() + } + return -1 +} + +func (this *ConnectionCache) Get(dest string) net.Conn { + this.Lock() + defer this.Unlock() + + list, found := this.cache[dest] + if !found { + return nil + } + + firstValid := FindFirstValid(list) + if firstValid == -1 { + delete(this.cache, dest) + return nil + } + res := list[firstValid].conn + list = list[firstValid+1:] + this.cache[dest] = list + return res +} diff --git a/transport/internet/ws/connection_test.go b/transport/internet/ws/connection_test.go new file mode 100644 index 00000000..5e594f4e --- /dev/null +++ b/transport/internet/ws/connection_test.go @@ -0,0 +1,19 @@ +package ws_test + +import ( + "net" + "testing" + + "github.com/v2ray/v2ray-core/testing/assert" + . "github.com/v2ray/v2ray-core/transport/internet/tcp" +) + +func TestRawConnection(t *testing.T) { + assert := assert.On(t) + + rawConn := RawConnection{net.TCPConn{}} + assert.Bool(rawConn.Reusable()).IsFalse() + + rawConn.SetReusable(true) + assert.Bool(rawConn.Reusable()).IsFalse() +} diff --git a/transport/internet/ws/dialer.go b/transport/internet/ws/dialer.go new file mode 100644 index 00000000..bb1e88b0 --- /dev/null +++ b/transport/internet/ws/dialer.go @@ -0,0 +1,119 @@ +package ws + +import ( + "fmt" + "net" + + "github.com/gorilla/websocket" + "github.com/v2ray/v2ray-core/common/log" + v2net "github.com/v2ray/v2ray-core/common/net" + "github.com/v2ray/v2ray-core/transport/internet" +) + +var ( + globalCache = NewConnectionCache() +) + +func Dial(src v2net.Address, dest v2net.Destination) (internet.Connection, error) { + log.Info("Dailing WS to ", dest) + if src == nil { + src = v2net.AnyIP + } + id := src.String() + "-" + dest.NetAddr() + var conn *wsconn + if dest.IsTCP() && effectiveConfig.ConnectionReuse { + connt := globalCache.Get(id) + if connt != nil { + conn = connt.(*wsconn) + } + } + if conn == nil { + var err error + conn, err = wsDial(src, dest) + if err != nil { + log.Warning("WS Dial failed:" + err.Error()) + return nil, err + } + } + return NewConnection(id, conn, globalCache), nil +} + +func init() { + internet.WSDialer = Dial +} + +func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { + //internet.DialToDest(src, dest) + commonDial := func(network, addr string) (net.Conn, error) { + return internet.DialToDest(src, dest) + } + + dialer := websocket.Dialer{NetDial: commonDial, ReadBufferSize: 65536, WriteBufferSize: 65536} + + effpto := func(dst v2net.Destination) string { + + if effectiveConfig.Pto != "" { + return effectiveConfig.Pto + } + + switch dst.Port().Value() { + /* + Since the value is not given explicitly, + We are guessing it now. + + HTTP Port: + 80 + 8080 + 8880 + 2052 + 2082 + 2086 + 2095 + + HTTPS Port: + 443 + 2053 + 2083 + 2087 + 2096 + 8443 + + if the port you are using is not well-known, + specify it to avoid this process. + + We will re return "CRASH"turn "unknown" if we can't guess it, cause Dial to fail. + */ + case 80: + case 8080: + case 8880: + case 2052: + case 2082: + case 2086: + case 2095: + return "ws" + case 443: + case 2053: + case 2083: + case 2087: + case 2096: + case 8443: + return "wss" + default: + return "unknown" + } + panic("Runtime unstable. Please report this bug to developers.") + }(dest) + + uri := func(dst v2net.Destination, pto string, path string) string { + return fmt.Sprintf("%v://%v:%v/%v", pto, dst.NetAddr(), dst.Port(), path) + }(dest, effpto, effectiveConfig.Path) + conn, _, err := dialer.Dial(uri, nil) + if err != nil { + return nil, err + } + return func() internet.Connection { + connv2ray := &wsconn{wsc: conn, connClosing: false} + connv2ray.setup() + return connv2ray + }().(*wsconn), nil +} diff --git a/transport/internet/ws/hub.go b/transport/internet/ws/hub.go new file mode 100644 index 00000000..cc4563c2 --- /dev/null +++ b/transport/internet/ws/hub.go @@ -0,0 +1,162 @@ +package ws + +import ( + "errors" + "net" + "net/http" + "strconv" + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/v2ray/v2ray-core/common/log" + v2net "github.com/v2ray/v2ray-core/common/net" + "github.com/v2ray/v2ray-core/transport/internet" +) + +var ( + ErrClosedListener = errors.New("Listener is closed.") +) + +type ConnectionWithError struct { + conn net.Conn + err error +} + +type WSListener struct { + sync.Mutex + acccepting bool + awaitingConns chan *ConnectionWithError +} + +func ListenWS(address v2net.Address, port v2net.Port) (internet.Listener, error) { + + l := &WSListener{ + acccepting: true, + awaitingConns: make(chan *ConnectionWithError, 32), + } + + err := l.listenws(address, port) + + return l, err +} + +func (wsl *WSListener) listenws(address v2net.Address, port v2net.Port) error { + + http.HandleFunc("/"+effectiveConfig.Path, func(w http.ResponseWriter, r *http.Request) { + log.Warning("WS:WSListener->listenws->(HandleFunc,lambda 2)! Accepting websocket") + con, err := wsl.converttovws(w, r) + if err != nil { + log.Warning("WS:WSListener->listenws->(HandleFunc,lambda 2)!" + err.Error()) + return + } + + select { + case wsl.awaitingConns <- &ConnectionWithError{ + conn: con, + err: err, + }: + log.Warning("WS:WSListener->listenws->(HandleFunc,lambda 2)! transferd websocket") + default: + if con != nil { + con.Close() + } + } + //con.retloc.Wait() + return + + }) + + errchan := make(chan error) + + go func() { + err := http.ListenAndServe(address.String()+":"+strconv.Itoa(int(port.Value())), nil) + errchan <- err + return + }() + + var err error + select { + case err = <-errchan: + case <-time.After(time.Second * 2): + //Should this listen fail after 2 sec, it could gone untracked. + } + + if err != nil { + log.Error("WS:WSListener->listenws->ListenAndServe!" + err.Error()) + } + + return err + +} + +func (wsl *WSListener) converttovws(w http.ResponseWriter, r *http.Request) (*wsconn, error) { + var upgrader = websocket.Upgrader{ + ReadBufferSize: 65536, + WriteBufferSize: 65536, + } + conn, err := upgrader.Upgrade(w, r, nil) + + if err != nil { + return nil, err + } + + wrapedConn := &wsconn{wsc: conn, connClosing: false} + wrapedConn.setup() + return wrapedConn, nil +} + +func (this *WSListener) Accept() (internet.Connection, error) { + for this.acccepting { + select { + case connErr, open := <-this.awaitingConns: + log.Info("WSListener: conn accepted") + if !open { + return nil, ErrClosedListener + } + if connErr.err != nil { + return nil, connErr.err + } + return NewConnection("", connErr.conn.(*wsconn), this), nil + case <-time.After(time.Second * 2): + } + } + return nil, ErrClosedListener +} + +func (this *WSListener) Recycle(dest string, conn *wsconn) { + this.Lock() + defer this.Unlock() + if !this.acccepting { + return + } + select { + case this.awaitingConns <- &ConnectionWithError{conn: conn}: + default: + conn.Close() + } +} + +func (this *WSListener) Addr() net.Addr { + return nil +} + +func (this *WSListener) Close() error { + this.Lock() + defer this.Unlock() + this.acccepting = false + + log.Warning("WSListener: Yet to support close listening HTTP service") + + close(this.awaitingConns) + for connErr := range this.awaitingConns { + if connErr.conn != nil { + go connErr.conn.Close() + } + } + return nil +} + +func init() { + internet.WSListenFunc = ListenWS +} diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go new file mode 100644 index 00000000..3215b731 --- /dev/null +++ b/transport/internet/ws/wsconn.go @@ -0,0 +1,199 @@ +package ws + +import ( + "bufio" + "io" + "net" + "sync" + "time" + + "github.com/v2ray/v2ray-core/common/log" + + "github.com/gorilla/websocket" +) + +type wsconn struct { + wsc *websocket.Conn + readBuffer *bufio.Reader + connClosing bool + reusable bool + retloc *sync.Cond + rlock *sync.Mutex + wlock *sync.Mutex +} + +func (ws *wsconn) Read(b []byte) (n int, err error) { + + //defer ws.rlock.Unlock() + //ws.checkifRWAfterClosing() + if ws.connClosing { + + return 0, io.EOF + } + getNewBuffer := func() error { + _, r, err := ws.wsc.NextReader() + if err != nil { + log.Warning("WS transport: ws connection NewFrameReader return " + err.Error()) + ws.connClosing = true + ws.Close() + return err + } + ws.readBuffer = bufio.NewReader(r) + return nil + } + + readNext := func(b []byte) (n int, err error) { + if ws.readBuffer == nil { + err = getNewBuffer() + if err != nil { + //ws.Close() + return 0, err + } + } + + n, err = ws.readBuffer.Read(b) + + if err == nil { + return n, err + } + + if err == io.EOF { + ws.readBuffer = nil + if n == 0 { + return ws.Read(b) + } + return n, nil + } + //ws.Close() + return n, err + + } + n, err = readNext(b) + + return n, err + +} + +func (ws *wsconn) Write(b []byte) (n int, err error) { + + //defer + //ws.checkifRWAfterClosing() + if ws.connClosing { + + return 0, io.EOF + } + writeWs := func(b []byte) (n int, err error) { + wr, err := ws.wsc.NextWriter(websocket.BinaryMessage) + if err != nil { + log.Warning("WS transport: ws connection NewFrameReader return " + err.Error()) + ws.connClosing = true + ws.Close() + return 0, err + } + n, err = wr.Write(b) + if err != nil { + //ws.Close() + return 0, err + } + err = wr.Close() + if err != nil { + //ws.Close() + return 0, err + } + return n, err + } + n, err = writeWs(b) + return n, err +} +func (ws *wsconn) Close() error { + ws.connClosing = true + err := ws.wsc.Close() + ws.retloc.Broadcast() + return err +} +func (ws *wsconn) LocalAddr() net.Addr { + return ws.wsc.LocalAddr() +} +func (ws *wsconn) RemoteAddr() net.Addr { + return ws.wsc.RemoteAddr() +} +func (ws *wsconn) SetDeadline(t time.Time) error { + return func() error { + errr := ws.SetReadDeadline(t) + errw := ws.SetWriteDeadline(t) + if errr == nil || errw == nil { + return nil + } + if errr != nil { + return errr + } + + return errw + }() +} +func (ws *wsconn) SetReadDeadline(t time.Time) error { + return ws.wsc.SetReadDeadline(t) +} +func (ws *wsconn) SetWriteDeadline(t time.Time) error { + return ws.wsc.SetWriteDeadline(t) +} + +func (ws *wsconn) checkifRWAfterClosing() { + if ws.connClosing { + log.Error("WS transport: Read or Write After Conn have been marked closing, this can be dangerous.") + //panic("WS transport: Read or Write After Conn have been marked closing. Please report this crash to developer.") + } +} + +func (ws *wsconn) setup() { + ws.connClosing = false + + ws.rlock = &sync.Mutex{} + ws.wlock = &sync.Mutex{} + + initConnectedCond := func() { + rsl := &sync.Mutex{} + ws.retloc = sync.NewCond(rsl) + } + + initConnectedCond() + //ws.pingPong() +} + +func (ws *wsconn) Reusable() bool { + return ws.reusable && !ws.connClosing +} + +func (ws *wsconn) SetReusable(reusable bool) { + if !effectiveConfig.ConnectionReuse { + return + } + ws.reusable = reusable +} + +func (ws *wsconn) pingPong() { + pongRcv := make(chan int, 0) + ws.wsc.SetPongHandler(func(data string) error { + pongRcv <- 0 + return nil + }) + + go func() { + for !ws.connClosing { + ws.wsc.WriteMessage(websocket.PingMessage, nil) + tick := time.NewTicker(time.Second * 3) + + select { + case <-pongRcv: + break + case <-tick.C: + ws.Close() + } + <-tick.C + tick.Stop() + } + + return + }() + +} From 16b529093d52878a4db2396c4b0f763f4aa8762c Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 21:56:51 +0800 Subject: [PATCH 11/47] Support Listen WebSocket with TLS --- transport/internet/ws/config_json.go | 4 ++++ transport/internet/ws/hub.go | 12 +++++++++++- 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/transport/internet/ws/config_json.go b/transport/internet/ws/config_json.go index 843e549b..dbb04cd5 100644 --- a/transport/internet/ws/config_json.go +++ b/transport/internet/ws/config_json.go @@ -9,6 +9,8 @@ func (this *Config) UnmarshalJSON(data []byte) error { ConnectionReuse bool `json:"connectionReuse"` Path string `json:"Path"` Pto string `json:"Pto"` + Cert string `json:"Cert"` + PrivKey string `json:"PrivKet"` } jsonConfig := &JsonConfig{ ConnectionReuse: true, @@ -21,5 +23,7 @@ func (this *Config) UnmarshalJSON(data []byte) error { this.ConnectionReuse = jsonConfig.ConnectionReuse this.Path = jsonConfig.Path this.Pto = jsonConfig.Pto + this.PrivKey = jsonConfig.PrivKey + this.Cert = jsonConfig.Cert return nil } diff --git a/transport/internet/ws/hub.go b/transport/internet/ws/hub.go index cc4563c2..412fb895 100644 --- a/transport/internet/ws/hub.go +++ b/transport/internet/ws/hub.go @@ -69,8 +69,18 @@ func (wsl *WSListener) listenws(address v2net.Address, port v2net.Port) error { errchan := make(chan error) + listenerfunc := func() error { + return http.ListenAndServe(address.String()+":"+strconv.Itoa(int(port.Value())), nil) + } + + if effectiveConfig.Pto == "wss" { + listenerfunc = func() error { + return http.ListenAndServeTLS(address.String()+":"+strconv.Itoa(int(port.Value())), effectiveConfig.Cert, effectiveConfig.PrivKey, nil) + } + } + go func() { - err := http.ListenAndServe(address.String()+":"+strconv.Itoa(int(port.Value())), nil) + err := listenerfunc() errchan <- err return }() From 427bed5ddbda78364a066081a7c444e7187eca03 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 22:07:47 +0800 Subject: [PATCH 12/47] typo~ --- transport/internet/ws/config_json.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/transport/internet/ws/config_json.go b/transport/internet/ws/config_json.go index dbb04cd5..eb479479 100644 --- a/transport/internet/ws/config_json.go +++ b/transport/internet/ws/config_json.go @@ -10,7 +10,7 @@ func (this *Config) UnmarshalJSON(data []byte) error { Path string `json:"Path"` Pto string `json:"Pto"` Cert string `json:"Cert"` - PrivKey string `json:"PrivKet"` + PrivKey string `json:"PrivKey"` } jsonConfig := &JsonConfig{ ConnectionReuse: true, From 1fe090dbfd5b48a020f436d92735f718762fdd9d Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sat, 13 Aug 2016 22:50:24 +0800 Subject: [PATCH 13/47] Verfiy Domain correctly --- transport/internet/ws/dialer.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/transport/internet/ws/dialer.go b/transport/internet/ws/dialer.go index bb1e88b0..480ec82e 100644 --- a/transport/internet/ws/dialer.go +++ b/transport/internet/ws/dialer.go @@ -1,6 +1,7 @@ package ws import ( + "crypto/tls" "fmt" "net" @@ -48,7 +49,9 @@ func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { return internet.DialToDest(src, dest) } - dialer := websocket.Dialer{NetDial: commonDial, ReadBufferSize: 65536, WriteBufferSize: 65536} + tlsconf := &tls.Config{ServerName: dest.Address().Domain()} + + dialer := websocket.Dialer{NetDial: commonDial, ReadBufferSize: 65536, WriteBufferSize: 65536, TLSClientConfig: tlsconf} effpto := func(dst v2net.Destination) string { From 3e9e9b520e93356a5db231b330a8b590bca92f7a Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sun, 14 Aug 2016 09:23:24 +0800 Subject: [PATCH 14/47] Remove useless test --- transport/internet/ws/connection_test.go | 19 ------------------- 1 file changed, 19 deletions(-) delete mode 100644 transport/internet/ws/connection_test.go diff --git a/transport/internet/ws/connection_test.go b/transport/internet/ws/connection_test.go deleted file mode 100644 index 5e594f4e..00000000 --- a/transport/internet/ws/connection_test.go +++ /dev/null @@ -1,19 +0,0 @@ -package ws_test - -import ( - "net" - "testing" - - "github.com/v2ray/v2ray-core/testing/assert" - . "github.com/v2ray/v2ray-core/transport/internet/tcp" -) - -func TestRawConnection(t *testing.T) { - assert := assert.On(t) - - rawConn := RawConnection{net.TCPConn{}} - assert.Bool(rawConn.Reusable()).IsFalse() - - rawConn.SetReusable(true) - assert.Bool(rawConn.Reusable()).IsFalse() -} From a0fda39274155ebef9eb440e20672f3706ed95bb Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sun, 14 Aug 2016 11:06:39 +0800 Subject: [PATCH 15/47] Fix crash --- transport/internet/ws/wsconn.go | 35 +++++++++++++++++++++++++++------ 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index 3215b731..9f86d815 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -2,6 +2,7 @@ package ws import ( "bufio" + "fmt" "io" "net" "sync" @@ -23,7 +24,7 @@ type wsconn struct { } func (ws *wsconn) Read(b []byte) (n int, err error) { - + ws.rlock.Lock() //defer ws.rlock.Unlock() //ws.checkifRWAfterClosing() if ws.connClosing { @@ -42,7 +43,19 @@ func (ws *wsconn) Read(b []byte) (n int, err error) { return nil } - readNext := func(b []byte) (n int, err error) { + /*It seems golang's support for recursive in anonymous func it yet to complete. + func1:=func(){ + func1() + } + won't work, failed to compile for it can't find func1. + + Should following work around panic, + readNext could have been called before the actual defination was made, + This is very unlikely. + */ + readNext := func(b []byte) (n int, err error) { panic("Runtime unstable. Please report this bug to developer.") } + + readNext = func(b []byte) (n int, err error) { if ws.readBuffer == nil { err = getNewBuffer() if err != nil { @@ -60,7 +73,7 @@ func (ws *wsconn) Read(b []byte) (n int, err error) { if err == io.EOF { ws.readBuffer = nil if n == 0 { - return ws.Read(b) + return readNext(b) } return n, nil } @@ -69,13 +82,19 @@ func (ws *wsconn) Read(b []byte) (n int, err error) { } n, err = readNext(b) - + ws.rlock.Unlock() return n, err } func (ws *wsconn) Write(b []byte) (n int, err error) { - + ws.wlock.Lock() + defer func() { + if r := recover(); r != nil { + fmt.Println("WS workaround: recover", r) + ws.wlock.Unlock() + } + }() //defer //ws.checkifRWAfterClosing() if ws.connClosing { @@ -103,10 +122,12 @@ func (ws *wsconn) Write(b []byte) (n int, err error) { return n, err } n, err = writeWs(b) + ws.wlock.Unlock() return n, err } func (ws *wsconn) Close() error { ws.connClosing = true + ws.wsc.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add((time.Second * 5))) err := ws.wsc.Close() ws.retloc.Broadcast() return err @@ -157,7 +178,7 @@ func (ws *wsconn) setup() { } initConnectedCond() - //ws.pingPong() + ws.pingPong() } func (ws *wsconn) Reusable() bool { @@ -185,8 +206,10 @@ func (ws *wsconn) pingPong() { select { case <-pongRcv: + //log.Debug("WS:Pong~" + ws.wsc.UnderlyingConn().RemoteAddr().String()) break case <-tick.C: + log.Debug("WS:Closing as ping is not responded~" + ws.wsc.UnderlyingConn().LocalAddr().String() + "-" + ws.wsc.UnderlyingConn().RemoteAddr().String()) ws.Close() } <-tick.C From 49441a50503375fc450358988d8db044d3f8f8b0 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sun, 14 Aug 2016 14:11:51 +0800 Subject: [PATCH 16/47] Docs --- transport/internet/ws/connection.go | 11 ---------- transport/internet/ws/dialer.go | 9 +++++--- transport/internet/ws/hub.go | 1 - transport/internet/ws/ws.go | 34 +++++++++++++++++++++++++++++ transport/internet/ws/wsconn.go | 28 +++++------------------- 5 files changed, 46 insertions(+), 37 deletions(-) create mode 100644 transport/internet/ws/ws.go diff --git a/transport/internet/ws/connection.go b/transport/internet/ws/connection.go index 141fdb32..2ba96df4 100644 --- a/transport/internet/ws/connection.go +++ b/transport/internet/ws/connection.go @@ -4,7 +4,6 @@ import ( "errors" "io" "net" - "reflect" "time" ) @@ -96,15 +95,5 @@ func (this *Connection) SysFd() (int, error) { } func getSysFd(conn net.Conn) (int, error) { - cv := reflect.ValueOf(conn) - switch ce := cv.Elem(); ce.Kind() { - case reflect.Struct: - netfd := ce.FieldByName("conn").FieldByName("fd") - switch fe := netfd.Elem(); fe.Kind() { - case reflect.Struct: - fd := fe.FieldByName("sysfd") - return int(fd.Int()), nil - } - } return 0, ErrInvalidConn } diff --git a/transport/internet/ws/dialer.go b/transport/internet/ws/dialer.go index 480ec82e..6b17dfeb 100644 --- a/transport/internet/ws/dialer.go +++ b/transport/internet/ws/dialer.go @@ -3,6 +3,7 @@ package ws import ( "crypto/tls" "fmt" + "io/ioutil" "net" "github.com/gorilla/websocket" @@ -44,7 +45,6 @@ func init() { } func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { - //internet.DialToDest(src, dest) commonDial := func(network, addr string) (net.Conn, error) { return internet.DialToDest(src, dest) } @@ -84,7 +84,7 @@ func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { if the port you are using is not well-known, specify it to avoid this process. - We will re return "CRASH"turn "unknown" if we can't guess it, cause Dial to fail. + We will return "CRASH"turn "unknown" if we can't guess it, cause Dial to fail. */ case 80: case 8080: @@ -110,8 +110,11 @@ func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { uri := func(dst v2net.Destination, pto string, path string) string { return fmt.Sprintf("%v://%v:%v/%v", pto, dst.NetAddr(), dst.Port(), path) }(dest, effpto, effectiveConfig.Path) - conn, _, err := dialer.Dial(uri, nil) + + conn, resp, err := dialer.Dial(uri, nil) if err != nil { + reason, reasonerr := ioutil.ReadAll(resp.Body) + log.Info(string(reason), reasonerr) return nil, err } return func() internet.Connection { diff --git a/transport/internet/ws/hub.go b/transport/internet/ws/hub.go index 412fb895..16fd3df7 100644 --- a/transport/internet/ws/hub.go +++ b/transport/internet/ws/hub.go @@ -62,7 +62,6 @@ func (wsl *WSListener) listenws(address v2net.Address, port v2net.Port) error { con.Close() } } - //con.retloc.Wait() return }) diff --git a/transport/internet/ws/ws.go b/transport/internet/ws/ws.go new file mode 100644 index 00000000..156e7f3c --- /dev/null +++ b/transport/internet/ws/ws.go @@ -0,0 +1,34 @@ +/*Package ws implements Websocket transport + +Websocket transport implements a HTTP(S) compliable, surveillance proof transport method with plausible deniability. + +To configure such a listener, set streamSettings to be ws. A http(s) listener will be listening at the port you have configured. + +There is additional configure can be made at transport configure. + +"wsSettings":{ + "Path":"ws", // the path our ws handler bind + "Pto": "wss/ws", // the transport protocol we are using ws or wss(listen ws with tls) + "Cert":"cert.pem", // if you have configured to use wss, configure your cert here + "PrivKey":"priv.pem" //if you have configured to use wss, configure your privatekey here + } + + +To configure such a Dialer, set streamSettings to be ws. + +There is additional configure can be made at transport configure. + +"wsSettings":{ + "Path":"ws", // the path our ws handler bind + "Pto": "wss/ws", // the transport protocol we are using ws or wss(listen ws with tls) + } + +It is worth noting that accepting a non-valid cert is not supported as a self-signed or invalid cert can be a sign of a website that is not correctly configured and lead to additional investigation. + + +This transport was disscussed at +https://github.com/v2ray/v2ray-core/issues/224 +https://trello.com/c/3uCCeBkC/8-add-websocket-transport + +*/ +package ws diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index 9f86d815..fa64ec88 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -18,7 +18,6 @@ type wsconn struct { readBuffer *bufio.Reader connClosing bool reusable bool - retloc *sync.Cond rlock *sync.Mutex wlock *sync.Mutex } @@ -59,7 +58,6 @@ func (ws *wsconn) Read(b []byte) (n int, err error) { if ws.readBuffer == nil { err = getNewBuffer() if err != nil { - //ws.Close() return 0, err } } @@ -77,7 +75,6 @@ func (ws *wsconn) Read(b []byte) (n int, err error) { } return n, nil } - //ws.Close() return n, err } @@ -89,14 +86,18 @@ func (ws *wsconn) Read(b []byte) (n int, err error) { func (ws *wsconn) Write(b []byte) (n int, err error) { ws.wlock.Lock() + /* + process can crash as websocket report "concurrent write to websocket connection" + even if lock is persent. + + It is yet to know how to prevent this but a workaround is the only choice. + */ defer func() { if r := recover(); r != nil { fmt.Println("WS workaround: recover", r) ws.wlock.Unlock() } }() - //defer - //ws.checkifRWAfterClosing() if ws.connClosing { return 0, io.EOF @@ -111,12 +112,10 @@ func (ws *wsconn) Write(b []byte) (n int, err error) { } n, err = wr.Write(b) if err != nil { - //ws.Close() return 0, err } err = wr.Close() if err != nil { - //ws.Close() return 0, err } return n, err @@ -129,7 +128,6 @@ func (ws *wsconn) Close() error { ws.connClosing = true ws.wsc.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add((time.Second * 5))) err := ws.wsc.Close() - ws.retloc.Broadcast() return err } func (ws *wsconn) LocalAddr() net.Addr { @@ -159,25 +157,12 @@ func (ws *wsconn) SetWriteDeadline(t time.Time) error { return ws.wsc.SetWriteDeadline(t) } -func (ws *wsconn) checkifRWAfterClosing() { - if ws.connClosing { - log.Error("WS transport: Read or Write After Conn have been marked closing, this can be dangerous.") - //panic("WS transport: Read or Write After Conn have been marked closing. Please report this crash to developer.") - } -} - func (ws *wsconn) setup() { ws.connClosing = false ws.rlock = &sync.Mutex{} ws.wlock = &sync.Mutex{} - initConnectedCond := func() { - rsl := &sync.Mutex{} - ws.retloc = sync.NewCond(rsl) - } - - initConnectedCond() ws.pingPong() } @@ -206,7 +191,6 @@ func (ws *wsconn) pingPong() { select { case <-pongRcv: - //log.Debug("WS:Pong~" + ws.wsc.UnderlyingConn().RemoteAddr().String()) break case <-tick.C: log.Debug("WS:Closing as ping is not responded~" + ws.wsc.UnderlyingConn().LocalAddr().String() + "-" + ws.wsc.UnderlyingConn().RemoteAddr().String()) From 5e576e11bba14997fd1ecd5245a1f510079195fc Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sun, 14 Aug 2016 20:40:59 +0800 Subject: [PATCH 17/47] fix crash --- transport/internet/ws/wsconn.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index fa64ec88..e23fc947 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -126,7 +126,9 @@ func (ws *wsconn) Write(b []byte) (n int, err error) { } func (ws *wsconn) Close() error { ws.connClosing = true + ws.wlock.Lock() ws.wsc.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add((time.Second * 5))) + ws.wlock.Unlock() err := ws.wsc.Close() return err } @@ -186,7 +188,9 @@ func (ws *wsconn) pingPong() { go func() { for !ws.connClosing { + ws.wlock.Lock() ws.wsc.WriteMessage(websocket.PingMessage, nil) + ws.wlock.Unlock() tick := time.NewTicker(time.Second * 3) select { From fed2933043ab793ab7b4e88d9c49d5d564fe9d0d Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Sun, 14 Aug 2016 20:41:26 +0800 Subject: [PATCH 18/47] fix crash --- transport/internet/ws/dialer.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/transport/internet/ws/dialer.go b/transport/internet/ws/dialer.go index 6b17dfeb..63fc0efd 100644 --- a/transport/internet/ws/dialer.go +++ b/transport/internet/ws/dialer.go @@ -113,8 +113,10 @@ func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { conn, resp, err := dialer.Dial(uri, nil) if err != nil { - reason, reasonerr := ioutil.ReadAll(resp.Body) - log.Info(string(reason), reasonerr) + if resp != nil { + reason, reasonerr := ioutil.ReadAll(resp.Body) + log.Info(string(reason), reasonerr) + } return nil, err } return func() internet.Connection { From d06c75f92671d9d5861c099dccbe235bd4fc24e2 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 10:25:20 +0800 Subject: [PATCH 19/47] Doc sync to code change --- transport/internet/ws/wsconn.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index e23fc947..a2904d29 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -24,8 +24,6 @@ type wsconn struct { func (ws *wsconn) Read(b []byte) (n int, err error) { ws.rlock.Lock() - //defer ws.rlock.Unlock() - //ws.checkifRWAfterClosing() if ws.connClosing { return 0, io.EOF @@ -90,7 +88,7 @@ func (ws *wsconn) Write(b []byte) (n int, err error) { process can crash as websocket report "concurrent write to websocket connection" even if lock is persent. - It is yet to know how to prevent this but a workaround is the only choice. + This problem should have been resolved. */ defer func() { if r := recover(); r != nil { From d250013c994ec532c891dad9360751bb3e6dadab Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 11:07:29 +0800 Subject: [PATCH 20/47] Increase conn keep-alive time --- transport/internet/ws/connection_cache.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/transport/internet/ws/connection_cache.go b/transport/internet/ws/connection_cache.go index e0743d75..e95f939a 100644 --- a/transport/internet/ws/connection_cache.go +++ b/transport/internet/ws/connection_cache.go @@ -5,6 +5,7 @@ import ( "sync" "time" + "github.com/v2ray/v2ray-core/common/log" "github.com/v2ray/v2ray-core/common/signal" ) @@ -66,7 +67,7 @@ func (this *ConnectionCache) Recycle(dest string, conn *wsconn) { aconn := &AwaitingConnection{ conn: conn, - expire: time.Now().Add(time.Second * 4), + expire: time.Now().Add(time.Second * 180), } var list []*AwaitingConnection @@ -108,5 +109,6 @@ func (this *ConnectionCache) Get(dest string) net.Conn { res := list[firstValid].conn list = list[firstValid+1:] this.cache[dest] = list + log.Debug("WS:Conn Cache used.") return res } From 2bf9edfe7d59435848f1b1fec766bab043ca36c9 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 12:23:15 +0800 Subject: [PATCH 21/47] Prevent unnesessary log --- transport/internet/ws/wsconn.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index a2904d29..08016da6 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -195,7 +195,9 @@ func (ws *wsconn) pingPong() { case <-pongRcv: break case <-tick.C: - log.Debug("WS:Closing as ping is not responded~" + ws.wsc.UnderlyingConn().LocalAddr().String() + "-" + ws.wsc.UnderlyingConn().RemoteAddr().String()) + if !ws.connClosing { + log.Debug("WS:Closing as ping is not responded~" + ws.wsc.UnderlyingConn().LocalAddr().String() + "-" + ws.wsc.UnderlyingConn().RemoteAddr().String()) + } ws.Close() } <-tick.C From 00aeb29e46e684f7cb7654513cdb051224c5aeee Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 12:31:55 +0800 Subject: [PATCH 22/47] typo --- transport/internet/ws/wsconn.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index 08016da6..e242eef2 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -40,13 +40,13 @@ func (ws *wsconn) Read(b []byte) (n int, err error) { return nil } - /*It seems golang's support for recursive in anonymous func it yet to complete. + /*It seems golang's support for recursive in anonymous func is yet to complete. func1:=func(){ func1() } won't work, failed to compile for it can't find func1. - Should following work around panic, + Should following workaround panic, readNext could have been called before the actual defination was made, This is very unlikely. */ From ee22082fb4c5dd85f97b4a5b85b498e04448e343 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 14:30:38 +0800 Subject: [PATCH 23/47] Fix host bug --- transport/internet/ws/dialer.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/transport/internet/ws/dialer.go b/transport/internet/ws/dialer.go index 63fc0efd..383dee67 100644 --- a/transport/internet/ws/dialer.go +++ b/transport/internet/ws/dialer.go @@ -108,7 +108,7 @@ func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { }(dest) uri := func(dst v2net.Destination, pto string, path string) string { - return fmt.Sprintf("%v://%v:%v/%v", pto, dst.NetAddr(), dst.Port(), path) + return fmt.Sprintf("%v://%v/%v", pto, dst.NetAddr(), path) }(dest, effpto, effectiveConfig.Path) conn, resp, err := dialer.Dial(uri, nil) From 7b3cf631c2056a201653704f57ce8e546084fc78 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 17:40:56 +0800 Subject: [PATCH 24/47] Removed inaccessable doc --- transport/internet/ws/ws.go | 1 - 1 file changed, 1 deletion(-) diff --git a/transport/internet/ws/ws.go b/transport/internet/ws/ws.go index 156e7f3c..23c2b8eb 100644 --- a/transport/internet/ws/ws.go +++ b/transport/internet/ws/ws.go @@ -28,7 +28,6 @@ It is worth noting that accepting a non-valid cert is not supported as a self-si This transport was disscussed at https://github.com/v2ray/v2ray-core/issues/224 -https://trello.com/c/3uCCeBkC/8-add-websocket-transport */ package ws From de28198234aee60b85ab504ba9cc7b234cbd2013 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 17:47:15 +0800 Subject: [PATCH 25/47] Guess protocol --- transport/internet/ws/dialer.go | 118 ++++++++++++++++++-------------- 1 file changed, 65 insertions(+), 53 deletions(-) diff --git a/transport/internet/ws/dialer.go b/transport/internet/ws/dialer.go index 383dee67..70e5ab81 100644 --- a/transport/internet/ws/dialer.go +++ b/transport/internet/ws/dialer.go @@ -53,59 +53,7 @@ func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { dialer := websocket.Dialer{NetDial: commonDial, ReadBufferSize: 65536, WriteBufferSize: 65536, TLSClientConfig: tlsconf} - effpto := func(dst v2net.Destination) string { - - if effectiveConfig.Pto != "" { - return effectiveConfig.Pto - } - - switch dst.Port().Value() { - /* - Since the value is not given explicitly, - We are guessing it now. - - HTTP Port: - 80 - 8080 - 8880 - 2052 - 2082 - 2086 - 2095 - - HTTPS Port: - 443 - 2053 - 2083 - 2087 - 2096 - 8443 - - if the port you are using is not well-known, - specify it to avoid this process. - - We will return "CRASH"turn "unknown" if we can't guess it, cause Dial to fail. - */ - case 80: - case 8080: - case 8880: - case 2052: - case 2082: - case 2086: - case 2095: - return "ws" - case 443: - case 2053: - case 2083: - case 2087: - case 2096: - case 8443: - return "wss" - default: - return "unknown" - } - panic("Runtime unstable. Please report this bug to developers.") - }(dest) + effpto := calcPto(dest) uri := func(dst v2net.Destination, pto string, path string) string { return fmt.Sprintf("%v://%v/%v", pto, dst.NetAddr(), path) @@ -125,3 +73,67 @@ func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { return connv2ray }().(*wsconn), nil } + +func calcPto(dst v2net.Destination) string { + + if effectiveConfig.Pto != "" { + return effectiveConfig.Pto + } + + switch dst.Port().Value() { + /* + Since the value is not given explicitly, + We are guessing it now. + + HTTP Port: + 80 + 8080 + 8880 + 2052 + 2082 + 2086 + 2095 + + HTTPS Port: + 443 + 2053 + 2083 + 2087 + 2096 + 8443 + + if the port you are using is not well-known, + specify it to avoid this process. + + We will return "CRASH"turn "unknown" if we can't guess it, cause Dial to fail. + */ + case 80: + fallthrough + case 8080: + fallthrough + case 8880: + fallthrough + case 2052: + fallthrough + case 2082: + fallthrough + case 2086: + fallthrough + case 2095: + return "ws" + case 443: + fallthrough + case 2053: + fallthrough + case 2083: + fallthrough + case 2087: + fallthrough + case 2096: + fallthrough + case 8443: + return "wss" + default: + return "unknown" + } +} From 4853a1a6d35c218bcb9e97bad5f01dacbbe9bf14 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 17:48:30 +0800 Subject: [PATCH 26/47] Increase ping time --- transport/internet/ws/wsconn.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index e242eef2..1f59b4cd 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -189,7 +189,7 @@ func (ws *wsconn) pingPong() { ws.wlock.Lock() ws.wsc.WriteMessage(websocket.PingMessage, nil) ws.wlock.Unlock() - tick := time.NewTicker(time.Second * 3) + tick := time.NewTicker(time.Second * 30) select { case <-pongRcv: From 863237b5184b8f721799b963f00f44862a02b209 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 17:53:11 +0800 Subject: [PATCH 27/47] Justify lock use --- transport/internet/ws/wsconn.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index 1f59b4cd..9c5658c1 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -160,6 +160,14 @@ func (ws *wsconn) SetWriteDeadline(t time.Time) error { func (ws *wsconn) setup() { ws.connClosing = false + /* + https://godoc.org/github.com/gorilla/websocket#Conn.NextReader + https://godoc.org/github.com/gorilla/websocket#Conn.NextWriter + + Both Read and write access are exclusive. + And in both case it will need a lock. + + */ ws.rlock = &sync.Mutex{} ws.wlock = &sync.Mutex{} From a36582b29451fe37aa659ac676a2fd3fa96abbf8 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 18:16:54 +0800 Subject: [PATCH 28/47] Added stopable listener --- transport/internet/ws/stopablehttplistener.go | 62 +++++++++++++++++++ 1 file changed, 62 insertions(+) create mode 100644 transport/internet/ws/stopablehttplistener.go diff --git a/transport/internet/ws/stopablehttplistener.go b/transport/internet/ws/stopablehttplistener.go new file mode 100644 index 00000000..832b04e2 --- /dev/null +++ b/transport/internet/ws/stopablehttplistener.go @@ -0,0 +1,62 @@ +package ws + +import ( + "errors" + "net" + "time" +) + +type StoppableListener struct { + *net.TCPListener //Wrapped listener + stop chan int //Channel used only to indicate listener should shutdown +} + +func NewStoppableListener(l net.Listener) (*StoppableListener, error) { + tcpL, ok := l.(*net.TCPListener) + + if !ok { + return nil, errors.New("Cannot wrap listener") + } + + retval := &StoppableListener{} + retval.TCPListener = tcpL + retval.stop = make(chan int) + + return retval, nil +} + +var StoppedError = errors.New("Listener stopped") + +func (sl *StoppableListener) Accept() (net.Conn, error) { + + for { + //Wait up to one second for a new connection + sl.SetDeadline(time.Now().Add(time.Second)) + + newConn, err := sl.TCPListener.Accept() + + //Check for the channel being closed + select { + case <-sl.stop: + return nil, StoppedError + default: + //If the channel is still open, continue as normal + } + + if err != nil { + netErr, ok := err.(net.Error) + + //If this is a timeout, then continue to wait for + //new connections + if ok && netErr.Timeout() && netErr.Temporary() { + continue + } + } + + return newConn, err + } +} + +func (sl *StoppableListener) Stop() { + close(sl.stop) +} From c8cd1320c74f9e7f42767f44b177da18edae301c Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 18:17:29 +0800 Subject: [PATCH 29/47] Use stopable listener for HTTP --- transport/internet/ws/hub.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/transport/internet/ws/hub.go b/transport/internet/ws/hub.go index 16fd3df7..a85e320e 100644 --- a/transport/internet/ws/hub.go +++ b/transport/internet/ws/hub.go @@ -27,6 +27,7 @@ type WSListener struct { sync.Mutex acccepting bool awaitingConns chan *ConnectionWithError + listener *StoppableListener } func ListenWS(address v2net.Address, port v2net.Port) (internet.Listener, error) { @@ -69,7 +70,15 @@ func (wsl *WSListener) listenws(address v2net.Address, port v2net.Port) error { errchan := make(chan error) listenerfunc := func() error { - return http.ListenAndServe(address.String()+":"+strconv.Itoa(int(port.Value())), nil) + ol, err := net.Listen("tcp", address.String()+":"+strconv.Itoa(int(port.Value()))) + if err != nil { + return err + } + wsl.listener, err = NewStoppableListener(ol) + if err != nil { + return err + } + return http.Serve(wsl.listener, nil) } if effectiveConfig.Pto == "wss" { From 062d05fdcfd5d482f2ede5cd4a7af318a5836304 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 18:43:17 +0800 Subject: [PATCH 30/47] Added Stoppable listener for TLS listener --- transport/internet/ws/hub.go | 7 ++++++- transport/internet/ws/stopabletlslistener.go | 18 ++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) create mode 100644 transport/internet/ws/stopabletlslistener.go diff --git a/transport/internet/ws/hub.go b/transport/internet/ws/hub.go index a85e320e..0e81edc1 100644 --- a/transport/internet/ws/hub.go +++ b/transport/internet/ws/hub.go @@ -83,7 +83,12 @@ func (wsl *WSListener) listenws(address v2net.Address, port v2net.Port) error { if effectiveConfig.Pto == "wss" { listenerfunc = func() error { - return http.ListenAndServeTLS(address.String()+":"+strconv.Itoa(int(port.Value())), effectiveConfig.Cert, effectiveConfig.PrivKey, nil) + var err error + wsl.listener, err = getstopableTLSlistener(effectiveConfig.Cert, effectiveConfig.PrivKey, address.String()+":"+strconv.Itoa(int(port.Value()))) + if err != nil { + return err + } + return http.Serve(wsl.listener, nil) } } diff --git a/transport/internet/ws/stopabletlslistener.go b/transport/internet/ws/stopabletlslistener.go new file mode 100644 index 00000000..58c54fd2 --- /dev/null +++ b/transport/internet/ws/stopabletlslistener.go @@ -0,0 +1,18 @@ +package ws + +import "crypto/tls" + +func getstopableTLSlistener(cert, key, listenaddr string) (*StoppableListener, error) { + cer, err := tls.LoadX509KeyPair(cert, key) + if err != nil { + return nil, err + } + + config := &tls.Config{Certificates: []tls.Certificate{cer}} + ln, err := tls.Listen("tcp", listenaddr, config) + if err != nil { + return nil, err + } + lns, err := NewStoppableListener(ln) + return lns, err +} From 273c8f6c576ca1c1a7df3e78d1d9c64b242e8157 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 18:47:44 +0800 Subject: [PATCH 31/47] Support Stop listener --- transport/internet/ws/hub.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/transport/internet/ws/hub.go b/transport/internet/ws/hub.go index 0e81edc1..9ac54de3 100644 --- a/transport/internet/ws/hub.go +++ b/transport/internet/ws/hub.go @@ -169,7 +169,7 @@ func (this *WSListener) Close() error { defer this.Unlock() this.acccepting = false - log.Warning("WSListener: Yet to support close listening HTTP service") + this.listener.Stop() close(this.awaitingConns) for connErr := range this.awaitingConns { From bfb4e0d9b00014b7efa72b142b7b597715b6f626 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 18:56:33 +0800 Subject: [PATCH 32/47] Apply coding style --- transport/internet/ws/wsconn.go | 82 ++++++++++++++++----------------- 1 file changed, 39 insertions(+), 43 deletions(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index 9c5658c1..e43b27ce 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -24,60 +24,56 @@ type wsconn struct { func (ws *wsconn) Read(b []byte) (n int, err error) { ws.rlock.Lock() + n, err = ws.read(b) + ws.rlock.Unlock() + return n, err + +} + +func (ws *wsconn) read(b []byte) (n int, err error) { + if ws.connClosing { return 0, io.EOF } - getNewBuffer := func() error { - _, r, err := ws.wsc.NextReader() + + n, err = ws.readNext(b) + return n, err +} + +func (ws *wsconn) getNewReadBuffer() error { + _, r, err := ws.wsc.NextReader() + if err != nil { + log.Warning("WS transport: ws connection NewFrameReader return " + err.Error()) + ws.connClosing = true + ws.Close() + return err + } + ws.readBuffer = bufio.NewReader(r) + return nil +} + +func (ws *wsconn) readNext(b []byte) (n int, err error) { + if ws.readBuffer == nil { + err = ws.getNewReadBuffer() if err != nil { - log.Warning("WS transport: ws connection NewFrameReader return " + err.Error()) - ws.connClosing = true - ws.Close() - return err + return 0, err } - ws.readBuffer = bufio.NewReader(r) - return nil } - /*It seems golang's support for recursive in anonymous func is yet to complete. - func1:=func(){ - func1() - } - won't work, failed to compile for it can't find func1. + n, err = ws.readBuffer.Read(b) - Should following workaround panic, - readNext could have been called before the actual defination was made, - This is very unlikely. - */ - readNext := func(b []byte) (n int, err error) { panic("Runtime unstable. Please report this bug to developer.") } - - readNext = func(b []byte) (n int, err error) { - if ws.readBuffer == nil { - err = getNewBuffer() - if err != nil { - return 0, err - } - } - - n, err = ws.readBuffer.Read(b) - - if err == nil { - return n, err - } - - if err == io.EOF { - ws.readBuffer = nil - if n == 0 { - return readNext(b) - } - return n, nil - } + if err == nil { return n, err - } - n, err = readNext(b) - ws.rlock.Unlock() + + if err == io.EOF { + ws.readBuffer = nil + if n == 0 { + return ws.readNext(b) + } + return n, nil + } return n, err } From a5ae6a0f4ebc2f6b2699fbaba4cd45727ec83e09 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 18:57:39 +0800 Subject: [PATCH 33/47] Remove workaround --- transport/internet/ws/wsconn.go | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index e43b27ce..89173015 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -2,7 +2,6 @@ package ws import ( "bufio" - "fmt" "io" "net" "sync" @@ -80,18 +79,7 @@ func (ws *wsconn) readNext(b []byte) (n int, err error) { func (ws *wsconn) Write(b []byte) (n int, err error) { ws.wlock.Lock() - /* - process can crash as websocket report "concurrent write to websocket connection" - even if lock is persent. - This problem should have been resolved. - */ - defer func() { - if r := recover(); r != nil { - fmt.Println("WS workaround: recover", r) - ws.wlock.Unlock() - } - }() if ws.connClosing { return 0, io.EOF From 4c093a210738dc488b277d98022fc5057f1cec43 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 18:59:14 +0800 Subject: [PATCH 34/47] Apply coding style --- transport/internet/ws/wsconn.go | 42 +++++++++++++++++---------------- 1 file changed, 22 insertions(+), 20 deletions(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index 89173015..53ec6a37 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -81,31 +81,33 @@ func (ws *wsconn) Write(b []byte) (n int, err error) { ws.wlock.Lock() if ws.connClosing { - return 0, io.EOF } - writeWs := func(b []byte) (n int, err error) { - wr, err := ws.wsc.NextWriter(websocket.BinaryMessage) - if err != nil { - log.Warning("WS transport: ws connection NewFrameReader return " + err.Error()) - ws.connClosing = true - ws.Close() - return 0, err - } - n, err = wr.Write(b) - if err != nil { - return 0, err - } - err = wr.Close() - if err != nil { - return 0, err - } - return n, err - } - n, err = writeWs(b) + + n, err = ws.write(b) ws.wlock.Unlock() return n, err } + +func (ws *wsconn) write(b []byte) (n int, err error) { + wr, err := ws.wsc.NextWriter(websocket.BinaryMessage) + if err != nil { + log.Warning("WS transport: ws connection NewFrameReader return " + err.Error()) + ws.connClosing = true + ws.Close() + return 0, err + } + n, err = wr.Write(b) + if err != nil { + return 0, err + } + err = wr.Close() + if err != nil { + return 0, err + } + return n, err +} + func (ws *wsconn) Close() error { ws.connClosing = true ws.wlock.Lock() From e382eaa59fa9df9e028095d83363ad78ecf3ee33 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 18:59:50 +0800 Subject: [PATCH 35/47] typo --- transport/internet/ws/wsconn.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index 53ec6a37..57b961eb 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -150,7 +150,7 @@ func (ws *wsconn) setup() { https://godoc.org/github.com/gorilla/websocket#Conn.NextReader https://godoc.org/github.com/gorilla/websocket#Conn.NextWriter - Both Read and write access are exclusive. + Both Read and write access are both exclusive. And in both case it will need a lock. */ From 0a49628555419ea3cd4e4c34efca0f21cef6c31b Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 19:13:18 +0800 Subject: [PATCH 36/47] Bettering ping method --- transport/internet/ws/wsconn.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/transport/internet/ws/wsconn.go b/transport/internet/ws/wsconn.go index 57b961eb..71693879 100644 --- a/transport/internet/ws/wsconn.go +++ b/transport/internet/ws/wsconn.go @@ -183,19 +183,18 @@ func (ws *wsconn) pingPong() { ws.wlock.Lock() ws.wsc.WriteMessage(websocket.PingMessage, nil) ws.wlock.Unlock() - tick := time.NewTicker(time.Second * 30) + tick := time.After(time.Second * 3) select { case <-pongRcv: break - case <-tick.C: + case <-tick: if !ws.connClosing { log.Debug("WS:Closing as ping is not responded~" + ws.wsc.UnderlyingConn().LocalAddr().String() + "-" + ws.wsc.UnderlyingConn().RemoteAddr().String()) } ws.Close() } - <-tick.C - tick.Stop() + <-time.After(time.Second * 27) } return From 2209294dfcd3a84f9b402554a76f1814edc7f14d Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 20:14:45 +0800 Subject: [PATCH 37/47] Set reuse limit --- transport/internet/ws/connection_cache.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/transport/internet/ws/connection_cache.go b/transport/internet/ws/connection_cache.go index e95f939a..cef3c1f8 100644 --- a/transport/internet/ws/connection_cache.go +++ b/transport/internet/ws/connection_cache.go @@ -67,7 +67,7 @@ func (this *ConnectionCache) Recycle(dest string, conn *wsconn) { aconn := &AwaitingConnection{ conn: conn, - expire: time.Now().Add(time.Second * 180), + expire: time.Now().Add(time.Second * 7), } var list []*AwaitingConnection From 40ec84e22264af6bc8ef0b7584f9f426884f3944 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 20:15:12 +0800 Subject: [PATCH 38/47] Dial test --- transport/internet/ws/ws_test.go | 109 +++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 transport/internet/ws/ws_test.go diff --git a/transport/internet/ws/ws_test.go b/transport/internet/ws/ws_test.go new file mode 100644 index 00000000..f72c0f99 --- /dev/null +++ b/transport/internet/ws/ws_test.go @@ -0,0 +1,109 @@ +package ws_test + +import ( + "testing" + "time" + + "github.com/v2ray/v2ray-core/testing/assert" + + . "github.com/v2ray/v2ray-core/transport/internet/ws" + + v2net "github.com/v2ray/v2ray-core/common/net" +) + +func Test_Connect_ws(t *testing.T) { + assert := assert.On(t) + (&Config{Pto: "ws", Path: ""}).Apply() + conn, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("echo.websocket.org"), 80)) + assert.Error(err).IsNil() + conn.Write([]byte("echo")) + s := make(chan int) + go func() { + buf := make([]byte, 4) + conn.Read(buf) + s <- 0 + }() + <-s + conn.Close() +} + +func Test_Connect_wss(t *testing.T) { + assert := assert.On(t) + (&Config{Pto: "wss", Path: ""}).Apply() + conn, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("echo.websocket.org"), 443)) + assert.Error(err).IsNil() + conn.Write([]byte("echo")) + s := make(chan int) + go func() { + buf := make([]byte, 4) + conn.Read(buf) + s <- 0 + }() + <-s + conn.Close() +} + +func Test_Connect_ws_guess(t *testing.T) { + assert := assert.On(t) + (&Config{Pto: "", Path: ""}).Apply() + conn, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("echo.websocket.org"), 80)) + assert.Error(err).IsNil() + conn.Write([]byte("echo")) + s := make(chan int) + go func() { + buf := make([]byte, 4) + conn.Read(buf) + s <- 0 + }() + <-s + conn.Close() +} + +func Test_Connect_wss_guess(t *testing.T) { + assert := assert.On(t) + (&Config{Pto: "", Path: ""}).Apply() + conn, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("echo.websocket.org"), 443)) + assert.Error(err).IsNil() + conn.Write([]byte("echo")) + s := make(chan int) + go func() { + buf := make([]byte, 4) + conn.Read(buf) + s <- 0 + }() + <-s + conn.Close() +} + +func Test_Connect_wss_guess_fail(t *testing.T) { + assert := assert.On(t) + (&Config{Pto: "", Path: ""}).Apply() + _, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("static.kkdev.org"), 443)) + assert.Error(err).IsNotNil() +} + +func Test_Connect_wss_guess_reuse(t *testing.T) { + assert := assert.On(t) + (&Config{Pto: "", Path: "", ConnectionReuse: true}).Apply() + i := 3 + for i != 0 { + conn, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("echo.websocket.org"), 443)) + assert.Error(err).IsNil() + conn.Write([]byte("echo")) + s := make(chan int) + go func() { + buf := make([]byte, 4) + conn.Read(buf) + s <- 0 + }() + <-s + if i == 0 { + conn.SetDeadline(time.Now()) + conn.SetReadDeadline(time.Now()) + conn.SetWriteDeadline(time.Now()) + conn.SetReusable(false) + } + conn.Close() + i-- + } +} From 02e3c17c97de69deb788bbc836fc541eff4b8caa Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 20:20:47 +0800 Subject: [PATCH 39/47] test:listen --- transport/internet/ws/ws_test.go | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/transport/internet/ws/ws_test.go b/transport/internet/ws/ws_test.go index f72c0f99..7eb97fe5 100644 --- a/transport/internet/ws/ws_test.go +++ b/transport/internet/ws/ws_test.go @@ -107,3 +107,18 @@ func Test_Connect_wss_guess_reuse(t *testing.T) { i-- } } + +func Test_listenWSAndDial(t *testing.T) { + assert := assert.On(t) + (&Config{Pto: "ws", Path: ""}).Apply() + listen, err := ListenWS(v2net.DomainAddress("localhost"), 13142) + assert.Error(err).IsNil() + go func() { + conn, err := listen.Accept() + assert.Error(err).IsNil() + conn.Close() + }() + conn, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("localhost"), 13142)) + assert.Error(err).IsNil() + conn.Close() +} From f7ce9d3be64f1b0586162cbd08b8b9c43a59e735 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 20:42:38 +0800 Subject: [PATCH 40/47] Add Developer Insecure Skip Verify for testing --- transport/internet/ws/config.go | 11 ++++++----- transport/internet/ws/config_json.go | 1 + transport/internet/ws/dialer.go | 2 +- 3 files changed, 8 insertions(+), 6 deletions(-) diff --git a/transport/internet/ws/config.go b/transport/internet/ws/config.go index e078d607..8ae70e2f 100644 --- a/transport/internet/ws/config.go +++ b/transport/internet/ws/config.go @@ -1,11 +1,12 @@ package ws type Config struct { - ConnectionReuse bool - Path string - Pto string - Cert string - PrivKey string + ConnectionReuse bool + Path string + Pto string + Cert string + PrivKey string + DeveloperInsecureSkipVerify bool } func (this *Config) Apply() { diff --git a/transport/internet/ws/config_json.go b/transport/internet/ws/config_json.go index eb479479..0ebc6237 100644 --- a/transport/internet/ws/config_json.go +++ b/transport/internet/ws/config_json.go @@ -25,5 +25,6 @@ func (this *Config) UnmarshalJSON(data []byte) error { this.Pto = jsonConfig.Pto this.PrivKey = jsonConfig.PrivKey this.Cert = jsonConfig.Cert + this.DeveloperInsecureSkipVerify = false return nil } diff --git a/transport/internet/ws/dialer.go b/transport/internet/ws/dialer.go index 70e5ab81..5cd3ecf8 100644 --- a/transport/internet/ws/dialer.go +++ b/transport/internet/ws/dialer.go @@ -49,7 +49,7 @@ func wsDial(src v2net.Address, dest v2net.Destination) (*wsconn, error) { return internet.DialToDest(src, dest) } - tlsconf := &tls.Config{ServerName: dest.Address().Domain()} + tlsconf := &tls.Config{ServerName: dest.Address().Domain(), InsecureSkipVerify: effectiveConfig.DeveloperInsecureSkipVerify} dialer := websocket.Dialer{NetDial: commonDial, ReadBufferSize: 65536, WriteBufferSize: 65536, TLSClientConfig: tlsconf} From 4b9e506dba9f2070a7bec7557f129275ca9253de Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 21:18:48 +0800 Subject: [PATCH 41/47] Added Testing Cert --- testing/tls/cert.pem | 15 +++++++++++++++ testing/tls/key.pem | 10 ++++++++++ 2 files changed, 25 insertions(+) create mode 100644 testing/tls/cert.pem create mode 100644 testing/tls/key.pem diff --git a/testing/tls/cert.pem b/testing/tls/cert.pem new file mode 100644 index 00000000..5468aa54 --- /dev/null +++ b/testing/tls/cert.pem @@ -0,0 +1,15 @@ +-----BEGIN CERTIFICATE----- +MIICSTCCAfOgAwIBAgIJAKvQIEezrxBNMA0GCSqGSIb3DQEBCwUAMH8xCzAJBgNV +BAYTAlYyMRAwDgYDVQQIDAdURVNUSU5HMRAwDgYDVQQHDAdURVNUSU5HMRAwDgYD +VQQKDAdURVNUSU5HMRAwDgYDVQQLDAdURVNUSU5HMRAwDgYDVQQDDAdURVNUSU5H +MRYwFAYJKoZIhvcNAQkBFgdURVNUSU5HMCAXDTE2MDgxNTEyNTAwNFoYDzY0ODcw +NTA3MTI1MDA0WjB/MQswCQYDVQQGEwJWMjEQMA4GA1UECAwHVEVTVElORzEQMA4G +A1UEBwwHVEVTVElORzEQMA4GA1UECgwHVEVTVElORzEQMA4GA1UECwwHVEVTVElO +RzEQMA4GA1UEAwwHVEVTVElORzEWMBQGCSqGSIb3DQEJARYHVEVTVElORzBcMA0G +CSqGSIb3DQEBAQUAA0sAMEgCQQDFGeGTGepVDwgLm5rFx8khAhbod6g3Xg7vU3M9 +lzowBeAOS6bpN8lnBEXo3U2brxB+okbRhNuSj3VQ4raX0iL1AgMBAAGjUDBOMB0G +A1UdDgQWBBRE81DrJv6nBXAF3JP4a3LTtwkp8TAfBgNVHSMEGDAWgBRE81DrJv6n +BXAF3JP4a3LTtwkp8TAMBgNVHRMEBTADAQH/MA0GCSqGSIb3DQEBCwUAA0EAkGX6 +sQvwHqNOdwise45dU8NvXwZsoqSQ2tdxrkB+SnKqEsMnRh/yPCSgzFkQVt53sYuf +HK8gD/wifGC5z39YlQ== +-----END CERTIFICATE----- diff --git a/testing/tls/key.pem b/testing/tls/key.pem new file mode 100644 index 00000000..e9abfef7 --- /dev/null +++ b/testing/tls/key.pem @@ -0,0 +1,10 @@ +-----BEGIN PRIVATE KEY----- +MIIBVQIBADANBgkqhkiG9w0BAQEFAASCAT8wggE7AgEAAkEAxRnhkxnqVQ8IC5ua +xcfJIQIW6HeoN14O71NzPZc6MAXgDkum6TfJZwRF6N1Nm68QfqJG0YTbko91UOK2 +l9Ii9QIDAQABAkEAgumaywKWgyJ1vIgAt8bnzxW9M3BueT/u+YTa8Ril3EiOtxDl +/aRtVJ/62r64Ymtq8BvYcEiopFKrUUKPaTfIrQIhAPkd9nDq7B4WepF8+pB0CyR0 +dpT0imCXooN4+utosrdDAiEAyowAwGcYULWiALgSi78gt7fRgp0GwTf80evFy/k0 +DWcCIQC9+IxrVarT0v6LHgyRxfyNQ0b+lnFD8b6bldF7Xa8TswIgA/Ptg9O/Prv8 +uGTfP8jwG4XD2fe0jQrJrVMbnhpz8JsCIHkkbC3ez+iPieasr8a+zEpreE8NjrEV +xs4xp6WZPsGp +-----END PRIVATE KEY----- From 856f2a2dd3a6b1598a26c7e7eb97aeee0754afc3 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 21:19:15 +0800 Subject: [PATCH 42/47] Fix assert fail --- transport/internet/ws/stopablehttplistener.go | 23 ++++++++----------- 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/transport/internet/ws/stopablehttplistener.go b/transport/internet/ws/stopablehttplistener.go index 832b04e2..2523caed 100644 --- a/transport/internet/ws/stopablehttplistener.go +++ b/transport/internet/ws/stopablehttplistener.go @@ -3,23 +3,23 @@ package ws import ( "errors" "net" - "time" ) type StoppableListener struct { - *net.TCPListener //Wrapped listener - stop chan int //Channel used only to indicate listener should shutdown + net.Listener //Wrapped listener + stop chan int //Channel used only to indicate listener should shutdown } func NewStoppableListener(l net.Listener) (*StoppableListener, error) { - tcpL, ok := l.(*net.TCPListener) - - if !ok { - return nil, errors.New("Cannot wrap listener") - } + /* + tcpL, ok := l.(*net.TCPListener) + if !ok { + return nil, errors.New("Cannot wrap listener") + } + */ retval := &StoppableListener{} - retval.TCPListener = tcpL + retval.Listener = l retval.stop = make(chan int) return retval, nil @@ -30,10 +30,7 @@ var StoppedError = errors.New("Listener stopped") func (sl *StoppableListener) Accept() (net.Conn, error) { for { - //Wait up to one second for a new connection - sl.SetDeadline(time.Now().Add(time.Second)) - - newConn, err := sl.TCPListener.Accept() + newConn, err := sl.Listener.Accept() //Check for the channel being closed select { From d7440bb1d60bd180b59b895d5cbaa1359ff41c02 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 21:19:53 +0800 Subject: [PATCH 43/47] Test:listen tls --- transport/internet/ws/ws_test.go | 35 +++++++++++++++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) diff --git a/transport/internet/ws/ws_test.go b/transport/internet/ws/ws_test.go index 7eb97fe5..42c5ba48 100644 --- a/transport/internet/ws/ws_test.go +++ b/transport/internet/ws/ws_test.go @@ -77,6 +77,10 @@ func Test_Connect_wss_guess(t *testing.T) { func Test_Connect_wss_guess_fail(t *testing.T) { assert := assert.On(t) + go func() { + <-time.After(time.Second * 5) + assert.Fail("Too slow") + }() (&Config{Pto: "", Path: ""}).Apply() _, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("static.kkdev.org"), 443)) assert.Error(err).IsNotNil() @@ -84,6 +88,10 @@ func Test_Connect_wss_guess_fail(t *testing.T) { func Test_Connect_wss_guess_reuse(t *testing.T) { assert := assert.On(t) + go func() { + <-time.After(time.Second * 5) + assert.Fail("Too slow") + }() (&Config{Pto: "", Path: "", ConnectionReuse: true}).Apply() i := 3 for i != 0 { @@ -110,15 +118,40 @@ func Test_Connect_wss_guess_reuse(t *testing.T) { func Test_listenWSAndDial(t *testing.T) { assert := assert.On(t) - (&Config{Pto: "ws", Path: ""}).Apply() + go func() { + <-time.After(time.Second * 5) + assert.Fail("Too slow") + }() + (&Config{Pto: "ws", Path: "ws"}).Apply() listen, err := ListenWS(v2net.DomainAddress("localhost"), 13142) assert.Error(err).IsNil() go func() { conn, err := listen.Accept() assert.Error(err).IsNil() conn.Close() + listen.Close() }() conn, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("localhost"), 13142)) assert.Error(err).IsNil() conn.Close() } + +func Test_listenWSAndDial_TLS(t *testing.T) { + assert := assert.On(t) + go func() { + <-time.After(time.Second * 5) + assert.Fail("Too slow") + }() + (&Config{Pto: "wss", Path: "wss", ConnectionReuse: true, DeveloperInsecureSkipVerify: true, PrivKey: "./../../../testing/tls/key.pem", Cert: "./../../../testing/tls/cert.pem"}).Apply() + listen, err := ListenWS(v2net.DomainAddress("localhost"), 13143) + assert.Error(err).IsNil() + go func() { + conn, err := listen.Accept() + assert.Error(err).IsNil() + conn.Close() + listen.Close() + }() + conn, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("localhost"), 13143)) + assert.Error(err).IsNil() + conn.Close() +} From 6c8b4f5db32b63ca5fc3080b4b08d6f6ea818d2f Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 21:21:09 +0800 Subject: [PATCH 44/47] Fix --- transport/internet/ws/connection_cache.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/transport/internet/ws/connection_cache.go b/transport/internet/ws/connection_cache.go index cef3c1f8..10b5bc10 100644 --- a/transport/internet/ws/connection_cache.go +++ b/transport/internet/ws/connection_cache.go @@ -34,7 +34,7 @@ func (this *ConnectionCache) Cleanup() { defer this.cleanupOnce.Reset() for len(this.cache) > 0 { - time.Sleep(time.Second * 4) + time.Sleep(time.Second * 7) this.Lock() for key, value := range this.cache { size := len(value) From 45d801b9225d5296321503c440f46e5a9dc38e14 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 21:29:15 +0800 Subject: [PATCH 45/47] WsTest:more case --- transport/internet/ws/ws_test.go | 49 ++++++++++++++++++++++++-------- 1 file changed, 37 insertions(+), 12 deletions(-) diff --git a/transport/internet/ws/ws_test.go b/transport/internet/ws/ws_test.go index 42c5ba48..c76f59cf 100644 --- a/transport/internet/ws/ws_test.go +++ b/transport/internet/ws/ws_test.go @@ -43,6 +43,22 @@ func Test_Connect_wss(t *testing.T) { conn.Close() } +func Test_Connect_wss_1_nil(t *testing.T) { + assert := assert.On(t) + (&Config{Pto: "wss", Path: ""}).Apply() + conn, err := Dial(nil, v2net.TCPDestination(v2net.DomainAddress("echo.websocket.org"), 443)) + assert.Error(err).IsNil() + conn.Write([]byte("echo")) + s := make(chan int) + go func() { + buf := make([]byte, 4) + conn.Read(buf) + s <- 0 + }() + <-s + conn.Close() +} + func Test_Connect_ws_guess(t *testing.T) { assert := assert.On(t) (&Config{Pto: "", Path: ""}).Apply() @@ -77,21 +93,20 @@ func Test_Connect_wss_guess(t *testing.T) { func Test_Connect_wss_guess_fail(t *testing.T) { assert := assert.On(t) - go func() { - <-time.After(time.Second * 5) - assert.Fail("Too slow") - }() (&Config{Pto: "", Path: ""}).Apply() _, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("static.kkdev.org"), 443)) assert.Error(err).IsNotNil() } +func Test_Connect_wss_guess_fail_port(t *testing.T) { + assert := assert.On(t) + (&Config{Pto: "", Path: ""}).Apply() + _, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("static.kkdev.org"), 179)) + assert.Error(err).IsNotNil() +} + func Test_Connect_wss_guess_reuse(t *testing.T) { assert := assert.On(t) - go func() { - <-time.After(time.Second * 5) - assert.Fail("Too slow") - }() (&Config{Pto: "", Path: "", ConnectionReuse: true}).Apply() i := 3 for i != 0 { @@ -118,10 +133,6 @@ func Test_Connect_wss_guess_reuse(t *testing.T) { func Test_listenWSAndDial(t *testing.T) { assert := assert.On(t) - go func() { - <-time.After(time.Second * 5) - assert.Fail("Too slow") - }() (&Config{Pto: "ws", Path: "ws"}).Apply() listen, err := ListenWS(v2net.DomainAddress("localhost"), 13142) assert.Error(err).IsNil() @@ -129,11 +140,25 @@ func Test_listenWSAndDial(t *testing.T) { conn, err := listen.Accept() assert.Error(err).IsNil() conn.Close() + conn, err = listen.Accept() + assert.Error(err).IsNil() + conn.Close() + conn, err = listen.Accept() + assert.Error(err).IsNil() + conn.Close() listen.Close() }() conn, err := Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("localhost"), 13142)) assert.Error(err).IsNil() conn.Close() + <-time.After(time.Second * 5) + conn, err = Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("localhost"), 13142)) + assert.Error(err).IsNil() + conn.Close() + <-time.After(time.Second * 15) + conn, err = Dial(v2net.AnyIP, v2net.TCPDestination(v2net.DomainAddress("localhost"), 13142)) + assert.Error(err).IsNil() + conn.Close() } func Test_listenWSAndDial_TLS(t *testing.T) { From 32e8deb6e8799b4e2f6dc5a0065f4a763dfef937 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 21:36:37 +0800 Subject: [PATCH 46/47] Simplify Design --- transport/internet/ws/stopablehttplistener.go | 38 ++----------------- 1 file changed, 4 insertions(+), 34 deletions(-) diff --git a/transport/internet/ws/stopablehttplistener.go b/transport/internet/ws/stopablehttplistener.go index 2523caed..99ff3a24 100644 --- a/transport/internet/ws/stopablehttplistener.go +++ b/transport/internet/ws/stopablehttplistener.go @@ -6,54 +6,24 @@ import ( ) type StoppableListener struct { - net.Listener //Wrapped listener - stop chan int //Channel used only to indicate listener should shutdown + net.Listener //Wrapped listener } func NewStoppableListener(l net.Listener) (*StoppableListener, error) { - /* - tcpL, ok := l.(*net.TCPListener) - if !ok { - return nil, errors.New("Cannot wrap listener") - } - */ retval := &StoppableListener{} retval.Listener = l - retval.stop = make(chan int) - return retval, nil } var StoppedError = errors.New("Listener stopped") func (sl *StoppableListener) Accept() (net.Conn, error) { + newConn, err := sl.Listener.Accept() + return newConn, err - for { - newConn, err := sl.Listener.Accept() - - //Check for the channel being closed - select { - case <-sl.stop: - return nil, StoppedError - default: - //If the channel is still open, continue as normal - } - - if err != nil { - netErr, ok := err.(net.Error) - - //If this is a timeout, then continue to wait for - //new connections - if ok && netErr.Timeout() && netErr.Temporary() { - continue - } - } - - return newConn, err - } } func (sl *StoppableListener) Stop() { - close(sl.stop) + sl.Listener.Close() } From 22519353c320757eef7e57aee82134a00d468a40 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Mon, 15 Aug 2016 21:55:43 +0800 Subject: [PATCH 47/47] Check Read --- transport/internet/ws/ws_test.go | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/transport/internet/ws/ws_test.go b/transport/internet/ws/ws_test.go index c76f59cf..25f7fdef 100644 --- a/transport/internet/ws/ws_test.go +++ b/transport/internet/ws/ws_test.go @@ -21,6 +21,10 @@ func Test_Connect_ws(t *testing.T) { go func() { buf := make([]byte, 4) conn.Read(buf) + str := string(buf) + if str != "echo" { + assert.Fail("Data mismatch") + } s <- 0 }() <-s @@ -37,6 +41,10 @@ func Test_Connect_wss(t *testing.T) { go func() { buf := make([]byte, 4) conn.Read(buf) + str := string(buf) + if str != "echo" { + assert.Fail("Data mismatch") + } s <- 0 }() <-s @@ -53,6 +61,10 @@ func Test_Connect_wss_1_nil(t *testing.T) { go func() { buf := make([]byte, 4) conn.Read(buf) + str := string(buf) + if str != "echo" { + assert.Fail("Data mismatch") + } s <- 0 }() <-s @@ -69,6 +81,10 @@ func Test_Connect_ws_guess(t *testing.T) { go func() { buf := make([]byte, 4) conn.Read(buf) + str := string(buf) + if str != "echo" { + assert.Fail("Data mismatch") + } s <- 0 }() <-s @@ -85,6 +101,10 @@ func Test_Connect_wss_guess(t *testing.T) { go func() { buf := make([]byte, 4) conn.Read(buf) + str := string(buf) + if str != "echo" { + assert.Fail("Data mismatch") + } s <- 0 }() <-s @@ -117,6 +137,10 @@ func Test_Connect_wss_guess_reuse(t *testing.T) { go func() { buf := make([]byte, 4) conn.Read(buf) + str := string(buf) + if str != "echo" { + assert.Fail("Data mismatch") + } s <- 0 }() <-s