@@ -10,11 +10,15 @@ import Reflex.Dom.Class
1010import Reflex.Dom.WebSocket.Foreign
1111
1212import Control.Concurrent
13+import Control.Concurrent.STM
14+import Control.Exception (catch, SomeException)
1315import Control.Lens
1416import Control.Monad hiding (forM, forM_, mapM, mapM_, sequence)
1517import Control.Monad.IO.Class
1618import Control.Monad.Ref
1719import Control.Monad.State
20+import Control.Arrow (second)
21+import Data.Maybe (isJust)
1822import Data.ByteString (ByteString)
1923import Data.Default
2024import Data.Dependent.Map (DSum (..))
@@ -41,6 +45,8 @@ webSocket url config = do
4145 currentSocketRef <- liftIO $ newIORef Nothing
4246 --TODO: Disconnect if value no longer needed
4347 (eOpen, eOpenTriggerRef) <- newEventWithTriggerRef
48+ payloadQueue <- liftIO newTQueueIO
49+ isOpen <- liftIO newEmptyTMVarIO
4450 let onMessage :: ByteString -> IO ()
4551 onMessage m = postGui $ do
4652 mt <- readRef eRecvTriggerRef
@@ -48,21 +54,34 @@ webSocket url config = do
4854 onOpen = postGui $ do
4955 mt <- readRef eOpenTriggerRef
5056 forM_ mt $ \t -> runWithActions [t :=> Identity ()]
57+ liftIO $ void $ atomically $ tryPutTMVar isOpen ()
58+ --TODO: Is the fork necessary, or do event handlers run in their own threads automatically?
59+ onClose = void $ forkIO $ do
60+ liftIO $ atomically $ tryTakeTMVar isOpen
61+ liftIO $ writeIORef currentSocketRef Nothing
62+ liftIO $ threadDelay 1000000
63+ start
5164 start = do
52− ws <- liftIO $ newWebSocket wv url onMessage onOpen $ do
53− void $ forkIO $ do --TODO: Is the fork necessary, or do event handlers run in their own threads automatically?
54− liftIO $ writeIORef currentSocketRef Nothing
55− liftIO $ threadDelay 1000000
56− start
65+ ws <- liftIO $ newWebSocket wv url onMessage onOpen onClose
5766 liftIO $ writeIORef currentSocketRef $ Just ws
5867 return ()
5968 schedulePostBuild $ liftIO start
60− performEvent_ $ ffor (_webSocketConfig_send config) $ \payloads -> forM_ payloads $ \payload -> do
69+ performEvent_ $ ffor (_webSocketConfig_send config) $ \payloads -> forM_ payloads $ \payload ->
70+ liftIO $ atomically $ writeTQueue payloadQueue payload
71+
72+ liftIO $ forkIO $ forever $ do
73+ payload <- atomically $ do
74+ pl <- readTQueue payloadQueue
75+ open <- tryReadTMVar isOpen
76+ if isJust open then return pl else retry
77+
6178 mws <- liftIO $ readIORef currentSocketRef
62− case mws of
63− Nothing -> return () -- Discard --TODO: should we do something better here? probably buffer it, since we handle reconnection logic; how do we verify that the server has received things?
64− Just ws -> do
65− liftIO $ webSocketSend ws payload
79+ succ <- case mws of
80+ Nothing -> return False
81+ Just ws -> liftIO ((webSocketSend ws payload >> return True)
82+ `catch`
83+ (\(_ :: SomeException) -> return False))
84+ unless succ $ atomically $ unGetTQueue payloadQueue payload
6685 return $ WebSocket eRecv eOpen
6786
6887makeLensesWith (lensRules & simpleLenses .~ True) ''WebSocketConfig