pubsub

client/src/components/vbox.component.jsx

@@ -78,7 +78,7 @@ const addState = connect(
 function Vbox(args) {
     const {
         combiner,
-        instance,
+        // instance,
         itemInfo,
         player,
         reclaiming,

client/src/events.jsx

@@ -6,16 +6,6 @@ const { TIMES } = require('./constants');
 const { getCombatSequence } = require('./utils');
 
 function registerEvents(store) {
-
-    // timeout handlers
-    store.subscribe(() => {
-        const { game, instance, ws } = store.getState();
-
-        if (!ws) return true;
-        if (!game) ws.clearGameStateTimeout();
-        if (!instance) ws.clearInstanceStateTimeout();
-    });
-
     function setPing(ping) {
         store.dispatch(actions.setPing(ping));
     }
@@ -54,11 +44,9 @@ function registerEvents(store) {
     function setGame(game) {
         const { game: currentGame, ws } = store.getState();
 
-        if (game) ws.startGameStateTimeout(game.id);
         if (game && currentGame) {
             if (game.resolved.length !== currentGame.resolved.length) {
                 // stop fetching the game state til animations are done
-                ws.clearGameStateTimeout();
                 const newRes = game.resolved.slice(currentGame.resolved.length);
                 let id = game.resolved.length - currentGame.resolved.length;
                 return eachSeries(newRes, (r, cb) => {
@@ -151,7 +139,6 @@ function registerEvents(store) {
         store.dispatch(actions.setItemEquip(null));
         store.dispatch(actions.setItemUnequip(null));
         store.dispatch(actions.setVboxHighlight([]));
-        store.dispatch(actions.setConstructDeleteId(null));
     }
 
     function setAccountInstances(v) {
@@ -161,8 +148,6 @@ function registerEvents(store) {
     function setInstance(v) {
         const { account, ws, instance } = store.getState();
         if (v) {
-            ws.startInstanceStateTimeout(v.id);
-
             const player = v.players.find(p => p.id === account.id);
             store.dispatch(actions.setPlayer(player));
 

client/src/socket.jsx

@@ -136,8 +136,6 @@ function createSocket(events) {
     // -------------
     function onAccount(login) {
         events.setAccount(login);
-        sendAccountConstructs();
-        sendAccountInstances();
     }
 
     function onAccountShop(shop) {
@@ -146,7 +144,6 @@ function createSocket(events) {
 
     function onAccountInstances(list) {
         events.setAccountInstances(list);
-        setTimeout(sendAccountInstances, 5000);
     }
 
     function onAccountConstructs(constructs) {
@@ -157,36 +154,12 @@ function createSocket(events) {
         events.setGame(game);
     }
 
-    let gameStateTimeout;
-    function startGameStateTimeout(id) {
-        clearTimeout(gameStateTimeout);
-        gameStateTimeout = setTimeout(() => sendGameState(id), 1000);
-        return true;
-    }
-
-    function clearGameStateTimeout() {
-        clearTimeout(gameStateTimeout);
-    }
-
-    let instanceStateTimeout;
-    function startInstanceStateTimeout(id) {
-        clearTimeout(instanceStateTimeout);
-        instanceStateTimeout = setTimeout(() => sendInstanceState(id), 1000);
-        return true;
-    }
-
     function onInstanceState(instance) {
         events.setInstance(instance);
-        return true;
     }
 
     function onOpenInstances(list) {
         events.setInstanceList(list);
-        return true;
-    }
-
-    function clearInstanceStateTimeout() {
-        clearTimeout(instanceStateTimeout);
     }
 
     function onItemInfo(info) {
@@ -277,8 +250,6 @@ function createSocket(events) {
 
     function connect() {
         if (ws) {
-            clearGameStateTimeout();
-            clearInstanceStateTimeout();
             clearTimeout(pongTimeout);
             ws.removeEventListener('open', onOpen);
             ws.removeEventListener('message', onMessage);
@@ -299,8 +270,6 @@ function createSocket(events) {
     }
 
     return {
-        clearGameStateTimeout,
-        clearInstanceStateTimeout,
         sendAccountConstructs,
         sendAccountInstances,
         sendGameState,
@@ -321,8 +290,6 @@ function createSocket(events) {
         sendItemInfo,
         sendMtxApply,
         sendMtxBuy,
-        startInstanceStateTimeout,
-        startGameStateTimeout,
         connect,
     };
 }

ops/migrations/20181020104420_games.js

@@ -54,18 +54,6 @@ exports.up = async knex => {
         table.unique(['account', 'instance']);
 
     });
-
-    await knex.schema.createTable('matchmaking', async table => {
-        table.uuid('id').primary();
-        table.index('id');
-        table.timestamps(true, true);
-
-        table.uuid('game').notNullable()
-        table.foreign('game')
-            .references('id')
-            .inTable('games')
-            .onDelete('NO ACTION');
-    });
 };
 
 exports.down = async () => {};

server/src/instance.rs

@@ -17,7 +17,7 @@ use construct::{Construct, construct_get};
 use mob::{bot_player, instance_mobs};
 use game::{Game, Phase, game_get, game_write};
 use item::{Item};
-use rpc::{RpcResult};
+use rpc::{RpcMessage};
 use img;
 
 #[derive(Debug,Clone,Copy,PartialEq,Serialize,Deserialize)]
@@ -756,7 +756,7 @@ pub fn instance_ready(tx: &mut Transaction, account: &Account, instance_id: Uuid
     instance_update(tx, instance)
 }
 
-pub fn instance_state(tx: &mut Transaction, _account: &Account, instance_id: Uuid) -> Result<RpcResult, Error> {
+pub fn instance_state(tx: &mut Transaction, _account: &Account, instance_id: Uuid) -> Result<RpcMessage, Error> {
     let instance = instance_get(tx, instance_id)?;
 
     if let Some(game_id) = instance.current_game_id() {
@@ -764,11 +764,11 @@ pub fn instance_state(tx: &mut Transaction, _account: &Account, instance_id: Uui
 
         // return the game until it's finished
         if game.phase != Phase::Finish {
-            return Ok(RpcResult::GameState(game))
+            return Ok(RpcMessage::GameState(game))
         }
     }
 
-    Ok(RpcResult::InstanceState(instance))
+    Ok(RpcMessage::InstanceState(instance))
 }
 
 pub fn instance_game_finished(tx: &mut Transaction, game: &Game, instance_id: Uuid) -> Result<(), Error> {

server/src/names.rs

@@ -1,63 +1,69 @@
 use rand::prelude::*;
 use rand::{thread_rng};
 
-const FIRSTS: [&'static str; 23] = [
+const FIRSTS: [&'static str; 25] = [
-    "fierce",
-    "obscure",
-    "mighty",
-    "rogue",
-    "inverted",
-    "recalcitrant",
-    "subterranean",
-    "brewing",
-    "nocturnal",
-    "convex",
-    "concave",
-    "piscine",
-    "dub",
     "borean",
-    "lurking",
+    "brewing",
-    "leafy",
-    "nutritious",
     "bristling",
-    "metallic",
+    "compressed",
-    "purified",
+    "concave",
-    "organic",
+    "convex",
     "distorted",
+    "dub",
+    "fierce",
+    "inverted",
+    "leafy",
+    "lurking",
+    "metallic",
+    "mighty",
+    "nocturnal",
+    "nutritious",
+    "obscure",
+    "organic",
+    "piscine",
+    "purified",
+    "recalcitrant",
+    "rogue",
+    "subterranean",
+    "sweet",
     "weary",
 ];
 
-const LASTS: [&'static str; 30] = [
+const LASTS: [&'static str; 34] = [
-    "kaffe",
-    "foilage",
-    "wildlife",
-    "design",
-    "assembly",
-    "layout",
-    "transmitter",
-    "lens",
     "artifact",
-    "frequency",
+    "assembly",
-    "entropy",
     "console",
-    "insulator",
-    "river",
-    "oak",
-    "replicant",
-    "mechanism",
     "construct",
-    "function",
+    "design",
-    "shape",
+    "entropy",
-    "form",
+    "foilage",
-    "poseidon",
-    "mountain",
-    "river",
     "forest",
-    "problem",
+    "form",
-    "warning",
+    "frequency",
+    "function",
     "information",
-    "witness",
+    "insulator",
+    "kaffe",
+    "layout",
+    "lens",
+    "mechanism",
+    "mountain",
+    "nectar",
+    "oak",
+    "plant",
+    "poseidon",
+    "problem",
+    "replicant",
+    "river",
+    "river",
+    "shape",
+    "signal",
+    "tower",
+    "transmitter",
     "traveller",
+    "warning",
+    "wildlife",
+    "witness",
 ];
 
 pub fn name() -> String {

server/src/rpc.rs

@@ -1,13 +1,13 @@
 use std::net::TcpStream;
 use std::time::{Instant};
 
-use tungstenite::protocol::WebSocket;
 
 use serde_cbor::{from_slice};
 use uuid::Uuid;
 use failure::Error;
 use failure::err_msg;
 
+use ws::{Ws};
 use pg::{Db};
 use construct::{Construct};
 use game::{Game, game_state, game_skill, game_ready};
@@ -20,7 +20,7 @@ use item::{Item, ItemInfoCtr, item_info};
 use mtx;
 
 #[derive(Debug,Clone,Serialize,Deserialize)]
-pub enum RpcResult {
+pub enum RpcMessage {
     AccountState(Account),
     AccountConstructs(Vec<Construct>),
     AccountInstances(Vec<Instance>),
@@ -53,7 +53,6 @@ enum RpcRequest {
     AccountState {},
     AccountShop {},
     AccountConstructs {},
-    AccountInstances {},
 
     InstanceList {},
     InstanceLobby { construct_ids: Vec<Uuid>, name: String, pve: bool, password: Option<String> },
@@ -69,7 +68,7 @@ enum RpcRequest {
     VboxReclaim { instance_id: Uuid, index: usize },
 }
 
-pub fn receive(data: Vec<u8>, db: &Db, _client: &mut WebSocket<TcpStream>, begin: Instant, account: &Option<Account>) -> Result<RpcResult, Error> {
+pub fn receive(data: Vec<u8>, db: &Db, _client: &mut Ws, begin: Instant, account: &Option<Account>) -> Result<RpcMessage, Error> {
     // cast the msg to this type to receive method name
     match from_slice::<RpcRequest>(&data) {
         Ok(v) => {
@@ -77,10 +76,10 @@ pub fn receive(data: Vec<u8>, db: &Db, _client: &mut WebSocket<TcpStream>, begin
             // non authenticated
             // non transactional reqs
             match v {
-                RpcRequest::Ping {} => return Ok(RpcResult::Pong(())),
+                RpcRequest::Ping {} => return Ok(RpcMessage::Pong(())),
-                RpcRequest::ItemInfo {} => return Ok(RpcResult::ItemInfo(item_info())),
+                RpcRequest::ItemInfo {} => return Ok(RpcMessage::ItemInfo(item_info())),
                 RpcRequest::DevResolve {a, b, skill } =>
-                    return Ok(RpcResult::DevResolutions(dev_resolve(a, b, skill))),
+                    return Ok(RpcMessage::DevResolutions(dev_resolve(a, b, skill))),
                 _ => (),
             };
 
@@ -97,63 +96,61 @@ pub fn receive(data: Vec<u8>, db: &Db, _client: &mut WebSocket<TcpStream>, begin
 
             let response = match v {
                 RpcRequest::AccountState {} =>
-                    return Ok(RpcResult::AccountState(account.clone())),
+                    return Ok(RpcMessage::AccountState(account.clone())),
                 RpcRequest::AccountConstructs {} =>
-                    Ok(RpcResult::AccountConstructs(account_constructs(&mut tx, &account)?)),
+                    Ok(RpcMessage::AccountConstructs(account_constructs(&mut tx, &account)?)),
-                RpcRequest::AccountInstances {} =>
-                    Ok(RpcResult::AccountInstances(account_instances(&mut tx, &account)?)),
 
                 // RpcRequest::AccountShop {} =>
-                //     Ok(RpcResult::AccountShop(mtx::account_shop(&mut tx, &account)?)),
+                //     Ok(RpcMessage::AccountShop(mtx::account_shop(&mut tx, &account)?)),
 
 
 
                 // RpcRequest::ConstructDelete" => handle_construct_delete(data, &mut tx, account),
 
                 RpcRequest::GameState { id } =>
-                    Ok(RpcResult::GameState(game_state(&mut tx, account, id)?)),
+                    Ok(RpcMessage::GameState(game_state(&mut tx, account, id)?)),
 
                 RpcRequest::GameSkill { game_id, construct_id, target_construct_id, skill } =>
-                    Ok(RpcResult::GameState(game_skill(&mut tx, account, game_id, construct_id, target_construct_id, skill)?)),
+                    Ok(RpcMessage::GameState(game_skill(&mut tx, account, game_id, construct_id, target_construct_id, skill)?)),
 
                 RpcRequest::GameReady { id } =>
-                    Ok(RpcResult::GameState(game_ready(&mut tx, account, id)?)),
+                    Ok(RpcMessage::GameState(game_ready(&mut tx, account, id)?)),
 
                 RpcRequest::InstanceList {} =>
-                    Ok(RpcResult::OpenInstances(instance_list(&mut tx)?)),
+                    Ok(RpcMessage::OpenInstances(instance_list(&mut tx)?)),
                 RpcRequest::InstanceLobby { construct_ids, name, pve, password } =>
-                    Ok(RpcResult::InstanceState(instance_new(&mut tx, account, construct_ids, name, pve, password)?)),
+                    Ok(RpcMessage::InstanceState(instance_new(&mut tx, account, construct_ids, name, pve, password)?)),
                 RpcRequest::InstanceJoin { instance_id, construct_ids } =>
-                    Ok(RpcResult::InstanceState(instance_join(&mut tx, account, instance_id, construct_ids)?)),
+                    Ok(RpcMessage::InstanceState(instance_join(&mut tx, account, instance_id, construct_ids)?)),
                 RpcRequest::InstanceReady { instance_id } =>
-                    Ok(RpcResult::InstanceState(instance_ready(&mut tx, account, instance_id)?)),
+                    Ok(RpcMessage::InstanceState(instance_ready(&mut tx, account, instance_id)?)),
                 RpcRequest::InstanceState { instance_id } =>
                     Ok(instance_state(&mut tx, account, instance_id)?),
 
                 RpcRequest::VboxAccept { instance_id, group, index } =>
-                    Ok(RpcResult::InstanceState(vbox_accept(&mut tx, account, instance_id, group, index)?)),
+                    Ok(RpcMessage::InstanceState(vbox_accept(&mut tx, account, instance_id, group, index)?)),
 
                 RpcRequest::VboxApply { instance_id, construct_id, index } =>
-                    Ok(RpcResult::InstanceState(vbox_apply(&mut tx, account, instance_id, construct_id, index)?)),
+                    Ok(RpcMessage::InstanceState(vbox_apply(&mut tx, account, instance_id, construct_id, index)?)),
 
                 RpcRequest::VboxCombine { instance_id, indices } =>
-                    Ok(RpcResult::InstanceState(vbox_combine(&mut tx, account, instance_id, indices)?)),
+                    Ok(RpcMessage::InstanceState(vbox_combine(&mut tx, account, instance_id, indices)?)),
 
                 RpcRequest::VboxDiscard { instance_id } =>
-                    Ok(RpcResult::InstanceState(vbox_discard(&mut tx, account, instance_id)?)),
+                    Ok(RpcMessage::InstanceState(vbox_discard(&mut tx, account, instance_id)?)),
 
                 RpcRequest::VboxReclaim { instance_id, index } =>
-                    Ok(RpcResult::InstanceState(vbox_reclaim(&mut tx, account, instance_id, index)?)),
+                    Ok(RpcMessage::InstanceState(vbox_reclaim(&mut tx, account, instance_id, index)?)),
 
                 RpcRequest::VboxUnequip { instance_id, construct_id, target } =>
-                    Ok(RpcResult::InstanceState(vbox_unequip(&mut tx, account, instance_id, construct_id, target)?)),
+                    Ok(RpcMessage::InstanceState(vbox_unequip(&mut tx, account, instance_id, construct_id, target)?)),
 
 
                 RpcRequest::MtxConstructApply { mtx, construct_id, name } =>
-                    Ok(RpcResult::AccountConstructs(mtx::apply(&mut tx, account, mtx, construct_id, name)?)),
+                    Ok(RpcMessage::AccountConstructs(mtx::apply(&mut tx, account, mtx, construct_id, name)?)),
 
                 RpcRequest::MtxBuy { mtx } =>
-                    Ok(RpcResult::AccountShop(mtx::buy(&mut tx, account, mtx)?)),
+                    Ok(RpcMessage::AccountShop(mtx::buy(&mut tx, account, mtx)?)),
 
                 _ => Err(format_err!("unknown request request={:?}", request)),
             };

server/src/ws.rs

@@ -17,14 +17,21 @@ use crossbeam_channel::{unbounded, Receiver};
 
 use serde_cbor::{to_vec};
 
+use failure::Error;
+use failure::{err_msg, format_err};
+
 use net::TOKEN_HEADER;
 use rpc;
+use rpc::{RpcMessage};
 
 use mtx;
 use pg::PgPool;
 use account;
+use account::Account;
 use pubsub::Message;
 
+pub type Ws = WebSocket<TcpStream>;
+
 #[derive(Debug,Clone,Serialize)]
 struct RpcError {
     err: String,
@@ -33,25 +40,91 @@ struct RpcError {
 #[derive(Debug)]
 struct Subscriptions {
     account: Option<Uuid>,
-    games: Vec<Uuid>,
+    game: Option<Uuid>,
-    instances: Vec<Uuid>,
+    instance: Option<Uuid>,
+    // account_instances: Vec<Uuid>,
 }
 
-fn handle_message(subs: &Subscriptions, m: Message, ws: &mut WebSocket<TcpStream>) {
+impl Subscriptions {
+    fn new(ws_pool: &PgPool, account: &Option<Account>, ws: &mut Ws) -> Result<Subscriptions, Error> {
+        if let Some(a) = account {
+            let db = ws_pool.get()?;
+            let mut tx = db.transaction()?;
+
+            // send account constructs
+            let account_constructs = account::account_constructs(&mut tx, a)?;
+            ws.write_message(Binary(to_vec(&rpc::RpcMessage::AccountConstructs(account_constructs))?))?;
+
+            // get account instances
+            // and send them to the client
+            let account_instances = account::account_instances(&mut tx, a)?;
+            // let instances = account_instances.iter().map(|i| i.id).collect::<Vec<Uuid>>();
+            ws.write_message(Binary(to_vec(&rpc::RpcMessage::AccountInstances(account_instances))?))?;
+
+            // get players
+            // add to games
+
+            tx.commit()?;
+
+            return Ok(Subscriptions {
+                account: Some(a.id),
+                game: None,
+                instance: None,
+            })
+        }
+
+        Ok(Subscriptions {
+            account: None,
+            game: None,
+            instance: None
+        })
+    }
+
+    fn update(&mut self, msg: &RpcMessage) -> Result<&mut Subscriptions, Error> {
+        match msg {
+            RpcMessage::AccountState(a) => self.account = Some(a.id),
+            RpcMessage::InstanceState(i) => self.instance = Some(i.id),
+            RpcMessage::GameState(g) => self.game = Some(g.id),
+            _ => (),
+        };
+
+        // info!("subscriptions updated {:?}", self);
+
+        Ok(self)
+    }
+}
+
+
+fn handle_message(subs: &Subscriptions, m: Message, ws: &mut Ws) {
     if let Some(msg) = match m {
         Message::Account(a) => {
-            match (a.id, &subs.account) {
+            match subs.account {
-                (id, Some(b)) => {
+                Some(wsa) => match wsa == a.id {
-                    if id == *b {
+                    true => Some(rpc::RpcMessage::AccountState(a)),
-                        Some(rpc::RpcResult::AccountState(a))
+                    false => None,
-                    } else {
+                },
-                        None
+                None => None,
-                    }
-                }
-                _ => None,
             }
         },
-        _ => None,
+        Message::Instance(i) => {
+            match subs.instance {
+                Some(ci) => match ci == i.id {
+                    true => Some(rpc::RpcMessage::InstanceState(i)),
+                    false => None,
+                },
+                None => None,
+            }
+        },
+        Message::Game(g) => {
+            match subs.game {
+                Some(cg) => match cg == g.id {
+                    true => Some(rpc::RpcMessage::GameState(g)),
+                    false => None,
+                },
+                None => None,
+            }
+        },
+        // _ => None,
     } {
         ws.write_message(Binary(to_vec(&msg).unwrap())).unwrap();
     }
@@ -101,12 +174,12 @@ pub fn start(pool: PgPool, psr: Receiver<Message>) {
 
                     match account::from_token(&db, t) {
                         Ok(a) => {
-                            let state = to_vec(&rpc::RpcResult::AccountState(a.clone())).unwrap();
+                            let state = to_vec(&rpc::RpcMessage::AccountState(a.clone())).unwrap();
                             websocket.write_message(Binary(state)).unwrap();
 
                             let mut tx = db.transaction().unwrap();
                             let shop = mtx::account_shop(&mut tx, &a).unwrap();
-                            let shop = to_vec(&rpc::RpcResult::AccountShop(shop)).unwrap();
+                            let shop = to_vec(&rpc::RpcMessage::AccountShop(shop)).unwrap();
 
                             websocket.write_message(Binary(shop)).unwrap();
 
@@ -124,13 +197,12 @@ pub fn start(pool: PgPool, psr: Receiver<Message>) {
                 None => None,
             };
 
-            let mut subs = Subscriptions {
+            let mut subs = match Subscriptions::new(&ws_pool, &account, &mut websocket) {
-                account: match account.as_ref() {
+                Ok(s) => s,
-                    Some(a) => Some(a.id),
+                Err(e) => {
-                    None => None,
+                    warn!("subscriptions error err={:?}", e);
+                    return;
                 },
-                games: vec![],
-                instances: vec![],
             };
 
             loop {
@@ -152,6 +224,8 @@ pub fn start(pool: PgPool, psr: Receiver<Message>) {
                                             debug!("{:?}", e);
                                             return;
                                         };
+
+                                        subs.update(&reply).unwrap();
                                     },
                                     Err(e) => {
                                         warn!("{:?}", e);