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.5.0.2 +version: 0.6.0.0 synopsis: Middleware to map LISTEN/NOTIFY messages to Websockets description: Please see README.md homepage: https://github.com/diogob/postgres-websockets#readme @@ -8,7 +8,7 @@ author: Diogo Biazus maintainer: diogo@biazus.ca copyright: 2016 Diogo Biazus -category: Web +category: Web, Database, PostgreSQL build-type: Simple -- extra-source-files: cabal-version: >=1.10 @@ -18,7 +18,6 @@ ghc-options: -Wall exposed-modules: PostgresWebsockets , PostgresWebsockets.Broadcast - , PostgresWebsockets.Database , PostgresWebsockets.HasqlBroadcast , PostgresWebsockets.Claims build-depends: base >= 4.7 && < 5 @@ -37,6 +36,7 @@ , aeson >= 0.11 , protolude >= 0.2 , hasql >= 0.19 + , hasql-notifications >= 0.1.0.0 && < 0.2 , either , stm-containers , stm @@ -77,7 +77,6 @@ main-is: Spec.hs other-modules: BroadcastSpec , ClaimsSpec - , DatabaseSpec , HasqlBroadcastSpec build-depends: base , protolude >= 0.2 @@ -89,6 +88,7 @@ , aeson , hasql , hasql-pool + , hasql-notifications >= 0.1.0.0 && < 0.2 , http-types , unordered-containers >= 0.2 , wai-extra diff --git a/src/PostgresWebsockets.hs b/src/PostgresWebsockets.hs --- a/src/PostgresWebsockets.hs +++ b/src/PostgresWebsockets.hs @@ -11,6 +11,7 @@ ) where import qualified Hasql.Pool as H +import qualified Hasql.Notifications as H import qualified Network.Wai as Wai import qualified Network.Wai.Handler.WebSockets as WS import qualified Network.WebSockets as WS @@ -25,7 +26,6 @@ import PostgresWebsockets.Broadcast (Multiplexer, onMessage) import qualified PostgresWebsockets.Broadcast as B import PostgresWebsockets.Claims -import PostgresWebsockets.Database import PostgresWebsockets.HasqlBroadcast (newHasqlBroadcaster, newHasqlBroadcasterOrError) @@ -38,7 +38,7 @@ 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 :: ByteString -> ByteString -> H.Pool -> Multiplexer -> Wai.Application -> Wai.Application +postgresWsMiddleware :: Text -> ByteString -> H.Pool -> Multiplexer -> Wai.Application -> Wai.Application postgresWsMiddleware = WS.websocketsOr WS.defaultConnectionOptions `compose` wsApp where @@ -48,7 +48,7 @@ -- when the websocket is closed a ConnectionClosed Exception is triggered -- this kills all children and frees resources for us -wsApp :: ByteString -> ByteString -> H.Pool -> Multiplexer -> WS.ServerApp +wsApp :: Text -> ByteString -> H.Pool -> Multiplexer -> WS.ServerApp wsApp dbChannel secret pool multi pendingConn = validateClaims requestChannel secret (toS jwtToken) >>= either rejectRequest forkSessions where @@ -74,7 +74,7 @@ onMessage multi ch $ WS.sendTextData conn . B.payload when (hasWrite mode) $ - let sendNotifications = void . notifyPool pool dbChannel + let sendNotifications = void . (H.notifyPool pool dbChannel) . toS in notifySession validClaims (toS ch) conn sendNotifications waitForever <- newEmptyMVar diff --git a/src/PostgresWebsockets/Database.hs b/src/PostgresWebsockets/Database.hs deleted file mode 100644 --- a/src/PostgresWebsockets/Database.hs +++ /dev/null @@ -1,98 +0,0 @@ -{-| This module encapsulates knowledge about the SQL commands and the Hasql interface. --} -module PostgresWebsockets.Database - ( notifyPool - , notify - , listen - , unlisten - , waitForNotifications - , PgIdentifier - , toPgIdentifier - ) where - -import Protolude hiding (replace) -import Hasql.Pool (Pool, UsageError, use) -import Hasql.Session (sql, run, statement) -import qualified Hasql.Session as S -import qualified Hasql.Statement as HST -import Hasql.Connection (Connection, withLibPQConnection) -import qualified Hasql.Decoders as HD -import qualified Hasql.Encoders as HE -import qualified Database.PostgreSQL.LibPQ as PQ -import Data.Either.Combinators -import qualified Data.ByteString.Char8 as B -import Data.ByteString.Search (replace) -import Data.Functor.Contravariant (contramap) - -newtype Error = NotifyError Text - --- | A wrapped bytestring that represents a properly escaped and quoted PostgreSQL identifier -newtype PgIdentifier = PgIdentifier ByteString deriving (Show) - --- | Given a PgIdentifier returns the wrapped bytestring -fromPgIdentifier :: PgIdentifier -> ByteString -fromPgIdentifier (PgIdentifier bs) = bs - --- | Given a bytestring returns a properly quoted and escaped PgIdentifier -toPgIdentifier :: ByteString -> PgIdentifier -toPgIdentifier x = PgIdentifier $ "\"" <> strictlyReplaceQuotes (trimNullChars x) <> "\"" - where - trimNullChars :: ByteString -> ByteString - trimNullChars = B.takeWhile (/= '\x0') - strictlyReplaceQuotes :: ByteString -> ByteString - strictlyReplaceQuotes = toS . replace "\"" ("\"\"" :: ByteString) - --- | 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 (statement (toS channel, toS mesg) callStatement) - where - mapError :: Either UsageError () -> Either Error () - mapError = mapLeft (NotifyError . show) - callStatement = HST.Statement ("SELECT pg_notify" <> "($1, $2)") encoder HD.noResult False - encoder = contramap fst (HE.param $ HE.nonNullable $ HE.text) <> contramap snd (HE.param $ HE.nonNullable $ 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.QueryError () -> Either Error () - mapError = mapLeft (NotifyError . show) - --- | Given a Hasql Connection and a channel sends a listen command to the database -listen :: Connection -> PgIdentifier -> IO () -listen con channel = - void $ withLibPQConnection con execListen - where - execListen pqCon = void $ PQ.exec pqCon $ "LISTEN " <> fromPgIdentifier channel - --- | Given a Hasql Connection and a channel sends a unlisten command to the database -unlisten :: Connection -> PgIdentifier -> IO () -unlisten con channel = - void $ withLibPQConnection con execListen - where - execListen pqCon = void $ PQ.exec pqCon $ "UNLISTEN " <> fromPgIdentifier channel - - -{- | Given a function that handles notifications and a Hasql connection forks a thread that listens on the database connection and calls the handler everytime a message arrives. - - The message handler passed as first argument needs two parameters channel and payload. --} - -waitForNotifications :: (ByteString -> ByteString -> IO()) -> Connection -> IO () -waitForNotifications sendNotification con = - withLibPQConnection con $ void . forever . pqFetch - where - pqFetch pqCon = do - mNotification <- PQ.notifies pqCon - case mNotification of - Nothing -> do - mfd <- PQ.socket pqCon - case mfd of - Nothing -> panic "Error checking for PostgreSQL notifications" - Just fd -> do - void $ threadWaitRead fd - void $ PQ.consumeInput pqCon - Just notification -> - sendNotification (PQ.notifyRelname notification) (PQ.notifyExtra notification) diff --git a/src/PostgresWebsockets/HasqlBroadcast.hs b/src/PostgresWebsockets/HasqlBroadcast.hs --- a/src/PostgresWebsockets/HasqlBroadcast.hs +++ b/src/PostgresWebsockets/HasqlBroadcast.hs @@ -14,19 +14,19 @@ import Protolude hiding (putErrLn) import Hasql.Connection +import Hasql.Notifications 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) -import PostgresWebsockets.Database import PostgresWebsockets.Broadcast {- | 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 -> ByteString -> IO Multiplexer +newHasqlBroadcaster :: Text -> ByteString -> IO Multiplexer newHasqlBroadcaster ch = newHasqlBroadcasterForConnection . tryUntilConnected where newHasqlBroadcasterForConnection = newHasqlBroadcasterForChannel ch @@ -34,7 +34,7 @@ {- | 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 -> ByteString -> IO (Either ByteString Multiplexer) +newHasqlBroadcasterOrError :: Text -> ByteString -> IO (Either ByteString Multiplexer) newHasqlBroadcasterOrError ch = acquire >=> (sequence . mapBoth show (newHasqlBroadcasterForConnection . return)) where @@ -78,7 +78,7 @@ @ -} -newHasqlBroadcasterForChannel :: ByteString -> IO Connection -> IO Multiplexer +newHasqlBroadcasterForChannel :: Text -> IO Connection -> IO Multiplexer newHasqlBroadcasterForChannel ch getCon = do multi <- newMultiplexer openProducer closeProducer void $ relayMessagesForever multi diff --git a/test/DatabaseSpec.hs b/test/DatabaseSpec.hs deleted file mode 100644 --- a/test/DatabaseSpec.hs +++ /dev/null @@ -1,27 +0,0 @@ -module DatabaseSpec (spec) where - -import Protolude - -import Data.Function (id) -import qualified Hasql.Connection as H -import Test.Hspec - -import PostgresWebsockets.Database - -spec :: Spec -spec = - describe "waitForNotifications" $ - 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 - - 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" - - readMVar notification `shouldReturn` ("test", "hello there")) diff --git a/test/HasqlBroadcastSpec.hs b/test/HasqlBroadcastSpec.hs --- a/test/HasqlBroadcastSpec.hs +++ b/test/HasqlBroadcastSpec.hs @@ -2,13 +2,11 @@ import Protolude -import Data.Function (id) - +import Data.Function (id) import Test.Hspec - import PostgresWebsockets.Broadcast import PostgresWebsockets.HasqlBroadcast -import PostgresWebsockets.Database +import Hasql.Notifications spec :: Spec spec = describe "newHasqlBroadcaster" $ do