diff --git a/ChangeLog.md b/ChangeLog.md
--- a/ChangeLog.md
+++ b/ChangeLog.md
@@ -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!
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -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
 
diff --git a/app/Main.hs b/app/Main.hs
--- a/app/Main.hs
+++ b/app/Main.hs
@@ -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: "
diff --git a/postgresql-replicant.cabal b/postgresql-replicant.cabal
--- a/postgresql-replicant.cabal
+++ b/postgresql-replicant.cabal
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant.hs b/src/Database/PostgreSQL/Replicant.hs
--- a/src/Database/PostgreSQL/Replicant.hs
+++ b/src/Database/PostgreSQL/Replicant.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/Connection.hs b/src/Database/PostgreSQL/Replicant/Connection.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/Connection.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/Message.hs b/src/Database/PostgreSQL/Replicant/Message.hs
--- a/src/Database/PostgreSQL/Replicant/Message.hs
+++ b/src/Database/PostgreSQL/Replicant/Message.hs
@@ -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)
 
diff --git a/src/Database/PostgreSQL/Replicant/Protocol.hs b/src/Database/PostgreSQL/Replicant/Protocol.hs
--- a/src/Database/PostgreSQL/Replicant/Protocol.hs
+++ b/src/Database/PostgreSQL/Replicant/Protocol.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/Queue.hs b/src/Database/PostgreSQL/Replicant/Queue.hs
deleted file mode 100644
--- a/src/Database/PostgreSQL/Replicant/Queue.hs
+++ /dev/null
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/ReplicationSlot.hs b/src/Database/PostgreSQL/Replicant/ReplicationSlot.hs
--- a/src/Database/PostgreSQL/Replicant/ReplicationSlot.hs
+++ b/src/Database/PostgreSQL/Replicant/ReplicationSlot.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/Settings.hs b/src/Database/PostgreSQL/Replicant/Settings.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/Settings.hs
@@ -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"
+  ]
diff --git a/src/Database/PostgreSQL/Replicant/Types/Lsn.hs b/src/Database/PostgreSQL/Replicant/Types/Lsn.hs
--- a/src/Database/PostgreSQL/Replicant/Types/Lsn.hs
+++ b/src/Database/PostgreSQL/Replicant/Types/Lsn.hs
@@ -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 ->
diff --git a/test/Spec.hs b/test/Spec.hs
--- a/test/Spec.hs
+++ b/test/Spec.hs
@@ -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"