StartReplication starts the replication process for the given slot name. This function blocks until replication is stopped via the Stop method, or an error occurs.
(sqlCtx *sql.Context, slotName string)
| 220 | // StartReplication starts the replication process for the given slot name. This function blocks until replication is |
| 221 | // stopped via the Stop method, or an error occurs. |
| 222 | func (r *LogicalReplicator) StartReplication(sqlCtx *sql.Context, slotName string) error { |
| 223 | sqlCtx.SetLogger(r.logger) |
| 224 | standbyMessageTimeout := 10 * time.Second |
| 225 | nextStandbyMessageDeadline := time.Now().Add(standbyMessageTimeout) |
| 226 | |
| 227 | lastWrittenLsn, err := SelectSubscriptionLsn(sqlCtx, r.subscription) |
| 228 | if err != nil { |
| 229 | return err |
| 230 | } |
| 231 | |
| 232 | state := &replicationState{} |
| 233 | state.reset(sqlCtx, slotName, lastWrittenLsn) |
| 234 | |
| 235 | // Switch to the `public` schema. |
| 236 | if _, err := adapter.ExecCatalog(sqlCtx, "USE public"); err != nil { |
| 237 | return err |
| 238 | } |
| 239 | sqlCtx.SetCurrentDatabase("public") |
| 240 | |
| 241 | var primaryConn *pgconn.PgConn |
| 242 | defer func() { |
| 243 | if primaryConn != nil { |
| 244 | _ = primaryConn.Close(context.Background()) |
| 245 | } |
| 246 | // We always shut down here and only here, so we do the cleanup on thread exit in exactly one place |
| 247 | r.shutdown(sqlCtx, state) |
| 248 | }() |
| 249 | |
| 250 | connErrCnt := 0 |
| 251 | handleErrWithRetry := func(err error, incrementErrorCount bool) error { |
| 252 | if err != nil { |
| 253 | r.logger.Warnf("Handle error: %v", err) |
| 254 | if incrementErrorCount { |
| 255 | connErrCnt++ |
| 256 | } |
| 257 | if connErrCnt < maxConsecutiveFailures { |
| 258 | r.logger.Warnf("Retrying (%d/%d) on error %v", connErrCnt, maxConsecutiveFailures, err) |
| 259 | if primaryConn != nil { |
| 260 | if err := primaryConn.Close(context.Background()); err != nil { |
| 261 | r.logger.Warnf("Failed to close connection: %v", err) |
| 262 | } |
| 263 | } |
| 264 | primaryConn = nil |
| 265 | return nil |
| 266 | } |
| 267 | } else { |
| 268 | connErrCnt = 0 |
| 269 | } |
| 270 | |
| 271 | return err |
| 272 | } |
| 273 | |
| 274 | sendStandbyStatusUpdate := func(state *replicationState) error { |
| 275 | // The StatusUpdate message wants us to respond with the current position in the WAL + 1: |
| 276 | // https://www.postgresql.org/docs/current/protocol-replication.html |
| 277 | err := pglogrepl.SendStandbyStatusUpdate(context.Background(), primaryConn, pglogrepl.StandbyStatusUpdate{ |
| 278 | WALWritePosition: state.lastReceivedLSN + 1, |
| 279 | WALFlushPosition: state.lastWrittenLSN + 1, |