func (c *Client) Session() (*yamux.Session, error) {
	if c.sess == nil {
		s, err := net.Dial("tcp", c.addr)
		if err != nil {
			return nil, err
		}

		if c.secure {
			sec, err := seconn.NewClient(s)
			if err != nil {
				return nil, err
			}

			c.conn = sec
		} else {
			c.conn = s
		}

		sess, err := yamux.Client(c.conn, muxConfig)
		if err != nil {
			return nil, err
		}

		c.sess = sess
	}

	return c.sess, nil
}

// NewClient creates a client from an already-open connection-like value.
// Dial is typically used instead.
func NewClient(conn io.ReadWriteCloser) (*Client, error) {
	// Create the yamux client so we can multiplex
	mux, err := yamux.Client(conn, nil)
	if err != nil {
		conn.Close()
		return nil, err
	}

	// Connect to the control stream.
	control, err := mux.Open()
	if err != nil {
		mux.Close()
		return nil, err
	}

	// Create the broker and start it up
	broker := newMuxBroker(mux)
	go broker.Run()

	// Build the client using our broker and control channel.
	return &Client{
		broker:  broker,
		control: rpc.NewClient(control),
	}, nil
}

// getNewConn is used to return a new connection
func (p *ConnPool) getNewConn(addr string, timeout time.Duration) (*Conn, error) {
	// Try to dial the conn
	conn, err := tls.DialWithDialer(&net.Dialer{Timeout: timeout}, "tcp", addr, p.config)
	if err != nil {
		return nil, err
	}

	// Setup the logger
	conf := yamux.DefaultConfig()
	conf.LogOutput = p.logOutput

	// Create a multiplexed session
	session, err := yamux.Client(conn, conf)
	if err != nil {
		conn.Close()
		return nil, err
	}

	// Wrap the connection
	c := &Conn{
		addr:     addr,
		session:  session,
		lastUsed: time.Now(),
		pool:     p,
	}
	return c, nil
}

func startTLSServer(config *Config) (net.Conn, chan error) {
	errc := make(chan error, 1)

	tlsConfigServer, err := config.IncomingTLSConfig()
	if err != nil {
		errc <- err
		return nil, errc
	}

	client, server := net.Pipe()

	// Use yamux to buffer the reads, otherwise it's easy to deadlock
	muxConf := yamux.DefaultConfig()
	serverSession, _ := yamux.Server(server, muxConf)
	clientSession, _ := yamux.Client(client, muxConf)
	clientConn, _ := clientSession.Open()
	serverConn, _ := serverSession.Accept()

	go func() {
		tlsServer := tls.Server(serverConn, tlsConfigServer)
		if err := tlsServer.Handshake(); err != nil {
			errc <- err
		}
		close(errc)
		// Because net.Pipe() is unbuffered, if both sides
		// Close() simultaneously, we will deadlock as they
		// both send an alert and then block. So we make the
		// server read any data from the client until error or
		// EOF, which will allow the client to Close(), and
		// *then* we Close() the server.
		io.Copy(ioutil.Discard, tlsServer)
		tlsServer.Close()
	}()
	return clientConn, errc
}

func Main() {
	kingpin.Parse()

	// Create websocket session for tunnels
	origin := fmt.Sprintf("http://%s/", *host)
	url := fmt.Sprintf("ws://%s/", *host)
	conn, err := websocket.Dial(url, "", origin)
	if err != nil {
		log.Fatal(fmt.Sprintf("Could not connect to subway server (%s)!", url))
	}

	session, err := yamux.Client(conn, nil)
	if err != nil {
		panic(err)
	}

	// Start REST api
	router := mux.NewRouter()
	router.HandleFunc("/", handlers.CreateTunnel)

	n := negroni.New()
	n.Use(negronilogrus.NewMiddleware())
	render := render.New(render.Options{})
	subway := context.CreateSubway(session, render)
	n.Use(subway)
	n.UseHandler(router)
	n.Run(fmt.Sprintf(":%v", *port))

}

