linemux

https://docs.rs/linemux/
Admin
A Rust library providing asynchronous, multiplexed tailing for log files with filesystem...
Tokens:18,180
Snippets:240
Trust Score:9.1
Update:4 months ago
Show doc for...
Tokens:
Raw
### Example Usage

Source: https://docs.rs/linemux/latest/src/linemux/lib.rs.html

Demonstrates how to use MuxedLines to tail multiple files asynchronously.

```rust
use linemux::MuxedLines;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut lines = MuxedLines::new()?;

    // Register some files to be tailed, whether they currently exist or not.
    lines.add_file("some/file.log").await?;
    lines.add_file("/some/other/file.log").await?;

    // Wait for `Line` event, which contains the line captured for a given
    // source path.
    while let Ok(Some(line)) = lines.next_line().await {
        println!("source: {}, line: {}", line.source().display(), line.line());
    }
    Ok(())
}

```

--------------------------------

### Event Handling Example

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

An example demonstrating how to assert the kind of an IO error within the event handling logic.

```rust
assert_eq!(io_error.kind(), io::ErrorKind::Other);
```

--------------------------------

### add_file_from_start function

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Adds a file to be watched, starting to read from the beginning. Returns the canonicalized path. If the file already exists, it will be read from the start. If it doesn't exist, it will be added to a pending list.

```rust
pub async fn add_file_from_start(&mut self, path: impl Into<PathBuf>) -> io::Result<PathBuf> {
        self._add_file(path, true).await
    }
```

--------------------------------

### Example

Source: https://docs.rs/linemux

Register some files to be tailed, whether they currently exist or not, and wait for Line events.

```Rust
use linemux::MuxedLines;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut lines = MuxedLines::new()?;

    // Register some files to be tailed, whether they currently exist or not.
    lines.add_file("some/file.log").await?;
    lines.add_file("/some/other/file.log").await?;

    // Wait for `Line` event, which contains the line captured for a given
    // source path.
    while let Ok(Some(line)) = lines.next_line().await {
        println!("source: {}, line: {}", line.source().display(), line.line());
    }
    Ok(())
}
```

--------------------------------

### Test Streaming from Start

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

This test verifies the streaming functionality of MuxedLines, ensuring that files can be added and read correctly, including handling initial content with `add_file_from_start`.

```rust
    async fn read_line(lines: &mut MuxedLines) -> Line {
        use tokio::time::timeout;
        timeout(Duration::from_millis(100), lines.next())
            .await
            .unwrap()
            .unwrap()
            .unwrap()
    }

    #[tokio::test]
    async fn test_streaming_from_start() {
        let tmp_dir = tempdir().unwrap();
        let tmp_dir_path = tmp_dir.path();

        // file starts off with "start\n"
        let file_path = tmp_dir_path.join("foo.txt");
        let mut file = File::create(&file_path)
            .await
            .expect("Failed to create file");
        file.write_all(b"start\n").await.expect("Failed to write");
        file.sync_all().await.expect("Failed to sync");

        let mut lines = MuxedLines::new().unwrap();
        lines.add_file(&file_path).await.unwrap();

        // add some extra data into the file
        file.write_all(b"foo\n").await.unwrap();
        file.sync_all().await.unwrap();

        // Now the files should be readable
        assert_eq!(lines.inner.readers.len(), 1);

        file.shutdown().await.unwrap();

        // assert that we don't read "start", since we didn't use `add_file_from_start`
        let line1 = read_line(&mut lines).await;
        assert!(line1.source().to_str().unwrap().contains("foo.txt"));
        assert_eq!(line1.line(), "foo");

        // assert that we do indeed read "start" by using `add_file_from_start`
        let mut lines = MuxedLines::new().unwrap();
        lines.add_file_from_start(&file_path).await.unwrap();
        let line1 = read_line(&mut lines).await;
        assert!(line1.source().to_str().unwrap().contains("foo.txt"));
        assert_eq!(line1.line(), "start");
    }
}
```

--------------------------------

### Next Line Implementation

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

The `next_line` method provides an asynchronous way to get the next line from the stream, waiting for new data or returning `Ok(None)` if no files were ever added, or an `Err`.

```Rust
    /// Returns the next line in the stream.
    ///
    /// Waits for the next line from the set of watched files, otherwise
    /// returns `Ok(None)` if no files were ever added, or `Err` for a given
    /// error.
    pub async fn next_line(&mut self) -> io::Result<Option<Line>> {
        use futures_util::future::poll_fn;

        poll_fn(|cx| Pin::new(&mut *self).poll_next_line(cx)).await
    }
```

--------------------------------

### Keyboard Shortcuts

Source: https://docs.rs/linemux/latest/help.html

A list of keyboard shortcuts available in Rustdoc for navigation and interaction.

```text
`?`
    Show this help dialog

`S` / `/`
    Focus the search field

`↑`
    Move up in search results

`↓`
    Move down in search results

`←` / `→`
    Switch result tab (when results focused)

`⏎`
    Go to active search result

`+` / `=`
    Expand all sections

`-`
    Collapse all sections

`_`
    Collapse all sections, including impl blocks
```

--------------------------------

### Search Tricks

