hickory_proto/xfer/
retry_dns_handle.rs1use std::pin::Pin;
11use std::task::{Context, Poll};
12
13use futures_util::stream::{Stream, StreamExt};
14
15use crate::error::{ProtoError, ProtoErrorKind};
16use crate::xfer::{DnsRequest, DnsResponse};
17use crate::DnsHandle;
18
19#[derive(Clone)]
32#[must_use = "queries can only be sent through a ClientHandle"]
33pub struct RetryDnsHandle<H>
34where
35 H: DnsHandle + Unpin + Send,
36 H::Error: RetryableError,
37{
38 handle: H,
39 attempts: usize,
40}
41
42impl<H> RetryDnsHandle<H>
43where
44 H: DnsHandle + Unpin + Send,
45 H::Error: RetryableError,
46{
47 pub fn new(handle: H, attempts: usize) -> Self {
54 Self { handle, attempts }
55 }
56}
57
58impl<H> DnsHandle for RetryDnsHandle<H>
59where
60 H: DnsHandle + Send + Unpin + 'static,
61 H::Error: RetryableError,
62{
63 type Response = Pin<Box<dyn Stream<Item = Result<DnsResponse, Self::Error>> + Send + Unpin>>;
64 type Error = <H as DnsHandle>::Error;
65
66 fn send<R: Into<DnsRequest>>(&self, request: R) -> Self::Response {
67 let request = request.into();
68
69 let stream = self.handle.send(request.clone());
72
73 Box::pin(RetrySendStream {
74 request,
75 handle: self.handle.clone(),
76 stream,
77 remaining_attempts: self.attempts,
78 })
79 }
80}
81
82struct RetrySendStream<H>
84where
85 H: DnsHandle,
86{
87 request: DnsRequest,
88 handle: H,
89 stream: <H as DnsHandle>::Response,
90 remaining_attempts: usize,
91}
92
93impl<H: DnsHandle + Unpin> Stream for RetrySendStream<H>
94where
95 <H as DnsHandle>::Error: RetryableError,
96{
97 type Item = Result<DnsResponse, <H as DnsHandle>::Error>;
98
99 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
100 loop {
103 match self.stream.poll_next_unpin(cx) {
104 Poll::Ready(Some(Err(e))) => {
105 if self.remaining_attempts == 0 || !e.should_retry() {
106 return Poll::Ready(Some(Err(e)));
107 }
108
109 if e.attempted() {
110 self.remaining_attempts -= 1;
111 }
112
113 let request = self.request.clone();
116 self.stream = self.handle.send(request);
117 }
118 poll => return poll,
119 }
120 }
121 }
122}
123
124pub trait RetryableError {
126 fn should_retry(&self) -> bool;
128 fn attempted(&self) -> bool;
130}
131
132impl RetryableError for ProtoError {
133 fn should_retry(&self) -> bool {
134 true
135 }
136
137 fn attempted(&self) -> bool {
138 !matches!(self.kind(), ProtoErrorKind::Busy)
139 }
140}
141
142#[cfg(test)]
143mod test {
144 use super::*;
145 use crate::error::*;
146 use crate::op::*;
147 use crate::xfer::FirstAnswer;
148 use futures_executor::block_on;
149 use futures_util::future::*;
150 use futures_util::stream::*;
151 use std::sync::{
152 atomic::{AtomicU16, Ordering},
153 Arc,
154 };
155 use DnsHandle;
156
157 #[derive(Clone)]
158 struct TestClient {
159 last_succeed: bool,
160 retries: u16,
161 attempts: Arc<AtomicU16>,
162 }
163
164 impl DnsHandle for TestClient {
165 type Response = Box<dyn Stream<Item = Result<DnsResponse, ProtoError>> + Send + Unpin>;
166 type Error = ProtoError;
167
168 fn send<R: Into<DnsRequest>>(&self, _: R) -> Self::Response {
169 let i = self.attempts.load(Ordering::SeqCst);
170
171 if (i > self.retries || self.retries - i == 0) && self.last_succeed {
172 let mut message = Message::new();
173 message.set_id(i);
174 return Box::new(once(ok(DnsResponse::from_message(message).unwrap())));
175 }
176
177 self.attempts.fetch_add(1, Ordering::SeqCst);
178 Box::new(once(err(ProtoError::from("last retry set to fail"))))
179 }
180 }
181
182 #[test]
183 fn test_retry() {
184 let handle = RetryDnsHandle::new(
185 TestClient {
186 last_succeed: true,
187 retries: 1,
188 attempts: Arc::new(AtomicU16::new(0)),
189 },
190 2,
191 );
192 let test1 = Message::new();
193 let result = block_on(handle.send(test1).first_answer()).expect("should have succeeded");
194 assert_eq!(result.id(), 1); }
196
197 #[test]
198 fn test_error() {
199 let client = RetryDnsHandle::new(
200 TestClient {
201 last_succeed: false,
202 retries: 1,
203 attempts: Arc::new(AtomicU16::new(0)),
204 },
205 2,
206 );
207 let test1 = Message::new();
208 assert!(block_on(client.send(test1).first_answer()).is_err());
209 }
210}