]>
Commit | Line | Data |
---|---|---|
15c0b25d AP |
1 | /* |
2 | * | |
3 | * Copyright 2014 gRPC authors. | |
4 | * | |
5 | * Licensed under the Apache License, Version 2.0 (the "License"); | |
6 | * you may not use this file except in compliance with the License. | |
7 | * You may obtain a copy of the License at | |
8 | * | |
9 | * http://www.apache.org/licenses/LICENSE-2.0 | |
10 | * | |
11 | * Unless required by applicable law or agreed to in writing, software | |
12 | * distributed under the License is distributed on an "AS IS" BASIS, | |
13 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
14 | * See the License for the specific language governing permissions and | |
15 | * limitations under the License. | |
16 | * | |
17 | */ | |
18 | ||
19 | // Package transport defines and implements message oriented communication | |
20 | // channel to complete various transactions (e.g., an RPC). | |
21 | package transport // import "google.golang.org/grpc/transport" | |
22 | ||
23 | import ( | |
24 | "fmt" | |
25 | "io" | |
26 | "net" | |
27 | "sync" | |
28 | ||
29 | "golang.org/x/net/context" | |
30 | "golang.org/x/net/http2" | |
31 | "google.golang.org/grpc/codes" | |
32 | "google.golang.org/grpc/credentials" | |
33 | "google.golang.org/grpc/keepalive" | |
34 | "google.golang.org/grpc/metadata" | |
35 | "google.golang.org/grpc/stats" | |
36 | "google.golang.org/grpc/status" | |
37 | "google.golang.org/grpc/tap" | |
38 | ) | |
39 | ||
40 | // recvMsg represents the received msg from the transport. All transport | |
41 | // protocol specific info has been removed. | |
42 | type recvMsg struct { | |
43 | data []byte | |
44 | // nil: received some data | |
45 | // io.EOF: stream is completed. data is nil. | |
46 | // other non-nil error: transport failure. data is nil. | |
47 | err error | |
48 | } | |
49 | ||
50 | // recvBuffer is an unbounded channel of recvMsg structs. | |
51 | // Note recvBuffer differs from controlBuffer only in that recvBuffer | |
52 | // holds a channel of only recvMsg structs instead of objects implementing "item" interface. | |
53 | // recvBuffer is written to much more often than | |
54 | // controlBuffer and using strict recvMsg structs helps avoid allocation in "recvBuffer.put" | |
55 | type recvBuffer struct { | |
56 | c chan recvMsg | |
57 | mu sync.Mutex | |
58 | backlog []recvMsg | |
59 | } | |
60 | ||
61 | func newRecvBuffer() *recvBuffer { | |
62 | b := &recvBuffer{ | |
63 | c: make(chan recvMsg, 1), | |
64 | } | |
65 | return b | |
66 | } | |
67 | ||
68 | func (b *recvBuffer) put(r recvMsg) { | |
69 | b.mu.Lock() | |
70 | defer b.mu.Unlock() | |
71 | if len(b.backlog) == 0 { | |
72 | select { | |
73 | case b.c <- r: | |
74 | return | |
75 | default: | |
76 | } | |
77 | } | |
78 | b.backlog = append(b.backlog, r) | |
79 | } | |
80 | ||
81 | func (b *recvBuffer) load() { | |
82 | b.mu.Lock() | |
83 | defer b.mu.Unlock() | |
84 | if len(b.backlog) > 0 { | |
85 | select { | |
86 | case b.c <- b.backlog[0]: | |
87 | b.backlog[0] = recvMsg{} | |
88 | b.backlog = b.backlog[1:] | |
89 | default: | |
90 | } | |
91 | } | |
92 | } | |
93 | ||
94 | // get returns the channel that receives a recvMsg in the buffer. | |
95 | // | |
96 | // Upon receipt of a recvMsg, the caller should call load to send another | |
97 | // recvMsg onto the channel if there is any. | |
98 | func (b *recvBuffer) get() <-chan recvMsg { | |
99 | return b.c | |
100 | } | |
101 | ||
102 | // recvBufferReader implements io.Reader interface to read the data from | |
103 | // recvBuffer. | |
104 | type recvBufferReader struct { | |
105 | ctx context.Context | |
106 | goAway chan struct{} | |
107 | recv *recvBuffer | |
108 | last []byte // Stores the remaining data in the previous calls. | |
109 | err error | |
110 | } | |
111 | ||
112 | // Read reads the next len(p) bytes from last. If last is drained, it tries to | |
113 | // read additional data from recv. It blocks if there no additional data available | |
114 | // in recv. If Read returns any non-nil error, it will continue to return that error. | |
115 | func (r *recvBufferReader) Read(p []byte) (n int, err error) { | |
116 | if r.err != nil { | |
117 | return 0, r.err | |
118 | } | |
119 | n, r.err = r.read(p) | |
120 | return n, r.err | |
121 | } | |
122 | ||
123 | func (r *recvBufferReader) read(p []byte) (n int, err error) { | |
124 | if r.last != nil && len(r.last) > 0 { | |
125 | // Read remaining data left in last call. | |
126 | copied := copy(p, r.last) | |
127 | r.last = r.last[copied:] | |
128 | return copied, nil | |
129 | } | |
130 | select { | |
131 | case <-r.ctx.Done(): | |
132 | return 0, ContextErr(r.ctx.Err()) | |
133 | case <-r.goAway: | |
134 | return 0, ErrStreamDrain | |
135 | case m := <-r.recv.get(): | |
136 | r.recv.load() | |
137 | if m.err != nil { | |
138 | return 0, m.err | |
139 | } | |
140 | copied := copy(p, m.data) | |
141 | r.last = m.data[copied:] | |
142 | return copied, nil | |
143 | } | |
144 | } | |
145 | ||
146 | // All items in an out of a controlBuffer should be the same type. | |
147 | type item interface { | |
148 | item() | |
149 | } | |
150 | ||
151 | // controlBuffer is an unbounded channel of item. | |
152 | type controlBuffer struct { | |
153 | c chan item | |
154 | mu sync.Mutex | |
155 | backlog []item | |
156 | } | |
157 | ||
158 | func newControlBuffer() *controlBuffer { | |
159 | b := &controlBuffer{ | |
160 | c: make(chan item, 1), | |
161 | } | |
162 | return b | |
163 | } | |
164 | ||
165 | func (b *controlBuffer) put(r item) { | |
166 | b.mu.Lock() | |
167 | defer b.mu.Unlock() | |
168 | if len(b.backlog) == 0 { | |
169 | select { | |
170 | case b.c <- r: | |
171 | return | |
172 | default: | |
173 | } | |
174 | } | |
175 | b.backlog = append(b.backlog, r) | |
176 | } | |
177 | ||
178 | func (b *controlBuffer) load() { | |
179 | b.mu.Lock() | |
180 | defer b.mu.Unlock() | |
181 | if len(b.backlog) > 0 { | |
182 | select { | |
183 | case b.c <- b.backlog[0]: | |
184 | b.backlog[0] = nil | |
185 | b.backlog = b.backlog[1:] | |
186 | default: | |
187 | } | |
188 | } | |
189 | } | |
190 | ||
191 | // get returns the channel that receives an item in the buffer. | |
192 | // | |
193 | // Upon receipt of an item, the caller should call load to send another | |
194 | // item onto the channel if there is any. | |
195 | func (b *controlBuffer) get() <-chan item { | |
196 | return b.c | |
197 | } | |
198 | ||
199 | type streamState uint8 | |
200 | ||
201 | const ( | |
202 | streamActive streamState = iota | |
203 | streamWriteDone // EndStream sent | |
204 | streamReadDone // EndStream received | |
205 | streamDone // the entire stream is finished. | |
206 | ) | |
207 | ||
208 | // Stream represents an RPC in the transport layer. | |
209 | type Stream struct { | |
210 | id uint32 | |
211 | // nil for client side Stream. | |
212 | st ServerTransport | |
213 | // ctx is the associated context of the stream. | |
214 | ctx context.Context | |
215 | // cancel is always nil for client side Stream. | |
216 | cancel context.CancelFunc | |
217 | // done is closed when the final status arrives. | |
218 | done chan struct{} | |
219 | // goAway is closed when the server sent GoAways signal before this stream was initiated. | |
220 | goAway chan struct{} | |
221 | // method records the associated RPC method of the stream. | |
222 | method string | |
223 | recvCompress string | |
224 | sendCompress string | |
225 | buf *recvBuffer | |
226 | trReader io.Reader | |
227 | fc *inFlow | |
228 | recvQuota uint32 | |
229 | ||
230 | // TODO: Remote this unused variable. | |
231 | // The accumulated inbound quota pending for window update. | |
232 | updateQuota uint32 | |
233 | ||
234 | // Callback to state application's intentions to read data. This | |
235 | // is used to adjust flow control, if need be. | |
236 | requestRead func(int) | |
237 | ||
238 | sendQuotaPool *quotaPool | |
239 | // Close headerChan to indicate the end of reception of header metadata. | |
240 | headerChan chan struct{} | |
241 | // header caches the received header metadata. | |
242 | header metadata.MD | |
243 | // The key-value map of trailer metadata. | |
244 | trailer metadata.MD | |
245 | ||
246 | mu sync.RWMutex // guard the following | |
247 | // headerOK becomes true from the first header is about to send. | |
248 | headerOk bool | |
249 | state streamState | |
250 | // true iff headerChan is closed. Used to avoid closing headerChan | |
251 | // multiple times. | |
252 | headerDone bool | |
253 | // the status error received from the server. | |
254 | status *status.Status | |
255 | // rstStream indicates whether a RST_STREAM frame needs to be sent | |
256 | // to the server to signify that this stream is closing. | |
257 | rstStream bool | |
258 | // rstError is the error that needs to be sent along with the RST_STREAM frame. | |
259 | rstError http2.ErrCode | |
260 | // bytesSent and bytesReceived indicates whether any bytes have been sent or | |
261 | // received on this stream. | |
262 | bytesSent bool | |
263 | bytesReceived bool | |
264 | } | |
265 | ||
266 | // RecvCompress returns the compression algorithm applied to the inbound | |
267 | // message. It is empty string if there is no compression applied. | |
268 | func (s *Stream) RecvCompress() string { | |
269 | return s.recvCompress | |
270 | } | |
271 | ||
272 | // SetSendCompress sets the compression algorithm to the stream. | |
273 | func (s *Stream) SetSendCompress(str string) { | |
274 | s.sendCompress = str | |
275 | } | |
276 | ||
277 | // Done returns a chanel which is closed when it receives the final status | |
278 | // from the server. | |
279 | func (s *Stream) Done() <-chan struct{} { | |
280 | return s.done | |
281 | } | |
282 | ||
283 | // GoAway returns a channel which is closed when the server sent GoAways signal | |
284 | // before this stream was initiated. | |
285 | func (s *Stream) GoAway() <-chan struct{} { | |
286 | return s.goAway | |
287 | } | |
288 | ||
289 | // Header acquires the key-value pairs of header metadata once it | |
290 | // is available. It blocks until i) the metadata is ready or ii) there is no | |
291 | // header metadata or iii) the stream is canceled/expired. | |
292 | func (s *Stream) Header() (metadata.MD, error) { | |
293 | var err error | |
294 | select { | |
295 | case <-s.ctx.Done(): | |
296 | err = ContextErr(s.ctx.Err()) | |
297 | case <-s.goAway: | |
298 | err = ErrStreamDrain | |
299 | case <-s.headerChan: | |
300 | return s.header.Copy(), nil | |
301 | } | |
302 | // Even if the stream is closed, header is returned if available. | |
303 | select { | |
304 | case <-s.headerChan: | |
305 | return s.header.Copy(), nil | |
306 | default: | |
307 | } | |
308 | return nil, err | |
309 | } | |
310 | ||
311 | // Trailer returns the cached trailer metedata. Note that if it is not called | |
312 | // after the entire stream is done, it could return an empty MD. Client | |
313 | // side only. | |
314 | func (s *Stream) Trailer() metadata.MD { | |
315 | s.mu.RLock() | |
316 | defer s.mu.RUnlock() | |
317 | return s.trailer.Copy() | |
318 | } | |
319 | ||
320 | // ServerTransport returns the underlying ServerTransport for the stream. | |
321 | // The client side stream always returns nil. | |
322 | func (s *Stream) ServerTransport() ServerTransport { | |
323 | return s.st | |
324 | } | |
325 | ||
326 | // Context returns the context of the stream. | |
327 | func (s *Stream) Context() context.Context { | |
328 | return s.ctx | |
329 | } | |
330 | ||
331 | // Method returns the method for the stream. | |
332 | func (s *Stream) Method() string { | |
333 | return s.method | |
334 | } | |
335 | ||
336 | // Status returns the status received from the server. | |
337 | func (s *Stream) Status() *status.Status { | |
338 | return s.status | |
339 | } | |
340 | ||
341 | // SetHeader sets the header metadata. This can be called multiple times. | |
342 | // Server side only. | |
343 | func (s *Stream) SetHeader(md metadata.MD) error { | |
344 | s.mu.Lock() | |
345 | defer s.mu.Unlock() | |
346 | if s.headerOk || s.state == streamDone { | |
347 | return ErrIllegalHeaderWrite | |
348 | } | |
349 | if md.Len() == 0 { | |
350 | return nil | |
351 | } | |
352 | s.header = metadata.Join(s.header, md) | |
353 | return nil | |
354 | } | |
355 | ||
356 | // SetTrailer sets the trailer metadata which will be sent with the RPC status | |
357 | // by the server. This can be called multiple times. Server side only. | |
358 | func (s *Stream) SetTrailer(md metadata.MD) error { | |
359 | if md.Len() == 0 { | |
360 | return nil | |
361 | } | |
362 | s.mu.Lock() | |
363 | defer s.mu.Unlock() | |
364 | s.trailer = metadata.Join(s.trailer, md) | |
365 | return nil | |
366 | } | |
367 | ||
368 | func (s *Stream) write(m recvMsg) { | |
369 | s.buf.put(m) | |
370 | } | |
371 | ||
372 | // Read reads all p bytes from the wire for this stream. | |
373 | func (s *Stream) Read(p []byte) (n int, err error) { | |
374 | // Don't request a read if there was an error earlier | |
375 | if er := s.trReader.(*transportReader).er; er != nil { | |
376 | return 0, er | |
377 | } | |
378 | s.requestRead(len(p)) | |
379 | return io.ReadFull(s.trReader, p) | |
380 | } | |
381 | ||
382 | // tranportReader reads all the data available for this Stream from the transport and | |
383 | // passes them into the decoder, which converts them into a gRPC message stream. | |
384 | // The error is io.EOF when the stream is done or another non-nil error if | |
385 | // the stream broke. | |
386 | type transportReader struct { | |
387 | reader io.Reader | |
388 | // The handler to control the window update procedure for both this | |
389 | // particular stream and the associated transport. | |
390 | windowHandler func(int) | |
391 | er error | |
392 | } | |
393 | ||
394 | func (t *transportReader) Read(p []byte) (n int, err error) { | |
395 | n, err = t.reader.Read(p) | |
396 | if err != nil { | |
397 | t.er = err | |
398 | return | |
399 | } | |
400 | t.windowHandler(n) | |
401 | return | |
402 | } | |
403 | ||
404 | // finish sets the stream's state and status, and closes the done channel. | |
405 | // s.mu must be held by the caller. st must always be non-nil. | |
406 | func (s *Stream) finish(st *status.Status) { | |
407 | s.status = st | |
408 | s.state = streamDone | |
409 | close(s.done) | |
410 | } | |
411 | ||
412 | // BytesSent indicates whether any bytes have been sent on this stream. | |
413 | func (s *Stream) BytesSent() bool { | |
414 | s.mu.Lock() | |
415 | defer s.mu.Unlock() | |
416 | return s.bytesSent | |
417 | } | |
418 | ||
419 | // BytesReceived indicates whether any bytes have been received on this stream. | |
420 | func (s *Stream) BytesReceived() bool { | |
421 | s.mu.Lock() | |
422 | defer s.mu.Unlock() | |
423 | return s.bytesReceived | |
424 | } | |
425 | ||
426 | // GoString is implemented by Stream so context.String() won't | |
427 | // race when printing %#v. | |
428 | func (s *Stream) GoString() string { | |
429 | return fmt.Sprintf("<stream: %p, %v>", s, s.method) | |
430 | } | |
431 | ||
432 | // The key to save transport.Stream in the context. | |
433 | type streamKey struct{} | |
434 | ||
435 | // newContextWithStream creates a new context from ctx and attaches stream | |
436 | // to it. | |
437 | func newContextWithStream(ctx context.Context, stream *Stream) context.Context { | |
438 | return context.WithValue(ctx, streamKey{}, stream) | |
439 | } | |
440 | ||
441 | // StreamFromContext returns the stream saved in ctx. | |
442 | func StreamFromContext(ctx context.Context) (s *Stream, ok bool) { | |
443 | s, ok = ctx.Value(streamKey{}).(*Stream) | |
444 | return | |
445 | } | |
446 | ||
447 | // state of transport | |
448 | type transportState int | |
449 | ||
450 | const ( | |
451 | reachable transportState = iota | |
452 | unreachable | |
453 | closing | |
454 | draining | |
455 | ) | |
456 | ||
457 | // ServerConfig consists of all the configurations to establish a server transport. | |
458 | type ServerConfig struct { | |
459 | MaxStreams uint32 | |
460 | AuthInfo credentials.AuthInfo | |
461 | InTapHandle tap.ServerInHandle | |
462 | StatsHandler stats.Handler | |
463 | KeepaliveParams keepalive.ServerParameters | |
464 | KeepalivePolicy keepalive.EnforcementPolicy | |
465 | InitialWindowSize int32 | |
466 | InitialConnWindowSize int32 | |
467 | } | |
468 | ||
469 | // NewServerTransport creates a ServerTransport with conn or non-nil error | |
470 | // if it fails. | |
471 | func NewServerTransport(protocol string, conn net.Conn, config *ServerConfig) (ServerTransport, error) { | |
472 | return newHTTP2Server(conn, config) | |
473 | } | |
474 | ||
475 | // ConnectOptions covers all relevant options for communicating with the server. | |
476 | type ConnectOptions struct { | |
477 | // UserAgent is the application user agent. | |
478 | UserAgent string | |
479 | // Authority is the :authority pseudo-header to use. This field has no effect if | |
480 | // TransportCredentials is set. | |
481 | Authority string | |
482 | // Dialer specifies how to dial a network address. | |
483 | Dialer func(context.Context, string) (net.Conn, error) | |
484 | // FailOnNonTempDialError specifies if gRPC fails on non-temporary dial errors. | |
485 | FailOnNonTempDialError bool | |
486 | // PerRPCCredentials stores the PerRPCCredentials required to issue RPCs. | |
487 | PerRPCCredentials []credentials.PerRPCCredentials | |
488 | // TransportCredentials stores the Authenticator required to setup a client connection. | |
489 | TransportCredentials credentials.TransportCredentials | |
490 | // KeepaliveParams stores the keepalive parameters. | |
491 | KeepaliveParams keepalive.ClientParameters | |
492 | // StatsHandler stores the handler for stats. | |
493 | StatsHandler stats.Handler | |
494 | // InitialWindowSize sets the intial window size for a stream. | |
495 | InitialWindowSize int32 | |
496 | // InitialConnWindowSize sets the intial window size for a connection. | |
497 | InitialConnWindowSize int32 | |
498 | } | |
499 | ||
500 | // TargetInfo contains the information of the target such as network address and metadata. | |
501 | type TargetInfo struct { | |
502 | Addr string | |
503 | Metadata interface{} | |
504 | } | |
505 | ||
506 | // NewClientTransport establishes the transport with the required ConnectOptions | |
507 | // and returns it to the caller. | |
508 | func NewClientTransport(ctx context.Context, target TargetInfo, opts ConnectOptions) (ClientTransport, error) { | |
509 | return newHTTP2Client(ctx, target, opts) | |
510 | } | |
511 | ||
512 | // Options provides additional hints and information for message | |
513 | // transmission. | |
514 | type Options struct { | |
515 | // Last indicates whether this write is the last piece for | |
516 | // this stream. | |
517 | Last bool | |
518 | ||
519 | // Delay is a hint to the transport implementation for whether | |
520 | // the data could be buffered for a batching write. The | |
521 | // Transport implementation may ignore the hint. | |
522 | Delay bool | |
523 | } | |
524 | ||
525 | // CallHdr carries the information of a particular RPC. | |
526 | type CallHdr struct { | |
527 | // Host specifies the peer's host. | |
528 | Host string | |
529 | ||
530 | // Method specifies the operation to perform. | |
531 | Method string | |
532 | ||
533 | // RecvCompress specifies the compression algorithm applied on | |
534 | // inbound messages. | |
535 | RecvCompress string | |
536 | ||
537 | // SendCompress specifies the compression algorithm applied on | |
538 | // outbound message. | |
539 | SendCompress string | |
540 | ||
541 | // Creds specifies credentials.PerRPCCredentials for a call. | |
542 | Creds credentials.PerRPCCredentials | |
543 | ||
544 | // Flush indicates whether a new stream command should be sent | |
545 | // to the peer without waiting for the first data. This is | |
546 | // only a hint. | |
547 | // If it's true, the transport may modify the flush decision | |
548 | // for performance purposes. | |
549 | // If it's false, new stream will never be flushed. | |
550 | Flush bool | |
551 | } | |
552 | ||
553 | // ClientTransport is the common interface for all gRPC client-side transport | |
554 | // implementations. | |
555 | type ClientTransport interface { | |
556 | // Close tears down this transport. Once it returns, the transport | |
557 | // should not be accessed any more. The caller must make sure this | |
558 | // is called only once. | |
559 | Close() error | |
560 | ||
561 | // GracefulClose starts to tear down the transport. It stops accepting | |
562 | // new RPCs and wait the completion of the pending RPCs. | |
563 | GracefulClose() error | |
564 | ||
565 | // Write sends the data for the given stream. A nil stream indicates | |
566 | // the write is to be performed on the transport as a whole. | |
567 | Write(s *Stream, data []byte, opts *Options) error | |
568 | ||
569 | // NewStream creates a Stream for an RPC. | |
570 | NewStream(ctx context.Context, callHdr *CallHdr) (*Stream, error) | |
571 | ||
572 | // CloseStream clears the footprint of a stream when the stream is | |
573 | // not needed any more. The err indicates the error incurred when | |
574 | // CloseStream is called. Must be called when a stream is finished | |
575 | // unless the associated transport is closing. | |
576 | CloseStream(stream *Stream, err error) | |
577 | ||
578 | // Error returns a channel that is closed when some I/O error | |
579 | // happens. Typically the caller should have a goroutine to monitor | |
580 | // this in order to take action (e.g., close the current transport | |
581 | // and create a new one) in error case. It should not return nil | |
582 | // once the transport is initiated. | |
583 | Error() <-chan struct{} | |
584 | ||
585 | // GoAway returns a channel that is closed when ClientTransport | |
586 | // receives the draining signal from the server (e.g., GOAWAY frame in | |
587 | // HTTP/2). | |
588 | GoAway() <-chan struct{} | |
589 | ||
590 | // GetGoAwayReason returns the reason why GoAway frame was received. | |
591 | GetGoAwayReason() GoAwayReason | |
592 | } | |
593 | ||
594 | // ServerTransport is the common interface for all gRPC server-side transport | |
595 | // implementations. | |
596 | // | |
597 | // Methods may be called concurrently from multiple goroutines, but | |
598 | // Write methods for a given Stream will be called serially. | |
599 | type ServerTransport interface { | |
600 | // HandleStreams receives incoming streams using the given handler. | |
601 | HandleStreams(func(*Stream), func(context.Context, string) context.Context) | |
602 | ||
603 | // WriteHeader sends the header metadata for the given stream. | |
604 | // WriteHeader may not be called on all streams. | |
605 | WriteHeader(s *Stream, md metadata.MD) error | |
606 | ||
607 | // Write sends the data for the given stream. | |
608 | // Write may not be called on all streams. | |
609 | Write(s *Stream, data []byte, opts *Options) error | |
610 | ||
611 | // WriteStatus sends the status of a stream to the client. WriteStatus is | |
612 | // the final call made on a stream and always occurs. | |
613 | WriteStatus(s *Stream, st *status.Status) error | |
614 | ||
615 | // Close tears down the transport. Once it is called, the transport | |
616 | // should not be accessed any more. All the pending streams and their | |
617 | // handlers will be terminated asynchronously. | |
618 | Close() error | |
619 | ||
620 | // RemoteAddr returns the remote network address. | |
621 | RemoteAddr() net.Addr | |
622 | ||
623 | // Drain notifies the client this ServerTransport stops accepting new RPCs. | |
624 | Drain() | |
625 | } | |
626 | ||
627 | // streamErrorf creates an StreamError with the specified error code and description. | |
628 | func streamErrorf(c codes.Code, format string, a ...interface{}) StreamError { | |
629 | return StreamError{ | |
630 | Code: c, | |
631 | Desc: fmt.Sprintf(format, a...), | |
632 | } | |
633 | } | |
634 | ||
635 | // connectionErrorf creates an ConnectionError with the specified error description. | |
636 | func connectionErrorf(temp bool, e error, format string, a ...interface{}) ConnectionError { | |
637 | return ConnectionError{ | |
638 | Desc: fmt.Sprintf(format, a...), | |
639 | temp: temp, | |
640 | err: e, | |
641 | } | |
642 | } | |
643 | ||
644 | // ConnectionError is an error that results in the termination of the | |
645 | // entire connection and the retry of all the active streams. | |
646 | type ConnectionError struct { | |
647 | Desc string | |
648 | temp bool | |
649 | err error | |
650 | } | |
651 | ||
652 | func (e ConnectionError) Error() string { | |
653 | return fmt.Sprintf("connection error: desc = %q", e.Desc) | |
654 | } | |
655 | ||
656 | // Temporary indicates if this connection error is temporary or fatal. | |
657 | func (e ConnectionError) Temporary() bool { | |
658 | return e.temp | |
659 | } | |
660 | ||
661 | // Origin returns the original error of this connection error. | |
662 | func (e ConnectionError) Origin() error { | |
663 | // Never return nil error here. | |
664 | // If the original error is nil, return itself. | |
665 | if e.err == nil { | |
666 | return e | |
667 | } | |
668 | return e.err | |
669 | } | |
670 | ||
671 | var ( | |
672 | // ErrConnClosing indicates that the transport is closing. | |
673 | ErrConnClosing = connectionErrorf(true, nil, "transport is closing") | |
674 | // ErrStreamDrain indicates that the stream is rejected by the server because | |
675 | // the server stops accepting new RPCs. | |
676 | ErrStreamDrain = streamErrorf(codes.Unavailable, "the server stops accepting new RPCs") | |
677 | ) | |
678 | ||
679 | // TODO: See if we can replace StreamError with status package errors. | |
680 | ||
681 | // StreamError is an error that only affects one stream within a connection. | |
682 | type StreamError struct { | |
683 | Code codes.Code | |
684 | Desc string | |
685 | } | |
686 | ||
687 | func (e StreamError) Error() string { | |
688 | return fmt.Sprintf("stream error: code = %s desc = %q", e.Code, e.Desc) | |
689 | } | |
690 | ||
691 | // wait blocks until it can receive from ctx.Done, closing, or proceed. | |
692 | // If it receives from ctx.Done, it returns 0, the StreamError for ctx.Err. | |
693 | // If it receives from done, it returns 0, io.EOF if ctx is not done; otherwise | |
694 | // it return the StreamError for ctx.Err. | |
695 | // If it receives from goAway, it returns 0, ErrStreamDrain. | |
696 | // If it receives from closing, it returns 0, ErrConnClosing. | |
697 | // If it receives from proceed, it returns the received integer, nil. | |
698 | func wait(ctx context.Context, done, goAway, closing <-chan struct{}, proceed <-chan int) (int, error) { | |
699 | select { | |
700 | case <-ctx.Done(): | |
701 | return 0, ContextErr(ctx.Err()) | |
702 | case <-done: | |
703 | // User cancellation has precedence. | |
704 | select { | |
705 | case <-ctx.Done(): | |
706 | return 0, ContextErr(ctx.Err()) | |
707 | default: | |
708 | } | |
709 | return 0, io.EOF | |
710 | case <-goAway: | |
711 | return 0, ErrStreamDrain | |
712 | case <-closing: | |
713 | return 0, ErrConnClosing | |
714 | case i := <-proceed: | |
715 | return i, nil | |
716 | } | |
717 | } | |
718 | ||
719 | // GoAwayReason contains the reason for the GoAway frame received. | |
720 | type GoAwayReason uint8 | |
721 | ||
722 | const ( | |
723 | // Invalid indicates that no GoAway frame is received. | |
724 | Invalid GoAwayReason = 0 | |
725 | // NoReason is the default value when GoAway frame is received. | |
726 | NoReason GoAwayReason = 1 | |
727 | // TooManyPings indicates that a GoAway frame with ErrCodeEnhanceYourCalm | |
728 | // was recieved and that the debug data said "too_many_pings". | |
729 | TooManyPings GoAwayReason = 2 | |
730 | ) |