postgresql-replicant 0.1.0.1 → 0.2.0.0

raw patch · 13 files changed

+254/−228 lines, 13 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +12/−1
• README.md +16/−1
• app/Main.hs +10/−8
• postgresql-replicant.cabal +10/−8
• src/Database/PostgreSQL/Replicant.hs +27/−55
• src/Database/PostgreSQL/Replicant/Connection.hs +71/−0
• src/Database/PostgreSQL/Replicant/Message.hs +3/−0
• src/Database/PostgreSQL/Replicant/Protocol.hs +32/−30
• src/Database/PostgreSQL/Replicant/Queue.hs +0/−77
• src/Database/PostgreSQL/Replicant/ReplicationSlot.hs +16/−14
• src/Database/PostgreSQL/Replicant/Settings.hs +28/−0
• src/Database/PostgreSQL/Replicant/Types/Lsn.hs +2/−1
• test/Spec.hs +27/−33

@@ -5,6 +5,17 @@  ## [Unreleased] +## [0.2.0.0] - 2021-07-27++This is a minor update that adds a replication connection type,+updates the root module API with more types, and is just a better+quality of life release.++### Added+- ReplicantConnection+- Connection handling functions to Replicant module+- Message types to main Replicant module+ ## [0.1.0.1] - 2021-06-22  This is a minor update to our experimental release, fixing some pretty@@ -16,7 +27,7 @@ - Minor cleanup, throw better errors instead of printing debug   information -## [0.1.0.0] - 2021-07-27+## [0.1.0.0] - 2021-05-04  This is an experimental release to test out this library and get it ready and polished for 1.0!

@@ -1,4 +1,4 @@-# postgres-replicant+# postgresql-replicant  A PostgreSQL streaming logical replication client for Haskell. @@ -58,12 +58,27 @@   let settings = PgSettings "my-user" "my-database" "localhost" "5432" "testing"   withLogicalStream settings $ \change -> do     print change+    return $ changeNextLSN change       `catch` \err ->       print err ```  Which should connect, create a `testing` replication slot, and start sending your callback the changes to your database as they arrive.++The type of the callback to `withLogicalStream` is:++    Change -> IO LSN++Note the return type.  The *LSN* or _Log Sequence Number_ is used by+replicant after running your callback to update its internal stream+state and tell PostgreSQL that you have consumed this change.  This+means that if the connection fails and replicant reconnects to the+_same_ slot it will restart the stream at the last *LSN* replicant was+able to successfully send to the server.++All you have to do is return `changeNextLSN change` at the end of your+callback.  ### WAL Output Plugins 

@@ -1,5 +1,6 @@ module Main where +import Data.Aeson import Data.Maybe import Control.Exception import System.Environment@@ -8,16 +9,17 @@  main :: IO () main = do-  dbUser <- maybe "postgresql" id <$> lookupEnv "PG_USER"-  dbName <- maybe "postgresql" id <$> lookupEnv "PG_DATABASE"-  dbHost <- maybe "localhost" id <$> lookupEnv "PG_HOST"-  dbPort <- maybe "5432" id <$> lookupEnv "PG_PORT"-  dbSlot <- maybe "replicant_test" id <$> lookupEnv "PG_SLOTNAME"-  dbUpdateDelay <- maybe "3000" id <$> lookupEnv "PG_UPDATEDELAY"-  let settings = PgSettings dbUser dbName dbHost dbPort dbSlot dbUpdateDelay+  dbUser <- fromMaybe "postgresql" <$> lookupEnv "PG_USER"+  dbName <- fromMaybe "postgresql" <$> lookupEnv "PG_DATABASE"+  dbHost <- fromMaybe "localhost" <$> lookupEnv "PG_HOST"+  dbPort <- fromMaybe "5432" <$> lookupEnv "PG_PORT"+  dbSlot <- fromMaybe "replicant_test" <$> lookupEnv "PG_SLOTNAME"+  dbUpdateDelay <- fromMaybe "3000" <$> lookupEnv "PG_UPDATEDELAY"+  let settings = PgSettings dbUser Nothing dbName dbHost dbPort dbSlot dbUpdateDelay   withLogicalStream settings $ \change -> do     putStrLn "Change received!"-    print change+    print $ encode change+    pure $ changeNextLSN change   `catch`   \exc -> do     putStrLn "Something bad happened: "

