diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/app/Main.hs b/app/Main.hs new file mode 100644 --- /dev/null +++ b/app/Main.hs @@ -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 diff --git a/pg-recorder.cabal b/pg-recorder.cabal new file mode 100644 --- /dev/null +++ b/pg-recorder.cabal @@ -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 diff --git a/src/PgRecorder.hs b/src/PgRecorder.hs new file mode 100644 --- /dev/null +++ b/src/PgRecorder.hs @@ -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: + +-} +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 () diff --git a/src/PgRecorder/Config.hs b/src/PgRecorder/Config.hs new file mode 100644 --- /dev/null +++ b/src/PgRecorder/Config.hs @@ -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 diff --git a/src/PgRecorder/Database.hs b/src/PgRecorder/Database.hs new file mode 100644 --- /dev/null +++ b/src/PgRecorder/Database.hs @@ -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) diff --git a/src/PgRecorder/Prelude.hs b/src/PgRecorder/Prelude.hs new file mode 100644 --- /dev/null +++ b/src/PgRecorder/Prelude.hs @@ -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) diff --git a/test/DatabaseSpec.hs b/test/DatabaseSpec.hs new file mode 100644 --- /dev/null +++ b/test/DatabaseSpec.hs @@ -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") diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover #-}