mirror of
https://github.com/digital-asset/daml.git
synced 2024-09-20 01:07:18 +03:00
kvutils: Remove the LedgerEntry trait; it's no longer necessary. (#5383)
* kvutils: Remove the LedgerEntry trait; it's no longer necessary. This was introduced to allow for heartbeats, which no longer exist. CHANGELOG_BEGIN CHANGELOG_END * kvutils: Make LedgerRecord a case class again. We used to store the envelope as an array of bytes, which doesn't have a value-based `equals` method and therefore should not be used in a case class. We now use a `ByteString`, so this is no longer an issue.
This commit is contained in:
Samir Talwar
GitHub
parent
33b55465cb
commit
1d9c7d2574
@ -11,7 +11,7 @@ import akka.stream.scaladsl.Source
|
||||
import com.codahale.metrics.MetricRegistry
|
||||
import com.daml.ledger.on.memory.InMemoryLedgerReaderWriter._
|
||||
import com.daml.ledger.on.memory.InMemoryState.MutableLog
|
||||
import com.daml.ledger.participant.state.kvutils.api.{LedgerEntry, LedgerReader, LedgerWriter}
|
||||
import com.daml.ledger.participant.state.kvutils.api.{LedgerReader, LedgerRecord, LedgerWriter}
|
||||
import com.daml.ledger.participant.state.kvutils.{Bytes, KVOffset, SequentialLogEntryId}
|
||||
import com.daml.ledger.participant.state.v1._
|
||||
import com.daml.ledger.validator.LedgerStateOperations.{Key, Value}
|
||||
@ -54,7 +54,7 @@ final class InMemoryLedgerReaderWriter(
|
||||
override def commit(correlationId: String, envelope: Bytes): Future[SubmissionResult] =
|
||||
committer.commit(correlationId, envelope, participantId)
|
||||
|
||||
override def events(startExclusive: Option[Offset]): Source[LedgerEntry, NotUsed] =
|
||||
override def events(startExclusive: Option[Offset]): Source[LedgerRecord, NotUsed] =
|
||||
dispatcher
|
||||
.startingAt(
|
||||
startExclusive
|
||||
@ -93,7 +93,7 @@ private class InMemoryLedgerStateOperations(
|
||||
}
|
||||
|
||||
override def appendToLog(key: Key, value: Value): Future[Index] =
|
||||
Future.successful(appendEntry(log, LedgerEntry.LedgerRecord(_, key, value)))
|
||||
Future.successful(appendEntry(log, LedgerRecord(_, key, value)))
|
||||
}
|
||||
|
||||
object InMemoryLedgerReaderWriter {
|
||||
@ -168,7 +168,7 @@ object InMemoryLedgerReaderWriter {
|
||||
headAtInitialization = StartIndex,
|
||||
))
|
||||
|
||||
private[memory] def appendEntry(log: MutableLog, createEntry: Offset => LedgerEntry): Int = {
|
||||
private[memory] def appendEntry(log: MutableLog, createEntry: Offset => LedgerRecord): Int = {
|
||||
val entryAtIndex = log.size
|
||||
val offset = KVOffset.fromLong(entryAtIndex.toLong)
|
||||
val entry = createEntry(offset)
|
||||
|
||||
@ -7,8 +7,7 @@ import java.util.concurrent.Semaphore
|
||||
|
||||
import com.daml.ledger.on.memory.InMemoryState._
|
||||
import com.daml.ledger.participant.state.kvutils.Bytes
|
||||
import com.daml.ledger.participant.state.kvutils.api.LedgerEntry
|
||||
import com.daml.ledger.participant.state.kvutils.api.LedgerEntry.LedgerRecord
|
||||
import com.daml.ledger.participant.state.kvutils.api.LedgerRecord
|
||||
import com.daml.ledger.participant.state.v1.Offset
|
||||
import com.google.protobuf.ByteString
|
||||
|
||||
@ -33,10 +32,10 @@ private[memory] class InMemoryState private (log: MutableLog, state: MutableStat
|
||||
}
|
||||
|
||||
object InMemoryState {
|
||||
type ImmutableLog = IndexedSeq[LedgerEntry]
|
||||
type ImmutableLog = IndexedSeq[LedgerRecord]
|
||||
type ImmutableState = collection.Map[StateKey, StateValue]
|
||||
|
||||
type MutableLog = mutable.Buffer[LedgerEntry] with ImmutableLog
|
||||
type MutableLog = mutable.Buffer[LedgerRecord] with ImmutableLog
|
||||
type MutableState = mutable.Map[StateKey, StateValue] with ImmutableState
|
||||
|
||||
type StateKey = Bytes
|
||||
|
||||
@ -12,7 +12,7 @@ import com.codahale.metrics.MetricRegistry
|
||||
import com.daml.ledger.on.sql.SqlLedgerReaderWriter._
|
||||
import com.daml.ledger.on.sql.queries.Queries
|
||||
import com.daml.ledger.participant.state.kvutils.DamlKvutils.DamlLogEntryId
|
||||
import com.daml.ledger.participant.state.kvutils.api.{LedgerEntry, LedgerReader, LedgerWriter}
|
||||
import com.daml.ledger.participant.state.kvutils.api.{LedgerReader, LedgerRecord, LedgerWriter}
|
||||
import com.daml.ledger.participant.state.kvutils.{Bytes, KVOffset}
|
||||
import com.daml.ledger.participant.state.v1._
|
||||
import com.daml.ledger.validator.LedgerStateOperations.{Key, Value}
|
||||
@ -66,7 +66,7 @@ final class SqlLedgerReaderWriter(
|
||||
|
||||
override def currentHealth(): HealthStatus = Healthy
|
||||
|
||||
override def events(startExclusive: Option[Offset]): Source[LedgerEntry, NotUsed] =
|
||||
override def events(startExclusive: Option[Offset]): Source[LedgerRecord, NotUsed] =
|
||||
dispatcher
|
||||
.startingAt(
|
||||
KVOffset.highestIndex(startExclusive.getOrElse(StartOffset)),
|
||||
|
||||
@ -10,7 +10,7 @@ import anorm._
|
||||
import com.daml.ledger.on.sql.Index
|
||||
import com.daml.ledger.on.sql.queries.Queries._
|
||||
import com.daml.ledger.participant.state.kvutils.KVOffset
|
||||
import com.daml.ledger.participant.state.kvutils.api.LedgerEntry
|
||||
import com.daml.ledger.participant.state.kvutils.api.LedgerRecord
|
||||
import com.daml.ledger.validator.LedgerStateOperations.{Key, Value}
|
||||
|
||||
import scala.collection.{breakOut, immutable}
|
||||
@ -28,14 +28,14 @@ trait CommonQueries extends Queries {
|
||||
override final def selectFromLog(
|
||||
start: Index,
|
||||
end: Index,
|
||||
): Try[immutable.Seq[(Index, LedgerEntry)]] = Try {
|
||||
): Try[immutable.Seq[(Index, LedgerRecord)]] = Try {
|
||||
SQL"SELECT sequence_no, entry_id, envelope FROM #$LogTable WHERE sequence_no > $start AND sequence_no <= $end ORDER BY sequence_no"
|
||||
.as(
|
||||
(long("sequence_no")
|
||||
~ getBytes("entry_id")
|
||||
~ getBytes("envelope")).map {
|
||||
case index ~ entryId ~ envelope =>
|
||||
index -> LedgerEntry.LedgerRecord(
|
||||
index -> LedgerRecord(
|
||||
KVOffset.fromLong(index),
|
||||
entryId,
|
||||
envelope,
|
||||
|
||||
@ -4,7 +4,7 @@
|
||||
package com.daml.ledger.on.sql.queries
|
||||
|
||||
import com.daml.ledger.on.sql.Index
|
||||
import com.daml.ledger.participant.state.kvutils.api.LedgerEntry
|
||||
import com.daml.ledger.participant.state.kvutils.api.LedgerRecord
|
||||
import com.daml.ledger.validator.LedgerStateOperations.{Key, Value}
|
||||
|
||||
import scala.collection.immutable
|
||||
@ -13,7 +13,7 @@ import scala.util.Try
|
||||
trait ReadQueries {
|
||||
def selectLatestLogEntryId(): Try[Option[Index]]
|
||||
|
||||
def selectFromLog(start: Index, end: Index): Try[immutable.Seq[(Index, LedgerEntry)]]
|
||||
def selectFromLog(start: Index, end: Index): Try[immutable.Seq[(Index, LedgerRecord)]]
|
||||
|
||||
def selectStateValuesByKeys(keys: Seq[Key]): Try[immutable.Seq[Option[Value]]]
|
||||
}
|
||||
|
||||
@ -22,7 +22,7 @@ class KeyValueParticipantStateReader(reader: LedgerReader)(implicit materializer
|
||||
.single(beginAfter.map(KVOffset.onlyKeepHighestIndex))
|
||||
.flatMapConcat(reader.events)
|
||||
.flatMapConcat {
|
||||
case LedgerEntry.LedgerRecord(offset, entryId, envelope) =>
|
||||
case LedgerRecord(offset, entryId, envelope) =>
|
||||
Envelope
|
||||
.open(envelope)
|
||||
.flatMap {
|
||||
|
||||
@ -1,38 +0,0 @@
|
||||
// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package com.daml.ledger.participant.state.kvutils.api
|
||||
|
||||
import com.daml.ledger.participant.state.kvutils.Bytes
|
||||
import com.daml.ledger.participant.state.v1.Offset
|
||||
|
||||
/**
|
||||
* Defines a log entry or update that may be read from the ledger.
|
||||
*/
|
||||
sealed trait LedgerEntry {
|
||||
def offset: Offset
|
||||
}
|
||||
|
||||
object LedgerEntry {
|
||||
|
||||
/**
|
||||
* A log entry read from the ledger.
|
||||
* @param offset offset of log entry
|
||||
* @param entryId opaque ID of log entry
|
||||
* @param envelope opaque contents of log entry
|
||||
*/
|
||||
final class LedgerRecord(
|
||||
val offset: Offset,
|
||||
val entryId: Bytes,
|
||||
val envelope: Bytes
|
||||
) extends LedgerEntry
|
||||
|
||||
object LedgerRecord {
|
||||
def apply(offset: Offset, entryId: Bytes, envelope: Bytes): LedgerRecord =
|
||||
new LedgerRecord(offset, entryId, envelope)
|
||||
|
||||
def unapply(record: LedgerRecord): Option[(Offset, Bytes, Bytes)] =
|
||||
Some((record.offset, record.entryId, record.envelope))
|
||||
}
|
||||
|
||||
}
|
||||
@ -23,13 +23,13 @@ trait LedgerReader extends ReportsHealth {
|
||||
*
|
||||
* In case an offset is not specified, all updates must be streamed from the oldest known state.
|
||||
* Each update is defined either as an opaque log entry ID and an
|
||||
* envelope ([[com.daml.ledger.participant.state.kvutils.api.LedgerEntry.LedgerRecord]]).
|
||||
* envelope ([[com.daml.ledger.participant.state.kvutils.api.LedgerRecord]]).
|
||||
*
|
||||
* @param startExclusive offset right after which updates must be streamed; in case not specified updates
|
||||
* must be returned from the beginning
|
||||
* @return stream of updates
|
||||
*/
|
||||
def events(startExclusive: Option[Offset]): Source[LedgerEntry, NotUsed]
|
||||
def events(startExclusive: Option[Offset]): Source[LedgerRecord, NotUsed]
|
||||
|
||||
/**
|
||||
* Get the ledger's ID from which this reader instance streams events.
|
||||
@ -45,7 +45,7 @@ object LedgerReader {
|
||||
/**
|
||||
* Default initial ledger configuration used by [[KeyValueParticipantStateReader]].
|
||||
*/
|
||||
val DefaultConfiguration = Configuration(
|
||||
val DefaultConfiguration: Configuration = Configuration(
|
||||
generation = 0,
|
||||
timeModel = TimeModel.reasonableDefault,
|
||||
maxDeduplicationTime = Duration.ofDays(1),
|
||||
|
||||
@ -0,0 +1,16 @@
|
||||
// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package com.daml.ledger.participant.state.kvutils.api
|
||||
|
||||
import com.daml.ledger.participant.state.kvutils.Bytes
|
||||
import com.daml.ledger.participant.state.v1.Offset
|
||||
|
||||
/**
|
||||
* A log entry read from the ledger.
|
||||
*
|
||||
* @param offset offset of log entry
|
||||
* @param entryId opaque ID of log entry
|
||||
* @param envelope opaque contents of log entry
|
||||
*/
|
||||
final case class LedgerRecord(offset: Offset, entryId: Bytes, envelope: Bytes)
|
||||
@ -10,7 +10,6 @@ import akka.NotUsed
|
||||
import akka.stream.scaladsl.{Sink, Source}
|
||||
import com.daml.ledger.participant.state.kvutils.DamlKvutils._
|
||||
import com.daml.ledger.participant.state.kvutils.api.KeyValueParticipantStateReaderSpec._
|
||||
import com.daml.ledger.participant.state.kvutils.api.LedgerEntry.LedgerRecord
|
||||
import com.daml.ledger.participant.state.kvutils.{Bytes, Envelope, KVOffset}
|
||||
import com.daml.ledger.participant.state.v1.{Offset, Update}
|
||||
import com.digitalasset.ledger.api.testing.utils.AkkaBeforeAndAfterAll
|
||||
@ -108,7 +107,7 @@ class KeyValueParticipantStateReaderSpec
|
||||
LedgerRecord(toOffset(3), aLogEntryId(3), aWrappedLogEntry)
|
||||
)
|
||||
|
||||
def getInstance(offset: Option[Offset], items: LedgerEntry*) =
|
||||
def getInstance(offset: Option[Offset], items: LedgerRecord*) =
|
||||
new KeyValueParticipantStateReader(readerStreamingFrom(offset = offset, items: _*))
|
||||
|
||||
val instances = records.tails.flatMap {
|
||||
@ -182,14 +181,14 @@ object KeyValueParticipantStateReaderSpec {
|
||||
.build
|
||||
.toByteString
|
||||
|
||||
private def readerStreamingFrom(offset: Option[Offset], items: LedgerEntry*): LedgerReader = {
|
||||
private def readerStreamingFrom(offset: Option[Offset], items: LedgerRecord*): LedgerReader = {
|
||||
val reader = mock[LedgerReader]
|
||||
val stream = Source.fromIterator(() => items.iterator)
|
||||
when(reader.events(offset)).thenReturn(stream)
|
||||
reader
|
||||
}
|
||||
|
||||
private def readerStreamingFromAnyOffset(items: LedgerEntry*): LedgerReader = {
|
||||
private def readerStreamingFromAnyOffset(items: LedgerRecord*): LedgerReader = {
|
||||
val reader = mock[LedgerReader]
|
||||
val stream = Source.fromIterator(() => items.iterator)
|
||||
when(reader.events(any[Option[Offset]]())).thenReturn(stream)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user