@@ -10,11 +10,15 @@ import Reflex.Dom.Class
10import Reflex.Dom.WebSocket.Foreign
10import Reflex.Dom.WebSocket.Foreign
12import Control.Concurrent
12import Control.Concurrent
13+import Control.Concurrent.STM
14+import Control.Exception (catch, SomeException)
13import Control.Lens
15import Control.Lens
14import Control.Monad hiding (forM, forM_, mapM, mapM_, sequence)
16import Control.Monad hiding (forM, forM_, mapM, mapM_, sequence)
15import Control.Monad.IO.Class
17import Control.Monad.IO.Class
16import Control.Monad.Ref
18import Control.Monad.Ref
17import Control.Monad.State
19import Control.Monad.State
20+import Control.Arrow (second)
21+import Data.Maybe (isJust)
18import Data.ByteString (ByteString)
22import Data.ByteString (ByteString)
19import Data.Default
23import Data.Default
20import Data.Dependent.Map (DSum (..))
24import Data.Dependent.Map (DSum (..))
@@ -41,6 +45,8 @@ webSocket url config = do
41 currentSocketRef <- liftIO $ newIORef Nothing
45 currentSocketRef <- liftIO $ newIORef Nothing
42 --TODO: Disconnect if value no longer needed
46 --TODO: Disconnect if value no longer needed
43 (eOpen, eOpenTriggerRef) <- newEventWithTriggerRef
47 (eOpen, eOpenTriggerRef) <- newEventWithTriggerRef
48+ payloadQueue <- liftIO newTQueueIO
49+ isOpen <- liftIO newEmptyTMVarIO
44 let onMessage :: ByteString -> IO ()
50 let onMessage :: ByteString -> IO ()
45 onMessage m = postGui $ do
51 onMessage m = postGui $ do
46 mt <- readRef eRecvTriggerRef
52 mt <- readRef eRecvTriggerRef
@@ -48,21 +54,34 @@ webSocket url config = do
48 onOpen = postGui $ do
54 onOpen = postGui $ do
49 mt <- readRef eOpenTriggerRef
55 mt <- readRef eOpenTriggerRef
50 forM_ mt $ \t -> runWithActions [t :=> Identity ()]
56 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
51 start = do
64 start = do
52− ws <- liftIO $ newWebSocket wv url onMessage onOpen $ do
65+ ws <- liftIO $ newWebSocket wv url onMessage onOpen onClose
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
57 liftIO $ writeIORef currentSocketRef $ Just ws
66 liftIO $ writeIORef currentSocketRef $ Just ws
59 schedulePostBuild $ liftIO start
68 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
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
61 mws <- liftIO $ readIORef currentSocketRef
78 mws <- liftIO $ readIORef currentSocketRef
62− case mws of
79+ succ <- 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?
80+ Nothing -> return False
65− liftIO $ webSocketSend ws payload
81+ Just ws -> liftIO ((webSocketSend ws payload >> return True)
83+ (\(_ :: SomeException) -> return False))
84+ unless succ $ atomically $ unGetTQueue payloadQueue payload
66 return $ WebSocket eRecv eOpen
85 return $ WebSocket eRecv eOpen
68makeLensesWith (lensRules & simpleLenses .~ True) ''WebSocketConfig
87makeLensesWith (lensRules & simpleLenses .~ True) ''WebSocketConfig