作用

从tranport.go的注释和之前的博文中可以看出transport的功能是提供异步通信。

// Package is an interface for synchronous communication
package transport

接口

type Socket interface {
    Recv(*Message) error
    Send(*Message) error
    Close() error
}

type Client interface {
    Socket
}

type Listener interface {
    Addr() string
    Close() error
    Accept(func(Socket)) error
}

// Transport is an interface which is used for communication between
// services. It uses socket send/recv semantics and had various
// implementations {HTTP, RabbitMQ, NATS, ...}
type Transport interface {
    Dial(addr string, opts ...DialOption) (Client, error)
    Listen(addr string, opts ...ListenOption) (Listener, error)
    String() string
}

从之上的代码可以看出：

• 当Transport作为客户端时：

  提供Dial主动连接功能，连接成功返回代表一个连接的Client，Client内嵌Socket，提供Send/Recv的网络收发功能。

• 当Transport作为服务端时：

  提供Listen监听功能，当Listener.Accept成功返回时，返回一个代表连接的Socket，提供Send/Recv的网络收发功能。

Send和Recv的数据是Message结构

type Message struct {
    Header map[string]string
    Body   []byte
}

由上可以看出Client是客户端语义，Socket是服务端语义。

transport在Go-micro内部的实现是httpTransport

Dial

从代码中可以看出本身Dial的逻辑很简单调用

        conn, err = tls.DialWithDialer
    或
        conn, err = net.DialTimeout

最终返回httpTransportClient对象

type httpTransportClient struct {
    ht       *httpTransport
    addr     string
    conn     net.Conn
    dialOpts DialOptions
    once     sync.Once

    sync.Mutex
    r    chan *http.Request
    bl   []*http.Request
    buff *bufio.Reader
}

httpTransportClient 实现Client 接口提供Send、Recv网络首发功能。

通过httpTransportClient 的Send代码可以看出，通过tpc的链接发送http请求。

func (h *httpTransportClient) Send(m *Message) error {
    header := make(http.Header)

    for k, v := range m.Header {
        header.Set(k, v)
    }

    reqB := bytes.NewBuffer(m.Body)
    defer reqB.Reset()
    buf := &buffer{
        reqB,
    }

    req := &http.Request{
        Method: "POST",
        URL: &url.URL{
            Scheme: "http",
            Host:   h.addr,
        },
        Header:        header,
        Body:          buf,
        ContentLength: int64(reqB.Len()),
        Host:          h.addr,
    }
    ///>>>>>>
    h.Lock()
    h.bl = append(h.bl, req)
    select {
    case h.r <- h.bl[0]:
        h.bl = h.bl[1:]
    default:
    }
    h.Unlock()
    ///<<<<<<<<
    // set timeout if its greater than 0
    if h.ht.opts.Timeout > time.Duration(0) {
        h.conn.SetDeadline(time.Now().Add(h.ht.opts.Timeout))   ///设置超时
    }

    return req.Write(h.conn)
}

其他的代码都好理解，唯独///>>>>>> ///<<<<<<<<中间的部分，为什么一直往h.bl中append,仅仅h.r可写的情况下remove.

猜想这里是不是应该这么写

    if !h.dialOpts.Stream {
        h.Lock()
        h.bl = append(h.bl, req)
        select {
            case h.r <- h.bl[0]:
            h.bl = h.bl[1:]
            default:
            }
        h.Unlock()
    }

func (h *httpTransportClient) Recv(m *Message) error {
    if m == nil {
        return errors.New("message passed in is nil")
    }

    var r *http.Request
    if !h.dialOpts.Stream {
        rc, ok := <-h.r
        if !ok {
            return io.EOF
        }
        r = rc
    }

    h.Lock()
    defer h.Unlock()
    if h.buff == nil {
        return io.EOF
    }

    // set timeout if its greater than 0
    if h.ht.opts.Timeout > time.Duration(0) {
        h.conn.SetDeadline(time.Now().Add(h.ht.opts.Timeout))
    }

    rsp, err := http.ReadResponse(h.buff, r)
    if err != nil {
        return err
    }
    defer rsp.Body.Close()

    b, err := ioutil.ReadAll(rsp.Body)
    if err != nil {
        return err
    }

    if rsp.StatusCode != 200 {
        return errors.New(rsp.Status + ": " + string(b))
    }

    m.Body = b

    if m.Header == nil {
        m.Header = make(map[string]string)
    }

    for k, v := range rsp.Header {
        if len(v) > 0 {
            m.Header[k] = v[0]
        } else {
            m.Header[k] = ""
        }
    }

    return nil
}

从if !h.dialOpts.Stream //这里也可以看出Stream的Send和对应的Recv直接不能插入其他的其他的Stream.Send。

Listener

根据传进来的addr，然后net/tls.Listen创建的Listener对象。

Accept

上面返回的Listener对象上进行Accept操作，如果有新的连接成功的话，创建Socket对应，用Socket调用注入的接口 fn.

func (h *httpTransportListener) Accept(fn func(Socket)) error