1 // Copyright 2018 The Go Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style
3 // license that can be found in the LICENSE file.
18 // Stream abstracts the transport mechanics from the JSON RPC protocol.
19 // A Conn reads and writes messages using the stream it was provided on
20 // construction, and assumes that each call to Read or Write fully transfers
21 // a single message, or returns an error.
22 // A stream is not safe for concurrent use, it is expected it will be used by
23 // a single Conn in a safe manner.
24 type Stream interface {
25 // Read gets the next message from the stream.
26 Read(context.Context) (Message, int64, error)
27 // Write sends a message to the stream.
28 Write(context.Context, Message) (int64, error)
29 // Close closes the connection.
30 // Any blocked Read or Write operations will be unblocked and return errors.
34 // Framer wraps a network connection up into a Stream.
35 // It is responsible for the framing and encoding of messages into wire form.
36 // NewRawStream and NewHeaderStream are implementations of a Framer.
37 type Framer func(conn net.Conn) Stream
39 // NewRawStream returns a Stream built on top of a net.Conn.
40 // The messages are sent with no wrapping, and rely on json decode consistency
41 // to determine message boundaries.
42 func NewRawStream(conn net.Conn) Stream {
45 in: json.NewDecoder(conn),
49 type rawStream struct {
54 func (s *rawStream) Read(ctx context.Context) (Message, int64, error) {
57 return nil, 0, ctx.Err()
60 var raw json.RawMessage
61 if err := s.in.Decode(&raw); err != nil {
64 msg, err := DecodeMessage(raw)
65 return msg, int64(len(raw)), err
68 func (s *rawStream) Write(ctx context.Context, msg Message) (int64, error) {
74 data, err := json.Marshal(msg)
76 return 0, fmt.Errorf("marshaling message: %v", err)
78 n, err := s.conn.Write(data)
82 func (s *rawStream) Close() error {
86 // NewHeaderStream returns a Stream built on top of a net.Conn.
87 // The messages are sent with HTTP content length and MIME type headers.
88 // This is the format used by LSP and others.
89 func NewHeaderStream(conn net.Conn) Stream {
92 in: bufio.NewReader(conn),
96 type headerStream struct {
101 func (s *headerStream) Read(ctx context.Context) (Message, int64, error) {
104 return nil, 0, ctx.Err()
107 var total, length int64
108 // read the header, stop on the first empty line
110 line, err := s.in.ReadString('\n')
111 total += int64(len(line))
113 return nil, total, fmt.Errorf("failed reading header line: %w", err)
115 line = strings.TrimSpace(line)
116 // check we have a header line
120 colon := strings.IndexRune(line, ':')
122 return nil, total, fmt.Errorf("invalid header line %q", line)
124 name, value := line[:colon], strings.TrimSpace(line[colon+1:])
126 case "Content-Length":
127 if length, err = strconv.ParseInt(value, 10, 32); err != nil {
128 return nil, total, fmt.Errorf("failed parsing Content-Length: %v", value)
131 return nil, total, fmt.Errorf("invalid Content-Length: %v", length)
134 // ignoring unknown headers
138 return nil, total, fmt.Errorf("missing Content-Length header")
140 data := make([]byte, length)
141 if _, err := io.ReadFull(s.in, data); err != nil {
142 return nil, total, err
145 msg, err := DecodeMessage(data)
146 return msg, total, err
149 func (s *headerStream) Write(ctx context.Context, msg Message) (int64, error) {
155 data, err := json.Marshal(msg)
157 return 0, fmt.Errorf("marshaling message: %v", err)
159 n, err := fmt.Fprintf(s.conn, "Content-Length: %v\r\n\r\n", len(data))
162 n, err = s.conn.Write(data)
168 func (s *headerStream) Close() error {
169 return s.conn.Close()