Source: https://docs.rs/linemux/latest/help.html

Tips for using the search functionality in Rustdoc, including type prefixes and exact name matching.

```text
Prefix searches with a type followed by a colon (e.g., `fn:`) to restrict the search to a given item kind.
Accepted kinds are: `fn`, `mod`, `struct`, `enum`, `trait`, `type`, `macro`, and `constant`.
Search functions by type signature (e.g., `vec -> usize` or `-> vec` or `String, enum:Cow -> bool`)
You can look for items with an exact name by putting double quotes around your request: `"string"`
Look for functions that accept or return slices and arrays by writing square brackets (e.g., `-> [u8]` or `[] -> Option`)
Look for items inside another one by searching for a path: `vec::Vec`
```

--------------------------------

### Event Handling Logic

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

This snippet shows the logic for handling file system events, including checking if a path exists, managing pending and watched files, and determining the event kind.

```rust
            // TODO: could be more intelligent/performant by checking event types
            if path_exists && self.pending_watched_files.contains(path) {
                let parent = path.parent().expect("Pending watched file needs a parent");
                let _ = self.remove_directory(parent);
                self.pending_watched_files.remove(path);
                let _ = self._add_file(path, false);
            }

            if !path_exists && self.watched_files.contains(path) {
                self.watched_files.remove(path);
                let _ = self._add_file(path, false);
            }

            if event_kind.is_remove() {
                self.pending_watched_files.contains(path)
            } else {
                self.watched_files.contains(path)
            }
        });
    }

    fn __poll_next_event(
        mut event_stream: Pin<&mut EventStream>,
        cx: &mut task::Context<'_>,
    ) -> task::Poll<Option<io::Result<notify::Event>>> {
        task::Poll::Ready(
            ready!(event_stream.poll_recv(cx)).map(|res| res.map_err(notify_to_io_error)),
        )
    }

    #[doc(hidden)]
    pub fn poll_next_event(
        mut self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
    ) -> task::Poll<io::Result<Option<notify::Event>>> {
        if self.is_empty() {
            return task::Poll::Ready(Ok(None));
        }

        let mut res = ready!(Self::__poll_next_event(
            Pin::new(&mut self.event_stream),
            cx
        ));

        if let Some(Ok(ref mut event)) = res {
            self.handle_event(event);
        }

        task::Poll::Ready(res.transpose())
    }

    /// Returns the next event in the stream.
    ///
    /// Waits for the next event from the set of watched files, otherwise
    /// returns `Ok(None)` if no files were ever added, or `Err` for a given
    /// error.
    pub async fn next_event(&mut self) -> io::Result<Option<notify::Event>> {
        use futures_util::future::poll_fn;

        poll_fn(|cx| Pin::new(&mut *self).poll_next_event(cx)).await
    }
}

impl Stream for MuxedEvents {
    type Item = io::Result<notify::Event>;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
    ) -> task::Poll<Option<Self::Item>> {
        self.poll_next_event(cx).map(Result::transpose)
    }
}

```

--------------------------------

### Test: Add Directory

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

Tests the `add_file` method when attempting to add a directory, which is expected to return an error.

```rust
    #[tokio::test]
    async fn test_add_directory() {
        let tmp_dir = tempdir().unwrap();
        let tmp_dir_path = tmp_dir.path();

        let mut watcher = MuxedEvents::new().unwrap();
        assert!(watcher.add_file(&tmp_dir_path).await.is_err());
    }

```

--------------------------------

### _add_file private implementation

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Internal implementation for adding files, handling both existing and non-existing files, and managing reader positions.

```rust
async fn _add_file(
        &mut self,
        path: impl Into<PathBuf>,
        from_start: bool,
    ) -> io::Result<PathBuf> {
        let source = path.into();
        let source = if from_start {
            self.events.add_file_initial_event(&source).await?
        } else {
            self.events.add_file(&source).await?
        };

        if self.reader_exists(&source) {
            return Ok(source);
        }

        if !source.exists() {
            let didnt_exist = self.inner.insert_pending(source.clone());

            // If this fails it's a bug
            assert!(didnt_exist);
        } else {
            let size = if from_start {
                0
            } else {
                metadata(&source).await?.len()
            };

            let reader = new_linereader(&source, Some(size)).await?;

            let inner_mut = &mut self.inner;
            inner_mut.insert_reader_position(source.clone(), size);
            let last = inner_mut.insert_reader(source.clone(), reader);

            // If this fails it's a bug
            assert!(last.is_none());
        }
        // TODO: prob need 'pending' for non-existent files like Events

        Ok(source)
    }
```

--------------------------------

### Test File Rollover

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Demonstrates how the MuxedLines reader handles file rollovers, including writing new content after truncation and ensuring the new content is read correctly.

