Skip to content

Commit d9aee90

Browse files
abaddonabaddon
committed
improvements
1 parent 2fc9b1a commit d9aee90

File tree

9 files changed

+79
-76
lines changed

9 files changed

+79
-76
lines changed

‎gradle/libs.versions.toml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,8 @@
66
kotlin = "2.1.20"
77
kotlinxCoroutines = "1.10.2"
88

9-
kcqrs-core = "0.1.2-SNAPSHOT"
10-
kcqrs-test = "0.1.2-SNAPSHOT"
9+
kcqrs-core = "0.2.0"
10+
kcqrs-test = "0.2.0"
1111

1212
kurrentdb-client = "1.0.1"
1313
jacksonModuleKotlin = "2.16.1"

‎src/main/kotlin/io/github/abaddon/kcqrs/eventstoredb/eventstore/EventStoreDBRepository.kt‎

Lines changed: 69 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -4,24 +4,24 @@ package io.github.abaddon.kcqrs.eventstoredb.eventstore
44
import io.github.abaddon.kcqrs.core.IAggregate
55
import io.github.abaddon.kcqrs.core.IIdentity
66
import io.github.abaddon.kcqrs.core.domain.messages.events.IDomainEvent
7-
import io.github.abaddon.kcqrs.core.helpers.log
7+
import io.github.abaddon.kcqrs.core.helpers.LoggerFactory.log
88
import io.github.abaddon.kcqrs.core.persistence.EventStoreRepository
99
import io.github.abaddon.kcqrs.core.projections.IProjection
1010
import io.github.abaddon.kcqrs.core.projections.IProjectionHandler
1111
import io.github.abaddon.kcqrs.eventstoredb.projection.EventStoreProjectionHandler
1212
import io.kurrent.dbclient.*
13-
import kotlinx.coroutines.CoroutineDispatcher
1413
import kotlinx.coroutines.withContext
1514
import java.security.InvalidParameterException
1615
import java.util.concurrent.CompletionException
16+
import kotlin.coroutines.CoroutineContext
1717

1818

