Method processValueUpdate

kv/memberlist/memberlist_client.go:1503–1543

(workerCh <-chan valueUpdate, key string)

}

func (m *KV) processValueUpdate(workerCh <-chan valueUpdate, key string) {
	for {
		select {
		case update := <-workerCh:
			// we have a value update! Let's merge it with our current version for given key
			mod, version, deleted, updated, err := m.mergeBytesValueForKey(key, update.value, update.codec, update.deleted, update.updateTime)

			changes := []string(nil)
			if mod != nil {
				changes = mod.MergeContent()
			}

			m.addReceivedMessage(Message{
				Time: time.Now(),
				Size: update.messageSize,
				Pair: KeyValuePair{
					Key:              key,
					Value:            update.value,
					Codec:            update.codec.CodecID(),
					Deleted:          deleted,
					UpdateTimeMillis: updateTimeMillis(updated),
				},
				Version: version,
				Changes: changes,
			})

			if err != nil {
				level.Error(m.logger).Log("msg", "failed to store received value", "key", key, "err", err)
			} else if version > 0 {
				m.notifyWatchers(key)

				// Don't resend original message, but only changes, if any.
				m.broadcastNewValue(key, mod, version, update.codec, false, deleted, updated)
			}

		case <-m.shutdown:
			// stop running on shutdown
			return
		}
	}
}

// GetBroadcasts is method from Memberlist Delegate interface
// It returns all pending broadcasts (within the size limit)