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

   1 /*
   2  * Copyright (c) 2020 PANTHEON.tech, 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 com.codahale.metrics.Histogram;
  13 import org.eclipse.jdt.annotation.NonNull;
  14 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.ReplayMessages;
  15 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.WriteMessages;
  16 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.WrittenMessages;
  17
  18 /**
  19  * Abstraction of a data journal. This provides a unified interface towards {@link SegmentedJournalActor}, allowing
  20  * specialization for various formats.
  21  */
  22 abstract class DataJournal {
  23     // Mirrors fields from associated actor
  24     final @NonNull String persistenceId;
  25     private final Histogram messageSize;
  26
  27     // Tracks largest message size we have observed either during recovery or during write
  28     private int largestObservedSize;
  29
  30     DataJournal(final String persistenceId, final Histogram messageSize) {
  31         this.persistenceId = requireNonNull(persistenceId);
  32         this.messageSize = requireNonNull(messageSize);
  33     }
  34
  35     final void recordMessageSize(final int size) {
  36         messageSize.update(size);
  37         updateLargestSize(size);
  38     }
  39
  40     final void updateLargestSize(final int size) {
  41         if (size > largestObservedSize) {
  42             largestObservedSize = size;
  43         }
  44     }
  45
  46     /**
  47      * Return the last sequence number completely written to the journal.
  48      *
  49      * @return Last written sequence number, {@code -1} if there are no in the journal.
  50      */
  51     abstract long lastWrittenSequenceNr();
  52
  53     /**
  54      * Delete all messages up to specified sequence number.
  55      *
  56      * @param sequenceNr Sequence number to delete to.
  57      */
  58     abstract void deleteTo(long sequenceNr);
  59
  60     /**
  61      * Delete all messages up to specified sequence number.
  62      *
  63      * @param sequenceNr Sequence number to compact to.
  64      */
  65     abstract void compactTo(long sequenceNr);
  66
  67     /**
  68      * Close this journal, freeing up resources associated with it.
  69      */
  70     abstract void close();
  71
  72     /**
  73      * Handle a request to replay messages.
  74      *
  75      * @param message Request message
  76      * @param fromSequenceNr Sequence number to replay from, adjusted for deletions
  77      */
  78     abstract void handleReplayMessages(@NonNull ReplayMessages message, long fromSequenceNr);
  79
  80     /**
  81      * Handle a request to store some messages.
  82      *
  83      * @param message {@link WriteMessages} message
  84      * @return a {@link WrittenMessages} object
  85      */
  86     abstract @NonNull WrittenMessages handleWriteMessages(@NonNull WriteMessages message);
  87
  88     /**
  89      * Flush all messages to durable storage.
  90      */
  91     abstract void flush();
  92 }