hasql-notifications (empty) → 0.1.0.0

raw patch · 8 files changed

+325/−0 lines, 8 filesdep +QuickCheckdep +basedep +bytestringsetup-changed

Dependencies added: QuickCheck, base, bytestring, contravariant, hasql, hasql-notifications, hasql-pool, hspec, postgresql-libpq, text

Files

• LICENSE +30/−0
• README.md +5/−0
• Setup.hs +2/−0
• app/Main.hs +21/−0
• hasql-notifications.cabal +59/−0
• src/Hasql/Notifications.hs +169/−0
• test/Hasql/NotificationsSpec.hs +38/−0
• test/Spec.hs +1/−0

@@ -0,0 +1,30 @@+Diogo Biazus (c) 2020++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Author name here nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,5 @@+# hasql-notifications++[![CircleCI](https://circleci.com/gh/diogob/hasql-notifications.svg?style=svg)](https://circleci.com/gh/diogob/hasql-notifications)++TBD

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,21 @@+module Main where++import System.IO+import System.Exit (die)++import Hasql.Connection+import Hasql.Notifications++main :: IO ()+main = do+    hSetBuffering stdout NoBuffering+    hSetBuffering stderr NoBuffering++    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"+

@@ -0,0 +1,59 @@+name:                hasql-notifications+version:             0.1.0.0+synopsis:            LISTEN/NOTIFY support for Hasql+description:         Use PostgreSQL Asynchronous notification support with your Hasql Types.+homepage:            https://github.com/githubuser/hasql-notifications#readme+license:             BSD3+license-file:        LICENSE+author:              Diogo Biazus+maintainer:          diogo@biazus.ca+copyright:           2020 Diogo Biazus+category:            Hasql, Database, PostgreSQL+build-type:          Simple+extra-source-files:  README.md+cabal-version:       >=1.10++library+  hs-source-dirs:      src+  exposed-modules:     Hasql.Notifications+  build-depends:       base >= 4.7 && < 5+                     , bytestring >= 0.10.8.2+                     , text >= 1.2.3.1 && < 1.3+                     , hasql-pool >= 0.4 && < 0.6+                     , bytestring >= 0.10+                     , postgresql-libpq >= 0.9 && < 1.0+                     , hasql >= 0.19+                     , contravariant >= 1.5.2 && < 1.6++  default-language:    Haskell2010+  ghc-options:         -Wall+  default-extensions: OverloadedStrings++executable hasql-notifications+  hs-source-dirs:      app+  main-is:             Main.hs+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , hasql+                     , hasql-notifications+  default-language:    Haskell2010+  default-extensions: OverloadedStrings++test-suite hasql-notifications-test+  type:                exitcode-stdio-1.0+  other-modules:       Hasql.NotificationsSpec+  hs-source-dirs:      test+  main-is:             Spec.hs+  build-depends:       base+                     , hasql+                     , hasql-notifications+                     , hspec+                     , bytestring+                     , QuickCheck+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  default-language:    Haskell2010+  default-extensions: OverloadedStrings++source-repository head+  type:     git+  location: https://github.com/diogob/hasql-notifications

@@ -0,0 +1,169 @@+{-|+  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++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.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)+import Control.Concurrent (threadWaitRead)+import Control.Exception (Exception, throw)++-- | A wrapped bytestring that represents a properly escaped and quoted PostgreSQL identifier+newtype PgIdentifier = PgIdentifier ByteString deriving (Show)++-- | Uncatchable exceptions thrown and never caught.+newtype FatalError = FatalError { fatalErrorMessage :: String }+  deriving (Show)++instance Exception FatalError++-- | 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 :: Text -> PgIdentifier+toPgIdentifier x =+  PgIdentifier $ "\"" <> T.encodeUtf8 (strictlyReplaceQuotes x) <> "\""+  where+    strictlyReplaceQuotes :: Text -> Text+    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 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)++-- | 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 con channel mesg =+   run (sql ("NOTIFY " <> fromPgIdentifier channel <> ", '" <> T.encodeUtf8 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++  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 ()+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 -- ^ Connection currently registerd by a previous 'listen' call+         -> PgIdentifier -- ^ Channel this connection will be deregistered from+         -> 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 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 ()+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)+    panic :: String -> a+    panic a = throw (FatalError a)

@@ -0,0 +1,38 @@+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++-- `main` is here so that this module can be run from GHCi on its own.  It is+-- not needed for automatic spec discovery.+main :: IO ()+main = hspec spec++spec :: Spec+spec = do+  describe "send and receive notification" $+    describe "when I send a notification to channel my handler is listening to" $+      it "should call our notification handler" $ do+        dbOrError <- acquire "postgres://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"++  describe "toPgIdenfier" $+    it "enclose text in quotes doubling existing ones" $+      fromPgIdentifier (toPgIdentifier "some \"identifier\"") `shouldBe` "\"some \"\"identifier\"\"\""

@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}