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.
(slotName string)
| 174 | // StartReplication starts the replication process for the given slot name. This function blocks until replication is |
| 175 | // stopped via the Stop method, or an error occurs. |
| 176 | func (r *LogicalReplicator) StartReplication(slotName string) error { |
| 177 | standbyMessageTimeout := 10 * time.Second |
| 178 | nextStandbyMessageDeadline := time.Now().Add(standbyMessageTimeout) |
| 179 | |
| 180 | lastWrittenLsn, err := r.readWALPosition() |
| 181 | if err != nil { |
| 182 | return err |
| 183 | } |
| 184 | |
| 185 | // TODO: we need to be able to re-establish this connection if it goes bad |
| 186 | replicationConn, err := pgx.Connect(context.Background(), r.replicationDns) |
| 187 | if err != nil { |
| 188 | return err |
| 189 | } |
| 190 | |
| 191 | state := &replicationState{ |
| 192 | lastWrittenLSN: lastWrittenLsn, |
| 193 | replicaConn: replicationConn, |
| 194 | relations: map[uint32]*pglogrepl.RelationMessageV2{}, |
| 195 | typeMap: pgtype.NewMap(), |
| 196 | } |
| 197 | |
| 198 | var primaryConn *pgconn.PgConn |
| 199 | defer func() { |
| 200 | if primaryConn != nil { |
| 201 | _ = primaryConn.Close(context.Background()) |
| 202 | } |
| 203 | if state.replicaConn != nil { |
| 204 | _ = state.replicaConn.Close(context.Background()) |
| 205 | } |
| 206 | // We always shut down here and only here, so we do the cleanup on thread exit in exactly one place |
| 207 | r.shutdown() |
| 208 | }() |
| 209 | |
| 210 | connErrCnt := 0 |
| 211 | handleErrWithRetry := func(err error, incrementErrorCount bool) error { |
| 212 | if err != nil { |
| 213 | if incrementErrorCount { |
| 214 | connErrCnt++ |
| 215 | } |
| 216 | if connErrCnt < maxConsecutiveFailures { |
| 217 | log.Printf("Error: %v. Retrying", err) |
| 218 | if primaryConn != nil { |
| 219 | _ = primaryConn.Close(context.Background()) |
| 220 | } |
| 221 | primaryConn = nil |
| 222 | return nil |
| 223 | } |
| 224 | } else { |
| 225 | connErrCnt = 0 |
| 226 | } |
| 227 | |
| 228 | return err |
| 229 | } |
| 230 | |
| 231 | sendStandbyStatusUpdate := func(state *replicationState) error { |
| 232 | // The StatusUpdate message wants us to respond with the current position in the WAL + 1: |
| 233 | // https://www.postgresql.org/docs/current/protocol-replication.html |