opendaylight/md-sal/sal-akka-segmented-journal/src/main/java/org/opendaylight/controller/akka/segjournal/DataJournalEntrySerdes.java

   1 /*
   2  * Copyright (c) 2019 Pantheon Technologies, s.r.o. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.akka.segjournal;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import akka.actor.ActorSystem;
  13 import akka.actor.ExtendedActorSystem;
  14 import akka.persistence.PersistentRepr;
  15 import akka.serialization.JavaSerializer;
  16 import com.google.common.base.VerifyException;
  17 import io.atomix.storage.journal.JournalSerdes.EntryInput;
  18 import io.atomix.storage.journal.JournalSerdes.EntryOutput;
  19 import io.atomix.storage.journal.JournalSerdes.EntrySerdes;
  20 import java.io.IOException;
  21 import java.util.concurrent.Callable;
  22 import org.opendaylight.controller.akka.segjournal.DataJournalEntry.FromPersistence;
  23 import org.opendaylight.controller.akka.segjournal.DataJournalEntry.ToPersistence;
  24
  25 /**
  26  * Kryo serializer for {@link DataJournalEntry}. Each {@link SegmentedJournalActor} has its own instance, as well as
  27  * a nested JavaSerializer to handle the payload.
  28  *
  29  * <p>
  30  * Since we are persisting only parts of {@link PersistentRepr}, this class asymmetric by design:
  31  * {@link #write(EntryOutput, DataJournalEntry)} only accepts {@link ToPersistence} subclass, which is a wrapper
  32  * around a {@link PersistentRepr}, while {@link #read(EntryInput)} produces an {@link FromPersistence}, which
  33  * needs further processing to reconstruct a {@link PersistentRepr}.
  34  */
  35 final class DataJournalEntrySerdes implements EntrySerdes<DataJournalEntry> {
  36     private final ExtendedActorSystem actorSystem;
  37
  38     DataJournalEntrySerdes(final ActorSystem actorSystem) {
  39         this.actorSystem = requireNonNull((ExtendedActorSystem) actorSystem);
  40     }
  41
  42     @Override
  43     public void write(final EntryOutput output, final DataJournalEntry entry) throws IOException {
  44         if (entry instanceof ToPersistence toPersistence) {
  45             final var repr = toPersistence.repr();
  46             output.writeString(repr.manifest());
  47             output.writeString(repr.writerUuid());
  48             output.writeObject(repr.payload());
  49         } else {
  50             throw new VerifyException("Unexpected entry " + entry);
  51         }
  52     }
  53
  54     @Override
  55     public DataJournalEntry read(final EntryInput input) throws IOException {
  56         return new FromPersistence(input.readString(), input.readString(),
  57             JavaSerializer.currentSystem().withValue(actorSystem, (Callable<Object>) input::readObject));
  58     }
  59 }