Copyright	(c) James King 2020 2021
License	BSD3
Maintainer	james@agentultra.com
Stability	experimental
Portability	POSIX
Safe Haskell	None
Language	Haskell2010

Database.PostgreSQL.Replicant

Contents

Types
Connection Handling
Functions

Description

Connect to a PostgreSQL server as a logical replication client and receive changes.

The basic API is this:

  withLogicalStream defaultSettings $ change -> do
    print change
    catch err -> do
      show err

This is a low-level library meant to give the primitives necessary to library authors to add streaming replication support. The API here to rather simplistic but should be hooked up to something like conduit to provide better ergonomics.

Synopsis

data Change = Change {
- changeNextLSN :: LSN
- changeDeltas :: [WalLogData]
}
data Column = Column {
- columnName :: !Text
- columnType :: !Text
- columnValue :: !WalValue
}
data Delete = Delete {
- deleteSchema :: !Text
- deleteTable :: !Text
- deleteColumns :: ![Column]
}
data Insert = Insert {
- insertSchema :: !String
- insertTable :: !String
- insertColumns :: ![Column]
}
data Message = Message {
- messageTransactional :: !Bool
- messagePrefix :: !Text
- messageContent :: !Text
}
data PgSettings = PgSettings {
- pgUser :: String
- pgPassword :: Maybe String
- pgDbName :: String
- pgHost :: String
- pgPort :: String
- pgSlotName :: String
- pgUpdateDelay :: String
}
data ReplicantConnection
data Update = Update {
- updateSchema :: !Text
- updateTable :: !Text
- updateColumns :: ![Column]
}
data WalLogData
- = WInsert !Insert
- | WUpdate !Update
- | WDelete !Delete
- | WMessage !Message
connect :: PgSettings -> IO ReplicantConnection
getConnection :: ReplicantConnection -> Connection
unsafeCreateConnection :: Connection -> ReplicantConnection
withLogicalStream :: PgSettings -> (Change -> IO LSN) -> IO ()

Types

data Change Source #

Constructors

Change
Fields 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.

Instances

Instances details