```rust
    #[tokio::test]
    async fn test_file_rollover() {
        use std::io::SeekFrom;
        use tokio::time::timeout;

        let mut _file1 = File::create("missing_file1.txt").await.unwrap();
        _file1.write_all(b"baz\n").await.unwrap();
        _file1.sync_all().await.unwrap();
        tokio::time::sleep(Duration::from_millis(100)).await;

        let mut lines = MuxedLines::new().unwrap();
        lines.add_file(&Path::new("missing_file1.txt")).await.unwrap();

        { // Read the first line
            let line1 = timeout(Duration::from_millis(100), lines.next())
                .await
                .unwrap()
                .unwrap()
                .unwrap();
            assert!(line1
                .source()
                .to_str()
                .unwrap()
                .contains("missing_file1.txt"));
            assert_eq!(line1.line(), "baz");
        }

        // Reset cursor
        _file1.seek(SeekFrom::Start(0)).await.unwrap();
        let _ = timeout(Duration::from_millis(100), lines.next()).await;

        // Roll over
        _file1.set_len(0).await.unwrap();

        // TODO: Can we still catch roll without flushing?
        let _ = timeout(Duration::from_millis(100), lines.next()).await;

        _file1.write_all(b"qux\n").await.unwrap();
        _file1.sync_all().await.unwrap();
        _file1.shutdown().await.unwrap();
        drop(_file1);
        tokio::time::sleep(Duration::from_millis(100)).await;
        {
            let line1 = timeout(Duration::from_millis(100), lines.next())
                .await
                .unwrap()
                .unwrap()
                .unwrap();
            assert!(line1
                .source()
                .to_str()
                .unwrap()
                .contains("missing_file1.txt"));
            assert_eq!(line1.line(), "qux");
        }
    }

```

--------------------------------

### Adding a File to Watch

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

The `add_file` function adds a file to the event watch. It canonicalizes the path, checks if it's a directory (which is not allowed), and then either adds it to pending files if it doesn't exist or directly watches it if it does. It also handles sending an initial event if requested.

```rust
fn _add_file(&mut self, path: impl Into<PathBuf>, initial_event: bool) -> io::Result<PathBuf> {
        let path = absolutify(path, true)?;

        // TODO: non-existent file that later gets created as a dir?
        if path.is_dir() {
            // on Linux this would be `EISDIR` (21) and maybe
            // `ERROR_DIRECTORY_NOT_SUPPORTED` (336) for windows?
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Is a directory",
            ));
        }

        // Make sure we aren't already watching the directory
        if self.watch_exists(&path) {
            return Ok(path);
        }

        if !path.exists() {
            let parent = path.parent().ok_or_else(|| {
                io::Error::new(io::ErrorKind::InvalidInput, "File needs a parent directory")
            })?;

            self.add_directory(parent)?;
            self.pending_watched_files.insert(path.clone());
        } else {
            Self::watch(self.inner.as_mut(), &path)?;

            self.watched_files.insert(path.clone());

            if initial_event {
                // Send an initial event for this file when requested.
                // This is useful if we wanted earlier lines in the file than
                // where it is up to now, and we want those events before the
                // next time this file is modified.
                self.event_stream_sender
                    .send(Ok(notify::Event {
                        attrs: notify::event::EventAttributes::new(),
                        kind: notify::EventKind::Create(notify::event::CreateKind::File),
                        paths: vec![path.clone()],
                    }))
                    .ok();
                // Errors here are not anything to worry about, so we .ok();
                // An error would just mean no one is listening.
            }
        }

        Ok(path)
    }
```

--------------------------------

### MuxedEvents::new constructor

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

Constructs a new MuxedEvents instance, setting up an unbounded channel for event communication and initializing the underlying notify::RecommendedWatcher.

```Rust
pub fn new() -> io::Result<Self> {
    let (tx, rx) = mpsc::unbounded_channel();
    let sender = tx.clone();

    let inner: notify::RecommendedWatcher = notify::RecommendedWatcher::new(
        move |res| {
            // The only way `send` can fail is if the receiver is dropped,
            // and `MuxedEvents` controls both. `unwrap` is not used,
            // however, since `Drop` idiosyncrasies could otherwise result
            // in a panic.
            let _ = tx.send(res);
        },
        notify::Config::default(),
    )
    .map_err(notify_to_io_error)?;

    Ok(MuxedEvents {
        inner: Box::new(inner),
        watched_directories: HashMap::new(),
        watched_files: HashSet::new(),
        pending_watched_files: HashSet::new(),
        event_stream: rx,
        event_stream_sender: sender,
    })
}
```

--------------------------------

### Test for adding missing files

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

This test verifies that the `add_file` method correctly handles cases where files do not exist initially, and then proceeds to read from them once they are created. It also checks that re-registering the same path is handled gracefully.

