postgres-websockets 0.10.0.0 → 0.11.1.0

raw patch · 11 files changed

+466/−364 lines, 11 filesdep +warp-tlsdep −asyncdep −contravariantdep −hspec-waidep ~aesondep ~base64-bytestringdep ~hasql-notificationsPVP ok

version bump matches the API change (PVP)

Dependencies added: warp-tls

Dependencies removed: async, contravariant, hspec-wai, hspec-wai-json, stringsearch

Dependency ranges changed: aeson, base64-bytestring, hasql-notifications, lens, network, postgresql-libpq, protolude, time, wai-extra

API changes (from Hackage documentation)

Files

• postgres-websockets.cabal +17/−22
• src/PostgresWebsockets/Broadcast.hs +113/−59
• src/PostgresWebsockets/Claims.hs +8/−7
• src/PostgresWebsockets/Config.hs +83/−66
• src/PostgresWebsockets/Context.hs +28/−22
• src/PostgresWebsockets/HasqlBroadcast.hs +90/−66
• src/PostgresWebsockets/Middleware.hs +9/−9
• src/PostgresWebsockets/Server.hs +24/−23
• test/ClaimsSpec.hs +5/−5
• test/HasqlBroadcastSpec.hs +2/−2
• test/ServerSpec.hs +87/−83

@@ -1,7 +1,7 @@ name:                postgres-websockets-version:             0.10.0.0+version:             0.11.1.0 synopsis:            Middleware to map LISTEN/NOTIFY messages to Websockets-description:         Please see README.md+description:         WAI middleware that adds websockets capabilites on top of PostgreSQL's asynchronous notifications using LISTEN and NOTIFY commands. Fully functioning server included. homepage:            https://github.com/diogob/postgres-websockets#readme license:             BSD3 license-file:        LICENSE@@ -38,25 +38,22 @@                      , lens >= 4.17.1                      , jose >= 0.6                      , unordered-containers >= 0.2-                     , postgresql-libpq >= 0.9 && < 1.0-                     , aeson >= 1.4.6.0 && < 1.5-                     , protolude >= 0.2.3 && < 0.3+                     , aeson >= 1.4.6.0 && < 1.6+                     , protolude >= 0.2.3 && < 0.4                      , hasql >= 1.4.1-                     , hasql-notifications >= 0.1.0.0 && < 0.2+                     , hasql-notifications >= 0.1.0.0 && < 0.3                      , either >= 5.0.1.1 && < 5.1                      , stm-containers >= 1.1.0.2 && < 1.2                      , stm >= 2.5.0.0 && < 2.6                      , retry >= 0.8.1.0 && < 0.9-                     , stringsearch >= 0.3.6.6 && < 0.4-                     , time >= 1.8.0.2 && < 1.9-                     , contravariant >= 1.5.2 && < 1.6+                     , time >= 1.8.0.2 && < 1.10                      , alarmclock >= 0.7.0.2 && < 0.8-                     , async >= 2.2.0 && < 2.3                      , envparse >= 0.4.1-                     , base64-bytestring >= 1.0.0.3 && < 1.1+                     , base64-bytestring >= 1.0.0.3 && < 1.2                      , bytestring >= 0.10                      , warp >= 3.2 && < 4-                     , wai-extra >= 3.0.29 && < 3.1+                     , warp-tls >= 3.2 && < 4+                     , wai-extra >= 3.0.29 && < 3.2                      , wai-app-static >= 3.1.7.1 && < 3.2                      , auto-update >= 0.1.6 && < 0.2 @@ -70,7 +67,7 @@   ghc-options:         -Wall -threaded -rtsopts -with-rtsopts=-N   build-depends:       base >= 4.7 && < 5                      , postgres-websockets-                     , protolude >= 0.2.3 && < 0.3+                     , protolude >= 0.2.3 && < 0.4   default-language:    Haskell2010   default-extensions: OverloadedStrings, NoImplicitPrelude, QuasiQuotes @@ -83,23 +80,21 @@                      , HasqlBroadcastSpec                      , ServerSpec   build-depends:       base-                     , protolude >= 0.2.3 && < 0.3+                     , protolude >= 0.2.3 && < 0.4                      , postgres-websockets                      , hspec >= 2.7.1 && < 2.8-                     , hspec-wai >= 0.9.2 && < 0.10-                     , hspec-wai-json >= 0.9.2 && < 0.10-                     , aeson >= 1.4.6.0 && < 1.5+                     , aeson >= 1.4.6.0 && < 1.6                      , hasql >= 0.19                      , hasql-pool >= 0.4-                     , hasql-notifications >= 0.1.0.0 && < 0.2+                     , hasql-notifications >= 0.1.0.0 && < 0.3                      , http-types >= 0.9-                     , time >= 1.8.0.2 && < 1.9+                     , time >= 1.8.0.2 && < 1.10                      , unordered-containers >= 0.2-                     , wai-extra >= 3.0.29 && < 3.1+                     , wai-extra >= 3.0.29 && < 3.2                      , stm >= 2.5.0.0 && < 2.6                      , websockets >= 0.12.7.0 && < 0.13-                     , network >= 2.8.0.1 && < 2.9-                     , lens >= 4.17.1 && < 4.18+                     , network >= 2.8.0.1 && < 3.2+                     , lens >= 4.17.1 && < 4.20                      , lens-aeson   ghc-options:         -Wall -threaded -rtsopts -with-rtsopts=-N   default-language:    Haskell2010

