Tauri 2.0 Async Rust Handling Asynchronous Operations

// Async Rust Fundamentals

// Synchronous vs Asynchronous:

// Synchronous (blocking)
fn fetch_data_sync() -> String {
    // Blocks thread until complete
    std::thread::sleep(std::time::Duration::from_secs(2));
    "Data".to_string()
}

// Asynchronous (non-blocking)
async fn fetch_data_async() -> String {
    // Yields control while waiting
    tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
    "Data".to_string()
}

// Key concepts:

// 1. Future trait
// Represents a value that will be available in the future
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct MyFuture {
    count: u32,
}

impl Future for MyFuture {
    type Output = u32;
    
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.count >= 3 {
            Poll::Ready(self.count)
        } else {
            self.count += 1;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

// 2. async fn syntax
// Returns impl Future<Output = T>
async fn async_function() -> i32 {
    42
}

// Equivalent to:
fn async_function_explicit() -> impl Future<Output = i32> {
    async { 42 }
}

// 3. await keyword
// Suspends execution until future is ready
async fn example() {
    let result = async_function().await;
    println!("Result: {}", result);
}

// 4. Tokio runtime
// Executes async code
use tokio;

#[tokio::main]
async fn main() {
    let result = async_function().await;
    println!("Result: {}", result);
}

// Or manually:
fn main() {
    let runtime = tokio::runtime::Runtime::new().unwrap();
    runtime.block_on(async {
        let result = async_function().await;
        println!("Result: {}", result);
    });
}

// 5. Spawning tasks
#[tokio::main]
async fn main() {
    // Spawn background task
    let handle = tokio::spawn(async {
        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
        "Task complete"
    });
    
    // Do other work
    println!("Doing other work...");
    
    // Wait for task to complete
    let result = handle.await.unwrap();
    println!("Result: {}", result);
}

// 6. Multiple async operations
async fn multiple_operations() {
    // Sequential (one after another)
    let a = fetch_data_async().await;
    let b = fetch_data_async().await;
    
    // Concurrent (both at same time)
    let (a, b) = tokio::join!(
        fetch_data_async(),
        fetch_data_async()
    );
    
    // Or with spawn:
    let handle_a = tokio::spawn(fetch_data_async());
    let handle_b = tokio::spawn(fetch_data_async());
    let (a, b) = (handle_a.await.unwrap(), handle_b.await.unwrap());
}

// 7. Error handling
async fn fallible_operation() -> Result<String, String> {
    if rand::random() {
        Ok("Success".to_string())
    } else {
        Err("Failed".to_string())
    }
}

async fn handle_errors() {
    // Using match
    match fallible_operation().await {
        Ok(value) => println!("Success: {}", value),
        Err(e) => println!("Error: {}", e),
    }
    
    // Using ?
    let result = fallible_operation().await?;
    
    // Using unwrap (panics on error)
    let result = fallible_operation().await.unwrap();
}

// 8. Timeout
use tokio::time::{timeout, Duration};

async fn operation_with_timeout() -> Result<String, String> {
    match timeout(Duration::from_secs(5), fetch_data_async()).await {
        Ok(result) => Ok(result),
        Err(_) => Err("Operation timed out".to_string()),
    }
}

// 9. Select (first to complete)
use tokio::select;

async fn first_complete() -> String {
    select! {
        result1 = fetch_data_async() => result1,
        result2 = fetch_data_async() => result2,
    }
}

// 10. Async traits (experimental)
use async_trait::async_trait;

#[async_trait]
trait DataFetcher {
    async fn fetch(&self) -> String;
}

struct HttpFetcher;

#[async_trait]
impl DataFetcher for HttpFetcher {
    async fn fetch(&self) -> String {
        // Async implementation
        fetch_data_async().await
    }
}

// Async Tauri commands
// src-tauri/src/main.rs

use tauri::State;
use tokio::time::{sleep, Duration};

// Simple async command
#[tauri::command]
async fn fetch_data() -> Result<String, String> {
    // Simulate API call
    sleep(Duration::from_secs(2)).await;
    Ok("Data fetched successfully".to_string())
}

// Async command with parameters
#[tauri::command]
async fn process_file(path: String) -> Result<String, String> {
    // Read file asynchronously
    let content = tokio::fs::read_to_string(&path)
        .await
        .map_err(|e| e.to_string())?;
    
    // Process content
    let processed = content.to_uppercase();
    
    Ok(processed)
}

// Async command with state
struct AppState {
    counter: tokio::sync::Mutex<u32>,
}

#[tauri::command]
async fn increment_counter(state: State<'_, AppState>) -> Result<u32, String> {
    let mut counter = state.counter.lock().await;
    *counter += 1;
    Ok(*counter)
}

// HTTP request in async command
use reqwest;

#[tauri::command]
async fn fetch_from_api(url: String) -> Result<String, String> {
    let response = reqwest::get(&url)
        .await
        .map_err(|e| e.to_string())?;
    
    let body = response.text()
        .await
        .map_err(|e| e.to_string())?;
    
    Ok(body)
}

// Multiple concurrent operations
#[tauri::command]
async fn fetch_multiple(urls: Vec<String>) -> Result<Vec<String>, String> {
    let futures = urls.into_iter().map(|url| async move {
        reqwest::get(&url)
            .await
            .ok()?
            .text()
            .await
            .ok()
    });
    
    let results = futures_util::future::join_all(futures).await;
    
    Ok(results.into_iter().filter_map(|r| r).collect())
}

// Progress reporting with events
use tauri::{Manager, Emitter};

#[tauri::command]
async fn download_file(
    app: tauri::AppHandle,
    url: String,
    path: String
) -> Result<(), String> {
    let client = reqwest::Client::new();
    let response = client.get(&url)
        .send()
        .await
        .map_err(|e| e.to_string())?;
    
    let total_size = response.content_length().unwrap_or(0);
    let mut downloaded: u64 = 0;
    
    let mut file = tokio::fs::File::create(&path)
        .await
        .map_err(|e| e.to_string())?;
    
    let mut stream = response.bytes_stream();
    
    use tokio_stream::StreamExt;
    use tokio::io::AsyncWriteExt;
    
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.map_err(|e| e.to_string())?;
        file.write_all(&chunk).await.map_err(|e| e.to_string())?;
        
        downloaded += chunk.len() as u64;
        
        let progress = if total_size > 0 {
            (downloaded as f64 / total_size as f64) * 100.0
        } else {
            0.0
        };
        
        // Emit progress event
        app.emit("download-progress", progress)
            .map_err(|e| e.to_string())?;
    }
    
    Ok(())
}

// Background task management
use std::sync::Arc;
use tokio::sync::Mutex;

struct BackgroundTasks {
    tasks: Arc<Mutex<Vec<tokio::task::JoinHandle<()>>>>,
}

impl BackgroundTasks {
    fn new() -> Self {
        Self {
            tasks: Arc::new(Mutex::new(Vec::new())),
        }
    }
    
