fn _reader_thread(&self, until_message: Option<usize>) -> Result<serde_json::Value, Error> {
        let mut raw_resp = String::new();
        let resp = match self.buf_reader.try_lock() {
            Ok(mut reader) => {
                trace!(
                    "Starting reader thread with `until_message` = {:?}",
                    until_message
                );

                if let Some(until_message) = until_message {
                    // If we are trying to start a reader thread but the corresponding sender is
                    // missing from the map, exit immediately. This can happen with batch calls,
                    // since the sender is shared for all the individual queries in a call. We
                    // might have already received a response for that id, but we don't know it
                    // yet. Exiting here forces the calling code to fallback to the sender-receiver
                    // method, and it should find a message there waiting for it.
                    if self.waiting_map.lock()?.get(&until_message).is_none() {
                        return Err(Error::CouldntLockReader);
                    }
                }

                // Loop over every message
                loop {
                    raw_resp.clear();

                    if let Err(e) = reader.read_line(&mut raw_resp) {
                        let error = Arc::new(e);
                        for (_, s) in self.waiting_map.lock().unwrap().drain() {
                            s.send(ChannelMessage::Error(error.clone()))?;
                        }
                        return Err(Error::SharedIOError(error));
                    }
                    trace!("<== {}", raw_resp);

                    let resp: serde_json::Value = serde_json::from_str(&raw_resp)?;

                    // Normally there is and id, but it's missing for spontaneous notifications
                    // from the server
                    let resp_id = resp["id"]
                        .as_str()
                        .and_then(|s| s.parse().ok())
                        .or_else(|| resp["id"].as_u64().map(|i| i as usize));
                    match resp_id {
                        Some(resp_id) if until_message == Some(resp_id) => {
                            // We have a valid id and it's exactly the one we were waiting for!
                            trace!(
                                "Reader thread {} received a response for its request",
                                resp_id
                            );

                            // Remove ourselves from the "waiting map"
                            let mut map = self.waiting_map.lock()?;
                            map.remove(&resp_id);

                            // If the map is not empty, we select a random thread to become the
                            // new reader thread.
                            if let Some(err) = map.values().find_map(|sender| {
                                sender
                                    .send(ChannelMessage::WakeUp)
                                    .map_err(|err| {
                                        warn!("Unable to wake up a thread, trying some other");
                                        err
                                    })
                                    .err()
                            }) {
                                error!("All the threads has failed, giving up");
                                return Err(err)?;
                            }

                            break Ok(resp);
                        }
                        Some(resp_id) => {
                            // We have an id, but it's not our response. Notify the thread and
                            // move on
                            trace!("Reader thread received response for {}", resp_id);

                            if let Some(sender) = self.waiting_map.lock()?.remove(&resp_id) {
                                sender.send(ChannelMessage::Response(resp))?;
                            } else {
                                warn!("Missing listener for {}", resp_id);
                            }
                        }
                        None => {
                            // No id, that's probably a notification.
                            let mut resp = resp;

                            if let Some(ref method) = resp["method"].take().as_str() {
                                self.handle_notification(method, resp["params"].take())?;
                            } else {
                                warn!("Unexpected response: {:?}", resp);
                            }
                        }
                    }
                }
            }
            Err(TryLockError::WouldBlock) => {
                // If we "WouldBlock" here it means that there's already a reader thread
                // running somewhere.
                Err(Error::CouldntLockReader)
            }
            Err(TryLockError::Poisoned(e)) => Err(e)?,
        };

        let resp = resp?;
        if let Some(err) = resp.get("error") {
            Err(Error::Protocol(err.clone()))
        } else {
            Ok(resp)
        }
    }
