Complete gracefully shuts down a replication, waiting for all in-flight revisions to be processed before stopping the replication
()
| 71 | // Complete gracefully shuts down a replication, waiting for all in-flight revisions to be processed |
| 72 | // before stopping the replication |
| 73 | func (apr *ActivePushReplicator) Complete() { |
| 74 | base.TracefCtx(apr.ctx, base.KeyReplicate, "ActivePushReplicator.Complete()") |
| 75 | apr.lock.Lock() |
| 76 | |
| 77 | // Wait for any pending changes responses to arrive and be processed |
| 78 | err := apr._waitForPendingChangesResponse() |
| 79 | if err != nil { |
| 80 | base.InfofCtx(apr.ctx, base.KeyReplicate, "Timeout waiting for pending changes response for replication %s - stopping: %v", apr.config.ID, err) |
| 81 | } |
| 82 | |
| 83 | _ = apr.forEachCollection(func(c *activeReplicatorCollection) error { |
| 84 | if err := c.Checkpointer.waitForExpectedSequences(); err != nil { |
| 85 | base.InfofCtx(apr.ctx, base.KeyReplicate, "Timeout draining replication %s - stopping: %v", apr.config.ID, err) |
| 86 | } |
| 87 | return nil |
| 88 | }) |
| 89 | |
| 90 | apr._stop() |
| 91 | |
| 92 | stopErr := apr._disconnect() |
| 93 | if stopErr != nil { |
| 94 | base.InfofCtx(apr.ctx, base.KeyReplicate, "Error attempting to stop replication %s: %v", apr.config.ID, stopErr) |
| 95 | } |
| 96 | apr.setState(ReplicationStateStopped) |
| 97 | |
| 98 | // unlock the replication before triggering callback, in case callback attempts to re-acquire the lock |
| 99 | onCompleteCallback := apr.onReplicatorComplete |
| 100 | apr._publishStatus() |
| 101 | apr.lock.Unlock() |
| 102 | |
| 103 | if onCompleteCallback != nil { |
| 104 | onCompleteCallback() |
| 105 | } |
| 106 | } |
| 107 | |
| 108 | // _getStatus returns current replicator status. Requires holding ActivePushReplicator.lock as a read lock. |
| 109 | func (apr *ActivePushReplicator) _getStatus() *ReplicationStatus { |
no test coverage detected