```Rust
    #[tokio::test]
    async fn test_add_missing_files() {
        use tokio::time::timeout;

        let tmp_dir = tempdir().unwrap();
        let tmp_dir_path = tmp_dir.path();

        let file_path1 = tmp_dir_path.join("missing_file1.txt");
        let file_path2 = tmp_dir_path.join("missing_file2.txt");

        let mut lines = MuxedLines::new().unwrap();
        lines.add_file(&file_path1).await.unwrap();
        lines.add_file(&file_path2).await.unwrap();

        // Registering the same path again should be fine
        lines.add_file(&file_path2).await.unwrap();

        assert_eq!(lines.inner.pending_readers.len(), 2);

        let mut _file1 = File::create(&file_path1)
            .await
            .expect("Failed to create file");

        if cfg!(target_os = "macos") {
            // XXX: OSX sometimes fails `readers.len() == 2` if no delay in between file creates.
            tokio::time::sleep(Duration::from_millis(100)).await;
        }

        let mut _file2 = File::create(&file_path2)
            .await
            .expect("Failed to create file");

        assert!(
            timeout(Duration::from_millis(100), lines.next())
                .await
                .is_err(),
            "Should not be any lines yet",
        );

        // Now the files should be readable
        assert_eq!(lines.inner.readers.len(), 2);

        _file1.write_all(b"foo\n").await.unwrap();
        _file1.sync_all().await.unwrap();
        _file1.shutdown().await.unwrap();
        drop(_file1);
        tokio::time::sleep(Duration::from_millis(100)).await;
        let line1 = timeout(Duration::from_millis(100), lines.next())
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        assert!(line1
            .source()
            .to_str()
            .unwrap()
            .contains("missing_file1.txt"));
        assert_eq!(line1.line(), "foo");

        _file2.write_all(b"bar\nbaz\n").await.unwrap();
        _file2.sync_all().await.unwrap();
        _file2.shutdown().await.unwrap();
        drop(_file2);
        tokio::time::sleep(Duration::from_millis(100)).await;
        {
            let line2 = timeout(Duration::from_millis(100), lines.next())
                .await
                .unwrap()
                .unwrap()
                .unwrap();
            assert!(line2
                .source()
                .to_str()
                .unwrap()
                .contains("missing_file2.txt"));
            assert_eq!(line2.line(), "bar");
        }
        {
            let line2 = timeout(Duration::from_millis(100), lines.next_line())
                .await
                .unwrap()
                .unwrap()
                .unwrap();
            assert!(line2
                .source()
                .to_str()
                .unwrap()
                .contains("missing_file2.txt"));
            assert_eq!(line2.line(), "baz");
        }

        drop(lines);
    }

```

--------------------------------

### add_file function

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Adds a file to be watched, returning its canonicalized path. If the file already exists, it continues reading from its current position. If it doesn't exist, it will be added to a pending list.

```rust
pub async fn add_file(&mut self, path: impl Into<PathBuf>) -> io::Result<PathBuf> {
        self._add_file(path, false).await
    }
```

--------------------------------

### Handling Incoming Events

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

This snippet outlines the `handle_event` function, which processes incoming `notify::Event` objects. It iterates through the paths associated with the event and applies filtering logic based on the event kind, such as ignoring file removal events or specific modification types.

```rust
fn handle_event(&mut self, event: &mut notify::Event) {
        let paths = &mut event.paths;
        let event_kind = &event.kind;

        // TODO: properly handle any errors encountered adding/removing stuff
        paths.retain(|path| {
            // Fixes a potential race when detecting file rotations.
            let path_exists =
                if let notify::EventKind::Remove(notify::event::RemoveKind::File) = &event_kind {
                    false
                } else if let notify::EventKind::Modify(notify::event::ModifyKind::Name(
                    notify::event::RenameMode::From,
                )) = &event_kind
                {

```

--------------------------------

### Absolutify Function

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

This function takes a path and a boolean indicating if it's a file, and returns an absolute path. It handles relative paths, symbolic links, and canonicalizes the path.

```rust
// TODO: maybe use with crate `path-absolutize`
fn absolutify(path: impl Into<PathBuf>, is_file: bool) -> io::Result<PathBuf> {
    let path = path.into();

    let (dir, maybe_filename) = if is_file {
        let parent = match path.parent() {
            None => std::env::current_dir()?,
            Some(path) => {
                if path == Path::new("") {
                    std::env::current_dir()?
                } else {
                    path.to_path_buf()
                }
            }
        };
        let filename = path
            .file_name()
            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "Filename not found in path"))?
            .to_os_string();

        (parent, Some(filename))
    } else {
        (path, None)
    };

    let dir = if let Ok(linked_dir) = dir.read_link() {
        linked_dir
    } else {
        dir
    };

    let dir = if let Ok(abs_dir) = dir.canonicalize() {
        abs_dir
    } else {
        dir
    };

    let path = if let Some(filename) = maybe_filename {
        dir.join(filename)
    } else {
        dir
    };

    Ok(path)
}

```

--------------------------------

### Test Empty Next Line

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Verifies that calling `next_line()` or `next()` on a MuxedLines instance with no files added returns `None`.

```rust
    #[tokio::test]
    async fn test_empty_next_line() {
        let mut watcher = MuxedLines::new().unwrap();

        // No files added, expect None
        assert!(watcher.next_line().await.unwrap().is_none());
        assert!(watcher.next().await.is_none());
    }

```

--------------------------------

### Test: Add Bad Filename

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

Tests the `add_file` method with a potentially problematic filename, specifically joining with '..'.

```rust
    #[tokio::test]
    async fn test_add_bad_filename() {
        let tmp_dir = tempdir().unwrap();
        let tmp_dir_path = tmp_dir.path();

        let mut watcher = MuxedEvents::new().unwrap();

        // This is not okay
        let file_path1 = tmp_dir_path.join("..");

```

--------------------------------

### Handling New Line Reader Creation

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

This snippet shows the creation of a new line reader future when a file is created or modified and rolled.

