Function startReplication

server/server.go:230–271

startReplication begins the background thread that replicates from Postgres, if one is configured.

(cfg *servercfg.DoltgresConfig, ssCfg doltservercfg.ServerConfig)

// startReplication begins the background thread that replicates from Postgres, if one is configured.
func startReplication(cfg *servercfg.DoltgresConfig, ssCfg doltservercfg.ServerConfig) (*logrepl.LogicalReplicator, error) {
	if cfg.PostgresReplicationConfig == nil {
		return nil, nil
	} else if cfg.PostgresReplicationConfig.PostgresDatabase == nil || *cfg.PostgresReplicationConfig.PostgresDatabase == "" {
		return nil, errors.Errorf("postgres replication database must be specified and not empty for replication")
	} else if cfg.PostgresReplicationConfig.PostgresUser == nil || *cfg.PostgresReplicationConfig.PostgresUser == "" {
		return nil, errors.Errorf("postgres replication user must be specified and not empty for replication")
	} else if cfg.PostgresReplicationConfig.PostgresPassword == nil || *cfg.PostgresReplicationConfig.PostgresPassword == "" {
		return nil, errors.Errorf("postgres replication password must be specified and not empty for replication")
	} else if cfg.PostgresReplicationConfig.PostgresPort == nil || *cfg.PostgresReplicationConfig.PostgresPort == 0 {
		return nil, errors.Errorf("postgres replication port must be specified and non-zero for replication")
	} else if cfg.PostgresReplicationConfig.SlotName == nil || *cfg.PostgresReplicationConfig.SlotName == "" {
		return nil, errors.Errorf("postgres replication slot name must be specified and not empty for replication")
	}

	walFilePath := filepath.Join(ssCfg.CfgDir(), "pg_wal_location")
	primaryDns := fmt.Sprintf(
		"postgres://%s:%s@%s:%d/%s",
		*cfg.PostgresReplicationConfig.PostgresUser,
		*cfg.PostgresReplicationConfig.PostgresPassword,
		*cfg.PostgresReplicationConfig.PostgresServerAddress,
		*cfg.PostgresReplicationConfig.PostgresPort,
		*cfg.PostgresReplicationConfig.PostgresDatabase,
	)

	replicationDns := fmt.Sprintf(
		"postgres://%s:%s@localhost:%d/%s",
		ssCfg.User(),
		ssCfg.Password(),
		ssCfg.Port(),
		"postgres", // TODO: this needs to come from config
	)

	replicator, err := logrepl.NewLogicalReplicator(walFilePath, primaryDns, replicationDns)
	if err != nil {
		return nil, err
	}

	cli.Println("Starting replication")
	go replicator.StartReplication(*cfg.PostgresReplicationConfig.SlotName)
	return replicator, nil
}

// configCliContext is a minimal implementation of CliContext that only supports Config()
type configCliContext struct {