[websocket] use Ref for better state management

spago.lock

@@ -62,6 +62,7 @@ workspace:
         - reactix: ">=0.6.1 <0.7.0"
         - record: ">=4.0.0 <5.0.0"
         - record-extra: ">=5.0.1 <6.0.0"
+        - refs
         - routing: ">=11.0.0 <12.0.0"
         - sequences: "*"
         - simple-json: ">=9.0.0 <10.0.0"

spago.yaml

@@ -105,6 +105,7 @@ package:
     - reactix: ">=0.6.1 <0.7.0"
     - record: ">=4.0.0 <5.0.0"
     - record-extra: ">=5.0.1 <6.0.0"
+    - refs
     - routing: ">=11.0.0 <12.0.0"
     - sequences: "*"
     - simple-json: ">=9.0.0 <10.0.0"

src/Gargantext/Components/App/App.purs

@@ -16,6 +16,7 @@ import Gargantext.Types (CacheParams, defaultCacheParams)
 import Gargantext.Utils (getter)
 import Gargantext.Utils.Reactix as R2
 import Reactix as R
+import Reactix.Utils as RU
 import Record as Record
 import Toestand as T
 
@@ -62,12 +63,14 @@ hydrateStoreCpt = here.component "hydrateStore" cpt where
       } _ = do
     -- | Computed
     -- |
+    wsNotification <- RU.hook $ \_ -> Notifications.emptyWSNotification
+    
     (state :: Record AppStore.State) <- pure $
       -- (cache options)
       { expandTableEdition: getter _.expandTableEdition cacheParams
       , showTree: getter _.showTree cacheParams
       -- (default options)
-      } `Record.merge` AppStore.options
+      } `Record.merge` (AppStore.options wsNotification)
 
     -- | Render
     -- |
@@ -102,13 +105,13 @@ mainAppCpt = here.component "main" cpt where
       R2.loadLocalStorageState R2.openNodesKey boxes.forestOpen
       T.listen (R2.listenLocalStorageState R2.openNodesKey) boxes.forestOpen
     R.useEffectOnce' $ do
-      ws' <- T.read boxes.wsNotification
+      ws <- T.read boxes.wsNotification
       -- TODO See G.C.Forest: we need a WS connection for every backend we're connected to
       (Sessions.Sessions { sessions }) <- T.read boxes.sessions
       let session = Seq.head sessions
       -- here.log2 "[mainApp] sessions" sessions'
-      ws <- Notifications.connect ws' "ws://localhost:8008/ws" session
-      T.write_ ws boxes.wsNotification
+      Notifications.connect ws "ws://localhost:8008/ws" session
+      -- T.write_ ws boxes.wsNotification
       let action = Notifications.InsertCallback (Notifications.UpdateTree (-1)) "some-uuid" (\_ -> here.log "callback!")
       Notifications.performAction ws action
     useHashRouter Router.router boxes.route -- Install router to window

src/Gargantext/Components/App/Store.purs

@@ -33,6 +33,7 @@ import Gargantext.Utils.Reactix as R2
 import Gargantext.Utils.Stores as Stores
 import Gargantext.Utils.Toestand as T2
 import Reactix as R
+import Record as Record
 import Toestand as T
 import Unsafe.Coerce (unsafeCoerce)
 
@@ -100,8 +101,9 @@ type State =
   , wsNotification      :: Notifications.WSNotification
   )
 
-options :: Record State
-options =
+options :: Notifications.WSNotification -> Record State
+options wsNotification =
+  { wsNotification } `Record.merge`
   { backend             : Nothing
   , errors              : []
   , expandTableEdition  : false
@@ -128,7 +130,6 @@ options =
   , theme               : Themes.defaultTheme
   , tileAxisXList       : mempty
   , tileAxisYList       : mempty
-  , wsNotification      : Notifications.emptyWSNotification
   }
 
 context :: R.Context (Record Store)

src/Gargantext/Components/Notifications.purs

@@ -10,9 +10,11 @@ import Data.Hashable (class Hashable, hash)
 import Data.HashMap as HM
 import Data.Maybe (Maybe(..), fromMaybe, isJust)
 import Data.Show.Generic (genericShow)
-import Data.Traversable (for)
+import Data.Traversable (for, traverse)
 import Data.Tuple (Tuple(..))
 import Effect (Effect)
+-- import Effect.AVar as AVar
+import Effect.Ref as Ref
 import Effect.Var (($=))
 import Effect.Var as Var
 import FFI.Simple ((.=), (..))
@@ -125,10 +127,10 @@ callTopic (State { callbacks }) topic = do
 
 
 data WSNotification =