    async fn spawn_task<F>(&self, future: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        let handle = tokio::spawn(future);
        self.tasks.lock().await.push(handle);
    }
    
    async fn cancel_all(&self) {
        let mut tasks = self.tasks.lock().await;
        for task in tasks.drain(..) {
            task.abort();
        }
    }
}

#[tauri::command]
async fn start_background_task(
    state: State<'_, BackgroundTasks>,
    app: tauri::AppHandle
) -> Result<(), String> {
    state.spawn_task(async move {
        loop {
            sleep(Duration::from_secs(60)).await;
            
            // Periodic task
            if let Err(e) = app.emit("heartbeat", ()) {
                eprintln!("Failed to emit heartbeat: {}", e);
                break;
            }
        }
    }).await;
    
    Ok(())
}

#[tauri::command]
async fn stop_all_tasks(
    state: State<'_, BackgroundTasks>
) -> Result<(), String> {
    state.cancel_all().await;
    Ok(())
}

// Main setup
fn main() {
    tauri::Builder::default()
        .manage(AppState {
            counter: tokio::sync::Mutex::new(0),
        })
        .manage(BackgroundTasks::new())
        .invoke_handler(tauri::generate_handler![
            fetch_data,
            process_file,
            increment_counter,
            fetch_from_api,
            fetch_multiple,
            download_file,
            start_background_task,
            stop_all_tasks,
        ])
        .run(tauri::generate_context!())
        .expect("error while running tauri application");
}

// Frontend usage
// src/App.jsx
import { invoke } from '@tauri-apps/api/core';
import { listen } from '@tauri-apps/api/event';

function App() {
  const [progress, setProgress] = useState(0);
  
  useEffect(() => {
    const unlisten = listen('download-progress', (event) => {
      setProgress(event.payload);
    });
    
    return () => {
      unlisten.then(fn => fn());
    };
  }, []);
  
  const downloadFile = async () => {
    try {
      await invoke('download_file', {
        url: 'https://example.com/file.zip',
        path: '/tmp/file.zip'
      });
      console.log('Download complete!');
    } catch (error) {
      console.error('Download failed:', error);
    }
  };
  
  return (
    <div>
      <button onClick={downloadFile}>Download</button>
      <progress value={progress} max={100} />
    </div>
  );
}

// Async channels for communication
use tokio::sync::{mpsc, oneshot, broadcast};

// 1. MPSC (Multiple Producer, Single Consumer)
#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<String>(100);
    
