<commit_before>use std::thread;

use tokio::sync::{mpsc, oneshot};
use tokio::runtime::Handle;
use serde::{Serialize, de::DeserializeOwned};
use ipc_channel::ipc::{IpcReceiver, IpcError};

/// Asynchronous wrapper over `IpcReceiver` that yields `Result<T, IpcError>`.
///
/// The `ipc_channel::ipc::IpcReceiver` type can't be shared between threads, so it can't be used
/// from `tokio::task::spawn_blocking`. This type uses a separate thread to manage receiving
/// values asynchronously. It implements the `Stream` trait, which allows it to be polled over
/// and over again for each received value.
#[derive(Debug)]
pub struct AsyncIpcReceiver<T: Serialize + DeserializeOwned + Send + 'static> {
    /// Each AsyncIpcReceiver::recv() call sends a oneshot channel that collects the result when
    /// it is ready. This ensures proper ordering even if the future from `recv` is dropped and has
    /// the added benefit of allowing `recv()` to take `&self` instead of `&mut self`. The second
    /// benefit allows some operations that use this struct to be lock-free.
    channel_sender: mpsc::UnboundedSender<oneshot::Sender<Result<T, IpcError>>>,
}

impl<T: Serialize + DeserializeOwned + Send + 'static> AsyncIpcReceiver<T> {
    pub fn new(ipc_receiver: IpcReceiver<T>) -> Self {
        let (channel_sender, mut receiver) = mpsc::unbounded_channel();

        let handle = Handle::current();
        thread::spawn(move || {
            loop {
                // This loop continues even if this call produces an error saying that IPC was
                // disconnected. This allows callers of `recv` to detect the disconnection.
                let next_value = ipc_receiver.recv();<commit_msg>AsyncIpcReceiver does not implement Stream<commit_after>use std::thread;

use tokio::sync::{mpsc, oneshot};
use tokio::runtime::Handle;
use serde::{Serialize, de::DeserializeOwned};
use ipc_channel::ipc::{IpcReceiver, IpcError};

/// Asynchronous wrapper over `IpcReceiver` that yields `Result<T, IpcError>`.
///
/// The `ipc_channel::ipc::IpcReceiver` type can't be shared between threads, so it can't be used
/// from `tokio::task::spawn_blocking`. This type uses a separate thread to manage receiving
/// values asynchronously.
#[derive(Debug)]
pub struct AsyncIpcReceiver<T: Serialize + DeserializeOwned + Send + 'static> {
    /// Each AsyncIpcReceiver::recv() call sends a oneshot channel that collects the result when
    /// it is ready. This ensures proper ordering even if the future from `recv` is dropped and has
    /// the added benefit of allowing `recv()` to take `&self` instead of `&mut self`. The second
    /// benefit allows some operations that use this struct to be lock-free.
    channel_sender: mpsc::UnboundedSender<oneshot::Sender<Result<T, IpcError>>>,
}

impl<T: Serialize + DeserializeOwned + Send + 'static> AsyncIpcReceiver<T> {
    pub fn new(ipc_receiver: IpcReceiver<T>) -> Self {
        let (channel_sender, mut receiver) = mpsc::unbounded_channel();

        let handle = Handle::current();
        thread::spawn(move || {
            loop {
                // This loop continues even if this call produces an error saying that IPC was
                // disconnected. This allows callers of `recv` to detect the disconnection.
                let next_value = ipc_receiver.recv();
{'fraction_non_alphanumeric': 0.11465543183869997, 'fraction_numerical': 0.0, 'mean_word_length': 3.661991584852735, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3090789', 'n_tokens_mistral': 954, 'n_tokens_neox': 939, 'n_words': 344}