hickory_net/xfer/
retry_dns_handle.rs1use core::pin::Pin;
11use core::task::{Context, Poll};
12
13use futures_util::stream::{Stream, StreamExt};
14
15use crate::xfer::{DnsHandle, DnsRequest, DnsResponse};
16use crate::{DnsError, NetError};
17
18#[derive(Clone)]
28#[must_use = "queries can only be sent through a ClientHandle"]
29#[allow(dead_code)]
30pub struct RetryDnsHandle<H> {
31 handle: H,
32 attempts: usize,
33}
34
35impl<H> RetryDnsHandle<H> {
36 pub fn new(handle: H, attempts: usize) -> Self {
43 Self { handle, attempts }
44 }
45}
46
47impl<H: DnsHandle> DnsHandle for RetryDnsHandle<H> {
48 type Response = Pin<Box<dyn Stream<Item = Result<DnsResponse, NetError>> + Send + Unpin>>;
49 type Runtime = H::Runtime;
50
51 fn send(&self, request: DnsRequest) -> Self::Response {
52 let stream = self.handle.send(request.clone());
55
56 Box::pin(RetrySendStream {
57 request,
58 handle: self.handle.clone(),
59 stream,
60 remaining_attempts: self.attempts,
61 })
62 }
63}
64
65struct RetrySendStream<H: DnsHandle> {
67 request: DnsRequest,
68 handle: H,
69 stream: <H as DnsHandle>::Response,
70 remaining_attempts: usize,
71}
72
73impl<H: DnsHandle> Stream for RetrySendStream<H> {
74 type Item = Result<DnsResponse, NetError>;
75
76 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
77 loop {
80 let err = match self.stream.poll_next_unpin(cx) {
81 Poll::Ready(Some(Err(e))) => e,
82 poll => return poll,
83 };
84
85 match (self.remaining_attempts, err) {
86 (0, err) => return Poll::Ready(Some(Err(err))),
88 (
90 _,
91 err @ NetError::NoConnections
92 | err @ NetError::Dns(DnsError::NoRecordsFound(_)),
93 ) => return Poll::Ready(Some(Err(err))),
94 (_, NetError::Busy) => {}
96 (_, _) => self.remaining_attempts -= 1,
98 }
99
100 let request = self.request.clone();
103 self.stream = self.handle.send(request);
104 }
105 }
106}
107
108#[cfg(all(test, feature = "tokio"))]
109mod test {
110 use core::sync::atomic::{AtomicU16, Ordering};
111 use std::sync::Arc;
112
113 use futures_executor::block_on;
114 use futures_util::future::{err, ok};
115 use futures_util::stream::{Stream, once};
116
117 use super::*;
118 use crate::proto::op::Message;
119 use crate::runtime::TokioRuntimeProvider;
120 use crate::xfer::{DnsHandle, DnsRequest, DnsResponse, FirstAnswer};
121 use test_support::subscribe;
122
123 #[derive(Clone)]
124 struct TestClient {
125 last_succeed: bool,
126 retries: u16,
127 attempts: Arc<AtomicU16>,
128 }
129
130 impl DnsHandle for TestClient {
131 type Response = Box<dyn Stream<Item = Result<DnsResponse, NetError>> + Send + Unpin>;
132 type Runtime = TokioRuntimeProvider;
133
134 fn send(&self, _: DnsRequest) -> Self::Response {
135 let i = self.attempts.load(Ordering::SeqCst);
136
137 if (i > self.retries || self.retries - i == 0) && self.last_succeed {
138 let mut message = Message::query();
139 message.metadata.id = i;
140 return Box::new(once(ok(
141 DnsResponse::from_message(message.into_response()).unwrap()
142 )));
143 }
144
145 self.attempts.fetch_add(1, Ordering::SeqCst);
146 Box::new(once(err(NetError::from("last retry set to fail"))))
147 }
148 }
149
150 #[test]
151 fn test_retry() {
152 subscribe();
153 let handle = RetryDnsHandle::new(
154 TestClient {
155 last_succeed: true,
156 retries: 1,
157 attempts: Arc::new(AtomicU16::new(0)),
158 },
159 2,
160 );
161 let test1 = DnsRequest::from(Message::query());
162 let result = block_on(handle.send(test1).first_answer()).expect("should have succeeded");
163 assert_eq!(result.id, 1); }
165
166 #[test]
167 fn test_error() {
168 subscribe();
169 let client = RetryDnsHandle::new(
170 TestClient {
171 last_succeed: false,
172 retries: 1,
173 attempts: Arc::new(AtomicU16::new(0)),
174 },
175 2,
176 );
177 let test1 = DnsRequest::from(Message::query());
178 assert!(block_on(client.send(test1).first_answer()).is_err());
179 }
180}