packages feed

pg-recorder (empty) → 0.2.0.0

raw patch · 10 files changed

+339/−0 lines, 10 filesdep +basedep +bytestringdep +contravariantsetup-changed

Dependencies added: base, bytestring, contravariant, either, hasql, hasql-pool, hspec, optparse-applicative, optparse-text, pg-recorder, postgresql-libpq, protolude, resource-pool, stringsearch, text

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Author name here (c) 2016++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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ app/Main.hs view
@@ -0,0 +1,20 @@+module Main where++import           PgRecorder+import           PgRecorder.Config+import           Protolude++import           System.IO                 (BufferMode (..), hSetBuffering)++main :: IO ()+main = do+  hSetBuffering stdout LineBuffering+  hSetBuffering stdin  LineBuffering+  hSetBuffering stderr NoBuffering++  conf <- readOptions+  let poolSettings = (10, 10, toS $ configDatabase conf)+      dispatcher = dispatcherFunction conf+  notificationHandler <- dbNotificationHandler poolSettings dispatcher+  putStrLn $ "Listening for notification on " <> configDatabase conf <> " using dispatcher: " <> dispatcher+  listenSession conf notificationHandler
+ pg-recorder.cabal view
@@ -0,0 +1,70 @@+name:                pg-recorder+version:             0.2.0.0+synopsis:            Initial project template from stack+description:         Please see README.md+homepage:            https://github.com/githubuser/pg-recorder#readme+license:             BSD3+license-file:        LICENSE+author:              Diogo Biazus+maintainer:          diogo@biazus.me+copyright:           2016 Diogo Biazus+category:            Web+build-type:          Simple+-- extra-source-files:+cabal-version:       >=1.10++library+  hs-source-dirs:      src+  ghc-options:         -Wall+  exposed-modules:     PgRecorder+                     , PgRecorder.Config+                     , PgRecorder.Database+  other-modules:       Paths_pg_recorder+                     , PgRecorder.Prelude+  build-depends:       base >= 4.7 && < 5+                     , protolude >= 0.1.6 && < 0.2+                     , optparse-applicative >= 0.12+                     , optparse-text >= 0.1.1.0+                     , resource-pool >= 0.2.3 && < 0.3+                     , text >= 1.2.2.1+                     , postgresql-libpq >= 0.9 && < 1.0+                     , hasql >= 0.19+                     , hasql-pool >= 0.4+                     , either >= 4.4 && < 5.0+                     , bytestring >= 0.10.8 && < 0.11+                     , stringsearch >= 0.3.6 && < 0.4+                     , contravariant++  default-language:    Haskell2010+  default-extensions:  OverloadedStrings, NoImplicitPrelude++executable pg-recorder+  hs-source-dirs:      app+  main-is:             Main.hs+  ghc-options:         -Wall -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , pg-recorder+                     , protolude >= 0.1.6 && < 0.2+  default-language:    Haskell2010+  default-extensions:  OverloadedStrings, NoImplicitPrelude++test-suite pg-recorder-test+  type:                exitcode-stdio-1.0+  hs-source-dirs:      test+  main-is:             Spec.hs+  other-modules:       DatabaseSpec+  build-depends:       base+                     , pg-recorder+                     , resource-pool >= 0.2.3 && < 0.3+                     , hspec >= 2.2.3 && < 2.5+                     , protolude >= 0.1.6 && < 0.2+                     , postgresql-libpq >= 0.9 && < 1.0+                     , hasql >= 0.19+                     , hasql-pool >= 0.4+  ghc-options:         -Wall -threaded -rtsopts -with-rtsopts=-N+  default-language:    Haskell2010+  default-extensions:  OverloadedStrings, NoImplicitPrelude++source-repository head+  type:     git+  location: https://github.com/githubuser/pg-recorder
+ src/PgRecorder.hs view
@@ -0,0 +1,43 @@+{-|+Module      : PgRecorder+Description : PgRecorder's main module++This is a haddock comment describing your library+For more information on how to write Haddock comments check the user guide:+<https://www.haskell.org/haddock/doc/html/index.html>+-}+module PgRecorder+    ( listenSession+    , dbNotificationHandler+    ) where++import PgRecorder.Prelude+import PgRecorder.Config+import PgRecorder.Database+import qualified Hasql.Pool as HP+import qualified Hasql.Connection                     as H++-- | Given a set of configurations and a way to handle notifications we loop forever fetching notifications and triggering the handler+listenSession :: AppConfig -> (ByteString -> ByteString -> IO ()) -> IO ()+listenSession conf withNotification = do+  con <- either (panic . show) id <$> H.acquire (toS $ configDatabase conf)+  listen con $ toPgIdentifier . toS $ channel conf+  waitForNotifications withNotification con++-- | Given a set of configurations creates a database connection pool and returns an IO database dispatcher to handle notifications+dbNotificationHandler :: HP.Settings -> Text -> IO (ByteString -> ByteString -> IO ())+dbNotificationHandler poolConfig dispatcher =+  dispatchNotificationToDb (toS dispatcher) <$> HP.acquire poolConfig++-- private functions++-- | Given a pool of database connections and a handler dispatches the handler to be executed in its own thread+dispatchNotificationToDb :: ByteString -> HP.Pool -> ByteString -> ByteString -> IO ()+dispatchNotificationToDb dispatcher pool chan notification =+  void $ async executeNotification+  where+    executeNotification = do+      r <- callProcedure pool (toPgIdentifier dispatcher) chan notification+      case r of+        Left e -> print e+        _ -> return ()
+ src/PgRecorder/Config.hs view
@@ -0,0 +1,46 @@+module PgRecorder.Config ( prettyVersion+                          , minimumPgVersion+                          , readOptions+                          , AppConfig (..)+                          ) where++import PgRecorder.Prelude++import qualified Data.Text                   as T+import           Data.Version                (versionBranch)+import           Options.Applicative+import           Options.Applicative.Text+import           Paths_pg_recorder          (version)++-- | Data type to store all command line options+data AppConfig = AppConfig { configDatabase  :: Text+                           , channel :: Text+                           , dispatcherFunction :: Text+                           }++argParser :: Parser AppConfig+argParser = AppConfig+  <$> argument text (help "(REQUIRED) database connection string, e.g. postgres://user:pass@host:port/db" <> metavar "DB_URL")+  <*> textOption    (long "channel"  <> short 'c' <> help "(REQUIRED) channel to listen to notifications for async commands" <> metavar "CHANNEL")+  <*> textOption    (long "dispatcher-function"  <> short 'f' <> help "(REQUIRED) function called to dispatch notifications for async commands" <> metavar "DISPATCHER_FUNCTION")++-- | User friendly version number+prettyVersion :: Text+prettyVersion = T.intercalate "." $ show <$> versionBranch version++-- | Tells the minimum PostgreSQL version required by this version of Haskell Tools+minimumPgVersion :: Integer+minimumPgVersion = 90500++-- | Function to read and parse options from the command line+readOptions :: IO AppConfig+readOptions = customExecParser parserPrefs opts+  where+    opts = info (helper <*> argParser) $+                    fullDesc+                    <> (progDesc . toS) (+                    ("pg-recorder " :: Text)+                    <> prettyVersion+                    <> (" / Records database notifications" :: Text)+                    )+    parserPrefs = prefs showHelpOnError
+ src/PgRecorder/Database.hs view
@@ -0,0 +1,88 @@+{-| This module encapsulates knowledge about the SQL commands and the Hasql interface.+-}+module PgRecorder.Database+  ( callProcedure+  , listen+  , unlisten+  , waitForNotifications+  , PgIdentifier+  , toPgIdentifier+  ) where++import Protolude+import Hasql.Pool (Pool, UsageError, use)+import Hasql.Session (query)+import Hasql.Query (statement)+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 deriving Show++-- | 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+callProcedure :: Pool -> PgIdentifier -> ByteString -> ByteString -> IO (Either Error ())+callProcedure pool procedure channel mesg =+   mapError <$> use pool (query (toS channel, toS mesg) callStatement)+   where+     mapError :: Either UsageError () -> Either Error ()+     mapError = mapLeft (NotifyError . show)+     callStatement = statement ("SELECT " <> fromPgIdentifier procedure <> "($1, $2)") encoder HD.unit False+     encoder = contramap fst (HE.value HE.text) <> contramap snd (HE.value HE.text)++-- | 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)
+ src/PgRecorder/Prelude.hs view
@@ -0,0 +1,13 @@+{-+Welcome to your custom Prelude+Export here everything that should always be in your library scope+For more info on what is exported by Protolude check:+https://github.com/sdiehl/protolude/blob/master/Symbols.md+-}+module PgRecorder.Prelude+    ( module Exports+    , id+    ) where++import Protolude as Exports+import Data.Function (id)
+ test/DatabaseSpec.hs view
@@ -0,0 +1,26 @@+module DatabaseSpec (spec) where++import Protolude++import           Data.Function                        (id)+import qualified Hasql.Connection                     as H+import qualified Hasql.Pool                           as HP+import Test.Hspec++import PgRecorder.Database++spec :: Spec+spec =+  describe "waitForNotifications" $+    it "does not block the connection ann trigger action upon notification" $ do+      conOrError <- H.acquire "postgres://localhost/postgrest_test"+      let con = either (panic . show) id conOrError :: H.Connection+      notification <- liftIO newEmptyMVar++      void $ forkIO $ waitForNotifications (curry $ putMVar notification) con+      listen con $ toPgIdentifier "test"++      pool <- HP.acquire (1, 1, "postgres://localhost/postgrest_test")+      void $ either (panic "error calling procedure") id <$> callProcedure pool (toPgIdentifier "pg_notify") "test" "hello there"++      readMVar notification `shouldReturn` ("test", "hello there")
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}