1 use crate::codec::UserError;
2 use crate::frame::Reason;
3 use crate::proto::{self, WindowSize};
4
5 use bytes::{Buf, Bytes};
rngfunctnull6 use http::HeaderMap;
7
8 use crate::PollExt;
9 use std::fmt;
10 #[cfg(feature = "stream")]
11 use std::pin::Pin;
12 use std::task::{Context, Poll};
13
14 /// Sends the body stream and trailers to the remote peer.
15 ///
16 /// # Overview
17 ///
18 /// A `SendStream` is provided by [`SendRequest`] and [`SendResponse`] once the
19 /// HTTP/2.0 message header has been sent sent. It is used to stream the message
20 /// body and send the message trailers. See method level documentation for more
21 /// details.
22 ///
23 /// The `SendStream` instance is also used to manage outbound flow control.
24 ///
25 /// If a `SendStream` is dropped without explicitly closing the send stream, a
26 /// `RST_STREAM` frame will be sent. This essentially cancels the request /
27 /// response exchange.
28 ///
29 /// The ways to explicitly close the send stream are:
30 ///
31 /// * Set `end_of_stream` to true when calling [`send_request`],
32 /// [`send_response`], or [`send_data`].
33 /// * Send trailers with [`send_trailers`].
34 /// * Explicitly reset the stream with [`send_reset`].
35 ///
36 /// # Flow control
37 ///
38 /// In HTTP/2.0, data cannot be sent to the remote peer unless there is
39 /// available window capacity on both the stream and the connection. When a data
40 /// frame is sent, both the stream window and the connection window are
41 /// decremented. When the stream level window reaches zero, no further data can
42 /// be sent on that stream. When the connection level window reaches zero, no
43 /// further data can be sent on any stream for that connection.
44 ///
45 /// When the remote peer is ready to receive more data, it sends `WINDOW_UPDATE`
46 /// frames. These frames increment the windows. See the [specification] for more
47 /// details on the principles of HTTP/2.0 flow control.
48 ///
49 /// The implications for sending data are that the caller **should** ensure that
50 /// both the stream and the connection has available window capacity before
51 /// loading the data to send into memory. The `SendStream` instance provides the
52 /// necessary APIs to perform this logic. This, however, is not an obligation.
53 /// If the caller attempts to send data on a stream when there is no available
54 /// window capacity, the library will buffer the data until capacity becomes
55 /// available, at which point the buffer will be flushed to the connection.
56 ///
57 /// **NOTE**: There is no bound on the amount of data that the library will
58 /// buffer. If you are sending large amounts of data, you really should hook
59 /// into the flow control lifecycle. Otherwise, you risk using up significant
60 /// amounts of memory.
61 ///
62 /// To hook into the flow control lifecycle, the caller signals to the library
63 /// that it intends to send data by calling [`reserve_capacity`], specifying the
64 /// amount of data, in octets, that the caller intends to send. After this,
65 /// `poll_capacity` is used to be notified when the requested capacity is
66 /// assigned to the stream. Once [`poll_capacity`] returns `Ready` with the number
67 /// of octets available to the stream, the caller is able to actually send the
68 /// data using [`send_data`].
69 ///
70 /// Because there is also a connection level window that applies to **all**
71 /// streams on a connection, when capacity is assigned to a stream (indicated by
72 /// `poll_capacity` returning `Ready`), this capacity is reserved on the
73 /// connection and will **not** be assigned to any other stream. If data is
74 /// never written to the stream, that capacity is effectively lost to other
75 /// streams and this introduces the risk of deadlocking a connection.
76 ///
77 /// To avoid throttling data on a connection, the caller should not reserve
78 /// capacity until ready to send data and once any capacity is assigned to the
79 /// stream, the caller should immediately send data consuming this capacity.
80 /// There is no guarantee as to when the full capacity requested will become
81 /// available. For example, if the caller requests 64 KB of data and 512 bytes
82 /// become available, the caller should immediately send 512 bytes of data.
83 ///
84 /// See [`reserve_capacity`] documentation for more details.
85 ///
86 /// [`SendRequest`]: client/struct.SendRequest.html
87 /// [`SendResponse`]: server/struct.SendResponse.html
88 /// [specification]: http://httpwg.org/specs/rfc7540.html#FlowControl
89 /// [`reserve_capacity`]: #method.reserve_capacity
90 /// [`poll_capacity`]: #method.poll_capacity
91 /// [`send_data`]: #method.send_data
92 /// [`send_request`]: client/struct.SendRequest.html#method.send_request
93 /// [`send_response`]: server/struct.SendResponse.html#method.send_response
94 /// [`send_data`]: #method.send_data
95 /// [`send_trailers`]: #method.send_trailers
96 /// [`send_reset`]: #method.send_reset
97 #[derive(Debug)]
98 pub struct SendStream<B: Buf> {
99 inner: proto::StreamRef<B>,
100 }
101
102 /// A stream identifier, as described in [Section 5.1.1] of RFC 7540.
103 ///
104 /// Streams are identified with an unsigned 31-bit integer. Streams
105 /// initiated by a client MUST use odd-numbered stream identifiers; those
106 /// initiated by the server MUST use even-numbered stream identifiers. A
107 /// stream identifier of zero (0x0) is used for connection control
108 /// messages; the stream identifier of zero cannot be used to establish a
109 /// new stream.
110 ///
111 /// [Section 5.1.1]: https://tools.ietf.org/html/rfc7540#section-5.1.1
112 #[derive(Debug, Clone, Eq, PartialEq, Hash)]
113 pub struct StreamId(u32);
114
115 /// Receives the body stream and trailers from the remote peer.
116 ///
117 /// A `RecvStream` is provided by [`client::ResponseFuture`] and
118 /// [`server::Connection`] with the received HTTP/2.0 message head (the response
119 /// and request head respectively).
120 ///
121 /// A `RecvStream` instance is used to receive the streaming message body and
122 /// any trailers from the remote peer. It is also used to manage inbound flow
123 /// control.
124 ///
125 /// See method level documentation for more details on receiving data. See
126 /// [`FlowControl`] for more details on inbound flow control.
127 ///
128 /// Note that this type implements [`Stream`], yielding the received data frames.
129 /// When this implementation is used, the capacity is immediately released when
130 /// the data is yielded. It is recommended to only use this API when the data
131 /// will not be retained in memory for extended periods of time.
132 ///
133 /// [`client::ResponseFuture`]: client/struct.ResponseFuture.html
134 /// [`server::Connection`]: server/struct.Connection.html
135 /// [`FlowControl`]: struct.FlowControl.html
136 /// [`Stream`]: https://docs.rs/futures/0.1/futures/stream/trait.Stream.html
137 #[must_use = "streams do nothing unless polled"]
138 pub struct RecvStream {
139 inner: FlowControl,
140 }
141
142 /// A handle to release window capacity to a remote stream.
143 ///
144 /// This type allows the caller to manage inbound data [flow control]. The
145 /// caller is expected to call [`release_capacity`] after dropping data frames.
146 ///
147 /// # Overview
148 ///
149 /// Each stream has a window size. This window size is the maximum amount of
150 /// inbound data that can be in-flight. In-flight data is defined as data that
151 /// has been received, but not yet released.
152 ///
153 /// When a stream is created, the window size is set to the connection's initial
154 /// window size value. When a data frame is received, the window size is then
155 /// decremented by size of the data frame before the data is provided to the
156 /// caller. As the caller finishes using the data, [`release_capacity`] must be
157 /// called. This will then increment the window size again, allowing the peer to
158 /// send more data.
159 ///
160 /// There is also a connection level window as well as the stream level window.
161 /// Received data counts against the connection level window as well and calls
162 /// to [`release_capacity`] will also increment the connection level window.
163 ///
164 /// # Sending `WINDOW_UPDATE` frames
165 ///
166 /// `WINDOW_UPDATE` frames will not be sent out for **every** call to
167 /// `release_capacity`, as this would end up slowing down the protocol. Instead,
168 /// `h2` waits until the window size is increased to a certain threshold and
169 /// then sends out a single `WINDOW_UPDATE` frame representing all the calls to
170 /// `release_capacity` since the last `WINDOW_UPDATE` frame.
171 ///
172 /// This essentially batches window updating.
173 ///
174 /// # Scenarios
175 ///
176 /// Following is a basic scenario with an HTTP/2.0 connection containing a
177 /// single active stream.
178 ///
179 /// * A new stream is activated. The receive window is initialized to 1024 (the
180 /// value of the initial window size for this connection).
181 /// * A `DATA` frame is received containing a payload of 400 bytes.
182 /// * The receive window size is reduced to 424 bytes.
183 /// * [`release_capacity`] is called with 200.
184 /// * The receive window size is now 624 bytes. The peer may send no more than
185 /// this.
186 /// * A `DATA` frame is received with a payload of 624 bytes.
187 /// * The window size is now 0 bytes. The peer may not send any more data.
188 /// * [`release_capacity`] is called with 1024.
189 /// * The receive window size is now 1024 bytes. The peer may now send more
190 /// data.
191 ///
192 /// [flow control]: ../index.html#flow-control
193 /// [`release_capacity`]: struct.FlowControl.html#method.release_capacity
194 #[derive(Clone, Debug)]
195 pub struct FlowControl {
196 inner: proto::OpaqueStreamRef,
197 }
198
199 /// A handle to send and receive PING frames with the peer.
200 // NOT Clone on purpose
201 pub struct PingPong {
202 inner: proto::UserPings,
203 }
204
205 /// Sent via [`PingPong`][] to send a PING frame to a peer.
206 ///
207 /// [`PingPong`]: struct.PingPong.html
208 pub struct Ping {
209 _p: (),
210 }
211
212 /// Received via [`PingPong`][] when a peer acknowledges a [`Ping`][].
213 ///
214 /// [`PingPong`]: struct.PingPong.html
215 /// [`Ping`]: struct.Ping.html
216 pub struct Pong {
217 _p: (),
218 }
219
220 // ===== impl SendStream =====
221
222 impl<B: Buf> SendStream<B> {
223 pub(crate) fn new(inner: proto::StreamRef<B>) -> Self {
224 SendStream { inner }
225 }
226
227 /// Requests capacity to send data.
228 ///
229 /// This function is used to express intent to send data. This requests
230 /// connection level capacity. Once the capacity is available, it is
231 /// assigned to the stream and not reused by other streams.
232 ///
233 /// This function may be called repeatedly. The `capacity` argument is the
234 /// **total** amount of requested capacity. Sequential calls to
235 /// `reserve_capacity` are *not* additive. Given the following:
236 ///
237 /// ```rust
238 /// # use h2::*;
239 /// # fn doc(mut send_stream: SendStream<&'static [u8]>) {
240 /// send_stream.reserve_capacity(100);
241 /// send_stream.reserve_capacity(200);
242 /// # }
243 /// ```
244 ///
245 /// After the second call to `reserve_capacity`, the *total* requested
246 /// capacity will be 200.
247 ///
248 /// `reserve_capacity` is also used to cancel previous capacity requests.
249 /// Given the following:
250 ///
251 /// ```rust
252 /// # use h2::*;
253 /// # fn doc(mut send_stream: SendStream<&'static [u8]>) {
254 /// send_stream.reserve_capacity(100);
255 /// send_stream.reserve_capacity(0);
256 /// # }
257 /// ```
258 ///
259 /// After the second call to `reserve_capacity`, the *total* requested
260 /// capacity will be 0, i.e. there is no requested capacity for the stream.
261 ///
262 /// If `reserve_capacity` is called with a lower value than the amount of
263 /// capacity **currently** assigned to the stream, this capacity will be
264 /// returned to the connection to be re-assigned to other streams.
265 ///
266 /// Also, the amount of capacity that is reserved gets decremented as data
267 /// is sent. For example:
268 ///
269 /// ```rust
270 /// # use h2::*;
271 /// # async fn doc(mut send_stream: SendStream<&'static [u8]>) {
272 /// send_stream.reserve_capacity(100);
273 ///
274 /// send_stream.send_data(b"hello", false).unwrap();
275 /// // At this point, the total amount of requested capacity is 95 bytes.
276 ///
277 /// // Calling `reserve_capacity` with `100` again essentially requests an
278 /// // additional 5 bytes.
279 /// send_stream.reserve_capacity(100);
280 /// # }
281 /// ```
282 ///
283 /// See [Flow control](struct.SendStream.html#flow-control) for an overview
284 /// of how send flow control works.
285 pub fn reserve_capacity(&mut self, capacity: usize) {
286 // TODO: Check for overflow
287 self.inner.reserve_capacity(capacity as WindowSize)
288 }
289
290 /// Returns the stream's current send capacity.
291 ///
292 /// This allows the caller to check the current amount of available capacity
293 /// before sending data.
294 pub fn capacity(&self) -> usize {
295 self.inner.capacity() as usize
296 }
297
298 /// Requests to be notified when the stream's capacity increases.
299 ///
300 /// Before calling this, capacity should be requested with
301 /// `reserve_capacity`. Once capacity is requested, the connection will
302 /// assign capacity to the stream **as it becomes available**. There is no
303 /// guarantee as to when and in what increments capacity gets assigned to
304 /// the stream.
305 ///
306 /// To get notified when the available capacity increases, the caller calls
307 /// `poll_capacity`, which returns `Ready(Some(n))` when `n` has been
308 /// increased by the connection. Note that `n` here represents the **total**
309 /// amount of assigned capacity at that point in time. It is also possible
310 /// that `n` is lower than the previous call if, since then, the caller has
311 /// sent data.
312 pub fn poll_capacity(&mut self, cx: &mut Context) -> Poll<Option<Result<usize, crate::Error>>> {
313 self.inner
314 .poll_capacity(cx)
315 .map_ok_(|w| w as usize)
316 .map_err_(Into::into)
317 }
318
319 /// Sends a single data frame to the remote peer.
320 ///
321 /// This function may be called repeatedly as long as `end_of_stream` is set
322 /// to `false`. Setting `end_of_stream` to `true` sets the end stream flag
323 /// on the data frame. Any further calls to `send_data` or `send_trailers`
324 /// will return an [`Error`].
325 ///
326 /// `send_data` can be called without reserving capacity. In this case, the
327 /// data is buffered and the capacity is implicitly requested. Once the
328 /// capacity becomes available, the data is flushed to the connection.
329 /// However, this buffering is unbounded. As such, sending large amounts of
330 /// data without reserving capacity before hand could result in large
331 /// amounts of data being buffered in memory.
332 ///
333 /// [`Error`]: struct.Error.html
334 pub fn send_data(&mut self, data: B, end_of_stream: bool) -> Result<(), crate::Error> {
335 self.inner
336 .send_data(data, end_of_stream)
337 .map_err(Into::into)
338 }
339
340 /// Sends trailers to the remote peer.
341 ///
342 /// Sending trailers implicitly closes the send stream. Once the send stream
343 /// is closed, no more data can be sent.
344 pub fn send_trailers(&mut self, trailers: HeaderMap) -> Result<(), crate::Error> {
345 self.inner.send_trailers(trailers).map_err(Into::into)
346 }
347
348 /// Resets the stream.
349 ///
350 /// This cancels the request / response exchange. If the response has not
351 /// yet been received, the associated `ResponseFuture` will return an
352 /// [`Error`] to reflect the canceled exchange.
353 ///
354 /// [`Error`]: struct.Error.html
355 pub fn send_reset(&mut self, reason: Reason) {
356 self.inner.send_reset(reason)
357 }
358
359 /// Polls to be notified when the client resets this stream.
360 ///
361 /// If stream is still open, this returns `Poll::Pending`, and
362 /// registers the task to be notified if a `RST_STREAM` is received.
363 ///
364 /// If a `RST_STREAM` frame is received for this stream, calling this
365 /// method will yield the `Reason` for the reset.
366 ///
367 /// # Error
368 ///
369 /// If connection sees an error, this returns that error instead of a
370 /// `Reason`.
371 pub fn poll_reset(&mut self, cx: &mut Context) -> Poll<Result<Reason, crate::Error>> {
372 self.inner.poll_reset(cx, proto::PollReset::Streaming)
373 }
374
375 /// Returns the stream ID of this `SendStream`.
376 ///
377 /// # Panics
378 ///
379 /// If the lock on the stream store has been poisoned.
380 pub fn stream_id(&self) -> StreamId {
381 StreamId::from_internal(self.inner.stream_id())
382 }
383 }
384
385 // ===== impl StreamId =====
386
387 impl StreamId {
388 pub(crate) fn from_internal(id: crate::frame::StreamId) -> Self {
389 StreamId(id.into())
390 }
391 }
392 // ===== impl RecvStream =====
393
394 impl RecvStream {
395 pub(crate) fn new(inner: FlowControl) -> Self {
396 RecvStream { inner }
397 }
398
399 /// Get the next data frame.
400 pub async fn data(&mut self) -> Option<Result<Bytes, crate::Error>> {
401 futures_util::future::poll_fn(move |cx| self.poll_data(cx)).await
402 }
403
404 /// Get optional trailers for this stream.
405 pub async fn trailers(&mut self) -> Result<Option<HeaderMap>, crate::Error> {
406 futures_util::future::poll_fn(move |cx| self.poll_trailers(cx)).await
407 }
408
409 #[doc(hidden)]
410 pub fn poll_data(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes, crate::Error>>> {
411 self.inner.inner.poll_data(cx).map_err_(Into::into)
412 }
413
414 #[doc(hidden)]
415 pub fn poll_trailers(
416 &mut self,
417 cx: &mut Context,
418 ) -> Poll<Result<Option<HeaderMap>, crate::Error>> {
419 match ready!(self.inner.inner.poll_trailers(cx)) {
420 Some(Ok(map)) => Poll::Ready(Ok(Some(map))),
421 Some(Err(e)) => Poll::Ready(Err(e.into())),
422 None => Poll::Ready(Ok(None)),
423 }
424 }
425
426 /// Returns true if the receive half has reached the end of stream.
427 ///
428 /// A return value of `true` means that calls to `poll` and `poll_trailers`
429 /// will both return `None`.
430 pub fn is_end_stream(&self) -> bool {
431 self.inner.inner.is_end_stream()
432 }
433
434 /// Get a mutable reference to this stream's `FlowControl`.
435 ///
436 /// It can be used immediately, or cloned to be used later.
437 pub fn flow_control(&mut self) -> &mut FlowControl {
438 &mut self.inner
439 }
440
441 /// Returns the stream ID of this stream.
442 ///
443 /// # Panics
444 ///
445 /// If the lock on the stream store has been poisoned.
446 pub fn stream_id(&self) -> StreamId {
447 self.inner.stream_id()
448 }
449 }
450
451 #[cfg(feature = "stream")]
452 impl futures_core::Stream for RecvStream {
453 type Item = Result<Bytes, crate::Error>;
454
455 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
456 self.poll_data(cx)
457 }
458 }
459
460 impl fmt::Debug for RecvStream {
461 fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
462 fmt.debug_struct("RecvStream")
463 .field("inner", &self.inner)
464 .finish()
465 }
466 }
467
468 impl Drop for RecvStream {
469 fn drop(&mut self) {
470 // Eagerly clear any received DATA frames now, since its no longer
471 // possible to retrieve them. However, this will be called
472 // again once *all* stream refs have been dropped, since
473 // this won't send a RST_STREAM frame, in case the user wishes to
474 // still *send* DATA.
475 self.inner.inner.clear_recv_buffer();
476 }
477 }
478
479 // ===== impl FlowControl =====
480
481 impl FlowControl {
482 pub(crate) fn new(inner: proto::OpaqueStreamRef) -> Self {
483 FlowControl { inner }
484 }
485
486 /// Returns the stream ID of the stream whose capacity will
487 /// be released by this `FlowControl`.
488 pub fn stream_id(&self) -> StreamId {
489 StreamId::from_internal(self.inner.stream_id())
490 }
491
492 /// Get the current available capacity of data this stream *could* receive.
493 pub fn available_capacity(&self) -> isize {
494 self.inner.available_recv_capacity()
495 }
496
497 /// Get the currently *used* capacity for this stream.
498 ///
499 /// This is the amount of bytes that can be released back to the remote.
500 pub fn used_capacity(&self) -> usize {
501 self.inner.used_recv_capacity() as usize
502 }
503
504 /// Release window capacity back to remote stream.
505 ///
506 /// This releases capacity back to the stream level and the connection level
507 /// windows. Both window sizes will be increased by `sz`.
508 ///
509 /// See [struct level] documentation for more details.
510 ///
511 /// # Errors
512 ///
513 /// This function errors if increasing the receive window size by `sz` would
514 /// result in a window size greater than the target window size. In other
515 /// words, the caller cannot release more capacity than data has been
516 /// received. If 1024 bytes of data have been received, at most 1024 bytes
517 /// can be released.
518 ///
519 /// [struct level]: #
520 pub fn release_capacity(&mut self, sz: usize) -> Result<(), crate::Error> {
521 if sz > proto::MAX_WINDOW_SIZE as usize {
522 return Err(UserError::ReleaseCapacityTooBig.into());
523 }
524 self.inner
525 .release_capacity(sz as proto::WindowSize)
526 .map_err(Into::into)
527 }
528 }
529
530 // ===== impl PingPong =====
531
532 impl PingPong {
533 pub(crate) fn new(inner: proto::UserPings) -> Self {
534 PingPong { inner }
535 }
536
537 /// Send a PING frame and wait for the peer to send the pong.
538 pub async fn ping(&mut self, ping: Ping) -> Result<Pong, crate::Error> {
539 self.send_ping(ping)?;
540 futures_util::future::poll_fn(|cx| self.poll_pong(cx)).await
541 }
542
543 #[doc(hidden)]
544 pub fn send_ping(&mut self, ping: Ping) -> Result<(), crate::Error> {
545 // Passing a `Ping` here is just to be forwards-compatible with
546 // eventually allowing choosing a ping payload. For now, we can
547 // just drop it.
548 drop(ping);
549
550 self.inner.send_ping().map_err(|err| match err {
551 Some(err) => err.into(),
552 None => UserError::SendPingWhilePending.into(),
553 })
554 }
555
556 #[doc(hidden)]
557 pub fn poll_pong(&mut self, cx: &mut Context) -> Poll<Result<Pong, crate::Error>> {
558 ready!(self.inner.poll_pong(cx))?;
559 Poll::Ready(Ok(Pong { _p: () }))
560 }
561 }
562
563 impl fmt::Debug for PingPong {
564 fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
565 fmt.debug_struct("PingPong").finish()
566 }
567 }
568
569 // ===== impl Ping =====
570
571 impl Ping {
572 /// Creates a new opaque `Ping` to be sent via a [`PingPong`][].
573 ///
574 /// The payload is "opaque", such that it shouldn't be depended on.
575 ///
576 /// [`PingPong`]: struct.PingPong.html
577 pub fn opaque() -> Ping {
578 Ping { _p: () }
579 }
580 }
581
582 impl fmt::Debug for Ping {
583 fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
584 fmt.debug_struct("Ping").finish()
585 }
586 }
587
588 // ===== impl Pong =====
589
590 impl fmt::Debug for Pong {
591 fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
592 fmt.debug_struct("Pong").finish()
593 }
594 }
595