sqlx-firebird 0.1.0-beta.1

//
// Copyright © 2023, RedSoft
// License: MIT
//

use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;

use futures_channel::oneshot;
use futures_intrusive::sync::{Mutex, MutexGuard};
use sqlx_core::error::Error;

use super::ConnectionState;
use crate::{FbConnectOptions, FbError, FbStatement};

static THREAD_ID: AtomicU64 = AtomicU64::new(0);

enum Command {
    Prepare {
        query: Box<str>,
        tx: oneshot::Sender<Result<FbStatement<'static>, Error>>,
    },
}

pub(crate) struct ConnectionWorker {
    command_tx: flume::Sender<Command>,
    pub(crate) shared: Arc<WorkerSharedState>,
}

pub(crate) struct WorkerSharedState {
    pub(crate) cached_statements_size: AtomicUsize,
    pub(crate) conn: Mutex<ConnectionState>,
}

impl ConnectionWorker {
    pub(crate) async fn establish(options: FbConnectOptions) -> Result<Self, Error> {
        let (establish_tx, establish_rx) = oneshot::channel();

        thread::Builder::new()
            .name(format!(
                "sqlx-firebird-worker-{}",
                THREAD_ID.fetch_add(1, Ordering::AcqRel)
            ))
            .spawn(move || {
                let (command_tx, command_rx) = flume::bounded(options.command_channel_size);

                let state = match options.establish() {
                    Ok(state) => state,
                    Err(e) => {
                        establish_tx.send(Err(e)).ok();
                        return;
                    },
                };

                let shared = Arc::new(WorkerSharedState {
                    cached_statements_size: AtomicUsize::new(0),
                    conn: Mutex::new(state, true),
                });

                let worker = Self {
                    command_tx,
                    shared: Arc::clone(&shared),
                };
                if establish_tx.send(Ok(worker)).is_err() {
                    return;
                }

                for cmd in command_rx {
                    match cmd {
                        Command::Prepare { query, tx } => {},
                    }
                }
            })?;

        establish_rx.await.map_err(|_| Error::WorkerCrashed)?
    }
}