package main
import (
	"bufio"
	"crypto/tls"
	"log"
	"net"
)
func main() {
	cert, err := tls.LoadX509KeyPair("server.pem", "server.key")
	if err != nil {
		log.Println(err)
		return
	}
	config := &tls.Config{Certificates: []tls.Certificate{cert}}
	ln, err := tls.Listen("tcp", ":4433", config)
	if err != nil {
		log.Println(err)
		return
	}
	defer ln.Close()
	for {
		conn, err := ln.Accept()
		if err != nil {
			log.Println(err)
			continue
		}
		go handleConn(conn)
	}
}
func handleConn(conn net.Conn) {
	defer conn.Close()
	r := bufio.NewReader(conn)
	for {
		msg, err := r.ReadString('\n')
		if err != nil {
			log.Println(err)
			return
		}
		println(msg)
		n, err := conn.Write([]byte("world\n"))
		if err != nil {
			log.Println(n, err)
			return
		}
	}
}

package main
import (
	"crypto/tls"
	"log"
)
func main() {
	conf := &tls.Config{
		InsecureSkipVerify: true,
	}
	conn, err := tls.Dial("tcp", "127.0.0.1:4433", conf)
	if err != nil {
		log.Println(err)
		return
	}
	defer conn.Close()
	n, err := conn.Write([]byte("hello\n"))
	if err != nil {
		log.Println(n, err)
		return
	}
	buf := make([]byte, 100)
	n, err = conn.Read(buf)
	if err != nil {
		log.Println(n, err)
		return
	}
	println(string(buf[:n]))
}

package main
import (
	"bufio"
	"crypto/tls"
	"crypto/x509"
	"io/ioutil"
	"log"
	"net"
)
func main() {
	cert, err := tls.LoadX509KeyPair("server.pem", "server.key")
	if err != nil {
		log.Println(err)
		return
	}
	certBytes, err := ioutil.ReadFile("client.pem")
	if err != nil {
		panic("Unable to read cert.pem")
	}
	clientCertPool := x509.NewCertPool()
	ok := clientCertPool.AppendCertsFromPEM(certBytes)
	if !ok {
		panic("failed to parse root certificate")
	}
	config := &tls.Config{
		Certificates: []tls.Certificate{cert},
		ClientAuth:   tls.RequireAndVerifyClientCert,
		ClientCAs:    clientCertPool,
	}
	ln, err := tls.Listen("tcp", ":443", config)
	if err != nil {
		log.Println(err)
		return
	}
	defer ln.Close()
	for {
		conn, err := ln.Accept()
		if err != nil {
			log.Println(err)
			continue
		}
		go handleConn(conn)
	}
}
func handleConn(conn net.Conn) {
	defer conn.Close()
	r := bufio.NewReader(conn)
	for {
		msg, err := r.ReadString('\n')
		if err != nil {
			log.Println(err)
			return
		}
		println(msg)
		n, err := conn.Write([]byte("world\n"))
		if err != nil {
			log.Println(n, err)
			return
		}
	}
}

package main
import (
	"crypto/tls"
	"crypto/x509"
	"io/ioutil"
	"log"
)
func main() {
	cert, err := tls.LoadX509KeyPair("client.pem", "client.key")
	if err != nil {
		log.Println(err)
		return
	}
	certBytes, err := ioutil.ReadFile("client.pem")
	if err != nil {
		panic("Unable to read cert.pem")
	}
	clientCertPool := x509.NewCertPool()
	ok := clientCertPool.AppendCertsFromPEM(certBytes)
	if !ok {
		panic("failed to parse root certificate")
	}
	conf := &tls.Config{
		RootCAs:            clientCertPool,
		Certificates:       []tls.Certificate{cert},
		InsecureSkipVerify: true,
	}
	conn, err := tls.Dial("tcp", "127.0.0.1:443", conf)
	if err != nil {
		log.Println(err)
		return
	}
	defer conn.Close()
	n, err := conn.Write([]byte("hello\n"))
	if err != nil {
		log.Println(n, err)
		return
	}
	buf := make([]byte, 100)
	n, err = conn.Read(buf)
	if err != nil {
		log.Println(n, err)
		return
	}
	println(string(buf[:n]))
}

type HashKey uint32

// 排序支持
type HashKeyOrder []HashKey
func (h HashKeyOrder) Len() int           { return len(h) }
func (h HashKeyOrder) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
func (h HashKeyOrder) Less(i, j int) bool { return h[i] < h[j] }

type HashRing struct {
    // 一致性hash 虚拟node，以及关联的节点
    ring       map[HashKey]string
    // 所有的虚拟node，（重新）生成之后，会排序，以便快速二分查找
    // 通过二分查找，找到比查询key小的最大key，然后查找ring
    sortedKeys []HashKey
    // 原始数据，所有的节点
    nodes      []string
    // 原始数据，所有的节点权重
    weights    map[string]int
}

func New(nodes []string) *HashRing {
    // 声明一个对象
    hashRing := &HashRing{
        ring:       make(map[HashKey]string),
        sortedKeys: make([]HashKey, 0),
        nodes:      nodes,
        weights:    make(map[string]int),
    }
    // 生成虚拟node
    hashRing.generateCircle()
    return hashRing
}
// 和上个函数比，增加权重
func NewWithWeights(weights map[string]int) *HashRing {
    nodes := make([]string, 0, len(weights))
    for node, _ := range weights {
        nodes = append(nodes, node)
    }
    hashRing := &HashRing{
        ring:       make(map[HashKey]string),
        sortedKeys: make([]HashKey, 0),
        nodes:      nodes,
        weights:    weights,
    }
    hashRing.generateCircle()
    return hashRing
}