// TODO: to enable this we have to find a way to allow concurrent read and writes to the
// underlying transport struct. This can be done pretty easily for TcpStream because it can be
// split into a "read" and a "write" object, but it's not as trivial for other types. Without
// such thing, this causes a deadlock, because the reader thread takes a lock on the
// `ClonableStream` before other threads can send a request to the server. They will block
// waiting for the reader to release the mutex, but this will never happen because the server
// didn't receive any request, so it has nothing to send back.
// pub fn reader_thread(&self) -> Result<(), Error> {
//     self._reader_thread(None).map(|_| ())
// }
to enable this we have to find a way to allow concurrent read and writes to the
underlying transport struct. This can be done pretty easily for TcpStream because it can be
split into a "read" and a "write" object, but it's not as trivial for other types. Without
such thing, this causes a deadlock, because the reader thread takes a lock on the
`ClonableStream` before other threads can send a request to the server. They will block
waiting for the reader to release the mutex, but this will never happen because the server
didn't receive any request, so it has nothing to send back.
pub fn reader_thread(&self) -> Result<(), Error> {
self._reader_thread(None).map(|_| ())
}
to enable this we have to find a way to allow concurrent read and writes to the underlying transport struct . This can be done pretty easily for TcpStream because it can be split into a " read " and a " write " object but it ' s not as trivial for other types . Without such thing this causes a deadlock because the reader thread takes a lock on the ` ClonableStream ` before other threads can send a request to the server . They will block waiting for the reader to release the mutex but this will never happen because the server didn ' t receive any request so it has nothing to send back . pub fn reader_thread ( &self ) - > Result< () Error > { self . _reader_thread ( None ) . map ( |_| () ) }
fn _reader_thread(&self, until_message: Option<usize>) -> Result<serde_json::Value, Error> {
        let mut raw_resp = String::new();
        let resp = match self.buf_reader.try_lock() {
            Ok(mut reader) => {
                trace!(
                    "Starting reader thread with `until_message` = {:?}",
                    until_message
                );
                if let Some(until_message) = until_message {
                    // If we are trying to start a reader thread but the corresponding sender is
                    // missing from the map, exit immediately. This can happen with batch calls,
                    // since the sender is shared for all the individual queries in a call. We
                    // might have already received a response for that id, but we don't know it
                    // yet. Exiting here forces the calling code to fallback to the sender-receiver
                    // method, and it should find a message there waiting for it.
                    if self.waiting_map.lock()?.get(&until_message).is_none() {
                        return Err(Error::CouldntLockReader);
                    }
                }
                // Loop over every message
                loop {
                    raw_resp.clear();
                    if let Err(e) = reader.read_line(&mut raw_resp) {
                        let error = Arc::new(e);
                        for (_, s) in self.waiting_map.lock().unwrap().drain() {
                            s.send(ChannelMessage::Error(error.clone()))?;
                        }
                        return Err(Error::SharedIOError(error));
                    }
                    trace!("<== {}", raw_resp);
                    let resp: serde_json::Value = serde_json::from_str(&raw_resp)?;
                    // Normally there is and id, but it's missing for spontaneous notifications
                    // from the server
                    let resp_id = resp["id"]
                        .as_str()
                        .and_then(|s| s.parse().ok())
                        .or_else(|| resp["id"].as_u64().map(|i| i as usize));
                    match resp_id {
                        Some(resp_id) if until_message == Some(resp_id) => {
                            // We have a valid id and it's exactly the one we were waiting for!
                            trace!(
                                "Reader thread {} received a response for its request",
                                resp_id
                            );
                            // Remove ourselves from the "waiting map"
                            let mut map = self.waiting_map.lock()?;
                            map.remove(&resp_id);
                            // If the map is not empty, we select a random thread to become the
                            // new reader thread.
                            if let Some(err) = map.values().find_map(|sender| {
                                sender
                                    .send(ChannelMessage::WakeUp)
                                    .map_err(|err| {
                                        warn!("Unable to wake up a thread, trying some other");
                                        err
                                    })
                                    .err()
                            }) {
                                error!("All the threads has failed, giving up");
                                return Err(err)?;
                            }
                            break Ok(resp);
                        }
                        Some(resp_id) => {
                            // We have an id, but it's not our response. Notify the thread and
                            // move on
                            trace!("Reader thread received response for {}", resp_id);
                            if let Some(sender) = self.waiting_map.lock()?.remove(&resp_id) {
                                sender.send(ChannelMessage::Response(resp))?;
                            } else {
                                warn!("Missing listener for {}", resp_id);
                            }
                        }
                        None => {
                            // No id, that's probably a notification.
                            let mut resp = resp;
                            if let Some(ref method) = resp["method"].take().as_str() {
                                self.handle_notification(method, resp["params"].take())?;
                            } else {
                                warn!("Unexpected response: {:?}", resp);
                            }
                        }
                    }
                }
            }
            Err(TryLockError::WouldBlock) => {
                // If we "WouldBlock" here it means that there's already a reader thread
                // running somewhere.
                Err(Error::CouldntLockReader)
            }
            Err(TryLockError::Poisoned(e)) => Err(e)?,
        };
        let resp = resp?;
        if let Some(err) = resp.get("error") {
            Err(Error::Protocol(err.clone()))
        } else {
            Ok(resp)
        }
    }