```rust
let reader_fut = Box::pin(new_linereader(path.clone(), None));

Some(HandleEventAwaitState::NewLineReader(reader_fut))
```

--------------------------------

### Line Struct

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Represents a line captured for a given source path, providing context.

```Rust
/// Line captured for a given source file.
///
/// Also provides the caller extra context, such as the source file.
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
pub struct Line {
    /// The path from where the line was read.
    source: PathBuf,
    /// The received line.
    line: String,
}

impl Line {
    /// Returns a reference to the path from where the line was read.
    pub fn source(&self) -> &Path {
        self.source.as_path()
    }

    /// Returns a reference to the line.
    pub fn line(&self) -> &str {
        self.line.as_str()
    }

    /// Returns the internal components that make up a `Line`. Hidden as the
    /// return signature may change.
    #[doc(hidden)]
    pub fn into_inner(self) -> (PathBuf, String) {
        let Line { source, line } = self;

        (source, line)
    }
}
```

--------------------------------

### Test Empty Next Event

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

This test verifies that calling `next_event()` or `next()` on a MuxedEvents watcher with no files added returns `None`.

```rust
    #[tokio::test]
    async fn test_empty_next_event() {
        let mut watcher = MuxedEvents::new().unwrap();

        // No files added, expect None
        assert!(watcher.next_event().await.unwrap().is_none());
        assert!(watcher.next().await.is_none());
    }

```

--------------------------------

### MuxedEvents::add_directory method

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

Adds a directory to be watched. It handles cases where the directory might already be watched to avoid duplicate events, and increments the watch count for the directory.

```Rust
fn add_directory(&mut self, path: impl AsRef<Path>) -> io::Result<()> {
    let path_ref = path.as_ref();

    // `watch` behavior is platform-specific, and on some (windows) can produce
    // duplicate events if called multiple times.
    if !self.watch_exists(path_ref) {
        NotifyWatcher::watch(
            self.inner.as_mut(),
            path_ref,
            notify::RecursiveMode::NonRecursive,
        )
        .map_err(notify_to_io_error)?;
    }

    let count = self
        .watched_directories
        .entry(path_ref.to_owned())
        .or_insert(0);
    *count += 1;

    Ok(())
}
```

--------------------------------

### Test Add Existing File

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Tests adding a file that already exists to MuxedLines, including handling potential timing issues on macOS and verifying file rename events.

```rust
    #[tokio::test]
    async fn test_add_existing_file() {
        use tokio::time::timeout;

        let tmp_dir = tempdir().unwrap();
        let tmp_dir_path = tmp_dir.path();

        let file_path1 = tmp_dir_path.join("foo.txt");
        let file_path2 = tmp_dir_path.join("bar.txt");

        let mut lines = MuxedLines::new().unwrap();

        lines.add_file(&file_path2).await.unwrap();

        assert_eq!(lines.inner.pending_readers.len(), 1);

        let mut _file1 = File::create(&file_path1)
            .await
            .expect("Failed to create file");

        if cfg!(target_os = "macos") {
            // XXX: OSX sometimes fails `readers.len() == 2` if no delay in between file creates.
            tokio::time::sleep(Duration::from_millis(100)).await;
        }

        _file1.write_all(b"foo\n").await.unwrap();
        _file1.sync_all().await.unwrap();
        tokio::time::sleep(Duration::from_millis(100)).await;

        tokio::fs::rename(&file_path1, &file_path2).await.unwrap();

        // Spin to handle the rename event
        let res = timeout(Duration::from_millis(100), lines.next_line()).await;
        if !cfg!(target_os = "macos") {
            assert!(res.is_err(), "res: {:?}", res);
        } else {
            // TODO: osx/kqueue is picking up the line written to __file1
        }

        // Now the files should be readable
    }

```

--------------------------------

### New Line Reader

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Creates a new line reader for a given path, optionally seeking to a specific position.

```Rust
async fn new_linereader(path: impl AsRef<Path>, seek_pos: Option<u64>) -> io::Result<LineReader> {
    let path = path.as_ref();
    let mut reader = File::open(path).await?;
    if let Some(pos) = seek_pos {
        reader.seek(io::SeekFrom::Start(pos)).await?;
    }
    let reader = BufReader::new(reader).lines();

    Ok(reader)
}
```

--------------------------------

### HandleEventState and HandleEventAwaitState

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Structures to manage the state during event handling, including asynchronous operations like fetching metadata or creating new line readers.

```Rust
type MetadataFuture = Pin<Box<dyn Future<Output = io::Result<std::fs::Metadata>> + Send + Sync>>;
type NewLineReaderFuture = Pin<Box<dyn Future<Output = io::Result<LineReader>> + Send + Sync>>;

struct HandleEventState {
    path_index: usize,
    await_state: HandleEventAwaitState,
}

impl HandleEventState {
    pub fn new() -> Self {
        HandleEventState {
            path_index: 0,
            await_state: Default::default(),
        }
    }
}

enum HandleEventAwaitState {
    Idle,
    Metadata(MetadataFuture),
    NewLineReader(NewLineReaderFuture),
}

impl Default for HandleEventAwaitState {
    fn default() -> Self {
        HandleEventAwaitState::Idle
    }
}

impl HandleEventAwaitState {
    pub fn replace(&mut self, new_state: Self) -> HandleEventAwaitState {
        let mut old_state = new_state;

        std::mem::swap(self, &mut old_state);

        old_state
    }
}
```