// getNewConn is used to return a new connection
func (p *ConnPool) getNewConn(region string, addr net.Addr, version int) (*Conn, error) {
	// Try to dial the conn
	conn, err := net.DialTimeout("tcp", addr.String(), 10*time.Second)
	if err != nil {
		return nil, err
	}

	// Cast to TCPConn
	if tcp, ok := conn.(*net.TCPConn); ok {
		tcp.SetKeepAlive(true)
		tcp.SetNoDelay(true)
	}

	// Check if TLS is enabled
	if p.tlsWrap != nil {
		// Switch the connection into TLS mode
		if _, err := conn.Write([]byte{byte(rpcTLS)}); err != nil {
			conn.Close()
			return nil, err
		}

		// Wrap the connection in a TLS client
		tlsConn, err := p.tlsWrap(region, conn)
		if err != nil {
			conn.Close()
			return nil, err
		}
		conn = tlsConn
	}

	// Write the multiplex byte to set the mode
	if _, err := conn.Write([]byte{byte(rpcMultiplex)}); err != nil {
		conn.Close()
		return nil, err
	}

	// Setup the logger
	conf := yamux.DefaultConfig()
	conf.LogOutput = p.logOutput

	// Create a multiplexed session
	session, err := yamux.Client(conn, conf)
	if err != nil {
		conn.Close()
		return nil, err
	}

	// Wrap the connection
	c := &Conn{
		refCount: 1,
		addr:     addr,
		session:  session,
		clients:  list.New(),
		lastUsed: time.Now(),
		version:  version,
		pool:     p,
	}
	return c, nil
}

func newMuxBrokerClient(rwc io.ReadWriteCloser) (*muxBroker, error) {
	s, err := yamux.Client(rwc, nil)
	if err != nil {
		return nil, err
	}

	return newMuxBroker(s), nil
}

func TestProvider_Disconnect(t *testing.T) {
	config := testProviderConfig()
	p, err := NewProvider(config)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer p.Shutdown()

	// Setup RPC client
	a, b := testConn(t)
	client, _ := yamux.Client(a, yamux.DefaultConfig())
	server, _ := yamux.Server(b, yamux.DefaultConfig())
	go p.handleSession(client, make(chan struct{}))

	stream, _ := server.Open()
	cc := msgpackrpc.NewCodec(false, false, stream)

	// Make the connect rpc
	args := &DisconnectRequest{
		NoRetry: true,
		Backoff: 300 * time.Second,
	}
	resp := &DisconnectResponse{}
	err = msgpackrpc.CallWithCodec(cc, "Client.Disconnect", args, resp)
	if err != nil {
		t.Fatalf("err: %v", err)
	}

	p.backoffLock.Lock()
	defer p.backoffLock.Unlock()

	if p.backoff != 300*time.Second {
		t.Fatalf("bad: %v", p.backoff)
	}
	if !p.noRetry {
		t.Fatalf("bad")
	}

	p.sessionLock.Lock()
	defer p.sessionLock.Unlock()

	if p.sessionID != "" {
		t.Fatalf("Bad: %v", p.sessionID)
	}
	if p.sessionAuth {
		t.Fatalf("Bad: %v", p.sessionAuth)
	}
}

func TestProvider_Connect(t *testing.T) {
	config := testProviderConfig()
	config.Service.Capabilities["foo"] = 1
	config.Handlers["foo"] = fooCapability(t)
	p, err := NewProvider(config)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer p.Shutdown()

	// Setup RPC client
	a, b := testConn(t)
	client, _ := yamux.Client(a, yamux.DefaultConfig())
	server, _ := yamux.Server(b, yamux.DefaultConfig())
	go p.handleSession(client, make(chan struct{}))

	stream, _ := server.Open()
	cc := msgpackrpc.NewCodec(false, false, stream)

	// Make the connect rpc
	args := &ConnectRequest{
		Capability: "foo",
		Meta: map[string]string{
			"zip": "zap",
		},
	}
	resp := &ConnectResponse{}
	err = msgpackrpc.CallWithCodec(cc, "Client.Connect", args, resp)
	if err != nil {
		t.Fatalf("err: %v", err)
	}

	// Should be successful!
	if !resp.Success {
		t.Fatalf("bad")
	}

	// At this point, we should be connected
	out := make([]byte, 9)
	n, err := stream.Read(out)
	if err != nil {
		t.Fatalf("err: %v %d", err, n)
	}

	if string(out) != "foobarbaz" {
		t.Fatalf("bad: %s", out)
	}
}

func openSession(d *Daemon) error {

	err := openWebsocket(d)
	if err != nil {
		d.log.Error("could not open websocket", log.Ctx{"error": err})
		return err
	}

	// Setup client side of yamux
	session, err := yamux.Client(d.wsconn, nil)
	if err != nil {
		d.log.Error("Yamux session failed!", log.Ctx{"error": err})
	}

	d.session = session

	return nil
}