fn _reader_thread ( & self , until_message : Option < usize > ) -> Result < serde_json :: Value , Error > { let mut raw_resp = String :: new ( ) ; let resp = match self . buf_reader . try_lock ( ) { Ok ( mut reader ) => { trace ! ( " " until_message ) ; if let Some ( until_message ) = until_message { // If we are trying to start a reader thread but the corresponding sender is // missing from the map, exit immediately. This can happen with batch calls, // since the sender is shared for all the individual queries in a call. We // might have already received a response for that id, but we don't know it // yet. Exiting here forces the calling code to fallback to the sender-receiver // method, and it should find a message there waiting for it. if self . waiting_map . lock ( ) ? . get ( & until_message ) . is_none ( ) { return Err ( Error :: CouldntLockReader ) ; } } // Loop over every message loop { raw_resp . clear ( ) ; if let Err ( e ) = reader . read_line ( & mut raw_resp ) { let error = Arc :: new ( e ) ; for ( _ , s ) in self . waiting_map . lock ( ) . unwrap ( ) . drain ( ) { s . send ( ChannelMessage :: Error ( error . clone ( ) ) ) ? ; } return Err ( Error :: SharedIOError ( error ) ) ; } trace ! ( " " raw_resp ) ; let resp : serde_json :: Value = serde_json :: from_str ( & raw_resp ) ? ; // Normally there is and id, but it's missing for spontaneous notifications // from the server let resp_id = resp [ " " ] . as_str ( ) . and_then ( | s | s . parse ( ) . ok ( ) ) . or_else ( | | resp [ " " ] . as_u64 ( ) . map ( | i | i as usize ) ) ; match resp_id { Some ( resp_id ) if until_message == Some ( resp_id ) => { // We have a valid id and it's exactly the one we were waiting for! trace ! ( " " resp_id ) ; // Remove ourselves from the "waiting map" let mut map = self . waiting_map . lock ( ) ? ; map . remove ( & resp_id ) ; // If the map is not empty, we select a random thread to become the // new reader thread. if let Some ( err ) = map . values ( ) . find_map ( | sender | { sender . send ( ChannelMessage :: WakeUp ) . map_err ( | err | { warn ! ( " " ) ; err } ) . err ( ) } ) { error ! ( " " ) ; return Err ( err ) ? ; } break Ok ( resp ) ; } Some ( resp_id ) => { // We have an id, but it's not our response. Notify the thread and // move on trace ! ( " " resp_id ) ; if let Some ( sender ) = self . waiting_map . lock ( ) ? . remove ( & resp_id ) { sender . send ( ChannelMessage :: Response ( resp ) ) ? ; } else { warn ! ( " " resp_id ) ; } } None => { // No id, that's probably a notification. let mut resp = resp ; if let Some ( ref method ) = resp [ " " ] . take ( ) . as_str ( ) { self . handle_notification ( method , resp [ " " ] . take ( ) ) ? ; } else { warn ! ( " " resp ) ; } } } } } Err ( TryLockError :: WouldBlock ) => { // If we "WouldBlock" here it means that there's already a reader thread // running somewhere. Err ( Error :: CouldntLockReader ) } Err ( TryLockError :: Poisoned ( e ) ) => Err ( e ) ? , } ; let resp = resp ? ; if let Some ( err ) = resp . get ( " " ) { Err ( Error :: Protocol ( err . clone ( ) ) ) } else { Ok ( resp ) } }
TODO: to enable this we have to find a way to allow concurrent read and writes to the
underlying transport struct.
TODO : to enable this we have to find a way to allow concurrent read and writes to the underlying transport struct .
&self
until_message
Option<usize>
&self
until_message
Option<usize>