Function getCommittedVersion

fdbserver/SimpleConfigConsumer.actor.cpp:25–71

#include "flow/actorcompiler.h" // must be last include

class SimpleConfigConsumerImpl {
	ConfigFollowerInterface cfi;
	Version lastSeenVersion{ 0 };
	double pollingInterval;
	Optional<double> compactionInterval;

	UID id;
	CounterCollection cc;
	Counter compactRequest;
	Counter successfulChangeRequest;
	Counter failedChangeRequest;
	Counter snapshotRequest;
	Future<Void> logger;

	ACTOR static Future<Void> compactor(SimpleConfigConsumerImpl* self, ConfigBroadcaster* broadcaster) {
		if (!self->compactionInterval.present()) {
			wait(Never());
			return Void();
		}
		loop {
			state Version compactionVersion = self->lastSeenVersion;
			wait(delayJittered(self->compactionInterval.get()));
			if (self->cfi.hostname.present()) {
				wait(retryGetReplyFromHostname(ConfigFollowerCompactRequest{ compactionVersion },
				                               self->cfi.hostname.get(),
				                               WLTOKEN_CONFIGFOLLOWER_COMPACT));
			} else {
				wait(self->cfi.compact.getReply(ConfigFollowerCompactRequest{ compactionVersion }));
			}
			++self->compactRequest;
			broadcaster->compact(compactionVersion);
		}
	}

	ACTOR static Future<Version> getCommittedVersion(SimpleConfigConsumerImpl* self) {
		state ConfigFollowerGetCommittedVersionReply committedVersionReply;
		if (self->cfi.hostname.present()) {
			wait(store(committedVersionReply,
			           retryGetReplyFromHostname(ConfigFollowerGetCommittedVersionRequest{},
			                                     self->cfi.hostname.get(),
			                                     WLTOKEN_CONFIGFOLLOWER_GETCOMMITTEDVERSION)));
		} else {
			wait(store(committedVersionReply,
			           self->cfi.getCommittedVersion.getReply(ConfigFollowerGetCommittedVersionRequest{})));
		}
		return committedVersionReply.lastCommitted;
	}

	ACTOR static Future<Void> fetchChanges(SimpleConfigConsumerImpl* self, ConfigBroadcaster* broadcaster) {
		wait(getSnapshotAndChanges(self, broadcaster));