func (h *HashRing) generateCircle() {
    // 获取总的权重
    totalWeight := 0
    for _, node := range h.nodes {
        if weight, ok := h.weights[node]; ok {
            totalWeight += weight
        } else {
            totalWeight += 1
        }
    }

    // 依次对每个节点生成若干虚拟node
    for _, node := range h.nodes {
        weight := 1

        // 是否指定了权重
        if _, ok := h.weights[node]; ok {
            weight = h.weights[node]
        }

        // 设置虚拟node的数量，40这个乘数因子可调整
        // 数量越大，分布越均匀，但是性能相对较低
        factor := math.Floor(float64(40*len(h.nodes)*weight) / float64(totalWeight))

        // 依次生成虚拟node（里面还会细分成多个，实现细节问题）
        for j := 0; j < int(factor); j++ {
            // 生成一个key（受factor影响）
            nodeKey := fmt.Sprintf("%s-%d", node, j)
            // 生成MD5（16大小的[]byte)
            bKey := hashDigest(nodeKey)
            // 原代码只计算了前面12Byte，可以用完16byte，生成4个虚拟node
            for i := 0; i < 4; i++ {
                // 生成一个虚拟node的key
                key := hashVal(bKey[i*4 : i*4+4])
                // 关联后端的节点
                h.ring[key] = node
                // 插入到查找的slice
                h.sortedKeys = append(h.sortedKeys, key)
            }
        }
    }

    // 排序，方便二分查找
    sort.Sort(HashKeyOrder(h.sortedKeys))
}

func (h *HashRing) GetNode(stringKey string) (node string, ok bool) {
    // 查找key所在的虚拟node
    pos, ok := h.GetNodePos(stringKey)
    if !ok {
        return "", false
    }
    // 返回后端节点
    return h.ring[h.sortedKeys[pos]], true
}

func (h *HashRing) GenKey(key string) HashKey {
    // 查询后端节点时，给key生成hash
    bKey := hashDigest(key)
    //MD5取前四位，比较简单
    return hashVal(bKey[0:4])
}

func (h *HashRing) GetNodePos(stringKey string) (pos int, ok bool) {
    // 无数据
    if len(h.ring) == 0 {
        return 0, false
    }

    // 生成
    key := h.GenKey(stringKey)

    nodes := h.sortedKeys
    // 二分查找key
    pos = sort.Search(len(nodes), func(i int) bool { return nodes[i] > key })

    // 找不到就第一个
    if pos == len(nodes) {
        // Wrap the search, should return first node
        return 0, true
    } else {
        // 否则返回找到的node
        return pos, true
    }
}

// size表示返回的副本数量
func (h *HashRing) GetNodes(stringKey string, size int) (nodes []string, ok bool) {
    // 获得key所在的虚拟node
    pos, ok := h.GetNodePos(stringKey)
    if !ok {
        return []string{}, false
    }

    // 没那么多节点
    if size > len(h.nodes) {
        return []string{}, false
    }

    // 由于节点关联的虚拟node交叉放置，下面的map用于判断是否已经get了这个节点
    returnedValues := make(map[string]bool, size)
    //mergedSortedKeys := append(h.sortedKeys[pos:], h.sortedKeys[:pos]...)
    // 已经get的节点
    resultSlice := make([]string, 0, size)

    // 从key所在的虚拟node开始查找，到尾自动从第一个开始
    for i := pos; i < pos+len(h.sortedKeys); i++ {
        key := h.sortedKeys[i%len(h.sortedKeys)]
        // 检查这个后端节点是否已经保存
        val := h.ring[key]
        if !returnedValues[val] {
            // 没有就保存起来，并且标记已保存
            returnedValues[val] = true
            resultSlice = append(resultSlice, val)
        }
        if len(returnedValues) == size {
            break
        }
    }

    return resultSlice, len(resultSlice) == size
}

func Main() {
    NewController().Run(config)
}

func NewController() *Controller {
    ctl := &Controller{
        Logger:  log.NewPrefixLogger("controller"),
        updates: util.NewBroadcast(),
        cmds:    make(chan command),
        views:   make([]mvc.View, 0),
        state:   make(chan mvc.State),
    }

    return ctl
}

func (ctl *Controller) Run(config *Configuration) {
    // Save the configuration
    ctl.config = config

    var model *ClientModel

    // 创建model（核心逻辑所在）
    if ctl.model == nil {
        // 不存在就创建
        model = ctl.SetupModel(config)
    } else {
        model = ctl.model.(*ClientModel)
    }

    // init the model
    var state mvc.State = model

    // 初始化web view
    var webView *web.WebView
    if config.InspectAddr != "disabled" {
        webView = web.NewWebView(ctl, config.InspectAddr)
        ctl.AddView(webView)
    }

    // 初始化term view
    var termView *term.TermView
    if config.LogTo != "stdout" {
        termView = term.NewTermView(ctl)
        ctl.AddView(termView)
    }

    // 将view关联到controller
    for _, protocol := range model.GetProtocols() {
        switch p := protocol.(type) {
        case *proto.Http:
            if termView != nil {
                ctl.AddView(termView.NewHttpView(p))
            }

            if webView != nil {
                ctl.AddView(webView.NewHttpView(p))
            }
        default:
        }
    }

    // 核心逻辑入口
    ctl.Go(ctl.model.Run)

    done := make(chan int)
    for {
        select {
        case obj := <-ctl.cmds:
            // 关闭command
            switch cmd := obj.(type) {
            case cmdQuit:
                msg := cmd.message
                go func() {
                    // 等待退出
                    ctl.doShutdown()
                    fmt.Println(msg)
                    done <- 1
                }()

            // 回放命令
            case cmdPlayRequest:
                ctl.Go(func() { ctl.model.PlayRequest(cmd.tunnel, cmd.payload) })
            }

        // 更新state
        case obj := <-updates:
            state = obj.(mvc.State)

        case ctl.state <- state:
        // 退出
        case <-done:
            return
        }
    }
}

func (c *ClientModel) Run() {
    // how long we should wait before we reconnect
    maxWait := 30 * time.Second
    wait := 1 * time.Second

    for {
        // 开始发起连接请求，响应报文等
        // 注意是阻塞的，若该函数返回了，说明掉线了
        c.control()

        // 失败后，第一次状态未变，此时重置wait的时间间隔
        if c.connStatus == mvc.ConnOnline {
            wait = 1 * time.Second
        }

        // sleep，避免无畏的不停请求，浪费服务器资源
        time.Sleep(wait)
        // 失败了继续加大重试的间隔
        wait = 2 * wait
        wait = time.Duration(math.Min(float64(wait), float64(maxWait)))
        // 设置状态
        c.connStatus = mvc.ConnReconnecting
        // 刷新各种view
        c.update()
    }
}

