Method apply

src/state/log.cpp:309–371

Future<Nothing> LogStorageProcess::apply(const list<Log::Entry>& entries)
{
  VLOG(2) << "Applying operations (" << entries.size() << " entries)";

  // Only read and apply entries past our index.
  foreach (const Log::Entry& entry, entries) {
    if (index.isNone() || index.get() < entry.position) {
      // Parse the Operation from the Log::Entry.
      Operation operation;

      google::protobuf::io::ArrayInputStream stream(
          entry.data.data(),
          entry.data.size());

      if (!operation.ParseFromZeroCopyStream(&stream)) {
        return Failure("Failed to deserialize Operation");
      }

      switch (operation.type()) {
        case Operation::SNAPSHOT: {
          CHECK(operation.has_snapshot());

          // Add or update (override) the snapshot.
          Snapshot snapshot(entry.position, operation.snapshot().entry());
          snapshots.put(snapshot.entry.name(), snapshot);
          break;
        }

        case Operation::DIFF: {
          CHECK(operation.has_diff());

          Option<Snapshot> snapshot =
            snapshots.get(operation.diff().entry().name());

          CHECK_SOME(snapshot);

          Try<Snapshot> patched = snapshot->patch(operation.diff());

          if (patched.isError()) {
            return Failure("Failed to apply the diff: " + patched.error());
          }

          // Replace the snapshot with the patched snapshot.
          snapshots.put(patched->entry.name(), patched.get());
          break;
        }

        case Operation::EXPUNGE: {
          CHECK(operation.has_expunge());
          snapshots.erase(operation.expunge().name());
          break;
        }

        default:
          return Failure("Unknown operation: " + stringify(operation.type()));
      }

      index = entry.position;