Custom lifetimes for callback request headers and futures about tokio-tungstenite HOT 6 CLOSED

pub fn run(&self, address: SocketAddr) {
    let mut core = Core::new().unwrap();
    let handle = core.handle();
    let socket = TcpListener::bind(&address, &handle).unwrap();
    println!("Listening on: {}", address);

    let server = socket.incoming().for_each(|(stream, addr)| {
        let engine_inner = self.engine.clone();
        let connections_inner = self.connections.clone();
        let auth_middleware_inner = self.auth_middleware.clone();
        let handle_inner = handle.clone();

        let mut headers: Headers = HashMap::new();
        let copy_headers_callback = |request: &Request| {
            for &(ref name, ref value) in request.headers.iter() {
                headers.insert(name.to_string(), value.clone());
            }
            Ok(None)
        };

        accept_hdr_async(stream, copy_headers_callback)
            // Process the messages
            .and_then(move |ws_stream| {

pub type Headers = HashMap<String, Box<[u8]>>;

pub struct Proxy {
    engine: Rc<RefCell<Engine>>,
    connections: Rc<RefCell<HashMap<SocketAddr, mpsc::UnboundedSender<Message>>>>,
    auth_middleware: Rc<RefCell<Box<Middleware>>>,
}


impl Proxy {
    pub fn new(router: Box<Router>, cli: &CliOptions) -> Proxy {
        let auth_middleware: Box<Middleware> = Box::new(EmptyMiddleware::new(cli));

        Proxy {
            engine: Rc::new(RefCell::new(Engine::new(router))),
            connections: Rc::new(RefCell::new(HashMap::new())),
            auth_middleware: Rc::new(RefCell::new(auth_middleware)),
        }
    }

    pub fn run(&self, address: SocketAddr) {
        let mut core = Core::new().unwrap();
        let handle = core.handle();
        let socket = TcpListener::bind(&address, &handle).unwrap();
        println!("Listening on: {}", address);

        let server = socket.incoming().for_each(|(stream, addr)| {
            let engine_inner = self.engine.clone();
            let connections_inner = self.connections.clone();
            let auth_middleware_inner = self.auth_middleware.clone();
            let handle_inner = handle.clone();

            let mut headers: Rc<RefCell<Headers>> = Rc::new(RefCell::new(HashMap::new()));
            let copy_headers_callback = |request: &Request| {
                let headers_inner = headers.clone();
                for &(ref name, ref value) in request.headers.iter() {
                    headers_inner.borrow_mut().insert(name.to_string(), value.clone());
                }
                Ok(None)
            };

            accept_hdr_async(stream, copy_headers_callback)
                // Process the messages
                .and_then(move |ws_stream| {
                    // Create a channel for the stream, which other sockets will use to
                    // send us messages. It could be used for broadcasting your data to
                    // another users in the future.
                    let (tx, rx) = mpsc::unbounded();
                    connections_inner.borrow_mut().insert(addr, tx);

                    // Split the WebSocket stream so that it will be possible to work
                    // with the reading and writing halves separately.
                    let (sink, stream) = ws_stream.split();

                    // TODO: fire it before starting the processing messages
                    let _auth_future = auth_middleware_inner.borrow().process_request(&headers.borrow(), &handle_inner);

                    // Read and process each message
                    let connections = connections_inner.clone();
                    let ws_reader = stream.for_each(move |message: Message| {
                        engine_inner.borrow().handle(&message, &addr, &connections);
                        Ok(())
                    });

                    // Write back prepared responses
                    let ws_writer = rx.fold(sink, |mut sink, msg| {
                        sink.start_send(msg).unwrap();
                        Ok(sink)
                    });

                    // Wait for either half to be done to tear down the other
                    let connection = ws_reader.map(|_| ()).map_err(|_| ())
                                              .select(ws_writer.map(|_| ()).map_err(|_| ()));

                    // Close the connection after using
                    handle_inner.spawn(connection.then(move |_| {
                        connections_inner.borrow_mut().remove(&addr);
                        println!("Connection {} closed.", addr);
                        Ok(())
                    }));

                    Ok(())
                })
                // An error occurred during the WebSocket handshake
                .or_else(|err| {
                    println!("{}", err.description());
                    Ok(())
                })
        });

        // Run the server
        core.run(server).unwrap();
    }
}