@@ -1,52 +1,77 @@-{-| -Module      : PostgresWebsockets.Broadcast-Description : Distribute messages from one producer to several consumers.+{-# LANGUAGE DeriveGeneric #-} -PostgresWebsockets functions to broadcast messages to several listening clients-This module provides a type called Multiplexer.-The multiplexer contains a map of channels and a producer thread.+-- |+-- Module      : PostgresWebsockets.Broadcast+-- Description : Distribute messages from one producer to several consumers.+--+-- PostgresWebsockets functions to broadcast messages to several listening clients+-- This module provides a type called Multiplexer.+-- The multiplexer contains a map of channels and a producer thread.+--+-- This module avoids any database implementation details, it is used by HasqlBroadcast where+-- the database logic is combined.+module PostgresWebsockets.Broadcast+  ( Multiplexer,+    Message (..),+    newMultiplexer,+    onMessage,+    relayMessages,+    relayMessagesForever,+    superviseMultiplexer, -This module avoids any database implementation details, it is used by HasqlBroadcast where-the database logic is combined.--}-module PostgresWebsockets.Broadcast ( Multiplexer (src)-                             , Message (..)-                             , newMultiplexer-                             , onMessage-                             , relayMessages-                             , relayMessagesForever-                             -- * Re-exports-                             , readTQueue-                             , writeTQueue-                             , readTChan-                             ) where+    -- * Re-exports+    readTQueue,+    writeTQueue,+    readTChan,+  )+where -import Protolude-import qualified StmContainers.Map as M import Control.Concurrent.STM.TChan import Control.Concurrent.STM.TQueue+import qualified Data.Aeson as A+import Protolude hiding (toS)+import Protolude.Conv (toS)+import qualified StmContainers.Map as M -import GHC.Show+data Message = Message+  { channel :: Text,+    payload :: Text+  }+  deriving (Eq, Show) -data Message = Message { channel :: ByteString-                       , payload :: ByteString-                       } deriving (Eq, Show)+data Multiplexer = Multiplexer+  { channels :: M.Map Text Channel,+    messages :: TQueue Message,+    producerThreadId :: MVar ThreadId,+    reopenProducer :: IO ThreadId+  } -data Multiplexer = Multiplexer { channels :: M.Map ByteString Channel-                               , src :: ThreadId-                               , messages :: TQueue Message-                               }+data MultiplexerSnapshot = MultiplexerSnapshot+  { channelsSize :: Int,+    messageQueueEmpty :: Bool,+    producerId :: Text+  }+  deriving (Generic) -instance Show Multiplexer where-  show Multiplexer{} = "Multiplexer"+data Channel = Channel+  { broadcast :: TChan Message,+    listeners :: Integer+  } -data Channel = Channel { broadcast :: TChan Message-                       , listeners :: Integer-                       }+instance A.ToJSON MultiplexerSnapshot +-- | Given a multiplexer derive a type that can be printed for debugging or logging purposes+takeSnapshot :: Multiplexer -> IO MultiplexerSnapshot+takeSnapshot multi =+  MultiplexerSnapshot <$> size <*> e <*> thread+  where+    size = atomically $ M.size $ channels multi+    thread = show <$> readMVar (producerThreadId multi)+    e = atomically $ isEmptyTQueue $ messages multi+ -- | Opens a thread that relays messages from the producer thread to the channels forever relayMessagesForever :: Multiplexer -> IO ThreadId-relayMessagesForever =  forkIO . forever . relayMessages+relayMessagesForever = forkIO . forever . relayMessages  -- | Reads the messages from the producer and relays them to the active listeners in their respective channels. relayMessages :: Multiplexer -> IO ()@@ -58,30 +83,59 @@       Nothing -> return ()       Just c -> writeTChan (broadcast c) m -newMultiplexer :: (TQueue Message -> IO a)-               -> (Either SomeException a -> IO ())-               -> IO Multiplexer+newMultiplexer ::+  (TQueue Message -> IO a) ->+  (Either SomeException a -> IO ()) ->+  IO Multiplexer newMultiplexer openProducer closeProducer = do   msgs <- newTQueueIO-  m <- liftA2 Multiplexer M.newIO (forkFinally (openProducer msgs) closeProducer)-  return $ m msgs+  let forkNewProducer = forkFinally (openProducer msgs) closeProducer+  tid <- forkNewProducer+  multiplexerMap <- M.newIO+  producerThreadId <- newMVar tid+  pure $ Multiplexer multiplexerMap msgs producerThreadId forkNewProducer -openChannel ::  Multiplexer -> ByteString -> STM Channel+-- |  Given a multiplexer, a number of milliseconds and an IO computation that returns a boolean+--      Runs the IO computation at every interval of milliseconds interval and reopens the multiplexer producer+--      if the resulting boolean is true+--      When interval is 0 this is NOOP, so the minimum interval is 1ms+--      Call this in case you want to ensure the producer thread is killed and restarted under a certain condition+superviseMultiplexer :: Multiplexer -> Int -> IO Bool -> IO ()+superviseMultiplexer multi msInterval shouldRestart = do+  void $+    forkIO $+      forever $ do+        threadDelay $ msInterval * 1000+        sr <- shouldRestart+        when sr $ do+          snapBefore <- takeSnapshot multi+          void $ killThread <$> readMVar (producerThreadId multi)+          new <- reopenProducer multi+          void $ swapMVar (producerThreadId multi) new+          snapAfter <- takeSnapshot multi+          putStrLn $+            "Restarting producer. Multiplexer updated: "+              <> A.encode snapBefore+              <> " -> "+              <> A.encode snapAfter++openChannel :: Multiplexer -> Text -> STM Channel openChannel multi chan = do-    c <- newBroadcastTChan-    let newChannel = Channel{ broadcast = c-                            , listeners = 0-                            }-    M.insert newChannel chan (channels multi)-    return newChannel+  c <- newBroadcastTChan+  let newChannel =+        Channel+          { broadcast = c,+            listeners = 0+          }+  M.insert newChannel chan (channels multi)+  return newChannel -{- |  Adds a listener to a certain multiplexer's channel.-      The listener must be a function that takes a 'TChan Message' and perform any IO action.-      All listeners run in their own thread.-      The first listener will open the channel, when a listener dies it will check if there acquire-      any others and close the channel when that's the case.--}-onMessage :: Multiplexer -> ByteString -> (Message -> IO()) -> IO ()+-- |  Adds a listener to a certain multiplexer's channel.+--      The listener must be a function that takes a 'TChan Message' and perform any IO action.+--      All listeners run in their own thread.+--      The first listener will open the channel, when a listener dies it will check if there acquire+--      any others and close the channel when that's the case.+onMessage :: Multiplexer -> Text -> (Message -> IO ()) -> IO () onMessage multi chan action = do   listener <- atomically $ openChannelWhenNotFound >>= addListener   void $ forkFinally (forever (atomically (readTChan listener) >>= action)) disposeListener@@ -91,13 +145,13 @@       let c = fromMaybe (panic $ "trying to remove listener from non existing channel: " <> toS chan) mC       M.delete chan (channels multi)       when (listeners c - 1 > 0) $-        M.insert Channel{ broadcast = broadcast c, listeners = listeners c - 1 } chan (channels multi)+        M.insert Channel {broadcast = broadcast c, listeners = listeners c - 1} chan (channels multi)     openChannelWhenNotFound =       M.lookup chan (channels multi) >>= \case-                                            Nothing -> openChannel multi chan-                                            Just ch -> return ch+        Nothing -> openChannel multi chan+        Just ch -> return ch     addListener ch = do       M.delete chan (channels multi)-      let newChannel = Channel{ broadcast = broadcast ch, listeners = listeners ch + 1}+      let newChannel = Channel {broadcast = broadcast ch, listeners = listeners ch + 1}       M.insert newChannel chan (channels multi)       dupTChan $ broadcast newChannel

@@ -11,7 +11,8 @@   ( ConnectionInfo,validateClaims   ) where -import Protolude+import Protolude hiding (toS)+import Protolude.Conv import Control.Lens import Crypto.JWT import Data.List@@ -22,13 +23,13 @@   type Claims = M.HashMap Text JSON.Value-type ConnectionInfo = ([ByteString], ByteString, Claims)+type ConnectionInfo = ([Text], Text, Claims)  {-| Given a secret, a token and a timestamp it validates the claims and returns     either an error message or a triple containing channel, mode and claims hashmap. -} validateClaims-  :: Maybe ByteString+  :: Maybe Text   -> ByteString   -> LByteString   -> UTCTime@@ -57,16 +58,16 @@   pure (validChannels, mode, cl')   where-  claimAsJSON :: Text -> Claims -> Maybe ByteString+  claimAsJSON :: Text -> Claims -> Maybe Text   claimAsJSON name cl = case M.lookup name cl of-    Just (JSON.String s) -> Just $ encodeUtf8 s+    Just (JSON.String s) -> Just s     _ -> Nothing -  claimAsJSONList :: Text -> Claims -> Maybe [ByteString]+  claimAsJSONList :: Text -> Claims -> Maybe [Text]   claimAsJSONList name cl = case M.lookup name cl of     Just channelsJson ->       case JSON.fromJSON channelsJson :: JSON.Result [Text] of-        JSON.Success channelsList -> Just $ encodeUtf8 <$> channelsList+        JSON.Success channelsList -> Just channelsList         _ -> Nothing     Nothing -> Nothing 

@@ -1,42 +1,47 @@-{-|-Module      : PostgresWebsockets.Config-Description : Manages PostgresWebsockets configuration options.--This module provides a helper function to read the command line-arguments using  the AppConfig type to store-them.  It also can be used to define other middleware configuration that-may be delegated to some sort of external configuration.--}-module PostgresWebsockets.Config -        ( prettyVersion-        , loadConfig-        , warpSettings-        , AppConfig (..)-        ) where+-- |+-- Module      : PostgresWebsockets.Config+-- Description : Manages PostgresWebsockets configuration options.+--+-- This module provides a helper function to read the command line+-- arguments using  the AppConfig type to store+-- them.  It also can be used to define other middleware configuration that+-- may be delegated to some sort of external configuration.+module PostgresWebsockets.Config+  ( prettyVersion,+    loadConfig,+    warpSettings,+    AppConfig (..),+  )+where -import           Env-import           Data.Text                   (intercalate, pack, replace, strip, stripPrefix)-import           Data.Version                (versionBranch)-import           Paths_postgres_websockets   (version)-import           Protolude hiding            (intercalate, (<>), optional, replace)-import           Data.String (IsString(..))-import           Network.Wai.Handler.Warp-import qualified Data.ByteString                      as BS-import qualified Data.ByteString.Base64               as B64+import qualified Data.ByteString as BS+import qualified Data.ByteString.Base64 as B64+import Data.String (IsString (..))+import Data.Text (intercalate, pack, replace, strip, stripPrefix)+import Data.Version (versionBranch)+import Env+import Network.Wai.Handler.Warp+import Paths_postgres_websockets (version)+import Protolude hiding (intercalate, optional, replace, toS, (<>))+import Protolude.Conv  -- | Config file settings for the server-data AppConfig = AppConfig {-    configDatabase          :: Text-  , configPath              :: Maybe Text-  , configHost              :: Text-  , configPort              :: Int-  , configListenChannel     :: Text-  , configMetaChannel       :: Maybe Text-  , configJwtSecret         :: ByteString-  , configJwtSecretIsBase64 :: Bool-  , configPool              :: Int-  , configRetries           :: Int+data AppConfig = AppConfig+  { configDatabase :: Text,+    configPath :: Maybe Text,+    configHost :: Text,+    configPort :: Int,+    configListenChannel :: Text,+    configMetaChannel :: Maybe Text,+    configJwtSecret :: ByteString,+    configJwtSecretIsBase64 :: Bool,+    configPool :: Int,+    configRetries :: Int,+    configReconnectInterval :: Maybe Int,+    configCertificateFile :: Maybe Text,+    configKeyFile :: Maybe Text   }+  deriving (Show)  -- | User friendly version number prettyVersion :: Text@@ -44,41 +49,53 @@  -- | Load all postgres-websockets config from Environment variables. This can be used to use just the middleware or to feed into warpSettings loadConfig :: IO AppConfig-loadConfig = readOptions >>= loadSecretFile >>= loadDatabaseURIFile+loadConfig =+  readOptions+    >>= verifyTLSConfig+    >>= loadSecretFile+    >>= loadDatabaseURIFile  -- | Given a shutdown handler and an AppConfig builds a Warp Settings to start a stand-alone server warpSettings :: (IO () -> IO ()) -> AppConfig -> Settings-warpSettings waitForShutdown AppConfig{..} =-      setHost (fromString $ toS configHost)-                  . setPort configPort-                  . setServerName (toS $ "postgres-websockets/" <> prettyVersion)-                  . setTimeout 3600-                  . setInstallShutdownHandler waitForShutdown-                  . setGracefulShutdownTimeout (Just 5)-                  $ defaultSettings-+warpSettings waitForShutdown AppConfig {..} =+  setHost (fromString $ toS configHost)+    . setPort configPort+    . setServerName (toS $ "postgres-websockets/" <> prettyVersion)+    . setTimeout 3600+    . setInstallShutdownHandler waitForShutdown+    . setGracefulShutdownTimeout (Just 5)+    $ defaultSettings  -- private  -- | Function to read and parse options from the environment readOptions :: IO AppConfig readOptions =-    Env.parse (header "You need to configure some environment variables to start the service.") $-      AppConfig <$> var (str <=< nonempty) "PGWS_DB_URI"  (help "String to connect to PostgreSQL")-                <*> optional (var str "PGWS_ROOT_PATH" (help "Root path to serve static files, unset to disable."))-                <*> var str "PGWS_HOST" (def "*4" <> helpDef show <> help "Address the server will listen for websocket connections")-                <*> var auto "PGWS_PORT" (def 3000 <> helpDef show <> help "Port the server will listen for websocket connections")-                <*> var str "PGWS_LISTEN_CHANNEL" (def "postgres-websockets-listener" <> helpDef show <> help "Master channel used in the database to send or read messages in any notification channel")-                <*> optional (var str "PGWS_META_CHANNEL" (help "Websockets channel used to send events about the server state changes."))-                <*> var str "PGWS_JWT_SECRET" (help "Secret used to sign JWT tokens used to open communications channels")-                <*> var auto "PGWS_JWT_SECRET_BASE64" (def False <> helpDef show <> help "Indicate whether the JWT secret should be decoded from a base64 encoded string")-                <*> var auto "PGWS_POOL_SIZE" (def 10 <> helpDef show <> help "How many connection to the database should be used by the connection pool")-                <*> var auto "PGWS_RETRIES" (def 5 <> helpDef show <> help "How many times it should try to connect to the database on startup before exiting with an error")+  Env.parse (header "You need to configure some environment variables to start the service.") $+    AppConfig <$> var (str <=< nonempty) "PGWS_DB_URI" (help "String to connect to PostgreSQL")+      <*> optional (var str "PGWS_ROOT_PATH" (help "Root path to serve static files, unset to disable."))+      <*> var str "PGWS_HOST" (def "*4" <> helpDef show <> help "Address the server will listen for websocket connections")+      <*> var auto "PGWS_PORT" (def 3000 <> helpDef show <> help "Port the server will listen for websocket connections")+      <*> var str "PGWS_LISTEN_CHANNEL" (def "postgres-websockets-listener" <> helpDef show <> help "Master channel used in the database to send or read messages in any notification channel")+      <*> optional (var str "PGWS_META_CHANNEL" (help "Websockets channel used to send events about the server state changes."))+      <*> var str "PGWS_JWT_SECRET" (help "Secret used to sign JWT tokens used to open communications channels")+      <*> var auto "PGWS_JWT_SECRET_BASE64" (def False <> helpDef show <> help "Indicate whether the JWT secret should be decoded from a base64 encoded string")+      <*> var auto "PGWS_POOL_SIZE" (def 10 <> helpDef show <> help "How many connection to the database should be used by the connection pool")+      <*> var auto "PGWS_RETRIES" (def 5 <> helpDef show <> help "How many times it should try to connect to the database on startup before exiting with an error")+      <*> optional (var auto "PGWS_CHECK_LISTENER_INTERVAL" (helpDef show <> help "Interval for supervisor thread to check if listener connection is alive. 0 to disable it."))+      <*> optional (var str "PGWS_CERTIFICATE_FILE" (helpDef show <> help "Certificate file to serve secure websockets connection (wss)."))+      <*> optional (var str "PGWS_KEY_FILE" (helpDef show <> help "Key file to serve secure websockets connection (wss).")) +verifyTLSConfig :: AppConfig -> IO AppConfig+verifyTLSConfig conf@AppConfig {..} = do+  when (isJust configCertificateFile /= isJust configKeyFile) $+    panic "PGWS_TLS_CERTIFICATE and PGWS_TLS_KEY must be set in tandem"+  pure conf+ loadDatabaseURIFile :: AppConfig -> IO AppConfig-loadDatabaseURIFile conf@AppConfig{..} =+loadDatabaseURIFile conf@AppConfig {..} =   case stripPrefix "@" configDatabase of-    Nothing       -> pure conf+    Nothing -> pure conf     Just filename -> setDatabase . strip <$> readFile (toS filename)   where     setDatabase uri = conf {configDatabase = uri}@@ -86,15 +103,16 @@ loadSecretFile :: AppConfig -> IO AppConfig loadSecretFile conf = extractAndTransform secret   where-    secret   = decodeUtf8 $ configJwtSecret conf-    isB64     = configJwtSecretIsBase64 conf+    secret = decodeUtf8 $ configJwtSecret conf+    isB64 = configJwtSecretIsBase64 conf      extractAndTransform :: Text -> IO AppConfig     extractAndTransform s =-      fmap setSecret $ transformString isB64 =<<-        case stripPrefix "@" s of-          Nothing       -> return . encodeUtf8 $ s-          Just filename -> chomp <$> BS.readFile (toS filename)+      fmap setSecret $+        transformString isB64+          =<< case stripPrefix "@" s of+            Nothing -> return . encodeUtf8 $ s+            Just filename -> chomp <$> BS.readFile (toS filename)       where         chomp bs = fromMaybe bs (BS.stripSuffix "\n" bs) @@ -104,11 +122,10 @@     transformString True t =       case B64.decode $ encodeUtf8 $ strip $ replaceUrlChars $ decodeUtf8 t of         Left errMsg -> panic $ pack errMsg-        Right bs    -> return bs+        Right bs -> return bs      setSecret bs = conf {configJwtSecret = bs}      -- replace: Replace every occurrence of one substring with another     replaceUrlChars =       replace "_" "/" . replace "-" "+" . replace "." "="-

@@ -1,39 +1,45 @@-{-|-Module      : PostgresWebsockets.Context-Description : Produce a context capable of running postgres-websockets sessions--}+-- |+-- Module      : PostgresWebsockets.Context+-- Description : Produce a context capable of running postgres-websockets sessions module PostgresWebsockets.Context-        ( Context (..)-        , mkContext-        ) where+  ( Context (..),+    mkContext,+  )+where -import Protolude+import Control.AutoUpdate+  ( defaultUpdateSettings,+    mkAutoUpdate,+    updateAction,+  ) import Data.Time.Clock (UTCTime, getCurrentTime)-import Control.AutoUpdate       ( defaultUpdateSettings-                                , mkAutoUpdate-                                , updateAction-                                ) import qualified Hasql.Pool as P--import PostgresWebsockets.Config ( AppConfig(..) )-import PostgresWebsockets.HasqlBroadcast (newHasqlBroadcaster) import PostgresWebsockets.Broadcast (Multiplexer)+import PostgresWebsockets.Config (AppConfig (..))+import PostgresWebsockets.HasqlBroadcast (newHasqlBroadcaster)+import Protolude hiding (toS)+import Protolude.Conv -data Context = Context {-    ctxConfig :: AppConfig-  , ctxPool :: P.Pool-  , ctxMulti :: Multiplexer-  , ctxGetTime :: IO UTCTime+data Context = Context+  { ctxConfig :: AppConfig,+    ctxPool :: P.Pool,+    ctxMulti :: Multiplexer,+    ctxGetTime :: IO UTCTime   }  -- | Given a configuration and a shutdown action (performed when the Multiplexer's listen connection dies) produces the context necessary to run sessions mkContext :: AppConfig -> IO () -> IO Context-mkContext conf@AppConfig{..} shutdown = do+mkContext conf@AppConfig {..} shutdownServer = do   Context conf     <$> P.acquire (configPool, 10, pgSettings)-    <*> newHasqlBroadcaster shutdown (toS configListenChannel) configRetries pgSettings+    <*> newHasqlBroadcaster shutdown (toS configListenChannel) configRetries configReconnectInterval pgSettings     <*> mkGetTime   where+    shutdown =+      maybe+        shutdownServer+        (const $ putText "Producer thread is dead")+        configReconnectInterval     mkGetTime :: IO (IO UTCTime)     mkGetTime = mkAutoUpdate defaultUpdateSettings {updateAction = getCurrentTime}     pgSettings = toS configDatabase

@@ -1,48 +1,50 @@-{-| -Module      : PostgresWebsockets.Broadcast-Description : Build a Hasql.Notifications based producer 'Multiplexer'.--Uses Broadcast module adding database as a source producer.-This module provides a function to produce a 'Multiplexer' from a Hasql 'Connection'.-The producer issues a LISTEN command upon Open commands and UNLISTEN upon Close.--}+-- |+-- Module      : PostgresWebsockets.Broadcast+-- Description : Build a Hasql.Notifications based producer 'Multiplexer'.+--+-- Uses Broadcast module adding database as a source producer.+-- This module provides a function to produce a 'Multiplexer' from a Hasql 'Connection'.+-- The producer issues a LISTEN command upon Open commands and UNLISTEN upon Close. module PostgresWebsockets.HasqlBroadcast-  ( newHasqlBroadcaster-  , newHasqlBroadcasterOrError-  -- re-export-  , acquire-  , relayMessages-  , relayMessagesForever-  ) where--import Protolude hiding (putErrLn)+  ( newHasqlBroadcaster,+    newHasqlBroadcasterOrError,+    -- re-export+    acquire,+    relayMessages,+    relayMessagesForever,+  )+where -import Hasql.Connection-import Hasql.Notifications-import Data.Aeson (decode, Value(..))-import Data.HashMap.Lazy (lookupDefault)+import Control.Retry (RetryStatus (..), capDelay, exponentialBackoff, retrying)+import Data.Aeson (Value (..), decode) import Data.Either.Combinators (mapBoth) import Data.Function (id)-import Control.Retry (RetryStatus(..), retrying, capDelay, exponentialBackoff)-+import Data.HashMap.Lazy (lookupDefault)+import GHC.Show+import Hasql.Connection+import qualified Hasql.Decoders as HD+import qualified Hasql.Encoders as HE+import Hasql.Notifications+import qualified Hasql.Session as H+import qualified Hasql.Statement as H import PostgresWebsockets.Broadcast+import Protolude hiding (putErrLn, show, toS)+import Protolude.Conv -{- | Returns a multiplexer from a connection URI, keeps trying to connect in case there is any error.-   This function also spawns a thread that keeps relaying the messages from the database to the multiplexer's listeners--}-newHasqlBroadcaster :: IO () -> Text -> Int -> ByteString -> IO Multiplexer-newHasqlBroadcaster onConnectionFailure ch maxRetries = newHasqlBroadcasterForConnection . tryUntilConnected maxRetries+-- | Returns a multiplexer from a connection URI, keeps trying to connect in case there is any error.+--   This function also spawns a thread that keeps relaying the messages from the database to the multiplexer's listeners+newHasqlBroadcaster :: IO () -> Text -> Int -> Maybe Int -> ByteString -> IO Multiplexer+newHasqlBroadcaster onConnectionFailure ch maxRetries checkInterval = newHasqlBroadcasterForConnection . tryUntilConnected maxRetries   where-    newHasqlBroadcasterForConnection = newHasqlBroadcasterForChannel onConnectionFailure ch+    newHasqlBroadcasterForConnection = newHasqlBroadcasterForChannel onConnectionFailure ch checkInterval -{- | Returns a multiplexer from a connection URI or an error message on the left case-   This function also spawns a thread that keeps relaying the messages from the database to the multiplexer's listeners--}+-- | Returns a multiplexer from a connection URI or an error message on the left case+--   This function also spawns a thread that keeps relaying the messages from the database to the multiplexer's listeners newHasqlBroadcasterOrError :: IO () -> Text -> ByteString -> IO (Either ByteString Multiplexer) newHasqlBroadcasterOrError onConnectionFailure ch =-  acquire >=> (sequence . mapBoth show (newHasqlBroadcasterForConnection . return))+  acquire >=> (sequence . mapBoth (toSL . show) (newHasqlBroadcasterForConnection . return))   where-    newHasqlBroadcasterForConnection = newHasqlBroadcasterForChannel onConnectionFailure ch+    newHasqlBroadcasterForConnection = newHasqlBroadcasterForChannel onConnectionFailure ch Nothing  tryUntilConnected :: Int -> ByteString -> IO Connection tryUntilConnected maxRetries =@@ -53,58 +55,80 @@     firstDelayInMicroseconds = 1000000     retryPolicy = capDelay maxDelayInMicroseconds $ exponentialBackoff firstDelayInMicroseconds     shouldRetry :: RetryStatus -> Either ConnectionError Connection -> IO Bool-    shouldRetry RetryStatus{..} con =+    shouldRetry RetryStatus {..} con =       case con of         Left err -> do-          putErrLn $ "Error connecting notification listener to database: " <> show err+          putErrLn $ "Error connecting notification listener to database: " <> (toS . show) err           pure $ rsIterNumber < maxRetries - 1         _ -> return False -{- | Returns a multiplexer from a channel and an IO Connection, listen for different database notifications on the provided channel using the connection produced.--   This function also spawns a thread that keeps relaying the messages from the database to the multiplexer's listeners--   To listen on channels *chat*--   @-   import Protolude-   import PostgresWebsockets.HasqlBroadcast-   import PostgresWebsockets.Broadcast-   import Hasql.Connection--   main = do-    conOrError <- H.acquire "postgres://localhost/test_database"-    let con = either (panic . show) id conOrError :: Connection-    multi <- newHasqlBroadcaster con--    onMessage multi "chat" (\ch ->-      forever $ fmap print (atomically $ readTChan ch)-   @--}-newHasqlBroadcasterForChannel :: IO () -> Text -> IO Connection -> IO Multiplexer-newHasqlBroadcasterForChannel onConnectionFailure ch getCon = do+-- | Returns a multiplexer from a channel and an IO Connection, listen for different database notifications on the provided channel using the connection produced.+--+--   This function also spawns a thread that keeps relaying the messages from the database to the multiplexer's listeners+--+--   To listen on channels *chat*+--+--   @+--   import Protolude+--   import PostgresWebsockets.HasqlBroadcast+--   import PostgresWebsockets.Broadcast+--   import Hasql.Connection+--+--   main = do+--    conOrError <- H.acquire "postgres://localhost/test_database"+--    let con = either (panic . show) id conOrError :: Connection+--    multi <- newHasqlBroadcaster con+--+--    onMessage multi "chat" (\ch ->+--      forever $ fmap print (atomically $ readTChan ch)+--   @+newHasqlBroadcasterForChannel :: IO () -> Text -> Maybe Int -> IO Connection -> IO Multiplexer+newHasqlBroadcasterForChannel onConnectionFailure ch checkInterval getCon = do   multi <- newMultiplexer openProducer $ const onConnectionFailure+  case checkInterval of+    Just i -> superviseMultiplexer multi i shouldRestart+    _ -> pure ()   void $ relayMessagesForever multi   return multi   where-    toMsg :: ByteString -> ByteString -> Message+    toMsg :: Text -> Text -> Message     toMsg c m = case decode (toS m) of-                   Just v -> Message (channelDef c v) m-                   Nothing -> Message c m+      Just v -> Message (channelDef c v) m+      Nothing -> Message c m -    lookupStringDef :: Text -> ByteString -> Value -> ByteString+    lookupStringDef :: Text -> Text -> Value -> Text     lookupStringDef key d (Object obj) =       case lookupDefault (String $ toS d) key obj of         String s -> toS s-        _ -> d-    lookupStringDef _ d _ = d+        _ -> toS d+    lookupStringDef _ d _ = toS d     channelDef = lookupStringDef "channel"+    shouldRestart = do+      con <- getCon+      not <$> isListening con ch+     openProducer msgQ = do       con <- getCon       listen con $ toPgIdentifier ch       waitForNotifications-        (\c m-> atomically $ writeTQueue msgQ $ toMsg c m)+        (\c m -> atomically $ writeTQueue msgQ $ toMsg (toS c) (toS m))         con  putErrLn :: Text -> IO () putErrLn = hPutStrLn stderr++isListening :: Connection -> Text -> IO Bool+isListening con ch = do+  resultOrError <- H.run session con+  pure $ fromRight False resultOrError+  where+    session = H.statement chPattern isListeningStatement+    chPattern = "listen%" <> ch <> "%"++isListeningStatement :: H.Statement Text Bool+isListeningStatement =+  H.Statement sql encoder decoder True+  where+    sql = "select exists (select * from pg_stat_activity where datname = current_database() and query ilike $1);"+    encoder = HE.param $ HE.nonNullable HE.text+    decoder = HD.singleRow (HD.column (HD.nonNullable HD.bool))

@@ -10,7 +10,8 @@   ( postgresWsMiddleware   ) where -import Protolude+import Protolude hiding (toS)+import Protolude.Conv import Data.Time.Clock (UTCTime) import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds, posixSecondsToUTCTime) import Control.Concurrent.AlarmClock (newAlarmClock, setAlarm)@@ -21,10 +22,9 @@ import qualified Network.WebSockets as WS  import qualified Data.Aeson as A-import qualified Data.ByteString.Char8 as BS+import qualified Data.Text as T import qualified Data.ByteString.Lazy as BL import qualified Data.HashMap.Strict as M-import qualified Data.Text.Encoding.Error as T  import PostgresWebsockets.Broadcast (onMessage) import PostgresWebsockets.Claims ( ConnectionInfo, validateClaims )@@ -63,8 +63,8 @@ wsApp Context{..} pendingConn =   ctxGetTime >>= validateClaims requestChannel (configJwtSecret ctxConfig) (toS jwtToken) >>= either rejectRequest forkSessions   where-    hasRead m = m == ("r" :: ByteString) || m == ("rw" :: ByteString)-    hasWrite m = m == ("w" :: ByteString) || m == ("rw" :: ByteString)+    hasRead m = m == ("r" :: Text) || m == ("rw" :: Text)+    hasWrite m = m == ("w" :: Text) || m == ("rw" :: Text)      rejectRequest :: Text -> IO ()     rejectRequest msg = do@@ -72,8 +72,8 @@       WS.rejectRequest pendingConn (toS msg)      -- the URI has one of the two formats - /:jwt or /:channel/:jwt-    pathElements = BS.split '/' $ BS.drop 1 $ WS.requestPath $ WS.pendingRequest pendingConn-    jwtToken = +    pathElements = T.split (== '/') $ T.drop 1 $ (toSL . WS.requestPath) $ WS.pendingRequest pendingConn+    jwtToken =       case length pathElements `compare` 1 of         GT -> headDef "" $ tailSafe pathElements         _ -> headDef "" pathElements@@ -102,7 +102,7 @@              case configMetaChannel ctxConfig of               Nothing -> pure ()-              Just ch -> sendMessageWithTimestamp $ connectionOpenMessage (toS $ BS.intercalate "," chs) ch+              Just ch -> sendMessageWithTimestamp $ connectionOpenMessage (toS $ T.intercalate "," chs) ch              when (hasRead mode) $               forM_ chs $ flip (onMessage ctxMulti) $ WS.sendTextData conn . B.payload@@ -116,7 +116,7 @@ -- Having both channel and claims as parameters seem redundant -- But it allows the function to ignore the claims structure and the source -- of the channel, so all claims decoding can be coded in the caller-notifySession :: WS.Connection -> (Text -> Text -> IO ()) -> [ByteString] -> IO ()+notifySession :: WS.Connection -> (Text -> Text -> IO ()) -> [Text] -> IO () notifySession wsCon sendToChannel chs =   withAsync (forever relayData) wait   where

@@ -1,42 +1,43 @@-{-|-Module      : PostgresWebsockets.Server-Description : Functions to start a full stand-alone PostgresWebsockets server.--}+-- |+-- Module      : PostgresWebsockets.Server+-- Description : Functions to start a full stand-alone PostgresWebsockets server. module PostgresWebsockets.Server-        ( serve -        ) where+  ( serve,+  )+where -import Protolude-import Network.Wai.Application.Static ( staticApp, defaultFileServerSettings )-import Network.Wai (Application, responseLBS) import Network.HTTP.Types (status200)-import Network.Wai.Handler.Warp ( runSettings )+import Network.Wai (Application, responseLBS)+import Network.Wai.Application.Static (defaultFileServerSettings, staticApp)+import Network.Wai.Handler.Warp (runSettings)+import Network.Wai.Handler.WarpTLS (runTLS, tlsSettings) import Network.Wai.Middleware.RequestLogger (logStdout)--import PostgresWebsockets.Middleware ( postgresWsMiddleware )-import PostgresWebsockets.Config ( AppConfig(..), warpSettings )-import PostgresWebsockets.Context ( mkContext )+import PostgresWebsockets.Config (AppConfig (..), warpSettings)+import PostgresWebsockets.Context (mkContext)+import PostgresWebsockets.Middleware (postgresWsMiddleware)+import Protolude  -- | Start a stand-alone warp server using the parameters from AppConfig and a opening a database connection pool. serve :: AppConfig -> IO ()-serve conf@AppConfig{..} = do+serve conf@AppConfig {..} = do   shutdownSignal <- newEmptyMVar-  let waitForShutdown cl = void $ forkIO (takeMVar shutdownSignal >> cl)-      appSettings = warpSettings waitForShutdown conf-   putStrLn $ ("Listening on port " :: Text) <> show configPort    let shutdown = putErrLn ("Broadcaster connection is dead" :: Text) >> putMVar shutdownSignal ()   ctx <- mkContext conf shutdown -  runSettings appSettings $-    postgresWsMiddleware ctx $-    logStdout $ maybe dummyApp staticApp' configPath-  die "Shutting down server..."+  let waitForShutdown cl = void $ forkIO (takeMVar shutdownSignal >> cl)+      appSettings = warpSettings waitForShutdown conf+      app = postgresWsMiddleware ctx $ logStdout $ maybe dummyApp staticApp' configPath +  case (configCertificateFile, configKeyFile) of+    (Just certificate, Just key) -> runTLS (tlsSettings (toS certificate) (toS key)) appSettings app+    _ -> runSettings appSettings app++  die "Shutting down server..."   where     staticApp' :: Text -> Application     staticApp' = staticApp . defaultFileServerSettings . toS     dummyApp :: Application     dummyApp _ respond =-        respond $ responseLBS status200 [("Content-Type", "text/plain")] "Hello, Web!"+      respond $ responseLBS status200 [("Content-Type", "text/plain")] "Hello, Web!"

@@ -21,12 +21,12 @@         `shouldReturn` Left "Token expired"     it "request any channel from a token that does not have channels or channel claims should succeed" $ do       time <- getCurrentTime-      validateClaims (Just (encodeUtf8 "test")) secret+      validateClaims (Just "test") secret         "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJtb2RlIjoiciJ9.jL5SsRFegNUlbBm8_okhHSujqLcKKZdDglfdqNl1_rY" time         `shouldReturn` Right (["test"], "r", M.fromList[("mode",String "r")])     it "requesting a channel that is set by and old style channel claim should work" $ do       time <- getCurrentTime-      validateClaims (Just (encodeUtf8 "test")) secret+      validateClaims (Just "test") secret         "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJtb2RlIjoiciIsImNoYW5uZWwiOiJ0ZXN0In0.1d4s-at2kWj8OSabHZHTbNh1dENF7NWy_r0ED3Rwf58" time         `shouldReturn` Right (["test"], "r", M.fromList[("mode",String "r"),("channel",String "test")])     it "no requesting channel should return all channels in the token" $ do@@ -37,18 +37,18 @@      it "requesting a channel from the channels claim shoud return only the requested channel" $ do       time <- getCurrentTime-      validateClaims (Just (encodeUtf8 "test")) secret+      validateClaims (Just "test") secret         "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJtb2RlIjoiciIsImNoYW5uZWxzIjpbInRlc3QiLCJ0ZXN0MiJdfQ.MumdJ5FpFX4Z6SJD3qsygVF0r9vqxfqhj5J30O32N0k" time         `shouldReturn` Right (["test"], "r", M.fromList[("mode",String "r"),("channels",  toJSON ["test"::Text, "test2"] )])      it "requesting a channel not from the channels claim shoud error" $ do       time <- getCurrentTime-      validateClaims (Just (encodeUtf8 "notAllowed")) secret+      validateClaims (Just "notAllowed") secret         "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJtb2RlIjoiciIsImNoYW5uZWxzIjpbInRlc3QiLCJ0ZXN0MiJdfQ.MumdJ5FpFX4Z6SJD3qsygVF0r9vqxfqhj5J30O32N0k" time         `shouldReturn` Left  "No allowed channels"      it "requesting a channel with no mode fails" $ do       time <- getCurrentTime-      validateClaims (Just (encodeUtf8 "test")) secret+      validateClaims (Just "test") secret         "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJjaGFubmVscyI6WyJ0ZXN0IiwidGVzdDIiXX0.akC1PEYk2DEZtLP2XjC6qXOGZJejmPx49qv-VeEtQYQ" time         `shouldReturn` Left "Missing mode"

@@ -15,11 +15,11 @@             <$> acquire connStr      it "relay messages sent to the appropriate database channel" $ do-      multi <- either (panic .show) id <$> newHasqlBroadcasterOrError (pure ()) "postgres-websockets" "postgres://localhost/postgres_ws_test"+      multi <- either (panic .show) id <$> newHasqlBroadcasterOrError (pure ()) "postgres-websockets" "postgres://postgres:roottoor@localhost:5432/postgres_ws_test"       msg <- liftIO newEmptyMVar       onMessage multi "test" $ putMVar msg -      con <- newConnection "postgres://localhost/postgres_ws_test"+      con <- newConnection "postgres://postgres:roottoor@localhost:5432/postgres_ws_test"       void $ notify con (toPgIdentifier "postgres-websockets") "{\"channel\": \"test\", \"payload\": \"hello there\"}"        readMVar msg `shouldReturn` Message "test" "{\"channel\": \"test\", \"payload\": \"hello there\"}"

@@ -1,51 +1,53 @@ module ServerSpec (spec) where -import Protolude--import Test.Hspec-import PostgresWebsockets-import PostgresWebsockets.Config- import Control.Lens import Data.Aeson.Lens-+import Network.Socket (withSocketsDo) import qualified Network.WebSockets as WS-import           Network.Socket (withSocketsDo)+import PostgresWebsockets+import PostgresWebsockets.Config+import Protolude+import Test.Hspec  testServerConfig :: AppConfig-testServerConfig = AppConfig-                    { configDatabase = "postgres://localhost/postgres"-                    , configPath = Nothing-                    , configHost = "*"-                    , configPort = 8080-                    , configListenChannel = "postgres-websockets-test-channel"-                    , configJwtSecret = "reallyreallyreallyreallyverysafe"-                    , configMetaChannel = Nothing-                    , configJwtSecretIsBase64 = False-                    , configPool = 10-                    , configRetries = 5-                    }+testServerConfig =+  AppConfig+    { configDatabase = "postgres://postgres:roottoor@localhost:5432/postgres_ws_test",+      configPath = Nothing,+      configHost = "*",+      configPort = 8080,+      configListenChannel = "postgres-websockets-test-channel",+      configJwtSecret = "reallyreallyreallyreallyverysafe",+      configMetaChannel = Nothing,+      configJwtSecretIsBase64 = False,+      configPool = 10,+      configRetries = 5,+      configReconnectInterval = Nothing,+      configCertificateFile = Nothing,+      configKeyFile = Nothing+    }  startTestServer :: IO ThreadId startTestServer = do-    threadId <- forkIO $ serve testServerConfig-    threadDelay 500000-    pure threadId+  threadId <- forkIO $ serve testServerConfig+  threadDelay 500000+  pure threadId  withServer :: IO () -> IO () withServer action =-  bracket startTestServer-          (\tid -> killThread tid >> threadDelay 500000)-          (const action)+  bracket+    startTestServer+    (\tid -> killThread tid >> threadDelay 500000)+    (const action)  sendWsData :: Text -> Text -> IO () sendWsData uri msg =-    withSocketsDo $-        WS.runClient-            "localhost"-            (configPort testServerConfig)-            (toS uri)-            (`WS.sendTextData` msg)+  withSocketsDo $+    WS.runClient+      "127.0.0.1"+      (configPort testServerConfig)+      (toS uri)+      (`WS.sendTextData` msg)  testChannel :: Text testChannel = "/test/eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJtb2RlIjoicncifQ.auy9z4-pqoVEAay9oMi1FuG7ux_C_9RQCH8-wZgej18"@@ -58,62 +60,64 @@  waitForWsData :: Text -> IO (MVar ByteString) waitForWsData uri = do-    msg <- newEmptyMVar-    void $ forkIO $-        withSocketsDo $-            WS.runClient-                "localhost"-                (configPort testServerConfig)-                (toS uri)-                (\c -> do-                    m <- WS.receiveData c-                    putMVar msg m-                )-    threadDelay 10000-    pure msg+  msg <- newEmptyMVar+  void $+    forkIO $+      withSocketsDo $+        WS.runClient+          "127.0.0.1"+          (configPort testServerConfig)+          (toS uri)+          ( \c -> do+              m <- WS.receiveData c+              putMVar msg m+          )+  threadDelay 10000+  pure msg  waitForMultipleWsData :: Int -> Text -> IO (MVar [ByteString]) waitForMultipleWsData messageCount uri = do-    msg <- newEmptyMVar-    void $ forkIO $-        withSocketsDo $-            WS.runClient-                "localhost"-                (configPort testServerConfig)-                (toS uri)-                (\c -> do-                    m <- replicateM messageCount (WS.receiveData c)-                    putMVar msg m-                )-    threadDelay 1000-    pure msg+  msg <- newEmptyMVar+  void $+    forkIO $+      withSocketsDo $+        WS.runClient+          "127.0.0.1"+          (configPort testServerConfig)+          (toS uri)+          ( \c -> do+              m <- replicateM messageCount (WS.receiveData c)+              putMVar msg m+          )+  threadDelay 1000+  pure msg  spec :: Spec spec = around_ withServer $-            describe "serve" $ do-                it "should be able to send messages to test server" $-                    sendWsData testChannel "test data"-                it "should be able to receive messages from test server" $ do-                    msg <- waitForWsData testChannel-                    sendWsData testChannel "test data"-                    msgJson <- takeMVar msg-                    (msgJson ^? key "payload" . _String) `shouldBe` Just "test data"-                it "should be able to send messages to multiple channels in one shot" $ do-                    msg <- waitForWsData testChannel-                    secondaryMsg <- waitForWsData secondaryChannel-                    sendWsData testAndSecondaryChannel "test data"-                    msgJson <- takeMVar msg-                    secondaryMsgJson <- takeMVar secondaryMsg+  describe "serve" $ do+    it "should be able to send messages to test server" $+      sendWsData testChannel "test data"+    it "should be able to receive messages from test server" $ do+      msg <- waitForWsData testChannel+      sendWsData testChannel "test data"+      msgJson <- takeMVar msg+      (msgJson ^? key "payload" . _String) `shouldBe` Just "test data"+    it "should be able to send messages to multiple channels in one shot" $ do+      msg <- waitForWsData testChannel+      secondaryMsg <- waitForWsData secondaryChannel+      sendWsData testAndSecondaryChannel "test data"+      msgJson <- takeMVar msg+      secondaryMsgJson <- takeMVar secondaryMsg -                    (msgJson ^? key "payload" . _String) `shouldBe` Just "test data"-                    (msgJson ^? key "channel" . _String) `shouldBe` Just "test"-                    (secondaryMsgJson ^? key "payload" . _String) `shouldBe` Just "test data"-                    (secondaryMsgJson ^? key "channel" . _String) `shouldBe` Just "secondary"-                it "should be able to receive from multiple channels in one shot" $ do-                    msgs <- waitForMultipleWsData 2 testAndSecondaryChannel-                    sendWsData testAndSecondaryChannel "test data"-                    msgsJson <- takeMVar msgs+      (msgJson ^? key "payload" . _String) `shouldBe` Just "test data"+      (msgJson ^? key "channel" . _String) `shouldBe` Just "test"+      (secondaryMsgJson ^? key "payload" . _String) `shouldBe` Just "test data"+      (secondaryMsgJson ^? key "channel" . _String) `shouldBe` Just "secondary"+    it "should be able to receive from multiple channels in one shot" $ do+      msgs <- waitForMultipleWsData 2 testAndSecondaryChannel+      sendWsData testAndSecondaryChannel "test data"+      msgsJson <- takeMVar msgs -                    forM_-                        msgsJson-                        (\msgJson -> (msgJson ^? key "payload" . _String) `shouldBe` Just "test data")+      forM_+        msgsJson+        (\msgJson -> (msgJson ^? key "payload" . _String) `shouldBe` Just "test data")