    // Spawn producer task
    tokio::spawn(async move {
        for i in 0..10 {
            tx.send(format!("Message {}", i)).await.unwrap();
        }
    });
    
    // Consumer
    while let Some(msg) = rx.recv().await {
        println!("Received: {}", msg);
    }
}

// Multiple producers
async fn multiple_producers() {
    let (tx, mut rx) = mpsc::channel::<String>(100);
    
    for i in 0..5 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(format!("Producer {} message", i)).await.unwrap();
        });
    }
    
    drop(tx); // Close channel when all producers done
    
    while let Some(msg) = rx.recv().await {
        println!("Received: {}", msg);
    }
}

// 2. Oneshot (Single value communication)
async fn request_response() {
    let (tx, rx) = oneshot::channel::<String>();
    
    tokio::spawn(async move {
        let result = fetch_data_async().await;
        tx.send(result).unwrap();
    });
    
    match rx.await {
        Ok(result) => println!("Result: {}", result),
        Err(_) => println!("Sender dropped"),
    }
}

// 3. Broadcast (Multiple consumers)
async fn broadcast_example() {
    let (tx, mut rx1) = broadcast::channel::<String>(100);
    let mut rx2 = tx.subscribe();
    let mut rx3 = tx.subscribe();
    
    tokio::spawn(async move {
        tx.send("Broadcast message".to_string()).unwrap();
    });
    
    tokio::spawn(async move {
        if let Ok(msg) = rx1.recv().await {
            println!("Receiver 1: {}", msg);
        }
    });
    
    tokio::spawn(async move {
        if let Ok(msg) = rx2.recv().await {
            println!("Receiver 2: {}", msg);
        }
    });
    
    tokio::spawn(async move {
        if let Ok(msg) = rx3.recv().await {
            println!("Receiver 3: {}", msg);
        }
    });
}

// Task coordination pattern
struct TaskManager {
    command_tx: mpsc::Sender<Command>,
}

enum Command {
    Process(String),
    Stop,
}

impl TaskManager {
    fn new() -> Self {
        let (tx, mut rx) = mpsc::channel::<Command>(100);
        
        tokio::spawn(async move {
            while let Some(cmd) = rx.recv().await {
                match cmd {
                    Command::Process(data) => {
                        println!("Processing: {}", data);
                        // Do work
                    }
                    Command::Stop => {
                        println!("Stopping");
                        break;
                    }
                }
            }
        });
        
        Self { command_tx: tx }
    }
    
    async fn process(&self, data: String) {
        self.command_tx.send(Command::Process(data)).await.unwrap();
    }
    
    async fn stop(&self) {
        self.command_tx.send(Command::Stop).await.unwrap();
    }
}

// Worker pool pattern
use std::sync::Arc;

struct WorkerPool {
    workers: Vec<mpsc::Sender<Task>>,
}

type Task = Box<dyn FnOnce() + Send + 'static>;

impl WorkerPool {
    fn new(size: usize) -> Self {
        let mut workers = Vec::new();
        
        for _ in 0..size {
            let (tx, mut rx) = mpsc::channel::<Task>(10);
            
            tokio::spawn(async move {
                while let Some(task) = rx.recv().await {
                    task();
                }
            });
            
            workers.push(tx);
        }
        
        Self { workers }
    }
    
    async fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let idx = rand::random::<usize>() % self.workers.len();
        self.workers[idx].send(Box::new(f)).await.unwrap();
    }
}

// Producer-consumer pattern
use tokio::sync::Semaphore;

struct BoundedQueue<T> {
    queue: Arc<Mutex<Vec<T>>>,
    semaphore: Arc<Semaphore>,
}

