diff --git a/app/Config.hs b/app/Config.hs --- a/app/Config.hs +++ b/app/Config.hs @@ -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 diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -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 diff --git a/postgres-websockets.cabal b/postgres-websockets.cabal --- a/postgres-websockets.cabal +++ b/postgres-websockets.cabal @@ -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 diff --git a/src/PostgresWebsockets.hs b/src/PostgresWebsockets.hs --- a/src/PostgresWebsockets.hs +++ b/src/PostgresWebsockets.hs @@ -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 diff --git a/src/PostgresWebsockets/Broadcast.hs b/src/PostgresWebsockets/Broadcast.hs --- a/src/PostgresWebsockets/Broadcast.hs +++ b/src/PostgresWebsockets/Broadcast.hs @@ -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 diff --git a/src/PostgresWebsockets/Claims.hs b/src/PostgresWebsockets/Claims.hs --- a/src/PostgresWebsockets/Claims.hs +++ b/src/PostgresWebsockets/Claims.hs @@ -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)) diff --git a/src/PostgresWebsockets/Database.hs b/src/PostgresWebsockets/Database.hs --- a/src/PostgresWebsockets/Database.hs +++ b/src/PostgresWebsockets/Database.hs @@ -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 diff --git a/src/PostgresWebsockets/HasqlBroadcast.hs b/src/PostgresWebsockets/HasqlBroadcast.hs --- a/src/PostgresWebsockets/HasqlBroadcast.hs +++ b/src/PostgresWebsockets/HasqlBroadcast.hs @@ -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 diff --git a/test/BroadcastSpec.hs b/test/BroadcastSpec.hs --- a/test/BroadcastSpec.hs +++ b/test/BroadcastSpec.hs @@ -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\"" diff --git a/test/DatabaseSpec.hs b/test/DatabaseSpec.hs --- a/test/DatabaseSpec.hs +++ b/test/DatabaseSpec.hs @@ -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")) diff --git a/test/HasqlBroadcastSpec.hs b/test/HasqlBroadcastSpec.hs --- a/test/HasqlBroadcastSpec.hs +++ b/test/HasqlBroadcastSpec.hs @@ -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\"}"