// Accepts new tunnel session in a blocking fashion. When it returns it closes the listener
// and returns with the error why it returned.
func Accept(accepted chan Tunnel, publicHostname string, listener net.Listener) error {
	defer listener.Close()

	for id := 0; ; id++ {

		conn, err := listener.Accept()
		if err != nil {
			log.Printf("FATAL: %v", err)
			return err
		}

		log.Printf("connection accepted %v\n", conn.RemoteAddr())

		// perform handshake
		go func(conn net.Conn, id int) {
			name := petname.Generate(2, "-")

			hostname := fmt.Sprintf("%v.%v", name, publicHostname)
			publicAddress := fmt.Sprintf("http://%v", hostname)

			reader := bufio.NewReader(conn)
			request, err := http.ReadRequest(reader)
			if err != nil {
				log.Println(err.Error())
				return
			}

			localAddress := request.Header.Get("X-Publichost-Local")
			session, err := yamux.Client(conn, nil)
			if err != nil {
				log.Println(err.Error())
				return
			}

			log.Printf("tunnel created %v->%v\n", publicAddress, localAddress)
			accepted <- Tunnel{
				id,
				hostname,
				session,
				localAddress,
			}
		}(conn, id)
	}
}

func TestProvider_Flash(t *testing.T) {
	config := testProviderConfig()
	buf := bytes.NewBuffer(nil)
	config.LogOutput = buf
	p, err := NewProvider(config)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	defer p.Shutdown()

	// Setup RPC client
	a, b := testConn(t)
	client, _ := yamux.Client(a, yamux.DefaultConfig())
	server, _ := yamux.Server(b, yamux.DefaultConfig())
	go p.handleSession(client, make(chan struct{}))

	stream, _ := server.Open()
	cc := msgpackrpc.NewCodec(false, false, stream)

	// Make the connect rpc
	args := &FlashRequest{
		Severity: "INFO",
		Message:  "TESTING",
	}
	resp := &FlashResponse{}
	err = msgpackrpc.CallWithCodec(cc, "Client.Flash", args, resp)
	if err != nil {
		t.Fatalf("err: %v", err)
	}

	// Wait until we are disconnected
	start := time.Now()
	for time.Now().Sub(start) < time.Second {
		if bytes.Contains(buf.Bytes(), []byte("TESTING")) {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	if !bytes.Contains(buf.Bytes(), []byte("TESTING")) {
		t.Fatalf("missing: %s", buf)
	}
}

// getNewConn is used to return a new connection
func (p *connPool) getNewConn(addr net.Addr) (*conn, error) {
	// Try to dial the conn
	con, err := net.DialTimeout("tcp", addr.String(), 10*time.Second)
	if err != nil {
		return nil, err
	}

	// Cast to TCPConn
	if tcp, ok := con.(*net.TCPConn); ok {
		tcp.SetKeepAlive(true)
		tcp.SetNoDelay(true)
	}

	// Write the multiplex byte to set the mode
	if _, err := con.Write([]byte{byte(rpcInternal)}); err != nil {
		con.Close()
		return nil, err
	}

	// Setup the logger
	conf := yamux.DefaultConfig()
	conf.LogOutput = p.logOutput

	// Create a multiplexed session
	session, err := yamux.Client(con, conf)
	if err != nil {
		con.Close()
		return nil, err
	}

	// Wrap the connection
	c := &conn{
		refCount: 1,
		addr:     addr,
		session:  session,
		clients:  list.New(),
		lastUsed: time.Now(),
		pool:     p,
	}
	return c, nil
}

// initClient does the common initialization
func initClient(conn net.Conn, opts *Opts) (*Client, error) {
	// Send the preamble
	_, err := conn.Write([]byte(clientPreamble))
	if err != nil {
		return nil, fmt.Errorf("preamble write failed: %v", err)
	}

	// Wrap the connection in yamux for multiplexing
	ymConf := yamux.DefaultConfig()
	if opts.LogOutput != nil {
		ymConf.LogOutput = opts.LogOutput
	}
	client, _ := yamux.Client(conn, ymConf)

	// Create the client
	c := &Client{
		conn:   conn,
		client: client,
	}
	return c, nil
}

// getNewConn is used to return a new connection
func (p *ConnPool) getNewConn(dc string, addr net.Addr, version int) (*Conn, error) {
	// Get a new, raw connection.
	conn, _, err := p.Dial(dc, addr)
	if err != nil {
		return nil, err
	}

	// Switch the multiplexing based on version
	var session muxSession
	if version < 2 {
		conn.Close()
		return nil, fmt.Errorf("cannot make client connection, unsupported protocol version %d", version)
	} else {
		// Write the Consul multiplex byte to set the mode
		if _, err := conn.Write([]byte{byte(rpcMultiplexV2)}); err != nil {
			conn.Close()
			return nil, err
		}

		// Setup the logger
		conf := yamux.DefaultConfig()
		conf.LogOutput = p.logOutput

		// Create a multiplexed session
		session, _ = yamux.Client(conn, conf)
	}

	// Wrap the connection
	c := &Conn{
		refCount: 1,
		addr:     addr,
		session:  session,
		clients:  list.New(),
		lastUsed: time.Now(),
		version:  version,
		pool:     p,
	}
	return c, nil
}

func upstreamServe(conn net.Conn) error {
	l, err := net.Listen("tcp", ":0")
	if err != nil {
		return err
	}
	session, err := yamux.Client(conn, nil)
	if err != nil {
		return err
	}

	control, err := session.Open()
	if err != nil {
		return err
	}

	laddr := l.Addr().String()
	if _, err := control.Write([]byte(laddr)); err != nil {
		return err
	}

	if err := control.Close(); err != nil {
		return err
	}

	for {
		incoming, err := l.Accept()
		if err != nil {
			return err
		}

		downstream, err := session.Open()
		if err != nil {
			return err
		}
		go handle(proxy(incoming, downstream))
	}

	return nil
}

// NewRPCClient creates a client from an already-open connection-like value.
// Dial is typically used instead.
func NewRPCClient(conn io.ReadWriteCloser, plugins map[string]Plugin) (*RPCClient, error) {
	// Create the yamux client so we can multiplex
	mux, err := yamux.Client(conn, nil)
	if err != nil {
		conn.Close()
		return nil, err
	}

	// Connect to the control stream.
	control, err := mux.Open()
	if err != nil {
		mux.Close()
		return nil, err
	}

	// Connect stdout, stderr streams
	stdstream := make([]net.Conn, 2)
	for i, _ := range stdstream {
		stdstream[i], err = mux.Open()
		if err != nil {
			mux.Close()
			return nil, err
		}
	}

	// Create the broker and start it up
	broker := newMuxBroker(mux)
	go broker.Run()

	// Build the client using our broker and control channel.
	return &RPCClient{
		broker:  broker,
		control: rpc.NewClient(control),
		plugins: plugins,
		stdout:  stdstream[0],
		stderr:  stdstream[1],
	}, nil
}

func testYamux(t *testing.T) (client *yamux.Session, server *yamux.Session) {
	l, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("err: %s", err)
	}

	// Server side
	doneCh := make(chan struct{})
	go func() {
		defer close(doneCh)
		conn, err := l.Accept()
		l.Close()
		if err != nil {
			t.Fatalf("err: %s", err)
		}

		server, err = yamux.Server(conn, nil)
		if err != nil {
			t.Fatalf("err: %s", err)
		}
	}()

	// Client side
	conn, err := net.Dial("tcp", l.Addr().String())
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	client, err = yamux.Client(conn, nil)
	if err != nil {
		t.Fatalf("err: %s", err)
	}

	// Wait for the server
	<-doneCh

	return
}

//.........这里部分代码省略.........
			Hidden: true,
		},
		cli.IntFlag{
			Name:   "nc",
			Value:  0,
			Hidden: true,
		},
	}
	myApp.Action = func(c *cli.Context) {
		log.Println("version:", VERSION)
		addr, err := net.ResolveTCPAddr("tcp", c.String("localaddr"))
		checkError(err)
		listener, err := net.ListenTCP("tcp", addr)
		checkError(err)
		pass := pbkdf2.Key([]byte(c.String("key")), []byte(SALT), 4096, 32, sha1.New)

		// kcp server
		nodelay, interval, resend, nc := c.Int("nodelay"), c.Int("interval"), c.Int("resend"), c.Int("nc")

		switch c.String("mode") {
		case "normal":
			nodelay, interval, resend, nc = 0, 30, 2, 1
		case "fast":
			nodelay, interval, resend, nc = 0, 20, 2, 1
		case "fast2":
			nodelay, interval, resend, nc = 1, 20, 2, 1
		case "fast3":
			nodelay, interval, resend, nc = 1, 10, 2, 1
		}

		log.Println("listening on:", listener.Addr())
		log.Println("encryption:", c.String("crypt"))
		log.Println("nodelay parameters:", nodelay, interval, resend, nc)
		log.Println("remote address:", c.String("remoteaddr"))
		log.Println("sndwnd:", c.Int("sndwnd"), "rcvwnd:", c.Int("rcvwnd"))
		log.Println("mtu:", c.Int("mtu"))
		log.Println("fec:", c.Int("fec"))
		log.Println("acknodelay:", c.Bool("acknodelay"))
		log.Println("dscp:", c.Int("dscp"))
		log.Println("conn:", c.Int("conn"))

		createConn := func() *yamux.Session {
			var block kcp.BlockCrypt
			switch c.String("crypt") {
			case "tea":
				block, _ = kcp.NewTEABlockCrypt(pass[:16])
			case "xor":
				block, _ = kcp.NewSimpleXORBlockCrypt(pass)
			case "none":
				block, _ = kcp.NewNoneBlockCrypt(pass)
			default:
				block, _ = kcp.NewAESBlockCrypt(pass)
			}
			kcpconn, err := kcp.DialWithOptions(c.Int("fec"), c.String("remoteaddr"), block)
			checkError(err)
			kcpconn.SetNoDelay(nodelay, interval, resend, nc)
			kcpconn.SetWindowSize(c.Int("sndwnd"), c.Int("rcvwnd"))
			kcpconn.SetMtu(c.Int("mtu"))
			kcpconn.SetACKNoDelay(c.Bool("acknodelay"))
			kcpconn.SetDSCP(c.Int("dscp"))

			// stream multiplex
			config := &yamux.Config{
				AcceptBacklog:          256,
				EnableKeepAlive:        true,
				KeepAliveInterval:      30 * time.Second,
				ConnectionWriteTimeout: 30 * time.Second,
				MaxStreamWindowSize:    16777216,
				LogOutput:              os.Stderr,
			}
			session, err := yamux.Client(kcpconn, config)
			checkError(err)
			return session
		}

		numconn := uint16(c.Int("conn"))
		var muxes []*yamux.Session
		for i := uint16(0); i < numconn; i++ {
			muxes = append(muxes, createConn())
		}

		rr := uint16(0)
		for {
			p1, err := listener.AcceptTCP()
			checkError(err)
			mux := muxes[rr%numconn]
			p2, err := mux.Open()
			if err != nil { // yamux failure
				log.Println(err)
				p1.Close()
				mux.Close()
				muxes[rr%numconn] = createConn()
				continue
			}
			go handleClient(p1, p2)
			rr++
		}
	}
	myApp.Run(os.Args)
}