1919
class EventStoreDBRepository<TAggregate : IAggregate>(
2020
eventStoreRepositoryConfig: EventStoreDBRepositoryConfig,
2121
private val funEmpty: (identity: IIdentity) -> TAggregate,
22-
dispatcher: CoroutineDispatcher
22+
coroutineContext: CoroutineContext
2323
) :
24-
EventStoreRepository<TAggregate>(dispatcher) {
24+
EventStoreRepository<TAggregate>(coroutineContext) {
2525
private val client: KurrentDBClient =
2626
KurrentDBClient.create(eventStoreRepositoryConfig.eventStoreDBClientSettings())
2727
private val MAX_READ_PAGE_SIZE: Long = eventStoreRepositoryConfig.maxReadPageSize
@@ -38,61 +38,76 @@ class EventStoreDBRepository<TAggregate : IAggregate>(
3838
}
3939
}
4040

41-
override suspend fun load(streamName: String, startFrom: Long): List<IDomainEvent> = withContext(coroutineContext) {
42-
val eventsFound = mutableListOf<IDomainEvent>()
43-
var currentRevision: Long = startFrom
44-
log.debug("loading events from stream {} with startRevision {}", streamName, startFrom)
45-
try {
46-
var hasMoreEvents = true
47-
while (hasMoreEvents) {
48-
val options = ReadStreamOptions.get()
49-
.forwards()
50-
.fromRevision(currentRevision)
51-
.maxCount(MAX_READ_PAGE_SIZE)
52-
val result = client.readStream(streamName, options).get()
53-
54-
val events = result.events
41+
override suspend fun load(streamName: String, startFrom: Long): Result<List<IDomainEvent>> =
42+
withContext(coroutineContext) {
43+
val eventsFound = mutableListOf<IDomainEvent>()
44+
var currentRevision: Long = startFrom
45+
log.debug("loading events from stream {} with startRevision {}", streamName, startFrom)
46+
val result = runCatching {
47+
var hasMoreEvents = true
48+
while (hasMoreEvents) {
49+
val options = ReadStreamOptions.get()
50+
.forwards()
51+
.fromRevision(currentRevision)
52+
.maxCount(MAX_READ_PAGE_SIZE)
53+
val result = client.readStream(streamName, options).get()
54+
55+
val events = result.events
56+
log.debug(
57+
"events received: {}, firstStreamPosition: {}, lastStreamPosition {}",
58+
events.size,
59+
result.firstStreamPosition,
60+
result.lastStreamPosition
61+
)
62+
val maxRevision = events.maxOfOrNull { event ->
63+
log.debug("event.originalEvent.revision, {}", event.originalEvent.revision)
64+
event.originalEvent.revision
65+
}
66+
log.debug("maxRevision is {}", maxRevision)
67+
if (events.isEmpty()) {
68+
hasMoreEvents = false
69+
log.debug("stream is empty")
70+
} else {
71+
eventsFound.addAll(events.toDomainEvents())
72+
currentRevision += events.size
73+
if (currentRevision != maxRevision) {
74+
log.warn(
75+
"currentRevision and maxRevision are different! {} and {}",
76+
currentRevision,
77+
maxRevision
78+
)
79+
}
80+
}
81+
}
5582
log.debug(
56-
"events received: {}, firstStreamPosition: {}, lastStreamPosition {}",
57-
events.size,
58-
result.firstStreamPosition,
59-
result.lastStreamPosition
83+
"end loading events from stream {} with startRevision {} getting {} events",
84+
streamName,
85+
startFrom,
86+
eventsFound.size
6087
)
61-
val maxRevision = events.maxOfOrNull { event ->
62-
log.debug("event.originalEvent.revision, {}", event.originalEvent.revision)
63-
event.originalEvent.revision
64-
}
65-
log.debug("maxRevision is {}", maxRevision)
66-
if (events.isEmpty()) {
67-
hasMoreEvents = false
68-
log.debug("stream is empty")
69-
} else {
70-
eventsFound.addAll(events.toDomainEvents())
71-
currentRevision += events.size
72-
if (currentRevision != maxRevision) {
73-
log.warn(
74-
"currentRevision and maxRevision are different! {} and {}",
75-
currentRevision,
76-
maxRevision
77-
)
88+
eventsFound
89+
}
90+
91+
when {
92+
result.isFailure -> {
93+
val ex = result.exceptionOrNull()!!
94+
log.debug("Error reading stream: {}", streamName, ex)
95+
when (ex.cause) {
96+
is StreamNotFoundException -> {
97+
log.debug("Stream not found: {}", streamName)
98+
Result.success(eventsFound)
99+
}
100+
101+
else -> {
102+
log.error("Error reading stream: {}", streamName, ex)
103+
result
104+
}
78105
}
79106
}
80-
}
81-
} catch (ex: CompletionException) {
82-
log.debug("Error reading stream: {}", streamName, ex)
83-
when (ex.cause) {
84-
is StreamNotFoundException -> log.debug("Stream not found: {}", streamName)
85-
else -> log.error("Error reading stream: {}", streamName, ex)
107+
108+
else -> result
86109
}
87110
}
88-
log.debug(
89-
"end loading events from stream {} with startRevision {} getting {} events",
90-
streamName,
91-
startFrom,
92-
eventsFound.size
93-
)
94-
eventsFound
95-
}
96111

97112
override fun aggregateIdStreamName(aggregateId: IIdentity): String {
98113
check(streamName.isNotEmpty()) { throw InvalidParameterException("Cannot get streamName empty") }
@@ -149,7 +164,7 @@ class EventStoreDBRepository<TAggregate : IAggregate>(
149164

150165
}
151166

152-
override suspend fun publish(persistResult: Result<Unit>, events: List<IDomainEvent>): Result<Unit> =
167+
override suspend fun publish(events: List<IDomainEvent>): Result<Unit> =
153168
Result.success(Unit)
154169

155170
}

‎src/main/kotlin/io/github/abaddon/kcqrs/eventstoredb/projection/EventStoreProjectionHandler.kt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import io.kurrent.dbclient.ResolvedEvent
55
import io.kurrent.dbclient.Subscription
66
import io.kurrent.dbclient.SubscriptionListener
77
import io.github.abaddon.kcqrs.core.domain.messages.events.IDomainEvent
8-
import io.github.abaddon.kcqrs.core.helpers.log
8+
import io.github.abaddon.kcqrs.core.helpers.LoggerFactory.log
99
import io.github.abaddon.kcqrs.core.persistence.IProjectionRepository
1010
import io.github.abaddon.kcqrs.core.projections.IProjection
1111
import io.github.abaddon.kcqrs.core.projections.IProjectionHandler

‎src/test/kotlin/io/github/abaddon/kcqrs/eventstoredb/eventstore/EventStoreDBRepositoryTest.kt‎

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
99
import kotlinx.coroutines.test.TestScope
1010
import kotlinx.coroutines.test.UnconfinedTestDispatcher
1111
import kotlinx.coroutines.test.runTest
12-
import org.junit.jupiter.api.AfterEach
1312
import org.junit.jupiter.api.BeforeEach
1413
import org.junit.jupiter.api.Test
1514
import java.util.*
@@ -39,12 +38,6 @@ internal class EventStoreDBRepositoryTest : WithEventStoreDBContainer() {
3938
)
4039
}
4140

42-
@AfterEach
43-
fun close(){
44-
repository.cleanup()
45-
}
46-
47-
4841
@Test
4942
fun `Given the initialise event when persist it then event is stored in the eventStore`() = testScope.runTest {
5043
//Given

‎src/test/kotlin/io/github/abaddon/kcqrs/eventstoredb/projection/EventStoreProjectionHandlerTest.kt‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -48,11 +48,6 @@ internal class EventStoreProjectionHandlerTest : WithEventStoreDBContainer() {
4848
)
4949
}
5050

51-
@AfterEach
52-
fun close() {
53-
repository.cleanup()
54-
}
55-
5651
@Test
5752
fun `Given an event, when it's applied to a projection then the updated projection is updated`() =
5853
testScope.runTest {

‎src/test/kotlin/io/github/abaddon/kcqrs/testHelpers/EventStoreContainer.kt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package io.github.abaddon.kcqrs.testHelpers
22

3-
import io.github.abaddon.kcqrs.core.helpers.log
3+
import io.github.abaddon.kcqrs.core.helpers.LoggerFactory.log
44
import org.testcontainers.containers.GenericContainer
55
import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy
66
import java.time.Duration

‎src/test/kotlin/io/github/abaddon/kcqrs/testHelpers/commands/DecreaseCounterCommand.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,8 @@ data class DecreaseCounterCommand(
88
override val aggregateID: CounterAggregateId,
99
val value: Int
1010
) : Command<CounterAggregateRoot>(aggregateID) {
11-
override fun execute(currentAggregate: CounterAggregateRoot?): CounterAggregateRoot {
11+
override fun execute(currentAggregate: CounterAggregateRoot?): Result<CounterAggregateRoot> = runCatching {
1212
requireNotNull(currentAggregate)
13-
return currentAggregate.decreaseCounter(value)
13+
currentAggregate.decreaseCounter(value)
1414
}
1515
}

‎src/test/kotlin/io/github/abaddon/kcqrs/testHelpers/commands/IncreaseCounterCommand.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,8 @@ data class IncreaseCounterCommand(
88
override val aggregateID: CounterAggregateId,
99
val value: Int
1010
) : Command<CounterAggregateRoot>(aggregateID) {
11-
override fun execute(currentAggregate: CounterAggregateRoot?): CounterAggregateRoot {
11+
override fun execute(currentAggregate: CounterAggregateRoot?): Result<CounterAggregateRoot> = runCatching {
1212
requireNotNull(currentAggregate)
13-
return currentAggregate.increaseCounter(value)
13+
currentAggregate.increaseCounter(value)
1414
}
1515
}

‎src/test/kotlin/io/github/abaddon/kcqrs/testHelpers/commands/InitialiseCounterCommand.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ data class InitialiseCounterCommand(
99
val value: Int
1010
) : Command<CounterAggregateRoot>(aggregateID) {
1111

12-
override fun execute(currentAggregate: CounterAggregateRoot?): CounterAggregateRoot {
13-
return CounterAggregateRoot.initialiseCounter(aggregateID, value)
12+
override fun execute(currentAggregate: CounterAggregateRoot?): Result<CounterAggregateRoot> = runCatching {
13+
CounterAggregateRoot.initialiseCounter(aggregateID, value)
1414
}
1515
}

0 commit comments

Comments
 (0)