You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2022/06/25 04:49:10 UTC
[james-project] 03/18: JAMES-3774 Migrate event-sourcing/event-store-cassandra to Cassandra driver 4
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 42aa736dc0804896adb13f9683d24d4f7124404d
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Jun 3 12:51:48 2022 +0700
JAMES-3774 Migrate event-sourcing/event-store-cassandra to Cassandra driver 4
---
.../cassandra/CassandraEventStoreModule.scala | 21 ++++----
.../eventstore/cassandra/EventStoreDao.scala | 61 ++++++++++------------
.../cassandra/CassandraEventStoreExtension.scala | 11 +---
3 files changed, 41 insertions(+), 52 deletions(-)
diff --git a/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/CassandraEventStoreModule.scala b/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/CassandraEventStoreModule.scala
index b366cb7aec..c92e350ba6 100644
--- a/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/CassandraEventStoreModule.scala
+++ b/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/CassandraEventStoreModule.scala
@@ -18,21 +18,22 @@
****************************************************************/
package org.apache.james.eventsourcing.eventstore.cassandra
+import com.datastax.oss.driver.api.core.`type`.DataTypes
+import com.datastax.oss.driver.api.querybuilder.SchemaBuilder
+import com.datastax.oss.driver.api.querybuilder.schema.CreateTableWithOptions
import org.apache.james.backends.cassandra.components.CassandraModule
import org.apache.james.backends.cassandra.utils.CassandraConstants
-import com.datastax.driver.core.DataType
-import com.datastax.driver.core.schemabuilder.{Create, SchemaBuilder}
object CassandraEventStoreModule {
val MODULE = CassandraModule.table(CassandraEventStoreTable.EVENTS_TABLE)
.comment("Store events of a EventSourcing aggregate")
- .options((options: Create.Options) => options
- .compactionOptions(SchemaBuilder.leveledStrategy())
- .caching(
- SchemaBuilder.KeyCaching.ALL,
- SchemaBuilder.rows(CassandraConstants.DEFAULT_CACHED_ROW_PER_PARTITION)))
- .statement(_.addPartitionKey(CassandraEventStoreTable.AGGREGATE_ID, DataType.varchar)
- .addClusteringColumn(CassandraEventStoreTable.EVENT_ID, DataType.cint)
- .addColumn(CassandraEventStoreTable.EVENT, DataType.text))
+ .options((options: CreateTableWithOptions) => options
+ .withCompaction(SchemaBuilder.leveledCompactionStrategy())
+ .withCaching(
+ true,
+ SchemaBuilder.RowsPerPartition.rows(CassandraConstants.DEFAULT_CACHED_ROW_PER_PARTITION)))
+ .statement(statement => _ => statement.withPartitionKey(CassandraEventStoreTable.AGGREGATE_ID, DataTypes.TEXT)
+ .withClusteringColumn(CassandraEventStoreTable.EVENT_ID, DataTypes.INT)
+ .withColumn(CassandraEventStoreTable.EVENT, DataTypes.TEXT))
.build
}
\ No newline at end of file
diff --git a/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/EventStoreDao.scala b/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/EventStoreDao.scala
index 60ded33f3e..33c8219bbe 100644
--- a/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/EventStoreDao.scala
+++ b/event-sourcing/event-store-cassandra/src/main/scala/org/apache/james/eventsourcing/eventstore/cassandra/EventStoreDao.scala
@@ -16,71 +16,68 @@
* specific language governing permissions and limitations *
* under the License. *
****************************************************************/
+
package org.apache.james.eventsourcing.eventstore.cassandra
-import com.datastax.driver.core._
-import com.datastax.driver.core.querybuilder.QueryBuilder
-import com.datastax.driver.core.querybuilder.QueryBuilder.{bindMarker, insertInto}
+import com.datastax.oss.driver.api.core.CqlSession
+import com.datastax.oss.driver.api.core.cql.{BatchStatementBuilder, BatchType, BoundStatement, PreparedStatement, Row, Statement}
+import com.datastax.oss.driver.api.querybuilder.QueryBuilder
+import com.datastax.oss.driver.api.querybuilder.QueryBuilder.{bindMarker, insertInto}
import javax.inject.Inject
-
-import org.apache.james.backends.cassandra.init.configuration.CassandraConsistenciesConfiguration
+import org.apache.james.backends.cassandra.init.configuration.JamesExecutionProfiles
import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor
import org.apache.james.eventsourcing.eventstore.History
import org.apache.james.eventsourcing.eventstore.cassandra.CassandraEventStoreTable.{AGGREGATE_ID, EVENT, EVENTS_TABLE, EVENT_ID}
import org.apache.james.eventsourcing.{AggregateId, Event}
-
import reactor.core.scala.publisher.{SFlux, SMono}
-class EventStoreDao @Inject() (val session: Session,
- val jsonEventSerializer: JsonEventSerializer,
- val consistenciesConfiguration: CassandraConsistenciesConfiguration) {
+class EventStoreDao @Inject() (val session: CqlSession,
+ val jsonEventSerializer: JsonEventSerializer) {
private val cassandraAsyncExecutor = new CassandraAsyncExecutor(session)
- private val consistencyLevel: ConsistencyLevel = consistenciesConfiguration.getLightweightTransaction
private val insert = prepareInsert(session)
private val select = prepareSelect(session)
+ private val executionProfile = JamesExecutionProfiles.getLWTProfile(session)
- private def prepareInsert(session: Session): PreparedStatement = {
+ private def prepareInsert(session: CqlSession): PreparedStatement =
session.prepare(
insertInto(EVENTS_TABLE)
.value(AGGREGATE_ID, bindMarker(AGGREGATE_ID))
.value(EVENT_ID, bindMarker(EVENT_ID))
.value(EVENT, bindMarker(EVENT))
- .ifNotExists)
- }
+ .ifNotExists
+ .build())
- private def prepareSelect(session: Session): PreparedStatement = {
+ private def prepareSelect(session: CqlSession): PreparedStatement =
session.prepare(QueryBuilder
- .select()
- .from(EVENTS_TABLE)
- .where(QueryBuilder.eq(AGGREGATE_ID, bindMarker(AGGREGATE_ID))))
- }
+ .selectFrom(EVENTS_TABLE)
+ .all()
+ .whereColumn(AGGREGATE_ID).isEqualTo(bindMarker(AGGREGATE_ID))
+ .build())
- private[cassandra] def appendAll(events: Iterable[Event]): SMono[Boolean] = {
+ private[cassandra] def appendAll(events: Iterable[Event]): SMono[Boolean] =
SMono(cassandraAsyncExecutor.executeReturnApplied(appendQuery(events))
.map(_.booleanValue()))
- }
- private def appendQuery(events: Iterable[Event]) = {
+ private def appendQuery(events: Iterable[Event]): Statement[_] =
if (events.size == 1)
insertEvent(events.head)
else {
- val batch: BatchStatement = new BatchStatement
- events.foreach((event: Event) => batch.add(insertEvent(event)))
- batch
+ val batch: BatchStatementBuilder = new BatchStatementBuilder(BatchType.LOGGED)
+ events.foreach((event: Event) => batch.addStatement(insertEvent(event)))
+ batch.build()
}
- }
- private def insertEvent(event: Event): BoundStatement = {
+ private def insertEvent(event: Event): BoundStatement =
insert
- .bind
+ .bind()
.setString(AGGREGATE_ID, event.getAggregateId.asAggregateKey)
.setInt(EVENT_ID, event.eventId.serialize)
.setString(EVENT, jsonEventSerializer.serialize(event))
- }
private[cassandra] def getEventsOfAggregate(aggregateId: AggregateId): SMono[History] = {
- val preparedStatement = select.bind.setString(AGGREGATE_ID, aggregateId.asAggregateKey)
- .setConsistencyLevel(consistencyLevel)
+ val preparedStatement = select.bind()
+ .setString(AGGREGATE_ID, aggregateId.asAggregateKey)
+ .setExecutionProfile(executionProfile)
val rows: SFlux[Row] = SFlux[Row](cassandraAsyncExecutor.executeRows(preparedStatement))
val events: SFlux[Event] = rows.map(toEvent)
@@ -90,7 +87,5 @@ class EventStoreDao @Inject() (val session: Session,
listEvents.map(History.of(_))
}
- private def toEvent(row: Row): Event = {
- jsonEventSerializer.deserialize(row.getString(EVENT))
- }
+ private def toEvent(row: Row): Event = jsonEventSerializer.deserialize(row.getString(EVENT))
}
\ No newline at end of file
diff --git a/event-sourcing/event-store-cassandra/src/test/scala/org/apache/james/eventsourcing/eventstore/cassandra/CassandraEventStoreExtension.scala b/event-sourcing/event-store-cassandra/src/test/scala/org/apache/james/eventsourcing/eventstore/cassandra/CassandraEventStoreExtension.scala
index f480cc0f1b..df5a0514f7 100644
--- a/event-sourcing/event-store-cassandra/src/test/scala/org/apache/james/eventsourcing/eventstore/cassandra/CassandraEventStoreExtension.scala
+++ b/event-sourcing/event-store-cassandra/src/test/scala/org/apache/james/eventsourcing/eventstore/cassandra/CassandraEventStoreExtension.scala
@@ -20,14 +20,7 @@ package org.apache.james.eventsourcing.eventstore.cassandra
import org.apache.james.backends.cassandra.CassandraClusterExtension
import org.apache.james.eventsourcing.eventstore.EventStore
-import org.junit.jupiter.api.extension.AfterAllCallback
-import org.junit.jupiter.api.extension.AfterEachCallback
-import org.junit.jupiter.api.extension.BeforeAllCallback
-import org.junit.jupiter.api.extension.BeforeEachCallback
-import org.junit.jupiter.api.extension.ExtensionContext
-import org.junit.jupiter.api.extension.ParameterContext
-import org.junit.jupiter.api.extension.ParameterResolutionException
-import org.junit.jupiter.api.extension.ParameterResolver
+import org.junit.jupiter.api.extension.{AfterAllCallback, AfterEachCallback, BeforeAllCallback, BeforeEachCallback, ExtensionContext, ParameterContext, ParameterResolutionException, ParameterResolver}
class CassandraEventStoreExtension(var cassandra: CassandraClusterExtension, val eventSerializer: JsonEventSerializer)
extends BeforeAllCallback with AfterAllCallback with BeforeEachCallback with AfterEachCallback with ParameterResolver {
@@ -43,7 +36,7 @@ class CassandraEventStoreExtension(var cassandra: CassandraClusterExtension, val
override def afterAll(context: ExtensionContext): Unit = cassandra.afterAll(context)
override def beforeEach(context: ExtensionContext): Unit = eventStoreDao =
- Some(new EventStoreDao(cassandra.getCassandraCluster.getConf, eventSerializer, cassandra.getCassandraConsistenciesConfiguration))
+ Some(new EventStoreDao(cassandra.getCassandraCluster.getConf, eventSerializer))
override def afterEach(context: ExtensionContext): Unit = cassandra.afterEach(context)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org