func (ctl *Controller) doShutdown() {
    ctl.Info("Shutting down")

    var wg sync.WaitGroup

    // wait for all of the views, plus the model
    wg.Add(len(ctl.views) + 1)

    // 关闭所有的view
    for _, v := range ctl.views {
        vClosure := v
        ctl.Go(func() {
            vClosure.Shutdown()
            wg.Done()
        })
    }

    // 关闭model（核心逻辑）
    ctl.Go(func() {
        ctl.model.Shutdown()
        wg.Done()
    })

    // 用WaitGroup等待多个goruntune
    wg.Wait()
}

func (c *ClientModel) control() {
    // establish control channel
    var (
        ctlConn conn.Conn
        err     error
    )

    // 向服务器发起连接
    if c.proxyUrl == "" {
        // simple non-proxied case, just connect to the server
        ctlConn, err = conn.Dial(c.serverAddr, "ctl", c.tlsConfig)
    } else {
        ctlConn, err = conn.DialHttpProxy(c.proxyUrl, c.serverAddr, "ctl", c.tlsConfig)
    }

    // authenticate with the server
    auth := &msg.Auth{
        ClientId:  c.id,
        OS:        runtime.GOOS,
        Arch:      runtime.GOARCH,
        Version:   version.Proto,
        MmVersion: version.MajorMinor(),
        User:      c.authToken,
    }

    // 发送连接报文
    if err = msg.WriteMsg(ctlConn, auth); err != nil {
        panic(err)
    }

    // 等待响应
    var authResp msg.AuthResp
    if err = msg.ReadMsgInto(ctlConn, &authResp); err != nil {
        panic(err)
    }

    if authResp.Error != "" {
        emsg := fmt.Sprintf("Failed to authenticate to server: %s", authResp.Error)
        c.ctl.Shutdown(emsg)
        return
    }

    c.id = authResp.ClientId
    c.serverVersion = authResp.MmVersion
    c.Info("Authenticated with server, client id: %v", c.id)
    // 更新view
    c.update()
    // 更新配置
    if err = SaveAuthToken(c.configPath, c.authToken); err != nil {
        c.Error("Failed to save auth token: %v", err)
    }

    // request tunnels
    reqIdToTunnelConfig := make(map[string]*TunnelConfiguration)
    for _, config := range c.tunnelConfig {
        // create the protocol list to ask for
        var protocols []string
        for proto, _ := range config.Protocols {
            protocols = append(protocols, proto)
        }

        // 注册tunnel
        reqTunnel := &msg.ReqTunnel{
            ReqId:      util.RandId(8),
            Protocol:   strings.Join(protocols, "+"),
            Hostname:   config.Hostname,
            Subdomain:  config.Subdomain,
            HttpAuth:   config.HttpAuth,
            RemotePort: config.RemotePort,
        }

        // 发送请求
        if err = msg.WriteMsg(ctlConn, reqTunnel); err != nil {
            panic(err)
        }

        // 存储这些配置，后面新建tunnel需要
        reqIdToTunnelConfig[reqTunnel.ReqId] = config
    }

    // 开始心跳
    lastPong := time.Now().UnixNano()
    c.ctl.Go(func() { c.heartbeat(&lastPong, ctlConn) })

    // main control loop
    for {
        var rawMsg msg.Message
        if rawMsg, err = msg.ReadMsg(ctlConn); err != nil {
            panic(err)
        }

        switch m := rawMsg.(type) {
        case *msg.ReqProxy:
            // 收到proxy请求，就向服务器发起一个proxy请求
            c.ctl.Go(c.proxy)

        case *msg.Pong:
            // 更新心跳
            atomic.StoreInt64(&lastPong, time.Now().UnixNano())

        case *msg.NewTunnel:
            // 注册tunnel确认
            if m.Error != "" {
                emsg := fmt.Sprintf("Server failed to allocate tunnel: %s", m.Error)
                c.Error(emsg)
                c.ctl.Shutdown(emsg)
                continue
            }

            tunnel := mvc.Tunnel{
                PublicUrl: m.Url,
                LocalAddr: reqIdToTunnelConfig[m.ReqId].Protocols[m.Protocol],
                Protocol:  c.protoMap[m.Protocol],
            }

            // 保存tunnel对象，用于view的数据呈现
            // 另外是申请proxy时，作校验
            c.tunnels[tunnel.PublicUrl] = tunnel
            // 更新状态
            c.connStatus = mvc.ConnOnline
            // 同步到view
            c.update()

        default:
            ctlConn.Warn("Ignoring unknown control message %v ", m)
        }
    }
}

func (c *ClientModel) heartbeat(lastPongAddr *int64, conn conn.Conn) {
    lastPing := time.Unix(atomic.LoadInt64(lastPongAddr)-1, 0)
    ping := time.NewTicker(pingInterval)
    pongCheck := time.NewTicker(time.Second)

    for {
        select {
        // 检查是否过期
        case <-pongCheck.C:
            lastPong := time.Unix(0, atomic.LoadInt64(lastPongAddr))
            needPong := lastPong.Sub(lastPing) < 0
            pongLatency := time.Since(lastPing)

            if needPong && pongLatency > maxPongLatency {
                c.Info("Last ping: %v, Last pong: %v", lastPing, lastPong)
                c.Info("Connection stale, haven't gotten PongMsg in %d seconds", int(pongLatency.Seconds()))
                return
            }

        // 定期发送心跳
        case <-ping.C:
            err := msg.WriteMsg(conn, &msg.Ping{})
            if err != nil {
                conn.Debug("Got error %v when writing PingMsg", err)
                return
            }
            lastPing = time.Now()
        }
    }
}