@@ -4,12 +4,12 @@ -- -- see: https://github.com/sol/hpack ----- hash: b9e10bcacd11fef9dd3cf878a5223b111e9355f146fc121702189290e50c23ed+-- hash: 79c5f5333f3b93e29420feb572c62237df61efdaa111a3a4425edb8247a2239f  name:           postgresql-replicant-version:        0.1.0.1+version:        0.2.0.0 synopsis:       PostgreSQL logical streaming replication library-description:    Please see the README on GitHub at <https://github.com/agentultra/postgres-replicant#readme>+description:    Please see the README on GitHub at <https://github.com/agentultra/postgresql-replicant#readme> category:       Experimental, Database homepage:       https://github.com/agentultra/postgresql-replicant#readme bug-reports:    https://github.com/agentultra/postgresql-replicant/issues@@ -31,13 +31,14 @@ library   exposed-modules:       Database.PostgreSQL.Replicant+      Database.PostgreSQL.Replicant.Connection       Database.PostgreSQL.Replicant.Exception       Database.PostgreSQL.Replicant.Message       Database.PostgreSQL.Replicant.PostgresUtils       Database.PostgreSQL.Replicant.Protocol-      Database.PostgreSQL.Replicant.Queue       Database.PostgreSQL.Replicant.ReplicationSlot       Database.PostgreSQL.Replicant.Serialize+      Database.PostgreSQL.Replicant.Settings       Database.PostgreSQL.Replicant.State       Database.PostgreSQL.Replicant.Types.Lsn       Database.PostgreSQL.Replicant.Util@@ -45,7 +46,7 @@       Paths_postgresql_replicant   hs-source-dirs:       src-  default-extensions: DeriveGeneric GADTs LambdaCase OverloadedStrings TypeApplications+  default-extensions: DeriveGeneric GADTs LambdaCase OverloadedStrings RecordWildCards TypeApplications   build-depends:       aeson     , async@@ -69,10 +70,11 @@       Paths_postgresql_replicant   hs-source-dirs:       app-  default-extensions: DeriveGeneric GADTs LambdaCase OverloadedStrings TypeApplications+  default-extensions: DeriveGeneric GADTs LambdaCase OverloadedStrings RecordWildCards TypeApplications   ghc-options: -threaded -rtsopts -with-rtsopts=-N   build-depends:-      base >=4.7 && <5+      aeson+    , base >=4.7 && <5     , postgresql-libpq     , postgresql-replicant   default-language: Haskell2010@@ -84,7 +86,7 @@       Paths_postgresql_replicant   hs-source-dirs:       test-  default-extensions: DeriveGeneric GADTs LambdaCase OverloadedStrings TypeApplications+  default-extensions: DeriveGeneric GADTs LambdaCase OverloadedStrings RecordWildCards TypeApplications   ghc-options: -threaded -rtsopts -with-rtsopts=-N   build-depends:       base >=4.7 && <5

