Function async_table_provider_thread

datafusion/ffi/src/tests/async_provider.rs:57–91

(
    mut shutdown: mpsc::Receiver<bool>,
    mut batch_request: mpsc::Receiver<bool>,
    batch_sender: broadcast::Sender<Option<RecordBatch>>,
    tokio_rt: &mpsc::Sender<Handle>,
)

}

fn async_table_provider_thread(
    mut shutdown: mpsc::Receiver<bool>,
    mut batch_request: mpsc::Receiver<bool>,
    batch_sender: broadcast::Sender<Option<RecordBatch>>,
    tokio_rt: &mpsc::Sender<Handle>,
) {
    let runtime = Arc::new(
        tokio::runtime::Builder::new_current_thread()
            .build()
            .expect("Unable to create tokio runtime"),
    );
    let _runtime_guard = runtime.enter();
    tokio_rt
        .blocking_send(runtime.handle().clone())
        .expect("Unable to send tokio runtime back to main thread");

    runtime.block_on(async move {
        let mut num_received = 0;
        while let Some(true) = batch_request.recv().await {
            let record_batch = match num_received {
                0 => Some(create_record_batch(1, 5)),
                1 => Some(create_record_batch(6, 1)),
                2 => Some(create_record_batch(7, 5)),
                _ => None,
            };
            num_received += 1;

            if batch_sender.send(record_batch).is_err() {
                break;
            }
        }
    });

    let _ = shutdown.blocking_recv();
}

pub fn start_async_provider() -> (AsyncTableProvider, Handle) {
    let (batch_request_tx, batch_request_rx) = mpsc::channel(10);