-  WSNotification { state :: State
+  WSNotification { state :: Ref.Ref State
 
                  -- TODO Implement a WS connection
-                 , connection :: Maybe WS.Connection
+                 , connection :: Ref.Ref (Maybe WS.Connection)
                    
                  -- This calls R.setRef :: R.Ref State -> Effect Unit
                  -- , insertCallback :: Topic -> UUID -> Effect Unit
@@ -136,37 +138,63 @@ data WSNotification =
                  -- , removeCallback :: Topic -> UUID -> Effect Unit
                  }
 
-emptyWSNotification :: WSNotification
-emptyWSNotification = WSNotification { state      : emptyState
-                                     , connection : Nothing }
+emptyWSNotification :: Effect WSNotification
+emptyWSNotification = do
+  state <- Ref.new emptyState
+  connection <- Ref.new Nothing
+  pure $ WSNotification { state
+                        , connection }
 
 isConnected :: WSNotification -> Effect Boolean
-isConnected (WSNotification { connection: Nothing }) = pure false
-isConnected (WSNotification { connection: Just (WS.Connection conn) }) = do
-  rs <- Var.get conn.readyState
-  pure $ rs == WS.Open
+isConnected (WSNotification { connection }) =do
+  mConn <- Ref.read connection
+  mRs <- traverse (\(WS.Connection conn) -> do
+      rs <- Var.get conn.readyState
+      pure rs -- $ rs == WS.Open
+           ) mConn
+  pure $ mRs == Just WS.Open
+  -- mConn <- AVar.tryRead connection
+  -- case mConn of
+  --   Nothing -> pure false
+  --   Just conn -> do
+-- isConnected (WSNotification { connection: Just (WS.Connection conn) }) = do
+--   rs <- Var.get conn.readyState
+--   pure $ rs == WS.Open
 
 send :: forall a. (JSON.WriteForeign a) => WSNotification -> a -> Effect Unit
-send (WSNotification { connection: Nothing }) _ = pure unit
-send (WSNotification { connection: Just (WS.Connection conn) }) d =
-  conn.send (WS.Message $ JSON.writeJSON d)
+-- send (WSNotification { connection: Nothing }) _ = pure unit
+-- send (WSNotification { connection: Just (WS.Connection conn) }) d =
+send (WSNotification { connection }) d = do
+  mConn <- Ref.read connection
+  void $ traverse (\(WS.Connection conn) -> do
+                      conn.send (WS.Message $ JSON.writeJSON d)
+                      ) mConn
+
+-- alterState :: WSNotification -> (State -> State) -> Effect Unit
+-- alterState (WSNotification ws') stateCb = do
+--   let state = ws' .. "state"
+--   -- here.log2 "[alterState] state" state
+--   void $ pure $ (ws' .= "state") (stateCb state)
 
 alterState :: WSNotification -> (State -> State) -> Effect Unit
 alterState (WSNotification ws') stateCb = do
-  let state = ws' .. "state"
-  -- here.log2 "[alterState] state" state
-  void $ pure $ (ws' .= "state") (stateCb state)
+  Ref.modify_ stateCb ws'.state
+    
 
-allSubscriptions :: WSNotification -> Array (Tuple Topic (Tuple UUID Callback))
-allSubscriptions (WSNotification ws') =
+allSubscriptions :: State -> Array (Tuple Topic (Tuple UUID Callback))
+allSubscriptions (State { callbacks }) =
   foldMapWithIndex outerFold callbacks
   where
-    State { callbacks } = ws' .. "state"
     innerFold :: Topic -> UUID -> Callback -> Array (Tuple Topic (Tuple UUID Callback))
     innerFold topic uuid cb = [Tuple topic (Tuple uuid cb)]
     outerFold :: Topic -> CallbacksHM -> Array (Tuple Topic (Tuple UUID Callback))
     outerFold topic uuidCb = foldMapWithIndex (innerFold topic) uuidCb
 
+allSubscriptionsWS :: WSNotification -> Effect (Array (Tuple Topic (Tuple UUID Callback)))
+allSubscriptionsWS (WSNotification ws') = do
+  state <- Ref.read ws'.state
+  pure $ allSubscriptions state
+
 data Action =
     InsertCallback Topic UUID Callback
   | RemoveCallback Topic UUID
@@ -203,52 +231,99 @@ performAction (WSNotification ws') (Call topic) = do
   here.log2 "[performAction Call] state" state
   callTopic state topic
 
+
+connect :: WSNotification -> String -> (Maybe Session) -> Effect Unit
+connect ws@(WSNotification ws') url session = do
+  mConn <- Ref.read ws'.connection
+  case mConn of
+    Just _conn -> pure unit
+    Nothing -> do
+      connection@(WS.Connection conn) <- WS.newWebSocket (WS.URL url) []
+      Ref.write (Just connection) ws'.connection
+      
+      let onmessage me = do
+            -- WARNING mutable state
+            s <- runExceptT $ F.readString (ME.data_ me)
+            case s of
+              Left err -> do
+                here.log2 "[connect] data received is not a string - was expecting a JSON string!" err
+              Right s' -> do
+                let parsed = JSON.readJSON s' :: JSON.E Notification
+                case parsed of
+                  Left err -> do
+                    here.log2 "[connect] Can't parse message" err
+                  Right (Notification topic) -> do
+                    here.log2 "[connect] notification" topic
+                    performAction ws (Call topic)
+                  -- Right parsed' -> do
+                  --   here.log2 "[connect] onmessage, F.readString" parsed'
+                
+      conn.onopen $= (\_ -> do
+                         -- authorize user first
+                         here.log2 "[connect] session" session
+                         case session of
+                           Just (Session { token }) ->
+                             send ws $ WSAuthorize token
+                           Nothing -> pure unit
+                         -- send pending subscriptions
+                         allSubs <- allSubscriptionsWS ws
+                         void $ for allSubs $ \(Tuple topic _) -> do
+                           let subscription = WSSubscribe topic
+                           here.log2 "[connect] pending subscription" subscription
+                           send ws subscription)
+      conn.onmessage $= onmessage
+
+      pure unit
+
 -- | NOTE This is very ugly. WebSockets events are asychronous. We
 -- | should wrap this similar as is done in 0MQ/nanomsg: make an
 -- | infinite loop and just wait for an incoming message. This should
 -- | reduce (to zero?) the need for SettableVar's etc (which is not
 -- | the functional way to do things).
-connect :: WSNotification -> String -> (Maybe Session) -> Effect WSNotification
-connect ws@(WSNotification { connection: Just _conn }) _ _ = pure ws
-connect (WSNotification ws') url session = do
-  connection@(WS.Connection conn) <- WS.newWebSocket (WS.URL url) []
-  let ws = WSNotification $ ws' { connection = Just connection } 
-  conn.onopen $= (\_ -> do
-                     -- authorize user first
-                     here.log2 "[connect] session" session
-                     case session of
-                       Just (Session { token }) ->
-                         send ws $ WSAuthorize token
-                       Nothing -> pure unit
-                     -- send pending subscriptions
-                     void $ for (allSubscriptions ws) $ \(Tuple topic _) -> do
-                       let subscription = WSSubscribe topic
-                       here.log2 "[connect] pending subscription" subscription
-                       send ws subscription)
-  conn.onmessage $= (onmessage ws)
-
-  pure ws
-
-  where
-    onmessage ws me = do
-      -- WARNING mutable state
-      s <- runExceptT $ F.readString (ME.data_ me)
-      case s of
-        Left err -> do
-          here.log2 "[connect] data received is not a string - was expecting a JSON string!" err
-        Right s' -> do
-          let parsed = JSON.readJSON s' :: JSON.E Notification
-          case parsed of
-            Left err -> do
-              here.log2 "[connect] Can't parse message" err
-            Right (Notification topic) -> do
-              here.log2 "[connect] notification" topic
-              performAction ws (Call topic)
-            Right parsed' -> do
-              here.log2 "[connect] onmessage, F.readString" parsed'
+-- connect' :: WSNotification -> String -> (Maybe Session) -> Effect WSNotification
+-- connect' ws@(WSNotification { connection: Just _conn }) _ _ = pure ws
+-- connect' (WSNotification ws') url session = do
+--   connection@(WS.Connection conn) <- WS.newWebSocket (WS.URL url) []
+--   let ws = WSNotification $ ws' { connection = Just connection } 
+--   conn.onopen $= (\_ -> do
+--                      -- authorize user first
+--                      here.log2 "[connect] session" session
+--                      case session of
+--                        Just (Session { token }) ->
+--                          send ws $ WSAuthorize token
+--                        Nothing -> pure unit
+--                      -- send pending subscriptions
+--                      void $ for (allSubscriptions ws) $ \(Tuple topic _) -> do
+--                        let subscription = WSSubscribe topic
+--                        here.log2 "[connect] pending subscription" subscription
+--                        send ws subscription)
+--   conn.onmessage $= (onmessage ws)
+
+--   pure ws
+
+--   where
+--     onmessage ws me = do
+--       -- WARNING mutable state
+--       s <- runExceptT $ F.readString (ME.data_ me)
+--       case s of
+--         Left err -> do
+--           here.log2 "[connect] data received is not a string - was expecting a JSON string!" err
+--         Right s' -> do
+--           let parsed = JSON.readJSON s' :: JSON.E Notification
+--           case parsed of
+--             Left err -> do
+--               here.log2 "[connect] Can't parse message" err
+--             Right (Notification topic) -> do
+--               here.log2 "[connect] notification" topic
+--               performAction ws (Call topic)
+--             Right parsed' -> do
+--               here.log2 "[connect] onmessage, F.readString" parsed'
 
 mkWSNotification :: String -> Effect WSNotification
-mkWSNotification url = connect emptyWSNotification url Nothing
+mkWSNotification url = do
+  ws <- emptyWSNotification
+  connect ws url Nothing
+  pure ws
 
 -- TODO useLoaderWithUpdate { loader, path, render, topic }
 -- topic can be used to bind with a reload state for a component.