func (c *ClientModel) proxy() {
    var (
        remoteConn conn.Conn
        err        error
    )

    // 尝试连接服务器
    if c.proxyUrl == "" {
        remoteConn, err = conn.Dial(c.serverAddr, "pxy", c.tlsConfig)
    } else {
        remoteConn, err = conn.DialHttpProxy(c.proxyUrl, c.serverAddr, "pxy", c.tlsConfig)
    }
    defer remoteConn.Close()

    // 发送响应报文
    err = msg.WriteMsg(remoteConn, &msg.RegProxy{ClientId: c.id})
    if err != nil {
        remoteConn.Error("Failed to write RegProxy: %v", err)
        return
    }

    // 等待具体的业务报文，
    // 之所以需要这么个报文，是因为需要这个报文中的内容来定位tunnel，进而获取本地端的参数
    // 因为客户端也是一个数据中转，它一样有下游节点
    var startPxy msg.StartProxy
    if err = msg.ReadMsgInto(remoteConn, &startPxy); err != nil {
        remoteConn.Error("Server failed to write StartProxy: %v", err)
        return
    }

    // 找到tunnel
    tunnel, ok := c.tunnels[startPxy.Url]
    if !ok {
        remoteConn.Error("Couldn't find tunnel for proxy: %s", startPxy.Url)
        return
    }

    // 连接本地端
    start := time.Now()
    localConn, err := conn.Dial(tunnel.LocalAddr, "prv", nil)
    if err != nil {
        remoteConn.Warn("Failed to open private leg %s: %v", tunnel.LocalAddr, err)

        if tunnel.Protocol.GetName() == "http" {
            // ...
        }
        return
    }
    defer localConn.Close()

    m := c.metrics
    m.proxySetupTimer.Update(time.Since(start))
    m.connMeter.Mark(1)
    // 更新view
    c.update()
    m.connTimer.Time(func() {
        localConn := tunnel.Protocol.WrapConn(localConn,  \
            mvc.ConnectionContext{Tunnel: tunnel, ClientAddr: startPxy.ClientAddr})
        // 数据交换
        bytesIn, bytesOut := conn.Join(localConn, remoteConn)
        m.bytesIn.Update(bytesIn)
        m.bytesOut.Update(bytesOut)
        m.bytesInCount.Inc(bytesIn)
        m.bytesOutCount.Inc(bytesOut)
    })
    c.update()
}

type Controller struct {
    // the model sends updates through this broadcast channel
    updates *util.Broadcast
}

func NewBroadcast() *Broadcast {
    b := &Broadcast{
        listeners: make([]chan interface{}, 0),
        reg:       make(chan (chan interface{})),
        unreg:     make(chan (chan interface{})),
        in:        make(chan interface{}),
    }

    go func() {
        for {
            select {
            // 取消注册
            case l := <-b.unreg:
                // remove L from b.listeners
                // this operation is slow: O(n) but not used frequently
                // unlike iterating over listeners
                oldListeners := b.listeners
                b.listeners = make([]chan interface{}, 0, len(oldListeners))
                // 这个删除操作很蛋疼
                for _, oldL := range oldListeners {
                    if l != oldL {
                        b.listeners = append(b.listeners, oldL)
                    }
                }
            // 注册操作
            case l := <-b.reg:
                b.listeners = append(b.listeners, l)
            // 刷新操作
            case item := <-b.in:
                for _, l := range b.listeners {
                    l <- item
                }
            }
        }
    }()

    return b
}

// 对外的刷新channel
func (b *Broadcast) In() chan interface{} {
    return b.in
}

// 生成一个注册channel，用于注册
func (b *Broadcast) Reg() chan interface{} {
    listener := make(chan interface{})
    b.reg <- listener
    return listener
}

// 取消注册一个channel
func (b *Broadcast) UnReg(listener chan interface{}) {
    b.unreg <- listener
}

func (ctl *Controller) Update(state mvc.State) {
    ctl.updates.In() <- state
}

func NewWebView(ctl mvc.Controller, addr string) *WebView {
    // handle web socket connections
    http.HandleFunc("/_ws", func(w http.ResponseWriter, r *http.Request) {
        // 注册一个channel
        msgs := wv.wsMessages.Reg()
        // 退出是取消注册
        defer wv.wsMessages.UnReg(msgs)
        // 监听这个channel
        for m := range msgs {
            err := conn.WriteMessage(websocket.TextMessage, m.([]byte))
            if err != nil {
                // connection is closed
                break
            }
        }
    })
    return wv
}

for _, protocol := range model.GetProtocols() {
    switch p := protocol.(type) {
    case *proto.Http:
        if termView != nil {
            ctl.AddView(termView.NewHttpView(p))
        }

        if webView != nil {
            ctl.AddView(webView.NewHttpView(p))
        }
    default:
    }
}

func (v *TermView) NewHttpView(p *proto.Http) *HttpView {
    return newTermHttpView(v.ctl, v, p, 0, 12)
}
func (wv *WebView) NewHttpView(proto *proto.Http) *WebHttpView {
    return newWebHttpView(wv.ctl, wv, proto)
}

func newWebHttpView(ctl mvc.Controller, wv *WebView, proto *proto.Http) *WebHttpView {
    whv := &WebHttpView{
        Logger:       log.NewPrefixLogger("view", "web", "http"),
        webview:      wv,
        ctl:          ctl,
        httpProto:    proto,
        idToTxn:      make(map[string]*SerializedTxn),
        HttpRequests: util.NewRing(20),
    }
    // 实时刷新
    ctl.Go(whv.updateHttp)
    whv.register()
    return whv
}