--------------------------------

### Handling Directory Watch Count

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

This snippet shows how the linemux library manages the count of watchers for a given directory. It decrements the count and removes the watch if the count reaches zero, or simply decrements the count otherwise.

```rust
if let Some(count) = self.watched_directories.get(path_ref).copied() {
            match count {
                0 => unreachable!(), // watch is removed if count == 1
                1 => {
                    // Remove from map first in case `unwatch` fails.
                    self.watched_directories.remove(path_ref);
                    Self::unwatch(self.inner.as_mut(), path_ref)?;
                }
                _ => {
                    let new_count = self
                        .watched_directories
                        .get_mut(path_ref)
                        .expect("path was not present but count > 1");
                    *new_count -= 1;
                }
            }
        }
```

--------------------------------

### Handling File Modification - Idle State

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

This snippet shows the initial handling of a file modification event when the state is idle, preparing to fetch metadata.

```rust
// Windows returns `Any` for file modification, so handle that
match (
    cfg!(target_os = "windows"),
    cfg!(target_os = "macos"),
    modify_event,
) {
    // This showed up while debugging kqueue, but unit tests passed without it
    // (_, true, notify::event::ModifyKind::Data(notify::event::DataChange::Size)) => {}
    (_, _, notify::event::ModifyKind::Data(_)) => {}
    (
        _,
        _,
        notify::event::ModifyKind::Name(notify::event::RenameMode::To),
    ) => {}
    (
        _,
        true,
        notify::event::ModifyKind::Name(notify::event::RenameMode::From),
    ) => {}
    (true, _, notify::event::ModifyKind::Any) => {}
    (_, _, _) => {
        state.path_index += 1;
        continue;
    }
}

let path = event.paths.get(state.path_index).expect("Got None Path");
let metadata_fut = Box::pin(metadata(path.clone()));
Some(HandleEventAwaitState::Metadata(metadata_fut))
```

--------------------------------

### Test add_directory failure

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Tests that attempting to add a directory to `MuxedLines` results in an error, as expected.

```rust
#[tokio::test]
async fn test_add_directory() {
    let tmp_dir = tempdir().unwrap();
    let tmp_dir_path = tmp_dir.path();

    let mut lines = MuxedLines::new().unwrap();
    assert!(lines.add_file(&tmp_dir_path).await.is_err());
}
```

--------------------------------

### Checking if Watcher is Empty

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

A simple function to determine if the linemux watcher is currently monitoring any files, either actively or pending.

```rust
fn is_empty(&self) -> bool {
        self.watched_files.is_empty() && self.pending_watched_files.is_empty()
    }
```

--------------------------------

### Test Add Missing Files

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

This test function verifies the behavior of MuxedEvents when adding and managing files that may not initially exist, including handling creation and deletion events.

```rust
    #[tokio::test]
    async fn test_add_missing_files() {
        use tokio::io::AsyncWriteExt;

        let tmp_dir = tempdir().unwrap();
        let tmp_dir_path = tmp_dir.path();
        let pathclone = absolutify(tmp_dir_path, false).unwrap();

        let file_path1 = tmp_dir_path.join("missing_file1.txt");
        let file_path2 = tmp_dir_path.join("missing_file2.txt");

        let mut watcher = MuxedEvents::new().unwrap();
        let _ = format!("{:?}", watcher);
        watcher.add_file(&file_path1).await.unwrap();
        watcher.add_file(&file_path2).await.unwrap();

        // Registering the same path again should be fine
        watcher.add_file(&file_path2).await.unwrap();

        assert_eq!(watcher.pending_watched_files.len(), 2);
        assert!(watcher.watched_directories.contains_key(&pathclone));

        // Flush possible directory creation event
        let _res = timeout(Duration::from_secs(1), watcher.next()).await;

        let expected_event = if cfg!(target_os = "windows") {
            notify::EventKind::Create(notify::event::CreateKind::Any)
        } else {
            notify::EventKind::Create(notify::event::CreateKind::File)
        };

        let mut _file1 = File::create(&file_path1)
            .await
            .expect("Failed to create file");
        let event1 = timeout(Duration::from_secs(1), watcher.next())
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        assert_eq!(event1.kind, expected_event,);

        let _file2 = File::create(&file_path2)
            .await
            .expect("Failed to create file");
        let event2 = timeout(Duration::from_secs(1), watcher.next())
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        assert_eq!(event2.kind, expected_event,);

        // Now the files should be watched properly
        assert_eq!(watcher.watched_files.len(), 2, "\nwatcher: {:?}", &watcher);
        assert!(
            !watcher.watched_directories.contains_key(&pathclone),
            "\nwatcher: {:?}",
            &watcher
        );

        // Explicitly close file to allow deletion event to propagate
        _file1.sync_all().await.unwrap();
        _file1.shutdown().await.unwrap();
        drop(_file1);
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Deleting a file should throw it back into pending
        tokio::fs::remove_file(&file_path1).await.unwrap();

        // Flush possible file deletion event
        let expected_event = {
            let remove_kind = if cfg!(target_os = "windows") || cfg!(target_os = "macos") {
                notify::event::RemoveKind::Any
            } else {
                notify::event::RemoveKind::File
            };
            notify::Event::new(notify::EventKind::Remove(remove_kind))
                .add_path(absolutify(file_path1, true).unwrap())
        };

        let mut events = vec![];
        tokio::time::timeout(tokio::time::Duration::from_millis(2000), async {
            loop {
                let event = watcher.next_event().await.unwrap().unwrap();
                if event == expected_event {
                    break;
                }
                events.push(event);
            }
        })
        .await
        .unwrap_or_else(|_| {
            panic!(
                "Did not receive expected event, events received: {:?}",
                events
            )
        });

        assert_eq!(watcher.watched_files.len(), 1, "\nwatcher: {:?}", &watcher);
        assert!(
            watcher.watched_directories.contains_key(&pathclone),
            "\nwatcher: {:?}",
            &watcher
        );

        drop(watcher);
    }

```

