Method Complete

db/active_replicator_pull.go:127–158

Complete gracefully shuts down a replication, waiting for all in-flight revisions to be processed before stopping the replication

()

// Complete gracefully shuts down a replication, waiting for all in-flight revisions to be processed
// before stopping the replication
func (apr *ActivePullReplicator) Complete() {
	base.TracefCtx(apr.ctx, base.KeyReplicate, "ActivePullReplicator.Complete()")
	apr.lock.Lock()
	_ = apr.forEachCollection(func(c *activeReplicatorCollection) error {
		base.TracefCtx(apr.ctx, base.KeyReplicate, "Before calling waitForExpectedSequences in Complete()")
		if err := c.Checkpointer.waitForExpectedSequences(); err != nil {
			base.InfofCtx(apr.ctx, base.KeyReplicate, "Couldn't drain replication %s - stopping anyway: %v", apr.config.ID, err)
		}
		base.TracefCtx(apr.ctx, base.KeyReplicate, "After calling waitForExpectedSequences in Complete()")
		return nil
	})

	apr._stop()

	base.TracefCtx(apr.ctx, base.KeyReplicate, "Calling disconnect from Complete() in active replicator pull")
	stopErr := apr._disconnect()
	if stopErr != nil {
		base.InfofCtx(apr.ctx, base.KeyReplicate, "Error attempting to stop replication %s: %v", apr.config.ID, stopErr)
	}
	apr.setState(ReplicationStateStopped)

	// unlock the replication before triggering callback, in case callback attempts to access replication information
	// from the replicator
	onCompleteCallback := apr.onReplicatorComplete

	apr._publishStatus()
	apr.lock.Unlock()

	if onCompleteCallback != nil {
		onCompleteCallback()
	}
}

// _getStatus returns current replicator status. Requires holding ActivePullReplicator.lock as a read lock.
func (apr *ActivePullReplicator) _getStatus() *ReplicationStatus {