You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: mdns_stream.rs
    path of file: ./repos/hickory-dns/crates/proto/src/multicast
    
    the code of the file until where you have to start completion: // Copyright 2015-2018 Benjamin Fry <benjaminfry@me.com>
//
// Licensed under the Apache License, Version 2.0, <LICENSE-APACHE or
// https://apache.org/licenses/LICENSE-2.0> or the MIT license <LICENSE-MIT or
// https://opensource.org/licenses/MIT>, at your option. This file may not be
// copied, modified, or distributed except according to those terms.

use std;
use std::io;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use futures_util::stream::{Stream, StreamExt};
use futures_util::{future, future::Future, ready, FutureExt, TryFutureExt};
use once_cell::sync::Lazy;
use rand;
use rand::distributions::{uniform::Uniform, Distribution};
use socket2::{self, Socket};
use tokio::net::UdpSocket;
use tracing::{debug, trace};

use crate::multicast::MdnsQueryType;
use crate::u
    
    fn passive_mdns_test(mdns_query_type: MdnsQueryType, mdns_addr: SocketAddr) {
        use std::time::Duration;

        let server_got_packet = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));

        let test_bytes: &'static [u8; 8] = b"DEADBEEF";
        let send_recv_times = 10;
        let server_got_packet_clone = server_got_packet.clone();

        // an in and out server
        let _server_handle = std::thread::Builder::new()
            .name("test_one_shot_mdns:server".to_string())
            .spawn(move || {
                let io_loop = runtime::Runtime::new().unwrap();
                let mut timeout = future::lazy(|_| tokio::time::sleep(Duration::from_millis(100)))
                    .flatten()
                    .boxed();

                // TTLs are 0 so that multicast test packets never leave the test host...
                // FIXME: this is hardcoded to index 5 for ipv6, which isn't going to be correct in most cases...
                let (server_stream_future, _server_sender) =
                    MdnsStream::new(mdns_addr, mdns_query_type, Some(1), None, Some(5));

                // For one-shot responses we are competing with a system mDNS responder, we will respond from a different port...
                let mut server_stream = io_loop
                    .block_on(server_stream_future)
                    .expect("could not create mDNS listener")
                    .into_future();

                for _ in 0..=send_recv_times {
                    // wait for some bytes...
                    match io_loop.block_on(
                        future::lazy(|_| future::select(server_stream, timeout)).flatten(),
                    ) {
                        Either::Left((_buffer_and_addr_stream_tmp, _timeout_tmp)) => {
                            // let (buffer_and_addr, stream_tmp) = buffer_and_addr_stream_tmp;

                            // server_stream = stream_tmp.into_future();
                            // timeout = timeout_tmp;
                            // let (buffer, addr) = buffer_and_addr.expect("no buffer received");

                            // assert_eq!(&buffer, test_bytes);
                            // println!("server got data! {}", addr);

                            server_got_packet_clone
                                .store(true, std::sync::atomic::Ordering::Relaxed);
                            return;
                        }
                        Either::Right(((), buffer_and_addr_stream_tmp)) => {
                            server_stream = buffer_and_addr_stream_tmp;
                            timeout =
                                future::lazy(|_| tokio::time::sleep(Duration::from_millis(100)))
                                    .flatten()
                                    .boxed();
                        }
                    }

                    // let the server turn for a bit... send the message
                    io_loop.block_on(tokio::time::sleep(Duration::from_millis(100)));
                }
            })
            .unwrap();

        // setup the client, which is going to run on the testing thread...
        let io_loop = runtime::Runtime::new().unwrap();
        // FIXME: this is hardcoded to index 5 for ipv6, which isn't going to be correct in most cases...
        let (stream, mut sender) =
            MdnsStream::new(mdns_addr, MdnsQueryType::OneShot, Some(1), None, Some(5));
        let mut stream = io_loop.block_on(stream).ok().unwrap().into_future();
        let mut timeout = future::lazy(|_| tokio::time::sleep(Duration::from_millis(100)))
            .flatten()
            .boxed();

        for _ in 0..send_recv_times {
            // test once
            sender
                .send(SerialMessage::new(test_bytes.to_vec(), mdns_addr))
                .unwrap();

            println!("client sending data!");

            // TODO: this lazy is probably unnecessary?
            let run_result =
                io_loop.block_on(future::lazy(|_| future::select(stream, timeout)).flatten());

            if server_got_packet.load(std::sync::atomic::Ordering::Relaxed) {
                return;
            }

            match run_result {
                Either::Left((buffer_and_addr_stream_tmp, timeout_tmp)) => {
                    let (_buffer_and_addr, stream_tmp) = buffer_and_addr_stream_tmp;
                    stream = stream_tmp.into_future();
                    timeout = timeout_tmp;
                }
                Either::Right(((), buffer_and_addr_stream_tmp)) => {
                    stream = buffer_and_addr_stream_tmp;
                    timeout = future::lazy(|_| tokio::time::sleep(Duration::from_millis(100)))
                        .flatten()
                        .boxed();
                }
            }
        }

        panic!("server never got packet.");
    }