hasql-notifications 0.2.0.6 → 0.2.1.0

raw patch · 3 files changed

+151/−136 lines, 3 filesdep ~textPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: text

API changes (from Hackage documentation)

Files

• hasql-notifications.cabal +2/−2
• src/Hasql/Notifications.hs +132/−121
• test/Hasql/NotificationsSpec.hs +17/−13

@@ -1,5 +1,5 @@ name:                hasql-notifications-version:             0.2.0.6+version:             0.2.1.0 synopsis:            LISTEN/NOTIFY support for Hasql description:         Use PostgreSQL Asynchronous notification support with your Hasql Types. homepage:            https://github.com/diogob/hasql-notifications@@ -18,7 +18,7 @@   exposed-modules:     Hasql.Notifications   build-depends:       base >= 4.7 && < 5                      , bytestring >= 0.10.8.2-                     , text >= 1.2.3.1 && < 2.2+                     , text >= 2 && < 2.2                      , hasql-pool >= 0.4 && < 0.11                      , bytestring >= 0.10                      , postgresql-libpq >= 0.9 && < 1.0

@@ -1,52 +1,51 @@ {-# LANGUAGE CPP #-}-{-|-  This module has functions to send commands LISTEN and NOTIFY to the database server.-  It also has a function to wait for and handle notifications on a database connection. -  For more information check the [PostgreSQL documentation](https://www.postgresql.org/docs/current/libpq-notify.html).---}+-- |+--  This module has functions to send commands LISTEN and NOTIFY to the database server.+--  It also has a function to wait for and handle notifications on a database connection.+--+--  For more information check the [PostgreSQL documentation](https://www.postgresql.org/docs/current/libpq-notify.html). module Hasql.Notifications-  ( notifyPool-  , notify-  , listen-  , unlisten-  , waitForNotifications-  , PgIdentifier-  , toPgIdentifier-  , fromPgIdentifier-  ) where+  ( notifyPool,+    notify,+    listen,+    unlisten,+    waitForNotifications,+    PgIdentifier,+    toPgIdentifier,+    fromPgIdentifier,+    FatalError (..),+  )+where -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 Control.Concurrent (threadDelay, threadWaitRead)+import Control.Exception (Exception, throw)+import Control.Monad (forever, unless, void)+import Data.ByteString.Char8 (ByteString)+import Data.Functor.Contravariant (contramap) import Data.Text (Text) import qualified Data.Text as T import qualified Data.Text.Encoding as T-import Data.ByteString.Char8 (ByteString)-import Data.Functor.Contravariant (contramap)-import Control.Monad (void, forever)-#if defined(mingw32_HOST_OS)-import           Control.Concurrent ( threadDelay )-#else-import Control.Concurrent (threadWaitRead)-#endif-import Control.Exception (Exception, throw)+import qualified Database.PostgreSQL.LibPQ as PQ+import Hasql.Connection (Connection, withLibPQConnection)+import qualified Hasql.Decoders as HD+import qualified Hasql.Encoders as HE+import Hasql.Pool (Pool, UsageError, use)+import Hasql.Session (run, sql, statement)+import qualified Hasql.Session as S+import qualified Hasql.Statement as HST  -- | A wrapped text that represents a properly escaped and quoted PostgreSQL identifier newtype PgIdentifier = PgIdentifier Text deriving (Show)  -- | Uncatchable exceptions thrown and never caught.-newtype FatalError = FatalError { fatalErrorMessage :: String }-  deriving (Show)+newtype FatalError = FatalError {fatalErrorMessage :: String}  instance Exception FatalError +instance Show FatalError where+  show = fatalErrorMessage+ -- | Given a PgIdentifier returns the wrapped text fromPgIdentifier :: PgIdentifier -> Text fromPgIdentifier (PgIdentifier bs) = bs@@ -60,101 +59,114 @@     strictlyReplaceQuotes = T.replace "\"" ("\"\"" :: Text)  -- | Given a Hasql Pool, a channel and a message sends a notify command to the database-notifyPool :: Pool -- ^ Pool from which the connection will be used to issue a NOTIFY command.-           -> Text -- ^ Channel where to send the notification-           -> Text -- ^ Payload to be sent with the notification-           -> IO (Either UsageError ())+notifyPool ::+  -- | Pool from which the connection will be used to issue a NOTIFY command.+  Pool ->+  -- | Channel where to send the notification+  Text ->+  -- | Payload to be sent with the notification+  Text ->+  IO (Either UsageError ()) notifyPool pool channel mesg =-   use pool (statement (channel, mesg) callStatement)-   where-     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)+  use pool (statement (channel, mesg) callStatement)+  where+    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 -- ^ Connection to be used to send the NOTIFY command-       -> PgIdentifier -- ^ Channel where to send the notification-       -> Text -- ^ Payload to be sent with the notification-       -> IO (Either S.QueryError ())+notify ::+  -- | Connection to be used to send the NOTIFY command+  Connection ->+  -- | Channel where to send the notification+  PgIdentifier ->+  -- | Payload to be sent with the notification+  Text ->+  IO (Either S.QueryError ()) notify con channel mesg =-   run (sql $ T.encodeUtf8 ("NOTIFY " <> fromPgIdentifier channel <> ", '" <> mesg <> "'")) con--{-| -  Given a Hasql Connection and a channel sends a listen command to the database.-  Once the connection sends the LISTEN command the server register its interest in the channel.-  Hence it's important to keep track of which connection was used to open the listen command.--  Example of listening and waiting for a notification:--  @-  import System.Exit (die)--  import Hasql.Connection-  import Hasql.Notifications+  run (sql $ T.encodeUtf8 ("NOTIFY " <> fromPgIdentifier channel <> ", '" <> mesg <> "'")) con -  main :: IO ()-  main = do-    dbOrError <- acquire "postgres://localhost/db_name"-    case dbOrError of-        Right db -> do-            let channelToListen = toPgIdentifier "sample-channel"-            listen db channelToListen-            waitForNotifications (\channel _ -> print $ "Just got notification on channel " <> channel) db-        _ -> die "Could not open database connection"-  @--}-listen :: Connection -- ^ Connection to be used to send the LISTEN command-       -> PgIdentifier -- ^ Channel this connection will be registered to listen to-       -> IO ()+-- |+--  Given a Hasql Connection and a channel sends a listen command to the database.+--  Once the connection sends the LISTEN command the server register its interest in the channel.+--  Hence it's important to keep track of which connection was used to open the listen command.+--+--  Example of listening and waiting for a notification:+--+--  @+--  import System.Exit (die)+--+--  import Hasql.Connection+--  import Hasql.Notifications+--+--  main :: IO ()+--  main = do+--    dbOrError <- acquire "postgres://localhost/db_name"+--    case dbOrError of+--        Right db -> do+--            let channelToListen = toPgIdentifier "sample-channel"+--            listen db channelToListen+--            waitForNotifications (\channel _ -> print $ "Just got notification on channel " <> channel) db+--        _ -> die "Could not open database connection"+--  @+listen ::+  -- | Connection to be used to send the LISTEN command+  Connection ->+  -- | Channel this connection will be registered to listen to+  PgIdentifier ->+  IO () listen con channel =   void $ withLibPQConnection con execListen   where     execListen pqCon = void $ PQ.exec pqCon $ T.encodeUtf8 $ "LISTEN " <> fromPgIdentifier channel  -- | Given a Hasql Connection and a channel sends a unlisten command to the database-unlisten :: Connection -- ^ Connection currently registerd by a previous 'listen' call-         -> PgIdentifier -- ^ Channel this connection will be deregistered from-         -> IO ()+unlisten ::+  -- | Connection currently registerd by a previous 'listen' call+  Connection ->+  -- | Channel this connection will be deregistered from+  PgIdentifier ->+  IO () unlisten con channel =   void $ withLibPQConnection con execListen   where     execListen pqCon = void $ PQ.exec pqCon $ T.encodeUtf8 $ "UNLISTEN " <> fromPgIdentifier channel --{-| -  Given a function that handles notifications and a Hasql connection it will listen -  on the database connection and call the handler everytime a message arrives.--  The message handler passed as first argument needs two parameters channel and payload.-  See an example of handling notification on a separate thread:--  @-  import Control.Concurrent.Async (async)-  import Control.Monad (void)-  import System.Exit (die)--  import Hasql.Connection-  import Hasql.Notifications--  notificationHandler :: ByteString -> ByteString -> IO()-  notificationHandler channel payload = -    void $ async do-      print $ "Handle payload " <> payload <> " in its own thread"--  main :: IO ()-  main = do-    dbOrError <- acquire "postgres://localhost/db_name"-    case dbOrError of-        Right db -> do-            let channelToListen = toPgIdentifier "sample-channel"-            listen db channelToListen-            waitForNotifications notificationHandler db-        _ -> die "Could not open database connection"-  @--}--waitForNotifications :: (ByteString -> ByteString -> IO()) -- ^ Callback function to handle incoming notifications-                     -> Connection -- ^ Connection where we will listen to-                     -> IO ()+-- |+--  Given a function that handles notifications and a Hasql connection it will listen+--  on the database connection and call the handler everytime a message arrives.+--+--  The message handler passed as first argument needs two parameters channel and payload.+--  See an example of handling notification on a separate thread:+--+--  @+--  import Control.Concurrent.Async (async)+--  import Control.Monad (void)+--  import System.Exit (die)+--+--  import Hasql.Connection+--  import Hasql.Notifications+--+--  notificationHandler :: ByteString -> ByteString -> IO()+--  notificationHandler channel payload =+--    void $ async do+--      print $ "Handle payload " <> payload <> " in its own thread"+--+--  main :: IO ()+--  main = do+--    dbOrError <- acquire "postgres://localhost/db_name"+--    case dbOrError of+--        Right db -> do+--            let channelToListen = toPgIdentifier "sample-channel"+--            listen db channelToListen+--            waitForNotifications notificationHandler db+--        _ -> die "Could not open database connection"+--  @+waitForNotifications ::+  -- | Callback function to handle incoming notifications+  (ByteString -> ByteString -> IO ()) ->+  -- | Connection where we will listen to+  Connection ->+  IO () waitForNotifications sendNotification con =   withLibPQConnection con $ void . forever . pqFetch   where@@ -164,16 +176,15 @@         Nothing -> do           mfd <- PQ.socket pqCon           case mfd of-            Nothing  -> panic "Error checking for PostgreSQL notifications"-#if defined(mingw32_HOST_OS)-            Just _ -> do-              void $ threadDelay 1000000-#else+            Nothing -> void $ threadDelay 1000000             Just fd -> do               void $ threadWaitRead fd-#endif-              void $ PQ.consumeInput pqCon++              result <- PQ.consumeInput pqCon+              unless result $ do+                mError <- PQ.errorMessage pqCon+                panic $ maybe "Error checking for PostgreSQL notifications" (T.unpack . T.decodeUtf8Lenient) mError         Just notification ->-           sendNotification (PQ.notifyRelname notification) (PQ.notifyExtra notification)+          sendNotification (PQ.notifyRelname notification) (PQ.notifyExtra notification)     panic :: String -> a     panic a = throw (FatalError a)

@@ -1,16 +1,14 @@ module Hasql.NotificationsSpec (main, spec) where -import Test.Hspec-import Test.QuickCheck- import Control.Concurrent (forkIO, killThread) import Control.Concurrent.MVar import Control.Monad (void)-import System.Exit (die) import Data.ByteString- import Hasql.Connection import Hasql.Notifications+import System.Exit (die)+import Test.Hspec+import Test.QuickCheck  -- `main` is here so that this module can be run from GHCi on its own.  It is -- not needed for automatic spec discovery.@@ -24,15 +22,21 @@       it "should call our notification handler" $ do         dbOrError <- acquire "postgres://postgres:roottoor@localhost/hasql_notifications_test"         case dbOrError of-            Right db -> do-                let channelToListen = toPgIdentifier "test-channel"-                mailbox <- newEmptyMVar :: IO (MVar ByteString)-                listen db channelToListen-                threadId <- forkIO $ waitForNotifications (\channel payload -> putMVar mailbox $ "Just got notification on channel " <> channel <> ": " <> payload) db-                notify db (toPgIdentifier "test-channel") "Payload"-                takeMVar mailbox `shouldReturn` "Just got notification on channel test-channel: Payload"-            _ -> die "Could not open database connection"+          Right db -> do+            let channelToListen = toPgIdentifier "test-channel"+            mailbox <- newEmptyMVar :: IO (MVar ByteString)+            listen db channelToListen+            threadId <- forkIO $ waitForNotifications (\channel payload -> putMVar mailbox $ "Just got notification on channel " <> channel <> ": " <> payload) db+            notify db (toPgIdentifier "test-channel") "Payload"+            takeMVar mailbox `shouldReturn` "Just got notification on channel test-channel: Payload"+          _ -> die "Could not open database connection" +  describe "FatalError show instance" $+    it "extracts message" $+      ( show $+          FatalError {fatalErrorMessage = "some message"}+      )+        `shouldBe` "some message"   describe "toPgIdenfier" $     it "enclose text in quotes doubling existing ones" $       fromPgIdentifier (toPgIdentifier "some \"identifier\"") `shouldBe` "\"some \"\"identifier\"\"\""