func main() {
	conn, err := net.Dial("tcp", *address)
	if err != nil {
		log.Fatalf("connection error: %v", err)
	}

	session, err := yamux.Client(conn, nil)
	if err != nil {
		log.Fatalf("session error: %v", err)
	}

	stdin := int(os.Stdin.Fd())
	if !terminal.IsTerminal(stdin) {
		log.Fatal("not on a terminal")
	}
	oldState, err := terminal.MakeRaw(stdin)
	if err != nil {
		log.Fatalf("unable to make terminal raw: %v", err)
	}
	defer terminal.Restore(stdin, oldState)

	done := make(chan struct{})

	controlChannel, err := session.Open()
	if err != nil {
		log.Fatalf("control channel open error: %v", err)
	}
	go func() {
		w := gob.NewEncoder(controlChannel)
		c := make(chan os.Signal, 1)
		signal.Notify(c, syscall.SIGWINCH)
		for {
			cols, rows, err := terminal.GetSize(stdin)
			if err != nil {
				log.Printf("getsize error: %v", err)
				break
			}
			win := struct {
				Rows, Cols int
			}{Rows: rows, Cols: cols}
			if err := w.Encode(win); err != nil {
				break
			}
			<-c
		}
		done <- struct{}{}
	}()

	dataChannel, err := session.Open()
	if err != nil {
		log.Fatalf("data channel open error: %v", err)
	}
	cp := func(dst io.Writer, src io.Reader) {
		io.Copy(dst, src)
		done <- struct{}{}
	}
	go cp(dataChannel, os.Stdin)
	go cp(os.Stdout, dataChannel)

	<-done
	session.Close() // closes controlChannel, dataChannel, session and conn
}