Refactor db.rs and start saving sessions

blastmud_game/src/db.rs

@@ -12,56 +12,66 @@ pub struct DBPool {
     pool: Pool
 }
 
-pub async fn record_listener_ping(listener: Uuid, pool: DBPool) -> DResult<()> {
+impl DBPool {
-    get_conn(pool).await?.execute(
+    pub async fn record_listener_ping(self: DBPool, listener: Uuid) -> DResult<()> {
-        "INSERT INTO listeners (listener, last_seen) \
+        self.get_conn().await?.execute(
-           VALUES ($1, NOW()) \
+            "INSERT INTO listeners (listener, last_seen) \
-           ON CONFLICT (listener) \
+             VALUES ($1, NOW()) \
-           DO UPDATE SET last_seen = EXCLUDED.last_seen", &[&listener]).await?;
+             ON CONFLICT (listener) \
-    Ok(())
+             DO UPDATE SET last_seen = EXCLUDED.last_seen", &[&listener]).await?;
-}
+        Ok(())
+    }
 
-pub async fn get_dead_listeners(pool: DBPool) -> DResult<Vec<Uuid>> {
+    pub async fn get_dead_listeners(self: DBPool) -> DResult<Vec<Uuid>> {
-    Ok(get_conn(pool).await?
+        Ok(self.get_conn().await?
-        .query("SELECT listener FROM listeners WHERE last_seen < NOW() - \
+           .query("SELECT listener FROM listeners WHERE last_seen < NOW() - \
-                INTERVAL '2 minutes'", &[])
+                   INTERVAL '2 minutes'", &[])
-       .await?.into_iter().map(|r| r.get(0)).collect())
+           .await?.into_iter().map(|r| r.get(0)).collect())
-}
+    }
 
-pub async fn cleanup_listener(pool: DBPool, listener: Uuid) -> DResult<()> {
+    pub async fn cleanup_listener(self: DBPool, listener: Uuid) -> DResult<()> {
-    let mut conn = get_conn(pool).await?;
+        let mut conn = self.get_conn().await?;
-    let tx = conn.transaction().await?;
+        let tx = conn.transaction().await?;
-    tx.execute("UPDATE users SET current_session = NULL, \
+        tx.execute("UPDATE users SET current_session = NULL, \
-                current_listener = NULL WHERE current_listener = $1",
+                    current_listener = NULL WHERE current_listener = $1",
-               &[&listener]).await?;
+                   &[&listener]).await?;
-    tx.execute("DELETE FROM sendqueue WHERE listener = $1",
+        tx.execute("DELETE FROM sendqueue WHERE listener = $1",
-               &[&listener]).await?;
+                   &[&listener]).await?;
-    tx.execute("DELETE FROM sessions WHERE listener = $1",
+        tx.execute("DELETE FROM sessions WHERE listener = $1",
-               &[&listener]).await?;
+                   &[&listener]).await?;
-    tx.execute("DELETE FROM listeners WHERE listener = $1",
+        tx.execute("DELETE FROM listeners WHERE listener = $1",
-               &[&listener]).await?;
+                   &[&listener]).await?;
-    tx.commit().await?;
+        tx.commit().await?;
-    Ok(())
+        Ok(())
-}
+    }
 
-pub async fn get_conn(DBPool { pool }: DBPool) ->
+    pub async fn start_session(self: DBPool, listener: Uuid, session: Uuid) -> DResult<()> {
-    DResult<Object> {
+        self.get_conn().await?.execute(
-    let conn = pool.get().await?;
+            "INSERT INTO sessions (session, listener, details) VALUES ($1, $2, '{}')",
-    conn.execute("SET synchronous_commit=off", &[]).await?;
+            &[&session, &listener]
-    Ok(conn)
+        ).await?;
-}
+        Ok(())
-
+    }
-pub fn start_pool(connstr: &str) -> DResult<DBPool> {
+    
-    let mgr_config = ManagerConfig {
+    pub async fn get_conn(self: DBPool) ->
-        recycling_method: RecyclingMethod::Fast
+        DResult<Object> {
-    };
+            let conn = self.pool.get().await?;
-    let mgr = Manager::from_config(
+            conn.execute("SET synchronous_commit=off", &[]).await?;
-        PgConfig::from_str(connstr)
+            Ok(conn)
-            .map_err(|e| Box::new(e) as Box<dyn Error + Send + Sync>)?,
+        }
-        NoTls, mgr_config
+    
-    );
+    pub fn start(connstr: &str) -> DResult<DBPool> {
-
+        let mgr_config = ManagerConfig {
-    Pool::builder(mgr).max_size(4).build()
+            recycling_method: RecyclingMethod::Fast
-        .map_err(|e| Box::new(e) as Box<dyn Error + Send + Sync>)
+        };
-        .map(|pool| DBPool { pool })
+        let mgr = Manager::from_config(
+            PgConfig::from_str(connstr)
+                .map_err(|e| Box::new(e) as Box<dyn Error + Send + Sync>)?,
+            NoTls, mgr_config
+        );
+        
+        Pool::builder(mgr).max_size(4).build()
+            .map_err(|e| Box::new(e) as Box<dyn Error + Send + Sync>)
+            .map(|pool| DBPool { pool })
+    }
 }

blastmud_game/src/main.rs

@@ -4,6 +4,7 @@ use std::error::Error;
 use log::{info, error, LevelFilter};
 use simple_logger::SimpleLogger;
 use tokio::signal::unix::{signal, SignalKind};
+use db::DBPool;
 
 mod db;
 mod listener;
@@ -35,10 +36,10 @@ async fn main() -> DResult<()> {
         Err(e)
     })?;
     let config = read_latest_config()?;
-    let pool = db::start_pool(&config.database_conn_string)?;
+    let pool = DBPool::start(&config.database_conn_string)?;
 
     // Test the database connection string works so we quit early if not...
-    let _ = db::get_conn(pool.clone()).await?.query("SELECT 1", &[]).await?;
+    let _ = pool.clone().get_conn().await?.query("SELECT 1", &[]).await?;
 
     info!("Database pool initialised");
 

blastmud_game/src/message_handler.rs

@@ -11,8 +11,10 @@ use log::info;
 pub async fn handle(listener: Uuid, msg: MessageFromListener, pool: db::DBPool, listener_map: ListenerMap)
                     -> DResult<()> {
     match msg {
-        ListenerPing { .. } => { db::record_listener_ping(listener, pool).await?; }
+        ListenerPing { .. } => { pool.record_listener_ping(listener).await?; }
-        SessionConnected { session: _, source: _ } => {}
+        SessionConnected { session, source: _ } => {
+            pool.start_session(listener, session).await?;
+        }
         SessionDisconnected { session: _ } => {}
         SessionSentLine { session, msg } => {
             info!("Awaiting listener lock");

blastmud_game/src/regular_tasks.rs

@@ -4,8 +4,8 @@ use crate::db;
 use log::warn;
 
 async fn cleanup_session_once(pool: db::DBPool) -> DResult<()> {
-    for listener in db::get_dead_listeners(pool.clone()).await? {
+    for listener in pool.clone().get_dead_listeners().await? {
-        db::cleanup_listener(pool.clone(), listener).await?;
+        pool.clone().cleanup_listener(listener).await?;
     }
     Ok(())
 }