Method replicate

onecache/replicator.go:277–324

replicate is a long lived process that replicates keys to the peer.

()

// replicate is a long lived process that replicates keys to the peer.
func (p *peerSync) replicate() {
	for {
		select {
		case <-p.endSync:
			return
		case <-p.r.quit:
			return
		case key := <-p.queue:
			p.lock.Lock()
			log, ok := p.tasks[key]
			delete(p.tasks, key)
			p.lock.Unlock()
			if !ok {
				continue
			}

			task, err := collapseOpLog(log)
			if err != nil {
				p.r.n.logger.Printf("[ERROR] onecache.replicator: key %v tasks invalid: %v\n", key, err)
				continue
			}

			// Ensure we are still the owner of the key.
			if keys := p.r.n.ring.getOwnedKeys([]string{key}); len(keys) == 0 {
				continue
			}

			p.lock.Lock()
			p.replicating = key
			p.lock.Unlock()

			switch task {
			case task_BACKFILL:
				p.replicateBackfill(key, p.name)
			case task_DIRTY_VALUE:
				p.replicateDirtyValue(key, p.name)
			case task_TOUCH:
				p.replicateTouch(key, p.name)
			case task_DELETE:
				p.replicateDelete(key, p.name)
			}

			p.lock.Lock()
			p.replicating = ""
			p.lock.Unlock()
		}
	}
}

// replicateDirtyValue replicates a key due to a dirty value.
func (p *peerSync) replicateDirtyValue(key, peer string) error {