websockets-simple 0.0.6.3 → 0.0.7
raw patch · 6 files changed
+245/−142 lines, 6 filesdep +profunctorsdep ~base
Dependencies added: profunctors
Dependency ranges changed: base
Files
- README.md +45/−0
- src/Network/WebSockets/Simple.hs +100/−78
- src/Network/WebSockets/Simple/PingPong.hs +7/−5
- src/Test/WebSockets/Simple.hs +6/−10
- test/Spec.hs +11/−4
- websockets-simple.cabal +76/−45
README.md view
@@ -1,1 +1,46 @@ # websockets-simple++Provides for a slightly more composable structure for websocket apps:++```haskell+data Input = Increment | Decrement+ deriving (FromJSON)++data Output = Value Int+ deriving (ToJSON)++myApp :: MonadBaseControl IO m => m (WebSocketsApp m Input Output)+myApp = do+ countRef <- liftIO $ newIORef 0+ emitter <- liftIO $ newIORef (Nothing :: Async ())++ let killEmitter = do+ mThread <- liftIO $ readIORef emitter+ case mThread of+ Nothing -> pure ()+ Just thread -> cancel thread++ pure WebSocketsApp+ { onOpen = \WebSocketsAppParams{send} ->+ liftBaseWith $ \runInBase -> do+ thread <- async $ forever $ do+ count <- readIORef countRef+ runInBase $ send $ Value count+ threadDelay 1000000 -- every second, emit the current value+ writeIORef emitter (Just thread)+ , onReceive = \WebSocketsAppParams{send,close} x -> do+ count <- liftIO $+ ( modifyIORef countRef $ case x of+ Increment -> (+ 1)+ Decrement -> (-) 1+ ) *> readIORef countRef+ if count >= 10 || count <= -10+ then close+ else send (Value count)+ , onClose = \mReason -> do+ killEmitter+ case mReason of+ Nothing -> liftIO $ writeIORef countRef 0+ Just _ -> pure ()+ }+```
src/Network/WebSockets/Simple.hs view
@@ -5,51 +5,68 @@ , ScopedTypeVariables , NamedFieldPuns , FlexibleContexts+ , InstanceSigs #-} -module Network.WebSockets.Simple where+module Network.WebSockets.Simple+ ( -- * Types+ WebSocketsApp (..), WebSocketsAppParams (..), WebSocketsAppThreads (..)+ , Network.WebSockets.ConnectionException (..), WebSocketsSimpleError (..)+ , -- * Running+ toClientAppT, toClientAppT', toServerAppT+ , -- * Utilities+ expBackoffStrategy+ , hoistWebSocketsApp+ ) where -import Network.WebSockets (DataMessage (..), sendTextData, sendClose, receiveDataMessage, acceptRequest, ConnectionException (CloseRequest))+import Network.WebSockets (DataMessage (..), sendTextData, sendClose, receiveDataMessage, acceptRequest, ConnectionException (..)) import Network.Wai.Trans (ServerAppT, ClientAppT)+import Data.Profunctor (Profunctor (..)) import Data.Aeson (ToJSON (..), FromJSON (..)) import qualified Data.Aeson as Aeson import Data.ByteString.Lazy (ByteString)-import Data.IORef (newIORef, writeIORef, readIORef)-import Data.Word (Word16) import Control.Monad (void, forever)+import Control.Monad.IO.Class (MonadIO (..)) import Control.Monad.Catch (Exception, throwM, MonadThrow, catch, MonadCatch) import Control.Monad.Trans.Control (MonadBaseControl (..)) import Control.Concurrent (threadDelay)-import Control.Concurrent.Async (Async, async)-import Control.Concurrent.STM (atomically)-import Control.Concurrent.STM.TChan (TChan, newTChanIO, writeTChan)+import Control.Concurrent.Async (Async, async, link)+import Control.Concurrent.STM (atomically, newTVarIO, readTVarIO, writeTVar) import GHC.Generics (Generic) import Data.Typeable (Typeable) -data WebSocketsAppParams send m = WebSocketsAppParams+data WebSocketsAppParams m send = WebSocketsAppParams { send :: send -> m () , close :: m () } deriving (Generic, Typeable) -data WebSocketsApp send receive m = WebSocketsApp- { onOpen :: WebSocketsAppParams send m -> m ()- , onReceive :: WebSocketsAppParams send m -> receive -> m ()- , onClose :: Maybe (Word16, ByteString) -> m () -- ^ Either was a clean close, with 'Network.WebSockets.CloseRequest' params, or was unclean.- -- Note that to implement backoff strategies, you should catch your 'Network.WebSockets.ConnectionException'- -- /outside/ this simple app, and only after you've 'Network.WebSockets.runClient' or server, because the- -- 'Network.WebSockets.Connection' will be different.+data WebSocketsApp m receive send = WebSocketsApp+ { onOpen :: WebSocketsAppParams m send -> m ()+ , onReceive :: WebSocketsAppParams m send -> receive -> m ()+ , onClose :: ConnectionException -> m () -- ^ Should be re-entrant; this exception is caught in all uses of 'send', even if used in a dead 'Network.WebSockets.Connection' in a lingering thread. } deriving (Generic, Typeable) +instance Profunctor (WebSocketsApp m) where+ dimap :: forall a b c d. (a -> b) -> (c -> d) -> WebSocketsApp m b c -> WebSocketsApp m a d+ dimap receiveF sendF WebSocketsApp{onOpen,onReceive,onClose} = WebSocketsApp+ { onOpen = \params -> onOpen (getParams params)+ , onReceive = \params r -> onReceive (getParams params) (receiveF r)+ , onClose = onClose+ }+ where+ getParams :: WebSocketsAppParams m d -> WebSocketsAppParams m c+ getParams WebSocketsAppParams{send,close} = WebSocketsAppParams{send = send . sendF,close} + hoistWebSocketsApp :: (forall a. m a -> n a) -> (forall a. n a -> m a)- -> WebSocketsApp send receive m- -> WebSocketsApp send receive n+ -> WebSocketsApp m receive send+ -> WebSocketsApp n receive send hoistWebSocketsApp f coF WebSocketsApp{onOpen,onReceive,onClose} = WebSocketsApp { onOpen = \WebSocketsAppParams{send,close} -> f $ onOpen WebSocketsAppParams{send = coF . send, close = coF close} , onReceive = \WebSocketsAppParams{send,close} r -> f $ onReceive WebSocketsAppParams{send = coF . send, close = coF close} r@@ -57,67 +74,79 @@ } --- | This can throw a 'WebSocketSimpleError' when json parsing fails. However, do note:--- the 'onOpen' is called once, but is still forked when called. Likewise, the 'onReceive'--- function is called /every time/ a (parsable) response is received from the other party,--- and is forked on every invocation.++instance Applicative m => Monoid (WebSocketsApp m receive send) where+ mempty = WebSocketsApp+ { onOpen = \_ -> pure ()+ , onReceive = \_ _ -> pure ()+ , onClose = \_ -> pure ()+ }+ mappend x y = WebSocketsApp+ { onOpen = \params -> onOpen x params *> onOpen y params+ , onReceive = \params r -> onReceive x params r *> onReceive y params r+ , onClose = \mE -> onClose x mE *> onClose y mE+ }+++-- | This can throw a 'WebSocketSimpleError' to the main thread via 'Control.Concurrent.Async.link' when json parsing fails. toClientAppT :: forall send receive m . ( ToJSON send , FromJSON receive+ , MonadIO m , MonadBaseControl IO m , MonadThrow m , MonadCatch m )- => WebSocketsApp send receive m- -> ClientAppT m (Maybe WebSocketsAppThreads)+ => WebSocketsApp m receive send+ -> ClientAppT m WebSocketsAppThreads toClientAppT WebSocketsApp{onOpen,onReceive,onClose} conn = do- let go = do- let send :: send -> m ()- send x = liftBaseWith $ \_ -> sendTextData conn (Aeson.encode x)-- close :: m ()- close = liftBaseWith $ \_ -> sendClose conn (Aeson.encode "requesting close")+ let send :: send -> m ()+ send x = liftIO (sendTextData conn (Aeson.encode x)) `catch` onClose - params :: WebSocketsAppParams send m- params = WebSocketsAppParams{send,close}+ close :: m ()+ close = liftIO (sendClose conn (Aeson.encode "requesting close")) `catch` onClose - onOpenThread <- liftBaseWith $ \runToBase ->- async $ void $ runToBase $ onOpen params+ params :: WebSocketsAppParams m send+ params = WebSocketsAppParams{send,close} - onReceiveThreads <- liftBaseWith $ \_ -> newTChanIO+ onOpen params - void $ forever $ do- data' <- liftBaseWith $ \_ -> receiveDataMessage conn+ receivingThread <- liftBaseWith $ \runInBase -> async $ forever $+ let go' = do+ data' <- receiveDataMessage conn let data'' = case data' of Text xs _ -> xs Binary xs -> xs case Aeson.decode data'' of Nothing -> throwM (JSONParseError data'')- Just received -> liftBaseWith $ \runToBase -> do- thread <- async $ void $ runToBase $ onReceive params received- atomically $ writeTChan onReceiveThreads thread-- pure $ Just WebSocketsAppThreads- { onOpenThread- , onReceiveThreads- }+ Just received -> runInBase (onReceive params received)+ in go' `catch` (runInBase . onClose) - onDisconnect :: ConnectionException -> m (Maybe WebSocketsAppThreads)- onDisconnect err = do- case err of- CloseRequest code reason -> onClose (Just (code,reason))- _ -> onClose Nothing- pure Nothing+ liftIO (link receivingThread) - go `catch` onDisconnect+ pure $ WebSocketsAppThreads+ { wsAppReceivingThread = receivingThread+ } -toClientAppT' :: (ToJSON send, FromJSON receive, MonadBaseControl IO m, MonadThrow m, MonadCatch m) => WebSocketsApp send receive m -> ClientAppT m ()+toClientAppT' :: ( ToJSON send+ , FromJSON receive+ , MonadIO m+ , MonadBaseControl IO m+ , MonadThrow m+ , MonadCatch m+ ) => WebSocketsApp m receive send -> ClientAppT m () toClientAppT' wsApp conn = void (toClientAppT wsApp conn) -toServerAppT :: (ToJSON send, FromJSON receive, MonadBaseControl IO m, MonadThrow m, MonadCatch m) => WebSocketsApp send receive m -> ServerAppT m+toServerAppT :: ( ToJSON send+ , FromJSON receive+ , MonadIO m+ , MonadBaseControl IO m+ , MonadThrow m+ , MonadCatch m+ ) => WebSocketsApp m receive send -> ServerAppT m toServerAppT wsApp pending = do conn <- liftBaseWith $ \_ -> acceptRequest pending toClientAppT' wsApp conn@@ -127,41 +156,34 @@ -- | A simple backoff strategy, which (per second), will increasingly delay at @2^soFar@, until @soFar >= 5minutes@, where it will then routinely poll every -- 5 minutes. expBackoffStrategy :: forall m a- . ( MonadBaseControl IO m- , MonadCatch m+ . ( MonadIO m )- => m a -- ^ The run app- -> m a-expBackoffStrategy app = do- soFarVar <- liftBaseWith $ \_ -> newIORef (0 :: Int)+ => m a -- ^ Action to call, like pinging a scoped channel to trigger the reconnect+ -> m (ConnectionException -> m a)+expBackoffStrategy action = do+ soFarVar <- liftIO $ newTVarIO (0 :: Int) let second = 1000000 - let go = app `catch` backoffStrat-- backoffStrat :: ConnectionException -> m a- backoffStrat _ = do- liftBaseWith $ \_ -> do- soFar <- readIORef soFarVar- let delay- | soFar >= 5 * 60 = 5 * 60- | otherwise = 2 ^ soFar- writeIORef soFarVar (soFar + delay)- threadDelay (delay * second)- go-- go+ pure $ \_ -> do+ liftIO $ do+ soFar <- readTVarIO soFarVar+ let delay+ | soFar >= 5 * 60 = 5 * 60+ | otherwise = 2 ^ soFar+ atomically $ writeTVar soFarVar (soFar + delay)+ threadDelay (delay * second)+ action -data WebSocketSimpleError+data WebSocketsSimpleError = JSONParseError ByteString deriving (Generic, Eq, Show) -instance Exception WebSocketSimpleError+instance Exception WebSocketsSimpleError -data WebSocketsAppThreads = WebSocketsAppThreads- { onOpenThread :: Async ()- , onReceiveThreads :: TChan (Async ())+newtype WebSocketsAppThreads = WebSocketsAppThreads+ { wsAppReceivingThread :: Async () }
src/Network/WebSockets/Simple/PingPong.hs view
@@ -11,10 +11,10 @@ import Data.Aeson.Types (Value (Array)) import Control.Monad.Trans.Control (MonadBaseControl (..)) import Control.Concurrent.Async.Every (every, reset)-import Control.Concurrent.STM (atomically)-import Control.Concurrent.STM.TVar (newTVarIO, readTVar, writeTVar)+import Control.Concurrent.STM (atomically, newTVarIO, readTVar, writeTVar) +-- | Uses the JSON literal @[]@ as the ping message newtype PingPong a = PingPong {getPingPong :: Maybe a} -- | Assumes @a@ isn't an 'Data.Aeson.Types.Array' of anything@@ -24,7 +24,9 @@ -- | Assumes @a@ isn't an 'Data.Aeson.Types.Array' of anything instance FromJSON a => FromJSON (PingPong a) where- parseJSON (Array _) = pure (PingPong Nothing)+ parseJSON x@(Array xs)+ | null xs = pure (PingPong Nothing)+ | otherwise = (PingPong . Just) <$> parseJSON x parseJSON x = (PingPong . Just) <$> parseJSON x @@ -32,8 +34,8 @@ pingPong :: ( MonadBaseControl IO m ) => Int -- ^ Delay in microseconds- -> WebSocketsApp send receive m- -> m (WebSocketsApp (PingPong send) (PingPong receive) m)+ -> WebSocketsApp m receive send+ -> m (WebSocketsApp m (PingPong receive) (PingPong send)) pingPong delay WebSocketsApp{onOpen,onReceive,onClose} = do counterVar <- liftBaseWith $ \_ -> newTVarIO Nothing
src/Test/WebSockets/Simple.hs view
@@ -7,15 +7,13 @@ module Test.WebSockets.Simple where --import Network.WebSockets.Simple (WebSocketsApp (..), WebSocketsAppParams (..))+import Network.WebSockets.Simple (WebSocketsApp (..), WebSocketsAppParams (..), ConnectionException (..)) import Control.Monad (forever, void) import Control.Monad.IO.Class (MonadIO (..)) import Control.Monad.Trans.Control (MonadBaseControl (..))-import Control.Concurrent (threadDelay) import Control.Concurrent.Async (Async, async) import Control.Concurrent.STM (atomically)-import Control.Concurrent.STM.TChan (TChan, newTChan, writeTChan, readTChan, tryReadTChan, isEmptyTChan)+import Control.Concurrent.STM.TChan (TChan, newTChan, writeTChan, readTChan) @@ -24,8 +22,8 @@ . ( MonadIO m , MonadBaseControl IO m )- => WebSocketsApp send receive m- -> WebSocketsApp receive send m+ => WebSocketsApp m receive send+ -> WebSocketsApp m send receive -> m (Async (), Async (), TChan send, TChan receive) runConnected sendsSreceivesR sendsRreceivesS = do (sendChan, receiveChan) <- liftIO $ atomically $ (,) <$> newTChan <*> newTChan@@ -38,11 +36,10 @@ close :: m () close = do- onClose sendsRreceivesS Nothing- onClose sendsSreceivesR Nothing+ onClose sendsRreceivesS ConnectionClosed+ onClose sendsSreceivesR ConnectionClosed sToR <- liftBaseWith $ \runInBase -> async $ forever $ do- _ <- atomically $ isEmptyTChan sendChan s <- atomically $ readTChan sendChan void $ runInBase $ onReceive sendsRreceivesS WebSocketsAppParams { send = sendToReceive@@ -50,7 +47,6 @@ } s rToS <- liftBaseWith $ \runInBase -> async $ forever $ do- _ <- atomically $ isEmptyTChan receiveChan r <- atomically $ readTChan receiveChan void $ runInBase $ onReceive sendsSreceivesR WebSocketsAppParams { send = sendToSend
test/Spec.hs view
@@ -5,6 +5,7 @@ module Main where import Network.WebSockets.Simple (WebSocketsApp (..), WebSocketsAppParams (..))+import Control.Concurrent (threadDelay) import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TChan (TChan, writeTChan, newTChanIO, readTChan) import Test.Tasty (defaultMain, testGroup)@@ -13,20 +14,26 @@ import Test.WebSockets.Simple (runConnected) -testReceivingApp :: TChan Int -> WebSocketsApp Int Int IO+testReceivingApp :: TChan Int -> WebSocketsApp IO Int Int testReceivingApp receivedChan = WebSocketsApp- { onOpen = \WebSocketsAppParams{send} ->+ { onOpen = \WebSocketsAppParams{send} -> do+ putStrLn "sending in 10 seconds..."+ threadDelay $ 10^6 * 10 send 0+ putStrLn "sent." , onReceive = \_ x -> atomically $ writeTChan receivedChan x , onClose = \_ -> pure () } -testSendingApp :: WebSocketsApp Int Int IO+testSendingApp :: WebSocketsApp IO Int Int testSendingApp = WebSocketsApp { onOpen = \_ -> pure ()- , onReceive = \WebSocketsAppParams{send} x ->+ , onReceive = \WebSocketsAppParams{send} x -> do+ putStrLn "received, sending in 10 seconds..."+ threadDelay $ 10^6 * 10 send x+ putStrLn "sent." , onClose = \_ -> pure () }
websockets-simple.cabal view
@@ -1,50 +1,81 @@-name: websockets-simple-version: 0.0.6.3-synopsis: Simpler interface to the websockets api--- description:-homepage: https://github.com/athanclark/websockets-simple#readme-license: BSD3-license-file: LICENSE-author: Athan Clark-maintainer: athan.clark@gmail.com-copyright: 2017 Athan Clark-category: Web-build-type: Simple-extra-source-files: README.md-cabal-version: >=1.10--library- hs-source-dirs: src- exposed-modules: Network.WebSockets.Simple- Network.WebSockets.Simple.PingPong- Test.WebSockets.Simple- build-depends: base >= 4.8 && < 5- , aeson- , async- , bytestring- , every >= 0.0.1- , exceptions- , monad-control- , stm- , transformers- , wai-transformers- , websockets >= 0.11- default-language: Haskell2010+-- This file has been generated from package.yaml by hpack version 0.21.2.+--+-- see: https://github.com/sol/hpack+--+-- hash: 0fd7a69684c73e2d69f006be6b9f3028e804149928f115366bc6ef246ebe1cf5 +name: websockets-simple+version: 0.0.7+synopsis: Composable websockets clients+description: See README at <https://github.com/athanclark/websockets-simple#readme>+category: Web+homepage: https://github.com/athanclark/websockets-simple#readme+bug-reports: https://github.com/athanclark/websockets-simple/issues+maintainer: Athan Clark <athan.clark@gmail.com>+license: BSD3+license-file: LICENSE+build-type: Simple+cabal-version: >= 1.10 -test-suite test- type: exitcode-stdio-1.0- hs-source-dirs: test- main-is: Spec.hs- build-depends: base- , websockets-simple- , stm- , tasty- , tasty-hspec- , hspec- ghc-options: -threaded -rtsopts -with-rtsopts=-N- default-language: Haskell2010+extra-source-files:+ README.md source-repository head- type: git+ type: git location: https://github.com/athanclark/websockets-simple++library+ exposed-modules:+ Network.WebSockets.Simple+ Network.WebSockets.Simple.PingPong+ Test.WebSockets.Simple+ other-modules:+ Paths_websockets_simple+ hs-source-dirs:+ src+ ghc-options: -Wall+ build-depends:+ aeson+ , async+ , base >=4.9 && <5+ , bytestring+ , every >=0.0.1+ , exceptions+ , monad-control+ , profunctors+ , stm+ , transformers+ , wai-transformers+ , websockets >=0.11+ default-language: Haskell2010++test-suite spec+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ Network.WebSockets.Simple+ Network.WebSockets.Simple.PingPong+ Test.WebSockets.Simple+ Paths_websockets_simple+ hs-source-dirs:+ test+ src+ ghc-options: -Wall+ build-depends:+ aeson+ , async+ , base >=4.9 && <5+ , bytestring+ , every >=0.0.1+ , exceptions+ , hspec+ , monad-control+ , profunctors+ , stm+ , tasty+ , tasty-hspec+ , transformers+ , wai-transformers+ , websockets >=0.11+ , websockets-simple+ default-language: Haskell2010