--------------------------------

### Test Line struct methods

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Tests the functionality of the `Line` struct, including accessing its source path, line content, and consuming it into its inner components.

```rust
#[test]
fn test_line_fns() {
    let source_path = "/some/path";
    let line_expected = "foo".to_string();

    let line = Line {
        source: PathBuf::from(&source_path),
        line: line_expected.clone(),
    };

    assert_eq!(line.source().to_str().unwrap(), source_path);

    let line_ref = line.line();
    assert_eq!(line_ref, line_expected.as_str());

    let (source_de, lines_de) = line.into_inner();
    assert_eq!(source_de, PathBuf::from(source_path));
    assert_eq!(lines_de, line_expected);
}
```

--------------------------------

### HandleEventAwaitState::NewLineReader

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

This snippet shows how the `HandleEventAwaitState::NewLineReader` state handles the result of a reader future, extracts a path, inserts the new reader into the inner state, and transitions to the `Idle` state.

```rust
HandleEventAwaitState::NewLineReader(ref mut reader_fut) => {
    let reader_res = ready!(reader_fut.as_mut().poll(cx));
    if let Ok(reader) = reader_res {
        let path = event.paths.get(state.path_index).expect("Got None Path");

        // Don't really care about old values, we got create
        let _previous_reader = inner.insert_reader(path.clone(), reader);
    }
    state.path_index += 1;
    Some(HandleEventAwaitState::Idle)
}
```

--------------------------------

### Inner Struct

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Internal state for managing readers and their positions.

```Rust
#[derive(Debug)]
struct Inner {
    reader_positions: HashMap<PathBuf, u64>,
    readers: HashMap<PathBuf, LineReader>,
    pending_readers: HashSet<PathBuf>,
}

impl Inner {
    pub fn new() -> Self {
        Inner {
            reader_positions: HashMap::new(),
            readers: HashMap::new(),
            pending_readers: HashSet::new(),
        }
    }

    pub fn reader_exists(&self, path: &Path) -> bool {
        // Make sure there isn't already a reader for the file
        self.readers.contains_key(path) || self.pending_readers.contains(path)
    }

    pub fn insert_pending(&mut self, path: PathBuf) -> bool {
        self.pending_readers.insert(path)
    }

    pub fn remove_pending(&mut self, path: &Path) -> bool {
        self.pending_readers.remove(path)
    }

    pub fn insert_reader(&mut self, path: PathBuf, reader: LineReader) -> Option<LineReader> {
        self.readers.insert(path, reader)
    }

    pub fn insert_reader_position(&mut self, path: PathBuf, pos: u64) -> Option<u64> {
        self.reader_positions.insert(path, pos)
    }

    pub fn is_empty(&self) -> bool {
        self.readers.is_empty() && self.pending_readers.is_empty()
    }
}
```

--------------------------------

### Line struct definition

Source: https://docs.rs/linemux/latest/linemux/struct.Line.html

Defines the Line struct with private fields.

```rust
pub struct Line { /* private fields */ }
```

--------------------------------

### Test Inner struct methods

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Tests various methods of the `Inner` struct, including checking for reader existence, inserting pending files, inserting readers, and removing pending files. It also tests inserting readers with specific positions.

```rust
#[tokio::test]
async fn test_inner_fns() {
    let dir = tempdir().unwrap();
    let source_path = dir.path().join("foo.txt");

    let mut inner = Inner::new();

    assert!(!inner.reader_exists(&source_path));
    assert!(inner.insert_pending(source_path.clone()));
    assert!(inner.reader_exists(&source_path));
    assert!(!inner.insert_pending(source_path.clone()));

    {
        let mut f = File::create(&source_path).await.unwrap();
        f.write_all(b"Hello, world!\nasdf\n").await.unwrap();
        f.sync_all().await.unwrap();
        f.shutdown().await.unwrap();
    }

    let linereader = new_linereader(&source_path, None).await.unwrap();
    assert!(inner
        .insert_reader(source_path.clone(), linereader)
        .is_none());
    assert!(inner
        .insert_reader_position(source_path.clone(), 0)
        .is_none());
    assert!(inner.remove_pending(&source_path));

    let linereader = new_linereader(&source_path, Some(3)).await.unwrap();
    assert!(inner
        .insert_reader(source_path.clone(), linereader)
        .is_some());
    assert_eq!(
        inner.insert_reader_position(source_path.clone(), 3),
        Some(0)
    );
}
```