func (whv *WebHttpView) updateHttp() {
    // open channels for incoming http state changes
    // and broadcasts
    txnUpdates := whv.httpProto.Txns.Reg()
    // 监听whv.httpProto.Txns
    for txn := range txnUpdates {
        // 获得实际的对象
        htxn := txn.(*proto.HttpTxn)

        // 。。。 刷新操作
}

type Http struct {
    // 可以看到Txns是一个广播对象
    Txns     *util.Broadcast
    reqGauge metrics.Gauge
    reqMeter metrics.Meter
    reqTimer metrics.Timer
}

localConn := tunnel.Protocol.WrapConn(localConn, param)

func (h *Http) WrapConn(c conn.Conn, ctx interface{}) conn.Conn {
    tee := conn.NewTee(c)
    // 用一个管道将获取Request和Response串起来
    lastTxn := make(chan *HttpTxn)
    // 读取Request
    go h.readRequests(tee, lastTxn, ctx)
    // 读取Response
    go h.readResponses(tee, lastTxn)
    return tee
}

func (h *Http) readRequests(tee *conn.Tee, lastTxn chan *HttpTxn, connCtx interface{}) {
    defer close(lastTxn)

    for {
        // 不停地从tee的写tee中解析Request
        req, err := http.ReadRequest(tee.WriteBuffer())
        if err != nil {
            // no more requests to be read, we're done
            break
        }

        // golang's ReadRequest/DumpRequestOut is broken. Fix up the request so it works later
        req.URL.Scheme = "http"
        req.URL.Host = req.Host

        // 生成一个HttpTxn对象
        txn := &HttpTxn{Start: time.Now(), ConnUserCtx: connCtx}
        txn.Req = &HttpRequest{Request: req}
        if req.Body != nil {
            txn.Req.BodyBytes, txn.Req.Body, err = extractBody(req.Body)
            if err != nil {
                tee.Warn("Failed to extract request body: %v", err)
            }
        }

        // 发送到Req/Resp共享的channel，通知resp逻辑解析Response
        lastTxn <- txn
        // 通知view刷新
        h.Txns.In() <- txn
    }
}

func (h *Http) readResponses(tee *conn.Tee, lastTxn chan *HttpTxn) {
    for txn := range lastTxn {
        // 当req解析完之后，会触发resp解析
        // 从tee的读tee中不停地解析Response
        resp, err := http.ReadResponse(tee.ReadBuffer(), txn.Req.Request)
        txn.Duration = time.Since(txn.Start)
        h.reqTimer.Update(txn.Duration)
        if err != nil {
            tee.Warn("Error reading response from server: %v", err)
            // no more responses to be read, we're done
            break
        }

        // 更新HttpTxn对象的Response
        txn.Resp = &HttpResponse{Response: resp}
        // apparently, Body can be nil in some cases
        if resp.Body != nil {
            txn.Resp.BodyBytes, txn.Resp.Body, err = extractBody(resp.Body)
            if err != nil {
                tee.Warn("Failed to extract response body: %v", err)
            }
        }

        // 通知view刷新
        h.Txns.In() <- txn
    }
}

func NewTee(conn Conn) *Tee {
    c := &Tee{
        rd:   nil,
        wr:   nil,
        Conn: conn,
    }

    c.readPipe.rd, c.readPipe.wr = io.Pipe()
    c.writePipe.rd, c.writePipe.wr = io.Pipe()

    // 读取的时候，一并拷贝一份到c.readPipe.wr，那么c.readPipe.rd就可读
    // 参考ReadBuffer
    c.rd = io.TeeReader(c.Conn, c.readPipe.wr)
    // 当写入的时候，一并写入c.writePipe.wr，那么c.writePipe.rz就可读
    // 参考WriteBuffer
    c.wr = io.MultiWriter(c.Conn, c.writePipe.wr)
    return c
}

func (c *Tee) ReadBuffer() *bufio.Reader {
    return bufio.NewReader(c.readPipe.rd)
}

func (c *Tee) WriteBuffer() *bufio.Reader {
    return bufio.NewReader(c.writePipe.rd)
}

// GLOBALS
var (
	// 所有的tunnel
	tunnelRegistry  *TunnelRegistry
	// 所有的controller
	controlRegistry *ControlRegistry

	// 参数
	opts	  *Options
	// 所有的tcp服务器
	listeners map[string]*conn.Listener
)

// ControlRegistry maps a client ID to Control structures
type ControlRegistry struct {
	controls map[string]*Control
	log.Logger
	sync.RWMutex
}
// TunnelRegistry maps a tunnel URL to Tunnel structures
type TunnelRegistry struct {
	tunnels  map[string]*Tunnel
	affinity *cache.LRUCache
	log.Logger
	sync.RWMutex
}

type Tunnel struct {
	// request that opened the tunnel
	req *msg.ReqTunnel

	// ...
	// 关联到control
	ctl *Control
	// ...
}

type Listener struct {
	net.Addr
	// 将请求accept的新tcp连接放入channel
	Conns chan *loggedConn
}

type loggedConn struct {
	tcp *net.TCPConn
	net.Conn
	log.Logger
	id  int32
	typ string
}

func wrapConn(conn net.Conn, typ string) *loggedConn {
	switch c := conn.(type) {
	case *vhost.HTTPConn:
		wrapped := c.Conn.(*loggedConn)
		return &loggedConn{wrapped.tcp, conn, wrapped.Logger, wrapped.id, wrapped.typ}
	case *loggedConn:
		return c
	case *net.TCPConn:
		wrapped := &loggedConn{c, conn, log.NewPrefixLogger(), rand.Int31(), typ}
		wrapped.AddLogPrefix(wrapped.Id())
		return wrapped
	}

	return nil
}

func Listen(addr, typ string, tlsCfg *tls.Config) (l *Listener, err error) {
	// listen for incoming connections
	listener, err := net.Listen("tcp", addr)
	// 。。。
	l = &Listener{
		Addr:  listener.Addr(),
		Conns: make(chan *loggedConn),
	}

	go func() {
		for {
			rawConn, err := listener.Accept()
			c := wrapConn(rawConn, typ)
			// 新的连接放入channel
			l.Conns <- c
		}
	}()
	return
}

func tunnelListener(addr string, tlsConfig *tls.Config) {
	// 建立tcp服务器
	listener, err := conn.Listen(addr, "tun", tlsConfig)
	if err != nil {
		panic(err)
	}

	// 从channel中等待新的请求到来
	for c := range listener.Conns {
		// 每个连接用新的goroutune
		go func(tunnelConn conn.Conn) {
			defer func() {
				// 自动处理异常
				if r := recover(); r != nil {
					tunnelConn.Info("tunnelListener failed with error %v: %s", r, debug.Stack())
				}
			}()

			// 读取消息
			var rawMsg msg.Message
			if rawMsg, err = msg.ReadMsg(tunnelConn); err != nil {
				return
			}

			switch m := rawMsg.(type) {
			case *msg.Auth:
				// 建立控制连接（controller）
				NewControl(tunnelConn, m)

			case *msg.RegProxy:
				// 建立数据连接
				NewProxy(tunnelConn, m)

			default:
				tunnelConn.Close()
			}
		}(c)
	}
}

func NewControl(ctlConn conn.Conn, authMsg *msg.Auth) {
	var err error

	// create the object
	c := &Control{
		auth:			authMsg,
		conn:			ctlConn,
		out:			 make(chan msg.Message),
		in:			  make(chan msg.Message),
		proxies:		 make(chan conn.Conn, 10),
		lastPing:		time.Now(),
		writerShutdown:  util.NewShutdown(),
		readerShutdown:  util.NewShutdown(),
		managerShutdown: util.NewShutdown(),
		shutdown:		util.NewShutdown(),
	}

	// 设置属性
	ctlConn.SetType("ctl")
	ctlConn.AddLogPrefix(c.id)

	// 版本判断
	if authMsg.Version != version.Proto {
		failAuth(fmt.Errorf("Incompatible versions. Server %s, client %s. Download a new version at http://ngrok.com", version.MajorMinor(), authMsg.Version))
		return
	}

	// 新增/更新control到全局Registry
	if replaced := controlRegistry.(c.id, c); replaced != nil {
		// 等待旧的完全关闭
		replaced.shutdown.WaitComplete()
	}

	// 新的goruntune监听写（需要最先开启，以便回复连接请求报文）
	go c.writer()

	// 回复连接建立确认报文
	c.out <- &msg.AuthResp{
		Version:   version.Proto,
		MmVersion: version.MajorMinor(),
		ClientId:  c.id,
	}

	// 预先申请一个proxy连接
	c.out <- &msg.ReqProxy{}

	// 一堆其他的后台goroutune
	go c.manager()
	go c.reader()
	go c.stopper()
}

// 发送控制消息
func (c *Control) writer() {
	for m := range c.out {
		if err := msg.WriteMsg(c.conn, m); err != nil {
			panic(err)
		}
	}
}

// 接收控制消息
func (c *Control) reader() {
	for {
		if msg, err := msg.ReadMsg(c.conn); err != nil {
			if err == io.EOF {
				c.conn.Info("EOF")
				return
			}
		} else {
			// 推送到c.in channel中
			c.in <- msg
		}
	}
}

func (c *Control) manager() {
	// 检查control超时
	reap := time.NewTicker(connReapInterval)

	for {
		select {
		case <-reap.C:
			// 检查是否超时
			if time.Since(c.lastPing) > pingTimeoutInterval {
				c.conn.Info("Lost heartbeat")
				c.shutdown.Begin()
			}
		case mRaw, ok := <-c.in:
			// 是否有新的消息
			if !ok {
				return
			}

			switch m := mRaw.(type) {
			case *msg.ReqTunnel:
				// 客户端注册一个新的tunnel
				c.registerTunnel(m)

			case *msg.Ping:
				// 回复心跳
				c.lastPing = time.Now()
				c.out <- &msg.Pong{}
			}
		}
	}
}

func (r *ControlRegistry) Add(clientId string, ctl *Control) (oldCtl *Control) {
	r.Lock()
	defer r.Unlock()

	oldCtl = r.controls[clientId]
	if oldCtl != nil {
		oldCtl.Replaced(ctl)
	}

	r.controls[clientId] = ctl
	r.Info("Registered control with id %s", clientId)
	return
}

type Control struct {
	// synchronizer for controlled shutdown of writer()
	writerShutdown *util.Shutdown

	// synchronizer for controlled shutdown of reader()
	readerShutdown *util.Shutdown

	// synchronizer for controlled shutdown of manager()
	managerShutdown *util.Shutdown

	// synchronizer for controller shutdown of entire Control
	shutdown *util.Shutdown
}

func (c *Control) reader() {
	// kill everything if the reader stops
	defer c.shutdown.Begin()
	// notify that we're done
	defer c.readerShutdown.Complete()
}

func (c *Control) writer() {
	// kill everything if the writer() stops
	defer c.shutdown.Begin()

	// notify that we've flushed all messages
	defer c.writerShutdown.Complete()
}

func (c *Control) manager() {
	// kill everything if the control manager stops
	defer c.shutdown.Begin()

	// notify that manager() has shutdown
	defer c.managerShutdown.Complete()
}

func (c *Control) stopper() {
	// 等待
	c.shutdown.WaitBegin()

	// 注销controller
	controlRegistry.Del(c.id)

	// 等待各种子goruntune注销
	// close会触发其他的goruntune退出
	close(c.in)
	c.managerShutdown.WaitComplete()

	// shutdown writer()
	close(c.out)
	c.writerShutdown.WaitComplete()

	// 关闭空置连接
	c.conn.Close()

	// 关闭各种tunnel
	for _, t := range c.tunnels {
		// 调用shutdown，
		t.Shutdown()
	}

	// 关闭各种proxy连接
	close(c.proxies)
	for p := range c.proxies {
		p.Close()
	}

	// 最终关闭
	c.shutdown.Complete()
	c.conn.Info("Shutdown complete")
}

// Register a new tunnel on this control connection
func (c *Control) registerTunnel(rawTunnelReq *msg.ReqTunnel) {
	// 若有多个tunnel，可以用一个报文一次性注册
	for _, proto := range strings.Split(rawTunnelReq.Protocol, "+") {
		tunnelReq := *rawTunnelReq
		tunnelReq.Protocol = proto

		c.conn.Debug("Registering new tunnel")
		t, err := NewTunnel(&tunnelReq, c)
		if err != nil {
			// 回复注册失败确认
			c.out <- &msg.NewTunnel{Error: err.Error()}
			return
		}

		// 注册到controller
		c.tunnels = append(c.tunnels, t)

		// 回复注册成功确认
		c.out <- &msg.NewTunnel{
			Url:	  t.url,
			Protocol: proto,
			ReqId:	rawTunnelReq.ReqId,
		}

		rawTunnelReq.Hostname = strings.Replace(t.url, proto+"://", "", 1)
	}
}

// Create a new tunnel from a registration message received
// on a control channel
func NewTunnel(m *msg.ReqTunnel, ctl *Control) (t *Tunnel, err error) {
	t = &Tunnel{
		req:	m,
		start:  time.Now(),
		ctl:	ctl,
		Logger: log.NewPrefixLogger(),
	}

	proto := t.req.Protocol
	switch proto {
	case "tcp":
		// ...
		return

	case "http", "https":
		l, ok := listeners[proto]
		// 注册vhost，之所以v，是因为多个url共享同一个端口，类似于nginx的server
		if err = registerVhost(t, proto, l.Addr.(*net.TCPAddr).Port); err != nil {
			return
		}

	default:
	}
	t.AddLogPrefix(t.Id())
	return
}

var defaultPortMap = map[string]int{
	"http":  80,
	"https": 443,
	"smtp":  25,
}

// Common functionality for registering virtually hosted protocols
func registerVhost(t *Tunnel, protocol string, servingPort int) (err error) {
	vhost := os.Getenv("VHOST")
	if vhost == "" {
		vhost = fmt.Sprintf("%s:%d", opts.domain, servingPort)
	}

	// Canonicalize virtual host by removing default port (e.g. :80 on HTTP)
	defaultPort, ok := defaultPortMap[protocol]
	if !ok {
		return fmt.Errorf("Couldn't find default port for protocol %s", protocol)
	}

	// 移除默认的端口（比如80可以忽略，81就必须明确地出现在连接中）
	defaultPortSuffix := fmt.Sprintf(":%d", defaultPort)
	if strings.HasSuffix(vhost, defaultPortSuffix) {
		vhost = vhost[0 : len(vhost)-len(defaultPortSuffix)]
	}

	// Canonicalize by always using lower-case
	vhost = strings.ToLower(vhost)

	// 从请求中获取host
	hostname := strings.ToLower(strings.TrimSpace(t.req.Hostname))
	if hostname != "" {
		t.url = fmt.Sprintf("%s://%s", protocol, hostname)
		// 注册tunnel
		return tunnelRegistry.Register(t.url, t)
	}

	// 未指定host，指定了subdomain，就通过服务器启动参数domain来拼装host
	subdomain := strings.ToLower(strings.TrimSpace(t.req.Subdomain))
	if subdomain != "" {
		t.url = fmt.Sprintf("%s://%s.%s", protocol, subdomain, vhost)
		// 注册tunnel
		return tunnelRegistry.Register(t.url, t)
	}

	// 随机生成一个url
	t.url, err = tunnelRegistry.RegisterRepeat(func() string {
		return fmt.Sprintf("%s://%x.%s", protocol, rand.Int31(), vhost)
	}, t)

	return
}

func (t *Tunnel) Shutdown() {
	// 取消注册
	tunnelRegistry.Del(t.url)
}

func NewControl(ctlConn conn.Conn, authMsg *msg.Auth) {
	// 预先申请一个proxy
	c.out <- &msg.ReqProxy{}
}

func tunnelListener(addr string, tlsConfig *tls.Config) {
	for c := range listener.Conns {
		go func(tunnelConn conn.Conn) {
			switch m := rawMsg.(type) {
			case *msg.RegProxy:
				// 新增一个proxy
				NewProxy(tunnelConn, m)
			}
		}(c)
	}
}

type Control struct {
	// proxy connections
	proxies chan conn.Conn
}

func NewControl(ctlConn conn.Conn, authMsg *msg.Auth) {
	c := &Control{
		// 10个元素的channel
		proxies:		 make(chan conn.Conn, 10),
	}
}

func NewProxy(pxyConn conn.Conn, regPxy *msg.RegProxy) {
	pxyConn.SetType("pxy")

	// 查询controller
	ctl := controlRegistry.Get(regPxy.ClientId)

	// 注册
	ctl.RegisterProxy(pxyConn)
}

func (c *Control) GetProxy() (proxyConn conn.Conn, err error) {
	var ok bool

	// get a proxy connection from the pool
	select {
	// 直接从channel中获取
	case proxyConn, ok = <-c.proxies:
		if !ok {
			err = fmt.Errorf("No proxy connections available, control is closing")
			return
		}
	default:
		//  没有的话，立即请求客户端
		if err = util.PanicToError(func() { c.out <- &msg.ReqProxy{} }); err != nil {
			return
		}

		// 继续从channle中获取
		select {
		case proxyConn, ok = <-c.proxies:
			if !ok {
				err = fmt.Errorf("No proxy connections available, control is closing")
				return
			}

		case <-time.After(pingTimeoutInterval):
			err = fmt.Errorf("Timeout trying to get proxy connection")
			return
		}
	}
	return
}

func Main() {
	// listen for http
	if opts.httpAddr != "" {
		listeners["http"] = startHttpListener(opts.httpAddr, nil)
	}
}

// Listens for new http(s) connections from the public internet
func startHttpListener(addr string, tlsCfg *tls.Config) (listener *conn.Listener) {
	// 创建服务器
	var err error
	if listener, err = conn.Listen(addr, "pub", tlsCfg); err != nil {
		panic(err)
	}

	proto := "http"
	if tlsCfg != nil {
		proto = "https"
	}

	log.Info("Listening for public %s connections on %v", proto, listener.Addr.String())
	go func() {
		for conn := range listener.Conns {
			// 每个连接都会调用下面的goruntune
			go httpHandler(conn, proto)
		}
	}()

	return
}

func httpHandler(c conn.Conn, proto string) {

	// 获取Http头
	vhostConn, err := vhost.HTTP(c)
	if err != nil {
		c.Warn("Failed to read valid %s request: %v", proto, err)
		c.Write([]byte(BadRequest))
		return
	}

	// 获取http参数
	host := strings.ToLower(vhostConn.Host())
	auth := vhostConn.Request.Header.Get("Authorization")

	// done reading mux data, free up the request memory
	vhostConn.Free()

	// We need to read from the vhost conn now since it mucked around reading the stream
	c = conn.Wrap(vhostConn, "pub")

	// 从全局的Registry中查找tunnel
	tunnel := tunnelRegistry.Get(fmt.Sprintf("%s://%s", proto, host))
	if tunnel == nil {
		c.Info("No tunnel found for hostname %s", host)
		c.Write([]byte(fmt.Sprintf(NotFound, len(host)+18, host)))
		return
	}

	// 检查认证
	if tunnel.req.HttpAuth != "" && auth != tunnel.req.HttpAuth {
		c.Info("Authentication failed: %s", auth)
		c.Write([]byte(NotAuthorized))
		return
	}

	// 数据交换
	tunnel.HandlePublicConnection(c)
}

const (
	initVhostBufSize = 1024 // allocate 1 KB up front to try to avoid resizing
)

type sharedConn struct {
	sync.Mutex
	net.Conn			   // the raw connection
	vhostBuf *bytes.Buffer // all of the initial data that has to be read in order to vhost a connection is saved here
}

func newShared(conn net.Conn) (*sharedConn, io.Reader) {
	c := &sharedConn{
		Conn:	 conn,
		// 分配一块内存，用于存储http头，以便原原本本的下发到客户端
		vhostBuf: bytes.NewBuffer(make([]byte, 0, initVhostBufSize)),
	}

	// 当从conn读取数据后，复制一份到vhostBuf
	return c, io.TeeReader(conn, c.vhostBuf)
}

func (c *sharedConn) Read(p []byte) (n int, err error) {
	c.Lock()
	// 已经读取到内存中的数据已经发送完
	if c.vhostBuf == nil {
		c.Unlock()
		return c.Conn.Read(p)
	}

	// 优先从buf中读取
	n, err = c.vhostBuf.Read(p)

	// end of the request buffer
	if err == io.EOF {
		// let the request buffer get garbage collected
		// and make sure we don't read from it again
		c.vhostBuf = nil

		// 读了一半继续从con中读取
		// continue reading from the connection
		var n2 int
		n2, err = c.Conn.Read(p[n:])

		// update total read
		n += n2
	}
	c.Unlock()
	return
}

func HTTP(conn net.Conn) (httpConn *HTTPConn, err error) {
	// 创建一个tee的conn
	c, rd := newShared(conn)

	// 解析Http头，不得不说go的系统库很牛X
	httpConn = &HTTPConn{sharedConn: c}
	if httpConn.Request, err = http.ReadRequest(bufio.NewReader(rd)); err != nil {
		return
	}

	// body不需要
	httpConn.Request.Body.Close()
	return
}

func (t *Tunnel) HandlePublicConnection(publicConn conn.Conn) {
	for i := 0; i < (2 * proxyMaxPoolSize); i++ {
		// 获取一个proxy
		if proxyConn, err = t.ctl.GetProxy(); err != nil {
			t.Warn("Failed to get proxy connection: %v", err)
			return
		}
		// 自动关闭和设置属性
		defer proxyConn.Close()
		t.Info("Got proxy connection %s", proxyConn.Id())
		proxyConn.AddLogPrefix(t.Id())

		// 发送数据，提示开始传输数据
		startPxyMsg := &msg.StartProxy{
			Url:		t.url,
			ClientAddr: publicConn.RemoteAddr().String(),
		}
		if err = msg.WriteMsg(proxyConn, startPxyMsg); err != nil {
			proxyConn.Warn("Failed to write StartProxyMessage: %v, attempt %d", err, i)
			proxyConn.Close()
		} else {
			// success
			break
		}
	}

	// 立即申请一个新的proxy
	util.PanicToError(func() { t.ctl.out <- &msg.ReqProxy{} })

	// 交换数据
	bytesIn, bytesOut := conn.Join(publicConn, proxyConn)
}

func Join(c Conn, c2 Conn) (int64, int64) {
	var wait sync.WaitGroup

	pipe := func(to Conn, from Conn, bytesCopied *int64) {
		defer to.Close()
		defer from.Close()
		defer wait.Done()

		var err error
		// 这个地方可以优化，在内部有一个应用层的buf作中转
		*bytesCopied, err = io.Copy(to, from)
		if err != nil {
			from.Warn("Copied %d bytes to %s before failing with error %v", *bytesCopied, to.Id(), err)
		} else {
			from.Debug("Copied %d bytes to %s", *bytesCopied, to.Id())
		}
	}

	wait.Add(2)
	var fromBytes, toBytes int64
	// 开启两个goruntune来实现双向交换
	go pipe(c, c2, &fromBytes)
	go pipe(c2, c, &toBytes)
	c.Info("Joined with connection %s", c2.Id())
	wait.Wait()
	return fromBytes, toBytes
}

func Listen(addr, typ string, tlsCfg *tls.Config) (l *Listener, err error) {
	// 监听tcp端口
	listener, err := net.Listen("tcp", addr)

	// 声明对象
	l = &Listener{
		Addr:  listener.Addr(),
		Conns: make(chan *loggedConn),
	}

	go func() {
		for {
			rawConn, err := listener.Accept()

			// 处理新连接
			c := wrapConn(rawConn, typ)
			// 若指定了tls配置（https tunnel必需）
			if tlsCfg != nil {
				// 直接将原来的裸tcp conn替换成被tls包装过的conn，所有的read/write会被tls层先做加工再下发
				c.Conn = tls.Server(c.Conn, tlsCfg)
			}
			l.Conns <- c
		}
	}()
	return
}

func Dial(addr, typ string, tlsCfg *tls.Config) (conn *loggedConn, err error) {
	var rawConn net.Conn
	if rawConn, err = net.Dial("tcp", addr); err != nil {
		return
	}

	conn = wrapConn(rawConn, typ)
	conn.Debug("New connection to: %v", rawConn.RemoteAddr())

	// 若指定了tls配置（https tunnel必需）
	if tlsCfg != nil {
		conn.StartTLS(tlsCfg)
	}
	return
}
func (c *loggedConn) StartTLS(tlsCfg *tls.Config) {
	// 直接将原来的裸tcp conn替换成被tls包装过的conn，所有的read/write会被tls层先做加工再下发
	c.Conn = tls.Client(c.Conn, tlsCfg)
}