diff --git a/postgres-websockets.cabal b/postgres-websockets.cabal --- a/postgres-websockets.cabal +++ b/postgres-websockets.cabal @@ -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 diff --git a/src/PostgresWebsockets/Broadcast.hs b/src/PostgresWebsockets/Broadcast.hs --- a/src/PostgresWebsockets/Broadcast.hs +++ b/src/PostgresWebsockets/Broadcast.hs @@ -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 diff --git a/src/PostgresWebsockets/Claims.hs b/src/PostgresWebsockets/Claims.hs --- a/src/PostgresWebsockets/Claims.hs +++ b/src/PostgresWebsockets/Claims.hs @@ -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 diff --git a/src/PostgresWebsockets/Config.hs b/src/PostgresWebsockets/Config.hs --- a/src/PostgresWebsockets/Config.hs +++ b/src/PostgresWebsockets/Config.hs @@ -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 "." "=" - diff --git a/src/PostgresWebsockets/Context.hs b/src/PostgresWebsockets/Context.hs --- a/src/PostgresWebsockets/Context.hs +++ b/src/PostgresWebsockets/Context.hs @@ -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 diff --git a/src/PostgresWebsockets/HasqlBroadcast.hs b/src/PostgresWebsockets/HasqlBroadcast.hs --- a/src/PostgresWebsockets/HasqlBroadcast.hs +++ b/src/PostgresWebsockets/HasqlBroadcast.hs @@ -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)) diff --git a/src/PostgresWebsockets/Middleware.hs b/src/PostgresWebsockets/Middleware.hs --- a/src/PostgresWebsockets/Middleware.hs +++ b/src/PostgresWebsockets/Middleware.hs @@ -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 diff --git a/src/PostgresWebsockets/Server.hs b/src/PostgresWebsockets/Server.hs --- a/src/PostgresWebsockets/Server.hs +++ b/src/PostgresWebsockets/Server.hs @@ -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!" diff --git a/test/ClaimsSpec.hs b/test/ClaimsSpec.hs --- a/test/ClaimsSpec.hs +++ b/test/ClaimsSpec.hs @@ -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" diff --git a/test/HasqlBroadcastSpec.hs b/test/HasqlBroadcastSpec.hs --- a/test/HasqlBroadcastSpec.hs +++ b/test/HasqlBroadcastSpec.hs @@ -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\"}" diff --git a/test/ServerSpec.hs b/test/ServerSpec.hs --- a/test/ServerSpec.hs +++ b/test/ServerSpec.hs @@ -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")