Eq Change Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods (==) :: Change -> Change -> Bool # (/=) :: Change -> Change -> Bool #
Show Change Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods showsPrec :: Int -> Change -> ShowS # show :: Change -> String # showList :: [Change] -> ShowS #
Generic Change Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Associated Types type Rep Change :: Type -> Type # Methods from :: Change -> Rep Change x # to :: Rep Change x -> Change #
ToJSON Change Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods toJSON :: Change -> Value # toEncoding :: Change -> Encoding # toJSONList :: [Change] -> Value # toEncodingList :: [Change] -> Encoding #
FromJSON Change Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods parseJSON :: Value -> Parser Change # parseJSONList :: Value -> Parser [Change] #
type Rep Change Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message type Rep Change = D1 ('MetaData "Change" "Database.PostgreSQL.Replicant.Message" "postgresql-replicant-0.2.0.1-LfvKLvORtF58sQk3dBptIQ" 'False) (C1 ('MetaCons "Change" 'PrefixI 'True) (S1 ('MetaSel ('Just "changeNextLSN") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedStrict) (Rec0 LSN) :*: S1 ('MetaSel ('Just "changeDeltas") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedStrict) (Rec0 [WalLogData])))

data Column Source #

Represents a single table column. We only support the wal2json logical decoder plugin and make no attempt to parse anything but JSON-like primitives.

Constructors

Column
Fields columnName :: !Text columnType :: !Text columnValue :: !WalValue

Instances

Instances details

Eq Column Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods (==) :: Column -> Column -> Bool # (/=) :: Column -> Column -> Bool #
Show Column Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods showsPrec :: Int -> Column -> ShowS # show :: Column -> String # showList :: [Column] -> ShowS #

data Delete Source #

Represents a single delete query in the logical replication format

Constructors

Delete
Fields deleteSchema :: !Text deleteTable :: !Text deleteColumns :: ![Column]

Instances

Instances details

Eq Delete Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods (==) :: Delete -> Delete -> Bool # (/=) :: Delete -> Delete -> Bool #
Show Delete Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods showsPrec :: Int -> Delete -> ShowS # show :: Delete -> String # showList :: [Delete] -> ShowS #
ToJSON Delete Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods toJSON :: Delete -> Value # toEncoding :: Delete -> Encoding # toJSONList :: [Delete] -> Value # toEncodingList :: [Delete] -> Encoding #
FromJSON Delete Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods parseJSON :: Value -> Parser Delete # parseJSONList :: Value -> Parser [Delete] #

data Insert Source #

Represents a single insert query in the logical replication format.

Constructors

Insert
Fields insertSchema :: !String insertTable :: !String insertColumns :: ![Column]

Instances

Instances details

Eq Insert Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods (==) :: Insert -> Insert -> Bool # (/=) :: Insert -> Insert -> Bool #
Show Insert Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods showsPrec :: Int -> Insert -> ShowS # show :: Insert -> String # showList :: [Insert] -> ShowS #
ToJSON Insert Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods toJSON :: Insert -> Value # toEncoding :: Insert -> Encoding # toJSONList :: [Insert] -> Value # toEncodingList :: [Insert] -> Encoding #
FromJSON Insert Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods parseJSON :: Value -> Parser Insert # parseJSONList :: Value -> Parser [Insert] #

data Message Source #

Occasionally the server may also send these for informational purposes and can be ignored. May be used internally.

Constructors

Message
Fields messageTransactional :: !Bool messagePrefix :: !Text messageContent :: !Text

Instances

Instances details

Eq Message Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods (==) :: Message -> Message -> Bool # (/=) :: Message -> Message -> Bool #
Show Message Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods showsPrec :: Int -> Message -> ShowS # show :: Message -> String # showList :: [Message] -> ShowS #
ToJSON Message Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods toJSON :: Message -> Value # toEncoding :: Message -> Encoding # toJSONList :: [Message] -> Value # toEncodingList :: [Message] -> Encoding #
FromJSON Message Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods parseJSON :: Value -> Parser Message # parseJSONList :: Value -> Parser [Message] #

data PgSettings Source #

Constructors

PgSettings
Fields 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`

Instances

Instances details

Eq PgSettings Source #
Instance details Defined in Database.PostgreSQL.Replicant.Settings Methods (==) :: PgSettings -> PgSettings -> Bool # (/=) :: PgSettings -> PgSettings -> Bool #
Show PgSettings Source #
Instance details Defined in Database.PostgreSQL.Replicant.Settings Methods showsPrec :: Int -> PgSettings -> ShowS # show :: PgSettings -> String # showList :: [PgSettings] -> ShowS #

data ReplicantConnection Source #

Instances

Instances details

Eq ReplicantConnection Source #
Instance details Defined in Database.PostgreSQL.Replicant.Connection Methods (==) :: ReplicantConnection -> ReplicantConnection -> Bool # (/=) :: ReplicantConnection -> ReplicantConnection -> Bool #

data Update Source #

Represents a single update query in the logical replication format.

Constructors

Update
Fields updateSchema :: !Text updateTable :: !Text updateColumns :: ![Column]

Instances

Instances details

Eq Update Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods (==) :: Update -> Update -> Bool # (/=) :: Update -> Update -> Bool #
Show Update Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods showsPrec :: Int -> Update -> ShowS # show :: Update -> String # showList :: [Update] -> ShowS #
ToJSON Update Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods toJSON :: Update -> Value # toEncoding :: Update -> Encoding # toJSONList :: [Update] -> Value # toEncodingList :: [Update] -> Encoding #
FromJSON Update Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods parseJSON :: Value -> Parser Update # parseJSONList :: Value -> Parser [Update] #

data WalLogData Source #

Constructors

WInsert !Insert
WUpdate !Update
WDelete !Delete
WMessage !Message

Instances

Instances details

Eq WalLogData Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods (==) :: WalLogData -> WalLogData -> Bool # (/=) :: WalLogData -> WalLogData -> Bool #
Show WalLogData Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods showsPrec :: Int -> WalLogData -> ShowS # show :: WalLogData -> String # showList :: [WalLogData] -> ShowS #
Generic WalLogData Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Associated Types type Rep WalLogData :: Type -> Type # Methods from :: WalLogData -> Rep WalLogData x # to :: Rep WalLogData x -> WalLogData #
ToJSON WalLogData Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods toJSON :: WalLogData -> Value # toEncoding :: WalLogData -> Encoding # toJSONList :: [WalLogData] -> Value # toEncodingList :: [WalLogData] -> Encoding #
FromJSON WalLogData Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message Methods parseJSON :: Value -> Parser WalLogData # parseJSONList :: Value -> Parser [WalLogData] #
type Rep WalLogData Source #
Instance details Defined in Database.PostgreSQL.Replicant.Message type Rep WalLogData = D1 ('MetaData "WalLogData" "Database.PostgreSQL.Replicant.Message" "postgresql-replicant-0.2.0.1-LfvKLvORtF58sQk3dBptIQ" 'False) ((C1 ('MetaCons "WInsert" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Insert)) :+: C1 ('MetaCons "WUpdate" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Update))) :+: (C1 ('MetaCons "WDelete" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Delete)) :+: C1 ('MetaCons "WMessage" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Message))))

Connection Handling

connect :: PgSettings -> IO ReplicantConnection Source #

Connect to the PostgreSQL server in replication mode

getConnection :: ReplicantConnection -> Connection Source #

unsafeCreateConnection :: Connection -> ReplicantConnection Source #

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.

Functions

withLogicalStream :: PgSettings -> (Change -> IO LSN) -> IO () Source #

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 changes you get represent the set of queries in a transaction: insert, update, and delete.

This function will create the replication slot, if it doesn't exist, or reconnect to it otherwise and restart the stream from where the replication slot left off.

This function can throw exceptions in IO and shut-down the socket in case of any error.