add unit test to expose RealtimeLines not timing out — radicle-ci-broker

Signed-off-by: Lars Wirzenius liw@liw.fi

fix: make RealtimeLines::line honor the max duration setting

Signed-off-by: Lars Wirzenius liw@liw.fi

         Self {
             max_duration,
             stdin_data: vec![],
             stdout: RealtimeLines::default(),
             stdout: RealtimeLines::new(max_duration),
         }
     }

 /// The buffer can be cloned and each clone is logically the same
 /// buffer. This allows separate producer and consumer threads to own
 /// the buffer, but internally it's the same one.
 #[derive(Default, Clone)]
 #[derive(Clone)]
 pub struct RealtimeLines {
     // We have an unlocked buffer protected by a mutex, and a
     // condition variable to signal consumers of new data or the
     // producer having finished.
     data: Arc<(Mutex<UnlockedBuf>, Condvar)>,
     started: Instant,
     max_duration: Duration,
 }
 impl RealtimeLines {
     fn new(max_duration: Duration) -> Self {
         Self {
             data: Arc::new((Mutex::new(UnlockedBuf::default()), Condvar::default())),
             started: Instant::now(),
             max_duration,
         }
     }
     /// Push some binary data to the buffer.
     ///
     /// The incoming data is a byte vector, as we may not receive

         loop {
             match buf.line() {
                 // We got a line: return it.
                 Some(line) => {
                     return Some(line);
                 }
                 // We've been reading for too long.
                 _ if self.started.elapsed() > self.max_duration => return None,
                 // We didn't get a line, but the input stream has
                 // finished. Return final partial line, if there is one,

                 None => {
                     buf = var.wait(buf).expect("wait for line");
                 }
                 // We got a line: return it.
                 Some(line) => {
                     return Some(line);
                 }
             }
         }
     }

     }
     #[test]
     fn yes_to_stdout_while_reading_with_realtimelines() -> Result<(), Box<dyn std::error::Error>> {
         let (running, mut stdout) = setup("exec yes", SHORT_TIMEOUT, None)?;
         while stdout.line().is_some() {}
         assert!(matches!(running.wait(), Err(TimeoutError::TimedOut)));
         Ok(())
     }
     #[test]
     fn kill() -> Result<(), Box<dyn std::error::Error>> {
         let (running, _) = setup(
             "exec sleep 1000",