postgres-websockets 0.4.2.1 → 0.5.0.0
raw patch · 11 files changed
+108/−207 lines, 11 filesdep +envparsedep ~hasql-pooldep ~websockets
Dependencies added: envparse
Dependency ranges changed: hasql-pool, websockets
Files
- app/Config.hs +14/−80
- app/Main.hs +11/−6
- postgres-websockets.cabal +5/−3
- src/PostgresWebsockets.hs +14/−16
- src/PostgresWebsockets/Broadcast.hs +8/−26
- src/PostgresWebsockets/Claims.hs +2/−2
- src/PostgresWebsockets/Database.hs +7/−7
- src/PostgresWebsockets/HasqlBroadcast.hs +29/−16
- test/BroadcastSpec.hs +2/−15
- test/DatabaseSpec.hs +8/−7
- test/HasqlBroadcastSpec.hs +8/−29
app/Config.hs view
@@ -20,19 +20,10 @@ ) where -import System.IO.Error (IOError)-import Control.Applicative-import qualified Data.Configurator as C-import qualified Data.Configurator.Types as C-import Data.Monoid-import Data.Text (intercalate, lines)-import Data.Text.Encoding (encodeUtf8)+import Env+import Data.Text (intercalate) import Data.Version (versionBranch)-import Options.Applicative hiding (str) import Paths_postgres_websockets (version)-import Text.Heredoc-import Text.PrettyPrint.ANSI.Leijen hiding ((<>))-import qualified Text.PrettyPrint.ANSI.Leijen as L import Protolude hiding (intercalate, (<>)) -- | Config file settings for the server@@ -41,7 +32,7 @@ , configPath :: Text , configHost :: Text , configPort :: Int- , configAuditChannel :: Maybe Text+ , configListenChannel :: Text , configJwtSecret :: ByteString , configJwtSecretIsBase64 :: Bool , configPool :: Int@@ -51,75 +42,18 @@ prettyVersion :: Text prettyVersion = intercalate "." $ map show $ versionBranch version --- | Function to read and parse options from the command line+-- | Function to read and parse options from the environment readOptions :: IO AppConfig-readOptions = do- -- First read the config file path from command line- cfgPath <- customExecParser parserPrefs opts- -- Now read the actual config file- conf <- catch- (C.load [C.Required cfgPath])- configNotfoundHint-- handle missingKeyHint $ do- -- db ----------------- cDbUri <- C.require conf "db-uri"- cPool <- C.lookupDefault 10 conf "db-pool"- -- server ------------- cPath <- C.require conf "server-root"- cHost <- C.lookupDefault "*4" conf "server-host"- cPort <- C.lookupDefault 3000 conf "server-port"- cAuditC <- C.lookup conf "audit-channel"- -- jwt ---------------- cJwtSec <- C.require conf "jwt-secret"- cJwtB64 <- C.lookupDefault False conf "secret-is-base64"-- return $ AppConfig cDbUri cPath cHost cPort cAuditC (encodeUtf8 cJwtSec) cJwtB64 cPool-- where- opts = info (helper <*> pathParser) $- fullDesc- <> progDesc (- "postgres-websockets "- <> toS prettyVersion- <> " / Connects websockets to PostgreSQL asynchronous notifications."- )- <> footerDoc (Just $- text "Example Config File:"- L.<> nest 2 (hardline L.<> exampleCfg)- )-- parserPrefs = prefs showHelpOnError-- configNotfoundHint :: IOError -> IO a- configNotfoundHint e = die $ "Cannot open config file:\n\t" <> show e-- missingKeyHint :: C.KeyError -> IO a- missingKeyHint (C.KeyError n) =- die $- "Required config parameter \"" <> n <> "\" is missing or of wrong type.\n"-- exampleCfg :: Doc- exampleCfg = vsep . map (text . toS) . lines $- [str|db-uri = "postgres://user:pass@localhost:5432/dbname"- |db-pool = 10- |- |server-root = "./client-example"- |server-host = "*4"- |server-port = 3000- |- |## choose a secret to enable JWT auth- |## (use "@filename" to load from separate file)- |# jwt-secret = "foo"- |# secret-is-base64 = false- |]---pathParser :: Parser FilePath-pathParser =- strArgument $- metavar "FILENAME" <>- help "Path to configuration file"+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")+ <*> var str "PGWS_ROOT_PATH" (def "./" <> helpDef show <> help "Root path to serve static files")+ <*> 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")+ <*> 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") data PgVersion = PgVersion { pgvNum :: Int32
app/Main.hs view
@@ -13,7 +13,7 @@ import Data.String (IsString (..)) import Data.Text (pack, replace, strip, stripPrefix) import Data.Text.Encoding (encodeUtf8, decodeUtf8)-import qualified Hasql.Query as H+import qualified Hasql.Statement as H import qualified Hasql.Session as H import qualified Hasql.Decoders as HD import qualified Hasql.Encoders as HE@@ -27,12 +27,12 @@ isServerVersionSupported :: H.Session Bool isServerVersionSupported = do- ver <- H.query () pgVersion+ ver <- H.statement () pgVersion return $ ver >= pgvNum minimumPgVersion where pgVersion =- H.statement "SELECT current_setting('server_version_num')::integer"- HE.unit (HD.singleRow $ HD.value HD.int4) False+ H.Statement "SELECT current_setting('server_version_num')::integer"+ HE.unit (HD.singleRow $ HD.column HD.int4) False main :: IO () main = do@@ -40,9 +40,14 @@ hSetBuffering stdin LineBuffering hSetBuffering stderr NoBuffering + putStrLn $ ("postgres-websockets " :: Text)+ <> prettyVersion+ <> " / Connects websockets to PostgreSQL asynchronous notifications."+ conf <- loadSecretFile =<< readOptions let host = configHost conf port = configPort conf+ listenChannel = toS $ configListenChannel conf pgSettings = toS (configDatabase conf) appSettings = setHost ((fromString . toS) host) . setPort port@@ -53,10 +58,10 @@ putStrLn $ ("Listening on port " :: Text) <> show (configPort conf) pool <- P.acquire (configPool conf, 10, pgSettings)- multi <- newHasqlBroadcaster pgSettings+ multi <- newHasqlBroadcaster listenChannel pgSettings runSettings appSettings $- postgresWsMiddleware (toS <$> configAuditChannel conf) (configJwtSecret conf) pool multi $+ postgresWsMiddleware listenChannel (configJwtSecret conf) pool multi $ logStdout $ staticApp $ defaultFileServerSettings $ toS $ configPath conf loadSecretFile :: AppConfig -> IO AppConfig
postgres-websockets.cabal view
@@ -1,5 +1,5 @@ name: postgres-websockets-version: 0.4.2.1+version: 0.5.0.0 synopsis: Middleware to map LISTEN/NOTIFY messages to Websockets description: Please see README.md homepage: https://github.com/diogob/postgres-websockets#readme@@ -22,10 +22,10 @@ , PostgresWebsockets.HasqlBroadcast , PostgresWebsockets.Claims build-depends: base >= 4.7 && < 5- , hasql-pool >= 0.4 && < 0.5+ , hasql-pool >= 0.4 && < 0.6 , text >= 1.2 && < 2 , wai >= 3.2 && < 4- , websockets >= 0.9 && < 0.11+ , websockets >= 0.9 && < 0.13 , wai-websockets >= 3.0 && < 4 , http-types >= 0.9 , bytestring >= 0.10@@ -53,6 +53,7 @@ hs-source-dirs: app main-is: Main.hs other-modules: Config+ , Paths_postgres_websockets ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N build-depends: base >= 4.7 && < 5 , transformers >= 0.4 && < 0.6@@ -72,6 +73,7 @@ , wai-app-static , heredoc , ansi-wl-pprint+ , envparse default-language: Haskell2010 default-extensions: OverloadedStrings, NoImplicitPrelude, QuasiQuotes
src/PostgresWebsockets.hs view
@@ -31,13 +31,14 @@ data Message = Message { claims :: A.Object+ , channel :: Text , payload :: Text } deriving (Show, Eq, Generic) instance A.ToJSON Message -- | Given a secret, a function to fetch the system time, a Hasql Pool and a Multiplexer this will give you a WAI middleware.-postgresWsMiddleware :: Maybe ByteString -> ByteString -> H.Pool -> Multiplexer -> Wai.Application -> Wai.Application+postgresWsMiddleware :: ByteString -> ByteString -> H.Pool -> Multiplexer -> Wai.Application -> Wai.Application postgresWsMiddleware = WS.websocketsOr WS.defaultConnectionOptions `compose` wsApp where@@ -47,8 +48,8 @@ -- when the websocket is closed a ConnectionClosed Exception is triggered -- this kills all children and frees resources for us-wsApp :: Maybe ByteString -> ByteString -> H.Pool -> Multiplexer -> WS.ServerApp-wsApp mAuditChannel secret pool multi pendingConn =+wsApp :: ByteString -> ByteString -> H.Pool -> Multiplexer -> WS.ServerApp+wsApp dbChannel secret pool multi pendingConn = validateClaims requestChannel secret (toS jwtToken) >>= either rejectRequest forkSessions where hasRead m = m == ("r" :: ByteString) || m == ("rw" :: ByteString)@@ -56,14 +57,13 @@ rejectRequest = WS.rejectRequest pendingConn . encodeUtf8 -- the URI has one of the two formats - /:jwt or /:channel/:jwt pathElements = BS.split '/' $ BS.drop 1 $ WS.requestPath $ WS.pendingRequest pendingConn- jwtToken + jwtToken | length pathElements > 1 = headDef "" $ tailSafe pathElements | length pathElements <= 1 = headDef "" pathElements requestChannel | length pathElements > 1 = Just $ headDef "" pathElements | length pathElements <= 1 = Nothing- notifySessionWithTime = notifySession- forkSessions (channel, mode, validClaims) = do+ forkSessions (ch, mode, validClaims) = do -- role claim defaults to anon if not specified in jwt -- We should accept only after verifying JWT conn <- WS.acceptRequest pendingConn@@ -71,15 +71,11 @@ WS.forkPingThread conn 30 when (hasRead mode) $- onMessage multi channel $ WS.sendTextData conn . B.payload+ onMessage multi ch $ WS.sendTextData conn . B.payload when (hasWrite mode) $- let sendNotifications = void . case mAuditChannel of- Nothing -> notifyPool pool channel- Just auditChannel -> \mesg ->- notifyPool pool channel mesg >>- notifyPool pool auditChannel mesg- in notifySessionWithTime validClaims conn sendNotifications+ let sendNotifications = void . notifyPool pool dbChannel+ in notifySession validClaims (toS ch) conn sendNotifications waitForever <- newEmptyMVar void $ takeMVar waitForever@@ -88,10 +84,11 @@ -- 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 :: A.Object+ -> Text -> WS.Connection -> (ByteString -> IO ()) -> IO ()-notifySession claimsToSend wsCon send =+notifySession claimsToSend ch wsCon send = withAsync (forever relayData) wait where relayData = jsonMsgWithTime >>= send@@ -100,9 +97,10 @@ -- we need to decode the bytestring to re-encode valid JSON for the notification jsonMsg :: M.HashMap Text A.Value -> ByteString -> ByteString- jsonMsg cl = BL.toStrict . A.encode . Message cl . decodeUtf8With T.lenientDecode+ jsonMsg cl = BL.toStrict . A.encode . Message cl ch . decodeUtf8With T.lenientDecode + claimsWithChannel = M.insert "channel" (A.String ch) claimsToSend claimsWithTime :: IO (M.HashMap Text A.Value) claimsWithTime = do time <- getPOSIXTime- return $ M.insert "message_delivered_at" (A.Number $ fromRational $ toRational time) claimsToSend+ return $ M.insert "message_delivered_at" (A.Number $ fromRational $ toRational time) claimsWithChannel
src/PostgresWebsockets/Broadcast.hs view
@@ -7,13 +7,10 @@ -} module PostgresWebsockets.Broadcast ( Multiplexer (src) , Message (..)- , SourceCommands (..) , newMultiplexer , onMessage , relayMessages , relayMessagesForever- , openChannelProducer- , closeChannelProducer -- * Re-exports , readTQueue , writeTQueue@@ -27,14 +24,12 @@ import GHC.Show -data SourceCommands = Open ByteString | Close ByteString deriving (Show) data Message = Message { channel :: ByteString- , payload :: ByteString- } deriving (Eq, Show)+ , payload :: ByteString+ } deriving (Eq, Show) data Multiplexer = Multiplexer { channels :: M.Map ByteString Channel , src :: ThreadId- , commands :: TQueue SourceCommands , messages :: TQueue Message } @@ -43,17 +38,8 @@ data Channel = Channel { broadcast :: TChan Message , listeners :: Integer- , close :: STM () } --- | Open a multiplexer's channel-openChannelProducer :: Multiplexer -> ByteString -> STM ()-openChannelProducer multi ch = writeTQueue (commands multi) (Open ch)---- | Close a multiplexer's channel-closeChannelProducer :: Multiplexer -> ByteString -> STM ()-closeChannelProducer multi chan = writeTQueue (commands multi) (Close chan)- -- | Opens a thread that relays messages from the producer thread to the channels forever relayMessagesForever :: Multiplexer -> IO ThreadId relayMessagesForever = forkIO . forever . relayMessages@@ -68,24 +54,21 @@ Nothing -> return () Just c -> writeTChan (broadcast c) m -newMultiplexer :: (TQueue SourceCommands -> TQueue Message -> IO a)+newMultiplexer :: (TQueue Message -> IO a) -> (Either SomeException a -> IO ()) -> IO Multiplexer newMultiplexer openProducer closeProducer = do- cmds <- newTQueueIO msgs <- newTQueueIO- m <- liftA2 Multiplexer M.newIO (forkFinally (openProducer cmds msgs) closeProducer)- return $ m cmds msgs+ m <- liftA2 Multiplexer M.newIO (forkFinally (openProducer msgs) closeProducer)+ return $ m msgs openChannel :: Multiplexer -> ByteString -> STM Channel openChannel multi chan = do c <- newBroadcastTChan let newChannel = Channel{ broadcast = c , listeners = 0- , close = closeChannelProducer multi chan } M.insert newChannel chan (channels multi)- openChannelProducer multi chan return newChannel {- | Adds a listener to a certain multiplexer's channel.@@ -103,15 +86,14 @@ mC <- M.lookup chan (channels multi) let c = fromMaybe (panic $ "trying to remove listener from non existing channel: " <> toS chan) mC M.delete chan (channels multi)- if listeners c - 1 > 0- then M.insert Channel{ broadcast = broadcast c, listeners = listeners c - 1, close = close c} chan (channels multi)- else closeChannelProducer multi chan+ when (listeners c - 1 > 0) $+ 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 addListener ch = do M.delete chan (channels multi)- let newChannel = Channel{ broadcast = broadcast ch, listeners = listeners ch + 1, close = close ch}+ let newChannel = Channel{ broadcast = broadcast ch, listeners = listeners ch + 1} M.insert newChannel chan (channels multi) dupTChan $ broadcast newChannel
src/PostgresWebsockets/Claims.hs view
@@ -89,8 +89,8 @@ hs256jwk :: ByteString -> JWK hs256jwk key = fromKeyMaterial km- & jwkUse .~ Just Sig- & jwkAlg .~ (Just $ JWSAlg HS256)+ & jwkUse ?~ Sig+ & jwkAlg ?~ JWSAlg HS256 where km = OctKeyMaterial (OctKeyParameters (JOSE.Types.Base64Octets key))
src/PostgresWebsockets/Database.hs view
@@ -12,9 +12,9 @@ import Protolude hiding (replace) import Hasql.Pool (Pool, UsageError, use)-import Hasql.Session (sql, run, query)+import Hasql.Session (sql, run, statement) import qualified Hasql.Session as S-import Hasql.Query (statement)+import qualified Hasql.Statement as HST import Hasql.Connection (Connection, withLibPQConnection) import qualified Hasql.Decoders as HD import qualified Hasql.Encoders as HE@@ -45,19 +45,19 @@ -- | Given a Hasql Pool, a channel and a message sends a notify command to the database notifyPool :: Pool -> ByteString -> ByteString -> IO (Either Error ()) notifyPool pool channel mesg =- mapError <$> use pool (query (toS channel, toS mesg) callStatement)+ mapError <$> use pool (statement (toS channel, toS mesg) callStatement) where mapError :: Either UsageError () -> Either Error () mapError = mapLeft (NotifyError . show)- callStatement = statement ("SELECT pg_notify" <> "($1, $2)") encoder HD.unit False- encoder = contramap fst (HE.value HE.text) <> contramap snd (HE.value HE.text)+ callStatement = HST.Statement ("SELECT pg_notify" <> "($1, $2)") encoder HD.unit False+ encoder = contramap fst (HE.param HE.text) <> contramap snd (HE.param HE.text) -- | Given a Hasql Connection, a channel and a message sends a notify command to the database notify :: Connection -> PgIdentifier -> ByteString -> IO (Either Error ()) notify con channel mesg = mapError <$> run (sql ("NOTIFY " <> fromPgIdentifier channel <> ", '" <> mesg <> "'")) con where- mapError :: Either S.Error () -> Either Error ()+ mapError :: Either S.QueryError () -> Either Error () mapError = mapLeft (NotifyError . show) -- | Given a Hasql Connection and a channel sends a listen command to the database@@ -82,7 +82,7 @@ waitForNotifications :: (ByteString -> ByteString -> IO()) -> Connection -> IO () waitForNotifications sendNotification con =- withLibPQConnection con $ void . forkIO . forever . pqFetch+ withLibPQConnection con $ void . forever . pqFetch where pqFetch pqCon = do mNotification <- PQ.notifies pqCon
src/PostgresWebsockets/HasqlBroadcast.hs view
@@ -11,9 +11,11 @@ , relayMessagesForever ) where -import Protolude+import Protolude hiding (putErrLn) import Hasql.Connection+import Data.Aeson (decode, Value(..))+import Data.HashMap.Lazy (lookupDefault) import Data.Either.Combinators (mapBoth) import Data.Function (id) import Control.Retry (RetryStatus, retrying, capDelay, exponentialBackoff)@@ -24,15 +26,19 @@ {- | 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 :: ByteString -> IO Multiplexer-newHasqlBroadcaster = newHasqlBroadcasterForConnection . tryUntilConnected+newHasqlBroadcaster :: ByteString -> ByteString -> IO Multiplexer+newHasqlBroadcaster ch = newHasqlBroadcasterForConnection . tryUntilConnected+ where+ newHasqlBroadcasterForConnection = newHasqlBroadcasterForChannel ch {- | 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 :: ByteString -> IO (Either ByteString Multiplexer)-newHasqlBroadcasterOrError =+newHasqlBroadcasterOrError :: ByteString -> ByteString -> IO (Either ByteString Multiplexer)+newHasqlBroadcasterOrError ch = acquire >=> (sequence . mapBoth show (newHasqlBroadcasterForConnection . return))+ where+ newHasqlBroadcasterForConnection = newHasqlBroadcasterForChannel ch tryUntilConnected :: ByteString -> IO Connection tryUntilConnected =@@ -50,7 +56,7 @@ return True _ -> return False -{- | Returns a multiplexer from an IO Connection, listen for different database notification channels using the connection produced.+{- | 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 @@ -72,24 +78,31 @@ @ -}-newHasqlBroadcasterForConnection :: IO Connection -> IO Multiplexer-newHasqlBroadcasterForConnection getCon = do+newHasqlBroadcasterForChannel :: ByteString -> IO Connection -> IO Multiplexer+newHasqlBroadcasterForChannel ch getCon = do multi <- newMultiplexer openProducer closeProducer void $ relayMessagesForever multi return multi where closeProducer _ = putErrLn "Broadcaster is dead"- openProducer cmds msgs = do+ toMsg :: ByteString -> ByteString -> Message+ toMsg c m = case decode (toS m) of+ Just v -> Message (channelDef c v) m+ Nothing -> Message c m++ lookupStringDef :: Text -> ByteString -> Value -> ByteString+ lookupStringDef key d (Object obj) =+ case lookupDefault (String $ toS d) key obj of+ String s -> toS s+ _ -> d+ lookupStringDef _ d _ = d+ channelDef = lookupStringDef "channel"+ openProducer msgs = do con <- getCon+ listen con $ toPgIdentifier ch waitForNotifications- (\c m-> atomically $ writeTQueue msgs $ Message c m)+ (\c m-> atomically $ writeTQueue msgs $ toMsg c m) con- forever $ do- cmd <- atomically $ readTQueue cmds- case cmd of- Open ch -> listen con $ toPgIdentifier ch- Close ch -> unlisten con $ toPgIdentifier ch- putErrLn :: Text -> IO () putErrLn = hPutStrLn stderr
test/BroadcastSpec.hs view
@@ -13,7 +13,7 @@ it "opens a separate thread for a producer function" $ do output <- newTQueueIO :: IO (TQueue ThreadId) - void $ liftIO $ newMultiplexer (\_ _-> do+ void $ liftIO $ newMultiplexer (\_-> do tid <- myThreadId atomically $ writeTQueue output tid ) (\_ -> return ())@@ -23,7 +23,7 @@ describe "relayMessages" $ it "relays a single message from producer to 1 listener on 1 test channel" $ do output <- newTQueueIO :: IO (TQueue Message)- multi <- liftIO $ newMultiplexer (\_ msgs->+ multi <- liftIO $ newMultiplexer (\msgs-> atomically $ writeTQueue msgs (Message "test" "payload")) (\_ -> return ()) void $ onMessage multi "test" $ atomically . writeTQueue output @@ -31,16 +31,3 @@ outMsg <- atomically $ readTQueue output outMsg `shouldBe` Message "test" "payload"- describe "onMessage" $- it "sends an open command to the producer" $ do- output <- newTQueueIO :: IO (TQueue Text)- multi <- liftIO $ newMultiplexer (\cmds _->- atomically $ do- cmd <- readTQueue cmds- writeTQueue output (show cmd :: Text)- ) (\_ -> return ())-- void $ onMessage multi "new channel" (\_ -> return ())-- outMsg <- atomically $ readTQueue output- outMsg `shouldBe` "Open \"new channel\""
test/DatabaseSpec.hs view
@@ -11,16 +11,17 @@ spec :: Spec spec = describe "waitForNotifications" $- it "does not block the connection ann trigger action upon notification" $ do+ it "does trigger action upon notification" $ do conOrError <- H.acquire "postgres://localhost/postgres_ws_test" let con = either (panic . show) id conOrError :: H.Connection notification <- liftIO newEmptyMVar - waitForNotifications (curry $ putMVar notification) con- listen con $ toPgIdentifier "test"+ race_+ (waitForNotifications (curry $ putMVar notification) con)+ (do listen con $ toPgIdentifier "test" - conOrError2 <- H.acquire "postgres://localhost/postgres_ws_test"- let con2 = either (panic . show) id conOrError2 :: H.Connection- void $ notify con2 (toPgIdentifier "test") "hello there"+ conOrError2 <- H.acquire "postgres://localhost/postgres_ws_test"+ let con2 = either (panic . show) id conOrError2 :: H.Connection+ void $ notify con2 (toPgIdentifier "test") "hello there" - readMVar notification `shouldReturn` ("test", "hello there")+ readMVar notification `shouldReturn` ("test", "hello there"))
test/HasqlBroadcastSpec.hs view
@@ -3,46 +3,25 @@ import Protolude import Data.Function (id)-import qualified Hasql.Query as H-import qualified Hasql.Session as H-import qualified Hasql.Decoders as HD-import qualified Hasql.Encoders as HE import Test.Hspec import PostgresWebsockets.Broadcast import PostgresWebsockets.HasqlBroadcast+import PostgresWebsockets.Database spec :: Spec spec = describe "newHasqlBroadcaster" $ do- let booleanQueryShouldReturn con query expected =- either (panic . show) id- <$> H.run query con- `shouldReturn` expected- newConnection connStr =+ let newConnection connStr = either (panic . show) id <$> acquire connStr - it "start listening on a database connection as we send an Open command" $ do- con <- newConnection "postgres://localhost/postgres_ws_test"- multi <- liftIO $ newHasqlBroadcaster "postgres://localhost/postgres_ws_test"-- atomically $ openChannelProducer multi "test channel"- threadDelay 1000000-- let statement = H.statement "SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE query ~* 'LISTEN \\\"test channel\\\"')"- HE.unit (HD.singleRow $ HD.value HD.bool) False- query = H.query () statement- booleanQueryShouldReturn con query True+ it "relay messages sent to the appropriate database channel" $ do+ multi <- either (panic .show) id <$> newHasqlBroadcasterOrError "postgres-websockets" "postgres://localhost/postgres_ws_test"+ msg <- liftIO newEmptyMVar+ onMessage multi "test" $ putMVar msg - it "stops listening on a database connection as we send a Close command" $ do con <- newConnection "postgres://localhost/postgres_ws_test"- multi <- liftIO $ newHasqlBroadcaster "postgres://localhost/postgres_ws_test"-- atomically $ closeChannelProducer multi "test channel"- threadDelay 1000000+ void $ notify con (toPgIdentifier "postgres-websockets") "{\"channel\": \"test\", \"payload\": \"hello there\"}" - let statement = H.statement "SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE query ~* 'UNLISTEN \\\"test channel\\\"')"- HE.unit (HD.singleRow $ HD.value HD.bool) False- query = H.query () statement- booleanQueryShouldReturn con query True+ readMVar msg `shouldReturn` Message "test" "{\"channel\": \"test\", \"payload\": \"hello there\"}"