--------------------------------

### Test for file rollover

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

This test checks the behavior of `MuxedLines` when a file is written to, read from, and then written to again, simulating a file rollover scenario. It ensures that lines are correctly processed even after the file has been written to multiple times.

```Rust
    #[tokio::test]
    async fn test_file_rollover() {
        use tokio::time::timeout;

        let tmp_dir = tempdir().unwrap();
        let tmp_dir_path = tmp_dir.path();

        let file_path1 = tmp_dir_path.join("missing_file1.txt");

        let mut lines = MuxedLines::new().unwrap();
        lines.add_file(&file_path1).await.unwrap();
        assert!(!lines.is_empty());

        let mut _file1 = File::create(&file_path1)
            .await
            .expect("Failed to create file");
        tokio::time::sleep(Duration::from_millis(100)).await;
        _file1.write_all(b"bar\nbaz\n").await.unwrap();
        _file1.sync_all().await.unwrap();
        tokio::time::sleep(Duration::from_millis(100)).await;
        {
            let line1 = timeout(Duration::from_millis(100), lines.next_line())
                .await
                .unwrap()
                .unwrap()
                .unwrap();
            assert!(line1
                .source()
                .to_str()
                .unwrap()
                .contains("missing_file1.txt"));
            assert_eq!(line1.line(), "bar");
        }
        {
            let line1 = timeout(Duration::from_millis(100), lines.next())
                .await
                .unwrap()
                .unwrap()
                .unwrap();
            assert!(line1
                .source()
                .to_str()
                .unwrap()
                .contains("missing_file1.txt"));
            assert_eq!(line1.line(), "baz");
        }
    }

```

--------------------------------

### Test Notify Error Conversion

Source: https://docs.rs/linemux/latest/src/linemux/events.rs.html

This test demonstrates the conversion of `notify::Error` types to `std::io::Error` using the `notify_to_io_error` function.

```rust
    #[test]
    fn test_notify_error() {
        use std::io;

        let notify_io_error = notify::Error::io(io::Error::new(io::ErrorKind::AddrInUse, "foobar"));
        let io_error = notify_to_io_error(notify_io_error);
        assert_eq!(io_error.kind(), io::ErrorKind::AddrInUse);

        let notify_custom_error = notify::Error::path_not_found();
        let io_error = notify_to_io_error(notify_custom_error);
    }

```

--------------------------------

### Test Operations in Transient State

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

Tests the behavior of MuxedLines when files are added while the reader is in a transient state, ensuring pending readers are handled correctly.

```rust
    #[tokio::test]
    async fn test_ops_in_transient_state() {
        use futures_util::future::poll_fn;
        use futures_util::stream::Stream;
        use tokio::time::timeout;

        let tmp_dir = tempdir().unwrap();
        let tmp_dir_path = tmp_dir.path();

        let file_path1 = tmp_dir_path.join("missing_file1.txt");

        let mut lines = MuxedLines::new().unwrap();
        lines.add_file(&file_path1).await.unwrap();

        let mut _file1 = File::create(&file_path1)
            .await
            .expect("Failed to create file");
        _file1.write_all(b"bar\n").await.unwrap();
        _file1.sync_all().await.unwrap();
        tokio::time::sleep(Duration::from_millis(100)).await;

        let maybe_pending =
            poll_fn(|cx| task::Poll::Ready(Pin::new(&mut lines).poll_next(cx))).await;
        assert!(maybe_pending.is_pending());

        // TODO: Deterministic state checking?
        //let maybe_pending = poll_fn(|cx| task::Poll::Ready(Pin::new(&mut lines).poll_next(cx))).await;
        //assert!(maybe_pending.is_pending());

        let file_path2 = tmp_dir_path.join("missing_file2.txt");
        lines.add_file(&file_path2).await.unwrap();

        // TODO: Find a way to guarantee this
        //assert_eq!(lines.inner.readers.len(), 1);

        // This should be guaranteed
        assert_eq!(lines.inner.pending_readers.len(), 1);

        {
            let line1 = timeout(Duration::from_millis(100), lines.next())
                .await
                .unwrap()
                .unwrap()
                .unwrap();
            assert!(line1
                .source()
                .to_str()
                .unwrap()
                .contains("missing_file1.txt"));
            assert_eq!(line1.line(), "bar");
        }
    }

```

--------------------------------

### Test add_bad_filename

Source: https://docs.rs/linemux/latest/src/linemux/reader.rs.html

This test is intended to check the behavior when adding a file with an invalid or problematic filename, though the specific assertion is cut off.

```rust
#[tokio::test]
async fn test_add_bad_filename() {
    let tmp_dir = tempdir().unwrap();
    let tmp_dir_path = tmp_dir.path();

    let mut lines = MuxedLines::new().unwrap();

    // This is not okay

```
Tokens:
Raw