golang源码分析：gopls（6）

golangLeetcode

发布于 2026-03-18 18:25:50

940

初始化完StreamServer后我们看看它是如何基于标准输入输出提供服务的。首先调用了golang.org/x/tools/internal/fakenet/conn.go

func NewConn(name string, in io.ReadCloser, out io.WriteCloser) net.Conn {
    c := &fakeConn{
        name:   name,
        reader: newFeeder(in.Read),
        writer: newFeeder(out.Write),
        in:     in,
        out:    out,
    }
    go c.reader.run()
    go c.writer.run()
    return c
}

里面有两个协程一个一个用于输入一个用于输出

func (f *connFeeder) run() {
    var b []byte
    for {
        // wait for an input request
        select {
        case b = <-f.input:
        case <-f.done:
            return
        }
        // invoke the underlying method
        n, err := f.source(b)
        // send the result back to the requester
        select {
        case f.result <- feedResult{n: n, err: err}:
        case <-f.done:
            return
        }
    }
}

type feedResult struct {
    n   int
    err error
}

type connFeeder struct {
    source func([]byte) (int, error)
    input  chan []byte
    result chan feedResult
    mu     sync.Mutex
    closed bool
    done   chan struct{}
}

然后把它装饰为一个HeaderStream

func NewHeaderStream(conn net.Conn) Stream {
    return &headerStream{
        conn: conn,
        in:   bufio.NewReader(conn),
    }
}

然后创立连接，最后调用ServeStream方法

func NewConn(s Stream) Conn {
    conn := &conn{
        stream:  s,
        pending: make(map[ID]chan *Response),
        done:    make(chan struct{}),
    }
    return conn
}

func (s *streamServer) ServeStream(ctx context.Context, conn jsonrpc2.Conn) error {
    client := protocol.ClientDispatcher(conn)
    session := cache.NewSession(ctx, s.cache)
    svr := s.serverForTest
    if svr == nil {
        options := settings.DefaultOptions(s.optionsOverrides)
        svr = server.New(session, client, options)
        if instance := debug.GetInstance(ctx); instance != nil {
            instance.AddService(svr, session)
        }
    }
    // Clients may or may not send a shutdown message. Make sure the server is
    // shut down.
    // TODO(rFindley): this shutdown should perhaps be on a disconnected context.
    defer func() {
        if err := svr.Shutdown(ctx); err != nil {
            event.Error(ctx, "error shutting down", err)
        }
    }()
    executable, err := os.Executable()
    if err != nil {
        log.Printf("error getting gopls path: %v", err)
        executable = ""
    }
    ctx = protocol.WithClient(ctx, client)
    conn.Go(ctx,
        protocol.Handlers(
            handshaker(session, executable, s.daemon,
                protocol.ServerHandler(svr,
                    jsonrpc2.MethodNotFound))))
    if s.daemon {
        log.Printf("Session %s: connected", session.ID())
        defer log.Printf("Session %s: exited", session.ID())
    }
    <-conn.Done()
    return conn.Err()
}

至此整个逻辑介绍完毕，如果是stdio模式，直接请求server处理，如果是rpc模式，需要两个协程分别做输入和输出的转发，最终完成功能从本地代理到远程的server。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-07-06，如有侵权请联系 cloudcommunity@tencent.com 删除

连接