impl<T: Send + 'static> BoundedQueue<T> {
    fn new(capacity: usize) -> Self {
        Self {
            queue: Arc::new(Mutex::new(Vec::new())),
            semaphore: Arc::new(Semaphore::new(capacity)),
        }
    }
    
    async fn push(&self, item: T) {
        let _permit = self.semaphore.acquire().await.unwrap();
        self.queue.lock().await.push(item);
    }
    
    async fn pop(&self) -> Option<T> {
        let mut queue = self.queue.lock().await;
        let item = queue.pop();
        if item.is_some() {
            self.semaphore.add_permits(1);
        }
        item
    }
}

// Tauri integration
#[tauri::command]
async fn start_processing(
    state: State<'_, TaskManager>
) -> Result<(), String> {
    state.process("Some data".to_string()).await;
    Ok(())
}

#[tauri::command]
async fn stop_processing(
    state: State<'_, TaskManager>
) -> Result<(), String> {
    state.stop().await;
    Ok(())
}

// Async error handling patterns
use thiserror::Error;

#[derive(Error, Debug)]
enum AppError {
    #[error("IO error: {0}")]
    Io(#[from] std::io::Error),
    
    #[error("HTTP error: {0}")]
    Http(#[from] reqwest::Error),
    
    #[error("Parse error: {0}")]
    Parse(String),
    
    #[error("Timeout")]
    Timeout,
}

type Result<T> = std::result::Result<T, AppError>;

// Using ? operator
async fn fetch_and_parse(url: &str) -> Result<serde_json::Value> {
    let response = reqwest::get(url).await?;
    let text = response.text().await?;
    let json = serde_json::from_str(&text)
        .map_err(|e| AppError::Parse(e.to_string()))?;
    Ok(json)
}

// Handling multiple errors
async fn complex_operation() -> Result<String> {
    let result = tokio::try_join!(
        fetch_and_parse("https://api1.example.com"),
        fetch_and_parse("https://api2.example.com"),
    )?;
    
    Ok(format!("{:?}", result))
}

// Timeout with error
use tokio::time::{timeout, Duration};

async fn with_timeout(url: &str) -> Result<String> {
    match timeout(Duration::from_secs(5), reqwest::get(url)).await {
        Ok(Ok(response)) => Ok(response.text().await?),
        Ok(Err(e)) => Err(AppError::Http(e)),
        Err(_) => Err(AppError::Timeout),
    }
}

// Retry logic
async fn retry_operation<F, Fut, T>(
    mut f: F,
    max_retries: u32
) -> Result<T>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T>>,
{
    let mut attempts = 0;
    
    loop {
        match f().await {
            Ok(result) => return Ok(result),
            Err(e) => {
                attempts += 1;
                if attempts >= max_retries {
                    return Err(e);
                }
                tokio::time::sleep(Duration::from_secs(2u64.pow(attempts))).await;
            }
        }
    }
}

// Usage
async fn fetch_with_retry(url: &str) -> Result<String> {
    retry_operation(
        || async { with_timeout(url).await },
        3
    ).await
}

// Error recovery
async fn fetch_with_fallback(primary: &str, fallback: &str) -> Result<String> {
    match fetch_and_parse(primary).await {
        Ok(result) => Ok(result.to_string()),
        Err(_) => {
            // Try fallback
            fetch_and_parse(fallback)
                .await
                .map(|r| r.to_string())
        }
    }
}

// Collecting errors
async fn fetch_all_with_errors(
    urls: Vec<String>
) -> (Vec<String>, Vec<AppError>) {
    let mut results = Vec::new();
    let mut errors = Vec::new();
    
    for url in urls {
        match fetch_and_parse(&url).await {
            Ok(data) => results.push(data.to_string()),
            Err(e) => errors.push(e),
        }
    }
    
    (results, errors)
}

// Tauri command with proper error handling
#[tauri::command]
async fn fetch_data_command(url: String) -> Result<String, String> {
    fetch_with_retry(&url)
        .await
        .map_err(|e| e.to_string())
}

// Panic handling in async
use std::panic::AssertUnwindSafe;
use futures::FutureExt;

async fn safe_operation<F, T>(f: F) -> Result<T>
where
    F: Future<Output = T>,
{
    match AssertUnwindSafe(f).catch_unwind().await {
        Ok(result) => Ok(result),
        Err(_) => Err(AppError::Parse("Panic occurred".to_string())),
    }
}