Method Start

db/active_replicator_common.go:144–173

Start starts the replicator, setting the state to ReplicationStateStarting and starting the status reporter.

(ctx context.Context)

// Start starts the replicator, setting the state to ReplicationStateStarting and starting the status reporter.
func (arc *activeReplicatorCommon) Start(ctx context.Context) error {
	arc.lock.Lock()
	defer arc.lock.Unlock()

	if arc.ctx != nil && arc.ctx.Err() == nil {
		return fmt.Errorf("Replicator is already running")
	}

	arc.setState(ReplicationStateStarting)
	logCtx := base.CorrelationIDLogCtx(ctx,
		arc.config.ID+"-"+string(arc.direction))
	arc.ctx, arc.ctxCancel = context.WithCancel(logCtx)

	arc.startStatusReporter(arc.ctx)

	err := arc.replicatorConnectFn()
	if err != nil {
		arc.setError(err)
		base.WarnfCtx(arc.ctx, "Couldn't connect: %s", err)
		if errors.Is(err, fatalReplicatorConnectError) {
			base.WarnfCtx(arc.ctx, "Stopping replication connection attempt")
			defer arc.ctxCancel()
		} else {
			base.InfofCtx(arc.ctx, base.KeyReplicate, "Attempting to reconnect in background: %v", err)
			arc.reconnect()
		}
	}
	arc._publishStatus()
	return err
}

// initCheckpointer starts a checkpointer. The remoteCheckpoints are only for collections and indexed by the blip collectionIdx. If using default collection only, replicationCheckpoints is an empty array.
func (arc *activeReplicatorCommon) _initCheckpointer(remoteCheckpoints []replicationCheckpoint) error {