@@ -1,5 +1,3 @@-{-# LANGUAGE RecordWildCards #-}- {-| Module      : Database.PostgreSQL.Replicant Description : A PostgreSQL streaming replication library@@ -28,8 +26,23 @@ -}  module Database.PostgreSQL.Replicant-    ( withLogicalStream+    (+    -- * Types+      Change (..)+    , Column (..)+    , Delete (..)+    , Insert (..)+    , Message (..)     , PgSettings (..)+    , ReplicantConnection+    , Update (..)+    , WalLogData (..)+    -- * Connection Handling+    , connect+    , getConnection+    , unsafeCreateConnection+    -- * Functions+    , withLogicalStream     ) where  import Control.Concurrent@@ -38,38 +51,17 @@ import qualified Data.ByteString.Char8 as B import qualified Data.Text as T import qualified Data.Text.Read as T-import Database.PostgreSQL.LibPQ-import Network.Socket.KeepAlive-import System.Posix.Types+import Database.PostgreSQL.LibPQ hiding (Column) +import Database.PostgreSQL.Replicant.Connection import Database.PostgreSQL.Replicant.Exception import Database.PostgreSQL.Replicant.Protocol import Database.PostgreSQL.Replicant.Message import Database.PostgreSQL.Replicant.ReplicationSlot+import Database.PostgreSQL.Replicant.Settings+import Database.PostgreSQL.Replicant.Types.Lsn import Database.PostgreSQL.Replicant.Util -data PgSettings-  = PgSettings-  { pgUser        :: String-  , pgDbName      :: String-  , pgHost        :: String-  , pgPort        :: String-  , pgSlotName    :: String-  , pgUpdateDelay :: String -- ^ Controls how frequently the-                            -- primaryKeepAlive thread updates-                            -- PostgresSQL in @ms@-  }-  deriving (Eq, Show)--pgConnectionString :: PgSettings -> ByteString-pgConnectionString PgSettings {..} = B.intercalate " "-  [ "user=" <> (B.pack pgUser)-  , "dbname=" <> (B.pack pgDbName)-  , "host=" <> (B.pack pgHost)-  , "port=" <> (B.pack pgPort)-  , "replication=database"-  ]- -- | Connect to a PostgreSQL database as a user with the replication -- attribute and start receiving changes using the logical replication -- protocol.  Logical replication happens at the query level so the@@ -82,36 +74,16 @@ -- -- This function can throw exceptions in @IO@ and shut-down the -- socket in case of any error.-withLogicalStream :: PgSettings -> (Change -> IO a) -> IO ()+withLogicalStream :: PgSettings -> (Change -> IO LSN) -> IO () withLogicalStream settings cb = do-  conn <- connectStart $ pgConnectionString settings-  mFd <- socket conn-  sockFd <- maybeThrow (ReplicantException "withLogicalStream: could not get socket fd") mFd-  pollResult <- pollConnectStart conn sockFd+  conn <- connect settings   let updateFreq = getUpdateDelay settings-  case pollResult of-    PollingFailed -> throwIO $ ReplicantException "withLogicalStream: Unable to connect to the database"-    PollingOk -> do-      maybeInfo <- identifySystemSync conn-      _ <- maybeThrow (ReplicantException "withLogicalStream: could not get system information") maybeInfo-      repSlot <- setupReplicationSlot conn $ B.pack . pgSlotName $ settings-      startReplicationStream conn (slotName repSlot) (slotRestart repSlot) updateFreq cb-      pure ()+  maybeInfo <- identifySystemSync conn+  _ <- maybeThrow (ReplicantException "withLogicalStream: could not get system information") maybeInfo+  repSlot <- setupReplicationSlot conn $ B.pack . pgSlotName $ settings+  startReplicationStream conn (slotName repSlot) (slotRestart repSlot) updateFreq cb+  pure ()   where-    pollConnectStart :: Connection -> Fd -> IO PollingStatus-    pollConnectStart conn fd@(Fd cint) = do-      pollStatus <- connectPoll conn-      case pollStatus of-        PollingReading -> do-          threadWaitRead fd-          pollConnectStart conn fd-        PollingWriting -> do-          threadWaitWrite fd-          pollConnectStart conn fd-        PollingOk -> do-          _ <- setKeepAlive cint $ KeepAlive True 60 2-          pure PollingOk-        PollingFailed -> pure PollingFailed     getUpdateDelay :: PgSettings -> Int     getUpdateDelay PgSettings {..} =       case T.decimal . T.pack $ pgUpdateDelay of

@@ -0,0 +1,71 @@+{-+Module : Database.PostgreSQL.Replicant.Connection+Description : Create replication handling connections to PostgreSQL++A ReplicantConnection is different from a regular Connection because+it uses a special mode that can send replication commands that regular+Connection objects cannots send.+-}+module Database.PostgreSQL.Replicant.Connection+  ( -- * Types+    ReplicantConnection+    -- * Constructor+  , connect+  , getConnection+  , unsafeCreateConnection+  )+where++import Control.Concurrent+import Control.Exception+import Database.PostgreSQL.LibPQ+import Network.Socket.KeepAlive+import System.Posix.Types++import Database.PostgreSQL.Replicant.Exception+import Database.PostgreSQL.Replicant.Settings+import Database.PostgreSQL.Replicant.Util++newtype ReplicantConnection+  = ReplicantConnection { getConnection :: Connection }+  deriving Eq++data ConnectResult+  = ConnectSuccess+  | ConnectFailure+  deriving (Eq, Show)++-- | Connect to the PostgreSQL server in replication mode+connect :: PgSettings -> IO ReplicantConnection+connect settings = do+  conn <- connectStart $ pgConnectionString settings+  mFd <- socket conn+  sockFd <- maybeThrow+    (ReplicantException "withLogicalStream: could not get socket fd") mFd+  pollResult <- pollConnectStart conn sockFd+  case pollResult of+    ConnectFailure -> throwIO+      $ ReplicantException "withLogicalStream: Unable to connect to the database"+    ConnectSuccess -> pure $ ReplicantConnection conn++pollConnectStart :: Connection -> Fd -> IO ConnectResult+pollConnectStart conn fd@(Fd cint) = do+  pollStatus <- connectPoll conn+  case pollStatus of+    PollingReading -> do+      threadWaitRead fd+      pollConnectStart conn fd+    PollingWriting -> do+      threadWaitWrite fd+      pollConnectStart conn fd+    PollingOk -> do+      _ <- setKeepAlive cint $ KeepAlive True 60 2+      pure ConnectSuccess+    PollingFailed -> pure ConnectFailure++-- | Unsafe function for wrapping regular libpq Connection.  This is+-- unsafe because the Connection needs to be set up to send+-- replication commands.  Improperly constructed connections can lead+-- to runtime exceptions.+unsafeCreateConnection :: Connection -> ReplicantConnection+unsafeCreateConnection = ReplicantConnection

@@ -392,7 +392,10 @@ data Change   = Change   { changeNextLSN :: LSN+    -- ^ Return this LSN in your callback to update the stream state+    -- in replicant   , changeDeltas  :: [WalLogData]+    -- ^ The list of WAL log changes in this transaction.   }   deriving (Eq, Generic, Show) 

@@ -26,6 +26,7 @@ import Data.Serialize hiding (flush) import Database.PostgreSQL.LibPQ +import Database.PostgreSQL.Replicant.Connection import Database.PostgreSQL.Replicant.Exception import Database.PostgreSQL.Replicant.Message import Database.PostgreSQL.Replicant.PostgresUtils@@ -49,9 +50,9 @@  -- | Synchronously execute the @IDENTIFY SYSTEM@ command which returns -- some basic system information about the server.-identifySystemSync :: Connection -> IO (Maybe IdentifySystem)+identifySystemSync :: ReplicantConnection -> IO (Maybe IdentifySystem) identifySystemSync conn = do-  result <- exec conn identifySystemCommand+  result <- exec (getConnection conn) identifySystemCommand   case result of     Just r -> do       resultStatus <- resultStatus r@@ -69,17 +70,19 @@                   pure $ Just (IdentifySystem s t logPosLsn d)             _ -> pure Nothing         _ -> do-          err <- fromMaybe "identifySystemSync: unknown error" <$> errorMessage conn+          err <- fromMaybe "identifySystemSync: unknown error"+            <$> errorMessage (getConnection conn)           throwIO $ ReplicantException (B.unpack err)     _ -> do-      err <- fromMaybe "identifySystemSync: unknown error" <$> errorMessage conn+      err <- fromMaybe "identifySystemSync: unknown error"+        <$> errorMessage (getConnection conn)       throwIO $ ReplicantException (B.unpack err)  -- | Create a @START_REPLICATION_SLOT@ query, escaping the slot name -- passed in by the user.-startReplicationCommand :: Connection -> ByteString -> LSN -> IO ByteString+startReplicationCommand :: ReplicantConnection -> ByteString -> LSN -> IO ByteString startReplicationCommand conn slotName systemLogPos = do-  escapedName <- escapeIdentifier conn slotName+  escapedName <- escapeIdentifier (getConnection conn) slotName   case escapedName of     Nothing -> throwIO $ ReplicantException $ "Invalid slot name: " ++ show slotName     Just escaped ->@@ -89,7 +92,7 @@       [ "START_REPLICATION SLOT "       , escaped       , " LOGICAL "-      , (toByteString systemLogPos)+      , toByteString systemLogPos       , " (\"include-lsn\" 'on')"       ] @@ -99,11 +102,11 @@ handleCopyOutData   :: TChan PrimaryKeepAlive   -> WalProgressState-  -> Connection-  -> (Change -> IO a)+  -> ReplicantConnection+  -> (Change -> IO LSN)   -> IO () handleCopyOutData chan walState conn cb = forever $ do-  d <- getCopyData conn False+  d <- getCopyData (getConnection conn) False   case d of     CopyOutRow row -> handleReplicationRow chan walState conn row cb     CopyOutError   -> handleReplicationError conn@@ -112,9 +115,9 @@ handleReplicationRow   :: TChan PrimaryKeepAlive   -> WalProgressState-  -> Connection+  -> ReplicantConnection   -> ByteString-  -> (Change -> IO a)+  -> (Change -> IO LSN)   -> IO () handleReplicationRow keepAliveChan walState _ row cb =   case decode @WalCopyData row of@@ -130,15 +133,14 @@             $ ReplicantException             $ "handleReplicationRow (parse error): " ++ err           Right walLogData -> do-            _ <- updateWalProgress walState (changeNextLSN walLogData)-            _ <- cb walLogData-            pure ()+            consumedLSN <- cb walLogData+            updateWalProgress walState consumedLSN       KeepAliveM keepAlive -> atomically $ writeTChan keepAliveChan keepAlive  -- | Used to re-throw an exception received from the server.-handleReplicationError :: Connection -> IO ()+handleReplicationError :: ReplicantConnection -> IO () handleReplicationError conn = do-  err <- errorMessage conn+  err <- errorMessage $ getConnection conn   throwIO (ReplicantException $ B.unpack . fromMaybe "Unknown error" $ err)   pure () @@ -149,16 +151,16 @@ -- race the /keep-alive/ and /copy data/ handler threads.  It will -- catch and rethrow exceptions from either thread if any fails or -- returns.-startReplicationStream :: Connection -> ByteString -> LSN -> Int -> (Change -> IO a) -> IO ()+startReplicationStream :: ReplicantConnection -> ByteString -> LSN -> Int -> (Change -> IO LSN) -> IO () startReplicationStream conn slotName systemLogPos _ cb = do   let initialWalProgress = WalProgress systemLogPos systemLogPos systemLogPos   walProgressState <- WalProgressState <$> newMVar initialWalProgress   replicationCommandQuery <- startReplicationCommand conn slotName systemLogPos-  result <- exec conn replicationCommandQuery+  result <- exec (getConnection conn) replicationCommandQuery   case result of     Nothing -> do       err <- fromMaybe "startReplicationStream: unknown error starting stream"-        <$> errorMessage conn+        <$> errorMessage (getConnection conn)       throwIO $ ReplicantException $ "startReplicationStream: " ++ B.unpack err     Just r  -> do       status <- resultStatus r@@ -170,18 +172,18 @@             (handleCopyOutData keepAliveChan walProgressState conn cb)             `catch`             \exc -> do-              finish conn+              finish (getConnection conn)               throwIO @SomeException exc           return ()         _ -> do-          err <- fromMaybe "startReplicationStream: unknown error entering COPY mode" <$> errorMessage conn+          err <- fromMaybe "startReplicationStream: unknown error entering COPY mode" <$> errorMessage (getConnection conn)           throwIO $ ReplicantException $ B.unpack err  -- | This listens on the channel for /primary keep-alive messages/ -- from the server and responds to them with the /update status/ -- message using the current WAL stream state.  It will attempt to -- buffer prior update messages when the socket is blocked.-keepAliveHandler :: Connection -> TChan PrimaryKeepAlive -> WalProgressState -> IO ()+keepAliveHandler :: ReplicantConnection -> TChan PrimaryKeepAlive -> WalProgressState -> IO () keepAliveHandler conn msgs walProgressState = forever $ do   mKeepAlive <- atomically $ tryReadTChan msgs   case mKeepAlive of@@ -196,7 +198,7 @@           sendStatusUpdate conn walProgressState  sendStatusUpdate-  :: Connection+  :: ReplicantConnection   -> WalProgressState   -> IO () sendStatusUpdate conn w@(WalProgressState walState) = do@@ -209,20 +211,20 @@         applied         timestamp         DoNotRespond-  copyResult <- putCopyData conn $ encode statusUpdate+  copyResult <- putCopyData (getConnection conn) $ encode statusUpdate   case copyResult of     CopyInOk -> do-      flushResult <- flush conn+      flushResult <- flush (getConnection conn)       case flushResult of         FlushOk -> pure ()         FlushFailed -> do-          err <- fromMaybe "sendStatusUpdate: error flushing message to server" <$> errorMessage conn+          err <- fromMaybe "sendStatusUpdate: error flushing message to server" <$> errorMessage (getConnection conn)           throwIO $ ReplicantException $ B.unpack err-        FlushWriting -> tryAgain conn w+        FlushWriting -> tryAgain (getConnection conn) w     CopyInError -> do-      err <- fromMaybe "sendStatusUpdate: unknown error sending COPY IN" <$> errorMessage conn+      err <- fromMaybe "sendStatusUpdate: unknown error sending COPY IN" <$> errorMessage (getConnection conn)       throwIO $ ReplicantException $ B.unpack err-    CopyInWouldBlock -> tryAgain conn w+    CopyInWouldBlock -> tryAgain (getConnection conn) w   where     tryAgain c ws = do       mSockFd <- socket c

@@ -1,77 +0,0 @@-{-|-Module      : Database.PostgreSQL.Replicant.Queue-Description : Bounded and unbounded FIFO queues-Copyright   : (c) James King, 2020, 2021-License     : BSD3-Maintainer  : james@agentultra.com-Stability   : experimental-Portability : POSIX--Shared FIFO queues--}-module Database.PostgreSQL.Replicant.Queue where--import Control.Concurrent.MVar-import Data.Sequence (Seq, ViewR (..), (<|), (|>))-import qualified Data.Sequence as S--data BoundedFifoQueueMeta a-  = BoundedFifoQueueMeta-  { boundedFifoQueueSize :: Int-  , boundedFifoQueue     :: Seq a-  }-  deriving (Eq, Show)--newtype BoundedFifoQueue a = BoundedFifoQueue (MVar (BoundedFifoQueueMeta a))--newtype BoundedQueueException a-  = BoundedQueueOverflow a-  deriving (Eq, Show)--emptyBounded :: Int -> IO (BoundedFifoQueue a)-emptyBounded size =-  BoundedFifoQueue <$> newMVar (BoundedFifoQueueMeta size S.empty)--enqueueBounded :: BoundedFifoQueue a -> a -> IO (Either (BoundedQueueException a) ())-enqueueBounded (BoundedFifoQueue mQueue) x = do-  b@(BoundedFifoQueueMeta size queue) <- takeMVar mQueue-  if size == S.length queue-    then pure $ Left $ BoundedQueueOverflow x-    else do-    putMVar mQueue $ b { boundedFifoQueue = x <| queue }-    pure $ Right ()--newtype FifoQueue a = FifoQueue (MVar (Seq a))--empty :: IO (FifoQueue a)-empty = FifoQueue <$> newMVar S.empty---- | Return @True@ if the queue is empty-null :: FifoQueue a -> IO Bool-null (FifoQueue mQueue) = do-  queue <- readMVar mQueue-  pure $ S.null queue---- | Remove an item from the end of the non-empty queue.-dequeue :: FifoQueue a -> IO (Maybe a)-dequeue (FifoQueue mQueue) = do-  queue <- takeMVar mQueue-  case S.viewr queue of-    S.EmptyR -> do-      putMVar mQueue queue-      pure Nothing-    rest :> x -> do-      putMVar mQueue rest-      pure $ Just x---- | Put an item on the front of the queue.-enqueue :: FifoQueue a -> a -> IO ()-enqueue (FifoQueue mQueue) x = do-  queue <- takeMVar mQueue-  putMVar mQueue $ x <| queue---- | Put an item on the end of the queue so that it will be dequeued first.-enqueueRight :: FifoQueue a -> a -> IO ()-enqueueRight (FifoQueue mQueue) x = do-  queue <- takeMVar mQueue-  putMVar mQueue $ queue |> x

@@ -14,10 +14,12 @@  import Control.Exception import Data.ByteString (ByteString)+import Data.Maybe import qualified Data.ByteString as B import qualified Data.ByteString.Char8 as B8 import Database.PostgreSQL.LibPQ +import Database.PostgreSQL.Replicant.Connection import Database.PostgreSQL.Replicant.Exception import Database.PostgreSQL.Replicant.Types.Lsn @@ -47,9 +49,9 @@ parseSlotActive "f" = Inactive parseSlotActive _   = Inactive -createReplicationSlotCommand :: Connection -> ByteString -> IO ByteString+createReplicationSlotCommand :: ReplicantConnection -> ByteString -> IO ByteString createReplicationSlotCommand conn slotName = do-  escapedName <- escapeIdentifier conn slotName+  escapedName <- escapeIdentifier (getConnection conn) slotName   case escapedName of     Nothing -> throwIO $ ReplicantException $ "Invalid slot name: " ++ show slotName     Just escaped ->@@ -64,10 +66,10 @@ -- | Create a replication slot using synchronous query execution. -- -- May throw an exception if the command fails.-createReplicationSlotSync :: Connection -> ByteString -> IO ReplicationSlotInfo+createReplicationSlotSync :: ReplicantConnection -> ByteString -> IO ReplicationSlotInfo createReplicationSlotSync conn slotName = do   createReplicationSlotQuery <- createReplicationSlotCommand conn slotName-  result <- exec conn createReplicationSlotQuery+  result <- exec (getConnection conn) createReplicationSlotQuery   case result of     Just r -> do       resultStatus <- resultStatus r@@ -82,18 +84,18 @@                 Left _ -> throwIO $ ReplicantException "createReplicationSlotSync: invalid LSN detected"                 Right lsn -> pure $ ReplicationSlotInfo s op Logical Active lsn             _ -> do-              err <- maybe "createReplicationSlotSync: unknown error" id <$> errorMessage conn+              err <- fromMaybe "createReplicationSlotSync: unknown error" <$> errorMessage (getConnection conn)               throwIO $ ReplicantException (B8.unpack err)         _ -> do-          err <- maybe "createReplicationSlotSync: unknown error" id <$> errorMessage conn+          err <- fromMaybe "createReplicationSlotSync: unknown error" <$> errorMessage (getConnection conn)           throwIO $ ReplicantException (B8.unpack err)     _ -> do-      err <- maybe "createReplicationSlotSync: unknown error" id <$> errorMessage conn+      err <- fromMaybe "createReplicationSlotSync: unknown error" <$> errorMessage (getConnection conn)       throwIO $ ReplicantException (B8.unpack err) -getReplicationSlotInfoCommand :: Connection -> ByteString -> IO ByteString+getReplicationSlotInfoCommand :: ReplicantConnection -> ByteString -> IO ByteString getReplicationSlotInfoCommand conn slotName = do-  escapedName <- escapeStringConn conn slotName+  escapedName <- escapeStringConn (getConnection conn) slotName   case escapedName of     Nothing -> throwIO $ ReplicantException $ "Invalid slot name: " ++ show slotName     Just escaped ->@@ -109,10 +111,10 @@ -- @Nothing@ when the requested slot cannot be found. -- -- May throw an exception if the command query fails.-getReplicationSlotSync :: Connection -> ByteString -> IO (Maybe ReplicationSlotInfo)+getReplicationSlotSync :: ReplicantConnection -> ByteString -> IO (Maybe ReplicationSlotInfo) getReplicationSlotSync conn slotName = do   replicationSlotInfoQuery <- getReplicationSlotInfoCommand conn slotName-  result <- exec conn replicationSlotInfoQuery+  result <- exec (getConnection conn) replicationSlotInfoQuery   case result of     Just r -> do       resultStatus <- resultStatus r@@ -135,16 +137,16 @@               _ ->  pure Nothing         _ -> pure Nothing     _ -> do-      err <- maybe "getReplicationSlotSync: unknown error" id <$> errorMessage conn+      err <- fromMaybe "getReplicationSlotSync: unknown error" <$> errorMessage (getConnection conn)       throwIO $ ReplicantException (B8.unpack err)  -- | Create replication slot or retrieve an existing slot. -- -- Can throw exceptions from @getReplicationSlotSync@ or -- @createReplicationSlotSync@.-setupReplicationSlot :: Connection -> ByteString -> IO ReplicationSlotInfo+setupReplicationSlot :: ReplicantConnection -> ByteString -> IO ReplicationSlotInfo setupReplicationSlot conn slotName = do   maybeSlot <- getReplicationSlotSync conn slotName   case maybeSlot of-    Just slot -> pure $ slot+    Just slot -> pure slot     Nothing   -> createReplicationSlotSync conn slotName

@@ -0,0 +1,28 @@+module Database.PostgreSQL.Replicant.Settings where++import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as B++data PgSettings+  = PgSettings+  { pgUser        :: String+  , pgPassword    :: Maybe String+  , pgDbName      :: String+  , pgHost        :: String+  , pgPort        :: String+  , pgSlotName    :: String+  , pgUpdateDelay :: String -- ^ Controls how frequently the+                            -- primaryKeepAlive thread updates+                            -- PostgresSQL in @ms@+  }+  deriving (Eq, Show)++pgConnectionString :: PgSettings -> ByteString+pgConnectionString PgSettings {..} = B.intercalate " "+  [ "user=" <> B.pack pgUser+  , maybe "" (\pgPass -> "pass=" <> B.pack pgPass) pgPassword+  , "dbname=" <> B.pack pgDbName+  , "host=" <> B.pack pgHost+  , "port=" <> B.pack pgPort+  , "replication=database"+  ]

@@ -33,6 +33,7 @@ import Data.ByteString.Lazy.Builder.ASCII (word32Hex) import Data.Serialize import Data.Word+import qualified Data.Text as T import qualified Data.Text.Encoding as T import GHC.Int @@ -51,7 +52,7 @@   get = fromInt64 <$> getInt64be  instance ToJSON LSN where-  toJSON = String . T.decodeUtf8 . toByteString+  toJSON = String . T.toUpper . T.decodeUtf8 . toByteString  instance FromJSON LSN where   parseJSON = withText "LSN" $ \txt ->

@@ -8,8 +8,8 @@ import GHC.Int  import Database.PostgreSQL.Replicant.Message+import Database.PostgreSQL.Replicant.Settings import Database.PostgreSQL.Replicant.Types.Lsn-import qualified Database.PostgreSQL.Replicant.Queue as Q  examplePrimaryKeepAliveMessage :: ByteString examplePrimaryKeepAliveMessage@@ -85,36 +85,30 @@           let (Right lsn) = fromByteString "16/3002D50"           (toByteString <$> (decode . encode @LSN $ lsn)) `shouldBe` Right "16/3002d50" -  describe "FifoQueue" $ do-    it "should dequeue Nothing if the queue is empty" $ do-      queue <- Q.empty @Int-      res <- Q.dequeue queue-      res `shouldBe` Nothing--    it "should respect first-in, first-out" $ do-      queue <- Q.empty-      Q.enqueue queue 1-      Q.enqueue queue 2-      Q.enqueue queue 3-      res <- Q.dequeue queue-      res `shouldBe` Just 1--    context "null" $ do-      it "should return True if the queue is empty" $ do-        queue <- Q.empty-        res <- Q.null queue-        res `shouldBe` True--      it "should return False if the queue has at least one element" $ do-        queue <- Q.empty-        Q.enqueue queue 1-        res <- Q.null queue-        res `shouldBe` False+  context "Settings" $ do+    describe "pgConnectionString" $ do+      it "should include the password when specified" $ do+        let settings+              = PgSettings+              { pgUser = "foo"+              , pgPassword = Just "bar"+              , pgDbName = "test"+              , pgHost = "hostname"+              , pgPort = "5432"+              , pgSlotName = "test-slot"+              , pgUpdateDelay = "43"+              }+        pgConnectionString settings `shouldBe` "user=foo pass=bar dbname=test host=hostname port=5432 replication=database" -    context "enqueueRight" $ do-      it "should enqueue an item at the head of the queue" $ do-        queue <- Q.empty-        Q.enqueue queue 1-        Q.enqueueRight queue 2-        res <- Q.dequeue queue-        res `shouldBe` Just 2+      it "should omit the password when not specified" $ do+        let settings+              = PgSettings+              { pgUser = "foo"+              , pgPassword = Nothing+              , pgDbName = "test"+              , pgHost = "hostname"+              , pgPort = "5432"+              , pgSlotName = "test-slot"+              , pgUpdateDelay = "43"+              }+        pgConnectionString settings `shouldBe` "user=foo  dbname=test host=hostname port=5432 replication=database"