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:11 UTC
[james-project] 04/18: JAMES-3774 Migrate server/mailrepository/mailrepository-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 2f1ccda75c4f3677c0d6440c792302742ee3ef30
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Fri Jun 3 15:42:01 2022 +0700
JAMES-3774 Migrate server/mailrepository/mailrepository-cassandra to Cassandra driver 4
---
.../cassandra/CassandraMailRepositoryCountDAO.java | 41 ++++----
.../cassandra/CassandraMailRepositoryKeysDAO.java | 67 +++++++-----
.../CassandraMailRepositoryMailDaoV2.java | 114 +++++++++++----------
.../cassandra/CassandraMailRepositoryModule.java | 63 ++++++------
.../CassandraMailRepositoryUrlModule.java | 12 +--
.../james/mailrepository/cassandra/UrlsDao.java | 37 ++++---
.../cassandra/CassandraMailRepositoryTest.java | 4 +-
...aMailRepositoryWithFakeImplementationsTest.java | 25 ++---
8 files changed, 189 insertions(+), 174 deletions(-)
diff --git a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryCountDAO.java b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryCountDAO.java
index c04f9c2dde..3f2c22a6a9 100644
--- a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryCountDAO.java
+++ b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryCountDAO.java
@@ -19,12 +19,10 @@
package org.apache.james.mailrepository.cassandra;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.decr;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.incr;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.update;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.selectFrom;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.update;
+import static com.datastax.oss.driver.api.querybuilder.relation.Relation.column;
import static org.apache.james.mailrepository.cassandra.MailRepositoryTable.COUNT;
import static org.apache.james.mailrepository.cassandra.MailRepositoryTable.COUNT_TABLE;
import static org.apache.james.mailrepository.cassandra.MailRepositoryTable.REPOSITORY_NAME;
@@ -36,9 +34,9 @@ import javax.inject.Inject;
import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
import org.apache.james.mailrepository.api.MailRepositoryUrl;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.Row;
import reactor.core.publisher.Mono;
@@ -50,7 +48,7 @@ public class CassandraMailRepositoryCountDAO {
private final PreparedStatement select;
@Inject
- public CassandraMailRepositoryCountDAO(Session session) {
+ public CassandraMailRepositoryCountDAO(CqlSession session) {
this.executor = new CassandraAsyncExecutor(session);
this.increment = prepareIncrement(session);
@@ -58,22 +56,25 @@ public class CassandraMailRepositoryCountDAO {
this.select = prepareSelect(session);
}
- private PreparedStatement prepareDecrement(Session session) {
+ private PreparedStatement prepareDecrement(CqlSession session) {
return session.prepare(update(COUNT_TABLE)
- .with(decr(COUNT))
- .where(eq(REPOSITORY_NAME, bindMarker(REPOSITORY_NAME))));
+ .decrement(COUNT)
+ .where(column(REPOSITORY_NAME).isEqualTo(bindMarker(REPOSITORY_NAME)))
+ .build());
}
- private PreparedStatement prepareIncrement(Session session) {
+ private PreparedStatement prepareIncrement(CqlSession session) {
return session.prepare(update(COUNT_TABLE)
- .with(incr(COUNT))
- .where(eq(REPOSITORY_NAME, bindMarker(REPOSITORY_NAME))));
+ .increment(COUNT)
+ .where(column(REPOSITORY_NAME).isEqualTo(bindMarker(REPOSITORY_NAME)))
+ .build());
}
- private PreparedStatement prepareSelect(Session session) {
- return session.prepare(select(COUNT)
- .from(COUNT_TABLE)
- .where(eq(REPOSITORY_NAME, bindMarker(REPOSITORY_NAME))));
+ private PreparedStatement prepareSelect(CqlSession session) {
+ return session.prepare(selectFrom(COUNT_TABLE)
+ .column(COUNT)
+ .where(column(REPOSITORY_NAME).isEqualTo(bindMarker(REPOSITORY_NAME)))
+ .build());
}
public Mono<Void> increment(MailRepositoryUrl url) {
diff --git a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryKeysDAO.java b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryKeysDAO.java
index a9f7fdb701..4e1b973e46 100644
--- a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryKeysDAO.java
+++ b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryKeysDAO.java
@@ -19,11 +19,12 @@
package org.apache.james.mailrepository.cassandra;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.delete;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.deleteFrom;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.insertInto;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.selectFrom;
+import static com.datastax.oss.driver.api.querybuilder.relation.Relation.column;
+import static java.util.List.of;
import static org.apache.james.mailrepository.cassandra.MailRepositoryTable.KEYS_TABLE_NAME;
import static org.apache.james.mailrepository.cassandra.MailRepositoryTable.MAIL_KEY;
import static org.apache.james.mailrepository.cassandra.MailRepositoryTable.REPOSITORY_NAME;
@@ -35,10 +36,10 @@ import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
import org.apache.james.mailrepository.api.MailKey;
import org.apache.james.mailrepository.api.MailRepositoryUrl;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.querybuilder.Delete;
-import com.datastax.driver.core.querybuilder.Insert;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.querybuilder.delete.Delete;
+import com.datastax.oss.driver.api.querybuilder.insert.Insert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -52,7 +53,7 @@ public class CassandraMailRepositoryKeysDAO {
private final boolean strongConsistency;
@Inject
- public CassandraMailRepositoryKeysDAO(Session session, CassandraConfiguration cassandraConfiguration) {
+ public CassandraMailRepositoryKeysDAO(CqlSession session, CassandraConfiguration cassandraConfiguration) {
this.strongConsistency = cassandraConfiguration.isMailRepositoryStrongConsistency();
this.executor = new CassandraAsyncExecutor(session);
@@ -61,50 +62,62 @@ public class CassandraMailRepositoryKeysDAO {
this.listKeys = prepareList(session);
}
- private PreparedStatement prepareList(Session session) {
- return session.prepare(select(MAIL_KEY)
- .from(KEYS_TABLE_NAME)
- .where(eq(REPOSITORY_NAME, bindMarker(REPOSITORY_NAME))));
+ private PreparedStatement prepareList(CqlSession session) {
+ return session.prepare(selectFrom(KEYS_TABLE_NAME)
+ .column(MAIL_KEY)
+ .where(column(REPOSITORY_NAME).isEqualTo(bindMarker(REPOSITORY_NAME)))
+ .build());
}
- private PreparedStatement prepareDelete(Session session) {
- Delete.Where deleteStatement = delete()
- .from(KEYS_TABLE_NAME)
- .where(eq(REPOSITORY_NAME, bindMarker(REPOSITORY_NAME)))
- .and(eq(MAIL_KEY, bindMarker(MAIL_KEY)));
+ private PreparedStatement prepareDelete(CqlSession session) {
+ Delete deleteStatement = deleteFrom(KEYS_TABLE_NAME)
+ .where(of(column(REPOSITORY_NAME).isEqualTo(bindMarker(REPOSITORY_NAME)),
+ column(MAIL_KEY).isEqualTo(bindMarker(MAIL_KEY))));
if (strongConsistency) {
- return session.prepare(deleteStatement.ifExists());
+ return session.prepare(deleteStatement.ifExists().build());
}
- return session.prepare(deleteStatement);
+ return session.prepare(deleteStatement.build());
}
- private PreparedStatement prepareInsert(Session session) {
+ private PreparedStatement prepareInsert(CqlSession session) {
Insert insertStatement = insertInto(KEYS_TABLE_NAME)
.value(REPOSITORY_NAME, bindMarker(REPOSITORY_NAME))
.value(MAIL_KEY, bindMarker(MAIL_KEY));
if (strongConsistency) {
- return session.prepare(insertStatement.ifNotExists());
+ return session.prepare(insertStatement.ifNotExists().build());
}
- return session.prepare(insertStatement);
+ return session.prepare(insertStatement.build());
}
public Mono<Boolean> store(MailRepositoryUrl url, MailKey key) {
- return executor.executeReturnApplied(insertKey.bind()
+ Mono<Boolean> operation = executor.executeReturnApplied(insertKey.bind()
.setString(REPOSITORY_NAME, url.asString())
.setString(MAIL_KEY, key.asString()));
+
+ if (strongConsistency) {
+ return operation;
+ } else {
+ return operation.switchIfEmpty(Mono.just(true));
+ }
}
public Flux<MailKey> list(MailRepositoryUrl url) {
return executor.executeRows(listKeys.bind()
- .setString(REPOSITORY_NAME, url.asString()))
+ .setString(REPOSITORY_NAME, url.asString()))
.map(row -> new MailKey(row.getString(MAIL_KEY)));
}
public Mono<Boolean> remove(MailRepositoryUrl url, MailKey key) {
- return executor.executeReturnApplied(deleteKey.bind()
+ Mono<Boolean> operation = executor.executeReturnApplied(deleteKey.bind()
.setString(REPOSITORY_NAME, url.asString())
.setString(MAIL_KEY, key.asString()));
+
+ if (strongConsistency) {
+ return operation;
+ } else {
+ return operation.switchIfEmpty(Mono.just(true));
+ }
}
}
diff --git a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryMailDaoV2.java b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryMailDaoV2.java
index 06c9c0a1f1..27461392c3 100644
--- a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryMailDaoV2.java
+++ b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryMailDaoV2.java
@@ -19,12 +19,12 @@
package org.apache.james.mailrepository.cassandra;
-import static com.datastax.driver.core.DataType.text;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.delete;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.deleteFrom;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.insertInto;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.selectFrom;
+import static com.datastax.oss.driver.api.querybuilder.relation.Relation.column;
+import static java.util.List.of;
import static org.apache.james.mailrepository.cassandra.MailRepositoryTableV2.ATTRIBUTES;
import static org.apache.james.mailrepository.cassandra.MailRepositoryTableV2.BODY_BLOB_ID;
import static org.apache.james.mailrepository.cassandra.MailRepositoryTableV2.CONTENT_TABLE_NAME;
@@ -69,12 +69,13 @@ import org.apache.mailet.Mail;
import org.apache.mailet.PerRecipientHeaders;
import org.apache.mailet.PerRecipientHeaders.Header;
-import com.datastax.driver.core.BoundStatement;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.TupleType;
-import com.datastax.driver.core.TupleValue;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.BoundStatementBuilder;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.data.TupleValue;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.TupleType;
import com.github.fge.lambdas.Throwing;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
@@ -136,7 +137,7 @@ public class CassandraMailRepositoryMailDaoV2 {
@Inject
@VisibleForTesting
- CassandraMailRepositoryMailDaoV2(Session session, BlobId.Factory blobIdFactory) {
+ CassandraMailRepositoryMailDaoV2(CqlSession session, BlobId.Factory blobIdFactory) {
this.executor = new CassandraAsyncExecutor(session);
this.insertMail = prepareInsert(session);
@@ -144,23 +145,23 @@ public class CassandraMailRepositoryMailDaoV2 {
this.selectMail = prepareSelect(session);
this.litBlobs = prepareListBlobs(session);
this.blobIdFactory = blobIdFactory;
- this.userHeaderNameHeaderValueTriple = session.getCluster().getMetadata().newTupleType(text(), text(), text());
+ this.userHeaderNameHeaderValueTriple = DataTypes.tupleOf(DataTypes.TEXT, DataTypes.TEXT, DataTypes.TEXT);
}
- private PreparedStatement prepareDelete(Session session) {
- return session.prepare(delete()
- .from(CONTENT_TABLE_NAME)
- .where(eq(REPOSITORY_NAME, bindMarker(REPOSITORY_NAME)))
- .and(eq(MAIL_KEY, bindMarker(MAIL_KEY))));
+ private PreparedStatement prepareDelete(CqlSession session) {
+ return session.prepare(deleteFrom(CONTENT_TABLE_NAME)
+ .where(of(column(REPOSITORY_NAME).isEqualTo(bindMarker(REPOSITORY_NAME)),
+ column(MAIL_KEY).isEqualTo(bindMarker(MAIL_KEY))))
+ .build());
}
- private PreparedStatement prepareListBlobs(Session session) {
+ private PreparedStatement prepareListBlobs(CqlSession session) {
return session.prepare(
- select(HEADER_BLOB_ID, BODY_BLOB_ID)
- .from(CONTENT_TABLE_NAME));
+ selectFrom(CONTENT_TABLE_NAME)
+ .columns(HEADER_BLOB_ID, BODY_BLOB_ID).build());
}
- private PreparedStatement prepareInsert(Session session) {
+ private PreparedStatement prepareInsert(CqlSession session) {
return session.prepare(insertInto(CONTENT_TABLE_NAME)
.value(REPOSITORY_NAME, bindMarker(REPOSITORY_NAME))
.value(MAIL_KEY, bindMarker(MAIL_KEY))
@@ -174,42 +175,44 @@ public class CassandraMailRepositoryMailDaoV2 {
.value(LAST_UPDATED, bindMarker(LAST_UPDATED))
.value(HEADER_BLOB_ID, bindMarker(HEADER_BLOB_ID))
.value(BODY_BLOB_ID, bindMarker(BODY_BLOB_ID))
- .value(PER_RECIPIENT_SPECIFIC_HEADERS, bindMarker(PER_RECIPIENT_SPECIFIC_HEADERS)));
+ .value(PER_RECIPIENT_SPECIFIC_HEADERS, bindMarker(PER_RECIPIENT_SPECIFIC_HEADERS))
+ .build());
}
- private PreparedStatement prepareSelect(Session session) {
+ private PreparedStatement prepareSelect(CqlSession session) {
return session.prepare(
- select(MAIL_PROPERTIES)
- .from(CONTENT_TABLE_NAME)
- .where(eq(REPOSITORY_NAME, bindMarker(REPOSITORY_NAME)))
- .and(eq(MAIL_KEY, bindMarker(MAIL_KEY))));
+ selectFrom(CONTENT_TABLE_NAME)
+ .columns(MAIL_PROPERTIES)
+ .where(of(column(REPOSITORY_NAME).isEqualTo(bindMarker(REPOSITORY_NAME)),
+ column(MAIL_KEY).isEqualTo(bindMarker(MAIL_KEY))))
+ .build());
}
public Mono<Void> store(MailRepositoryUrl url, Mail mail, BlobId headerId, BlobId bodyId) {
return Mono.fromCallable(() -> {
- BoundStatement boundStatement = insertMail.bind()
- .setString(REPOSITORY_NAME, url.asString())
- .setString(MAIL_KEY, mail.getName())
- .setString(HEADER_BLOB_ID, headerId.asString())
- .setString(BODY_BLOB_ID, bodyId.asString())
- .setString(STATE, mail.getState())
- .setList(RECIPIENTS, asStringList(mail.getRecipients()))
- .setString(REMOTE_ADDR, mail.getRemoteAddr())
- .setString(REMOTE_HOST, mail.getRemoteHost())
- .setTimestamp(LAST_UPDATED, mail.getLastUpdated())
- .setMap(ATTRIBUTES, toRawAttributeMap(mail))
- .setList(PER_RECIPIENT_SPECIFIC_HEADERS, toTupleList(mail.getPerRecipientSpecificHeaders()));
-
- Optional.ofNullable(mail.getErrorMessage())
- .ifPresent(errorMessage -> boundStatement.setString(MailRepositoryTable.ERROR_MESSAGE, mail.getErrorMessage()));
-
- mail.getMaybeSender()
- .asOptional()
- .map(MailAddress::asString)
- .ifPresent(mailAddress -> boundStatement.setString(MailRepositoryTable.SENDER, mailAddress));
-
- return boundStatement;
- })
+ BoundStatementBuilder statement = insertMail.boundStatementBuilder()
+ .setString(REPOSITORY_NAME, url.asString())
+ .setString(MAIL_KEY, mail.getName())
+ .setString(HEADER_BLOB_ID, headerId.asString())
+ .setString(BODY_BLOB_ID, bodyId.asString())
+ .setString(STATE, mail.getState())
+ .setList(RECIPIENTS, asStringList(mail.getRecipients()), String.class)
+ .setString(REMOTE_ADDR, mail.getRemoteAddr())
+ .setString(REMOTE_HOST, mail.getRemoteHost())
+ .setInstant(LAST_UPDATED, Optional.ofNullable(mail.getLastUpdated()).map(Date::toInstant).orElse(null))
+ .setMap(ATTRIBUTES, toRawAttributeMap(mail), String.class, String.class)
+ .setList(PER_RECIPIENT_SPECIFIC_HEADERS, toTupleList(mail.getPerRecipientSpecificHeaders()), TupleValue.class);
+
+ Optional.ofNullable(mail.getErrorMessage())
+ .ifPresent(errorMessage -> statement.setString(MailRepositoryTable.ERROR_MESSAGE, mail.getErrorMessage()));
+
+ mail.getMaybeSender()
+ .asOptional()
+ .map(MailAddress::asString)
+ .ifPresent(mailAddress -> statement.setString(MailRepositoryTable.SENDER, mailAddress));
+
+ return statement.build();
+ })
.flatMap(executor::executeVoid);
}
@@ -237,7 +240,10 @@ public class CassandraMailRepositoryMailDaoV2 {
String remoteHost = row.getString(REMOTE_HOST);
String errorMessage = row.getString(ERROR_MESSAGE);
String name = row.getString(MAIL_KEY);
- Date lastUpdated = row.getTimestamp(LAST_UPDATED);
+ Date lastUpdated = Optional.ofNullable(row.getInstant(LAST_UPDATED))
+ .map(Date::from)
+ .orElse(null);
+
Map<String, String> rawAttributes = row.getMap(ATTRIBUTES, String.class, String.class);
PerRecipientHeaders perRecipientHeaders = fromList(row.getList(PER_RECIPIENT_SPECIFIC_HEADERS, TupleValue.class));
@@ -310,7 +316,7 @@ public class CassandraMailRepositoryMailDaoV2 {
Flux<BlobId> listBlobs() {
return executor.executeRows(litBlobs.bind())
- .flatMapIterable(row -> ImmutableList.of(
+ .flatMapIterable(row -> List.of(
blobIdFactory.from(row.getString(HEADER_BLOB_ID)),
blobIdFactory.from(row.getString(BODY_BLOB_ID))
));
diff --git a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryModule.java b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryModule.java
index 08d5baeff7..8d0a44a2c7 100644
--- a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryModule.java
+++ b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryModule.java
@@ -19,56 +19,53 @@
package org.apache.james.mailrepository.cassandra;
-import static com.datastax.driver.core.DataType.bigint;
-import static com.datastax.driver.core.DataType.counter;
-import static com.datastax.driver.core.DataType.list;
-import static com.datastax.driver.core.DataType.map;
-import static com.datastax.driver.core.DataType.text;
-import static com.datastax.driver.core.DataType.timestamp;
+import static com.datastax.oss.driver.api.core.type.DataTypes.BIGINT;
+import static com.datastax.oss.driver.api.core.type.DataTypes.COUNTER;
+import static com.datastax.oss.driver.api.core.type.DataTypes.TEXT;
+import static com.datastax.oss.driver.api.core.type.DataTypes.TIMESTAMP;
+import static com.datastax.oss.driver.api.core.type.DataTypes.listOf;
+import static com.datastax.oss.driver.api.core.type.DataTypes.mapOf;
+import static com.datastax.oss.driver.api.core.type.DataTypes.tupleOf;
import org.apache.james.backends.cassandra.components.CassandraModule;
-import com.datastax.driver.core.CodecRegistry;
-import com.datastax.driver.core.ProtocolVersion;
-import com.datastax.driver.core.TupleType;
-
public interface CassandraMailRepositoryModule {
CassandraModule MODULE = CassandraModule.builder()
.type(MailRepositoryTable.HEADER_TYPE)
.statement(statement -> statement
- .addColumn(MailRepositoryTable.HEADER_NAME, text())
- .addColumn(MailRepositoryTable.HEADER_VALUE, text()))
+ .withField(MailRepositoryTable.HEADER_NAME, TEXT)
+ .withField(MailRepositoryTable.HEADER_VALUE, TEXT))
.table(MailRepositoryTable.COUNT_TABLE)
.comment("Projection maintaining per mail repository mail counter")
- .statement(statement -> statement
- .addPartitionKey(MailRepositoryTable.REPOSITORY_NAME, text())
- .addColumn(MailRepositoryTable.COUNT, counter()))
+ .statement(statement -> types -> statement
+ .withPartitionKey(MailRepositoryTable.REPOSITORY_NAME, TEXT)
+ .withColumn(MailRepositoryTable.COUNT, COUNTER))
.table(MailRepositoryTable.KEYS_TABLE_NAME)
.comment("Per-mailRepository mail key list")
- .statement(statement -> statement
- .addPartitionKey(MailRepositoryTable.REPOSITORY_NAME, text())
- .addClusteringColumn(MailRepositoryTable.MAIL_KEY, text()))
+ .statement(statement -> types -> statement
+ .withPartitionKey(MailRepositoryTable.REPOSITORY_NAME, TEXT)
+ .withClusteringColumn(MailRepositoryTable.MAIL_KEY, TEXT))
.table(MailRepositoryTableV2.CONTENT_TABLE_NAME)
.comment("Stores the mails for a given repository. " +
"Content is stored with other blobs. " +
"This v2 version was introduced to support multiple headers for each user. " +
"The attributes are store as Json introduced in Mailet API v3.2.")
- .statement(statement -> statement
- .addPartitionKey(MailRepositoryTable.REPOSITORY_NAME, text())
- .addPartitionKey(MailRepositoryTable.MAIL_KEY, text())
- .addColumn(MailRepositoryTable.MESSAGE_SIZE, bigint())
- .addColumn(MailRepositoryTable.STATE, text())
- .addColumn(MailRepositoryTable.HEADER_BLOB_ID, text())
- .addColumn(MailRepositoryTable.BODY_BLOB_ID, text())
- .addColumn(MailRepositoryTable.ATTRIBUTES, map(text(), text()))
- .addColumn(MailRepositoryTable.ERROR_MESSAGE, text())
- .addColumn(MailRepositoryTable.SENDER, text())
- .addColumn(MailRepositoryTable.RECIPIENTS, list(text()))
- .addColumn(MailRepositoryTable.REMOTE_HOST, text())
- .addColumn(MailRepositoryTable.REMOTE_ADDR, text())
- .addColumn(MailRepositoryTable.LAST_UPDATED, timestamp())
- .addColumn(MailRepositoryTable.PER_RECIPIENT_SPECIFIC_HEADERS, list(TupleType.of(ProtocolVersion.NEWEST_SUPPORTED, CodecRegistry.DEFAULT_INSTANCE, text(), text(), text()))))
+ .statement(statement -> types -> statement
+ .withPartitionKey(MailRepositoryTable.REPOSITORY_NAME, TEXT)
+ .withPartitionKey(MailRepositoryTable.MAIL_KEY, TEXT)
+ .withColumn(MailRepositoryTable.MESSAGE_SIZE, BIGINT)
+ .withColumn(MailRepositoryTable.STATE, TEXT)
+ .withColumn(MailRepositoryTable.HEADER_BLOB_ID, TEXT)
+ .withColumn(MailRepositoryTable.BODY_BLOB_ID, TEXT)
+ .withColumn(MailRepositoryTable.ATTRIBUTES, mapOf(TEXT, TEXT))
+ .withColumn(MailRepositoryTable.ERROR_MESSAGE, TEXT)
+ .withColumn(MailRepositoryTable.SENDER, TEXT)
+ .withColumn(MailRepositoryTable.RECIPIENTS, listOf(TEXT))
+ .withColumn(MailRepositoryTable.REMOTE_HOST, TEXT)
+ .withColumn(MailRepositoryTable.REMOTE_ADDR, TEXT)
+ .withColumn(MailRepositoryTable.LAST_UPDATED, TIMESTAMP)
+ .withColumn(MailRepositoryTable.PER_RECIPIENT_SPECIFIC_HEADERS, listOf(tupleOf(TEXT, TEXT, TEXT))))
.build();
}
diff --git a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryUrlModule.java b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryUrlModule.java
index 6d7b17ba22..716a9a160b 100644
--- a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryUrlModule.java
+++ b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryUrlModule.java
@@ -19,20 +19,18 @@
package org.apache.james.mailrepository.cassandra;
-import static com.datastax.driver.core.DataType.text;
+import static com.datastax.oss.driver.api.core.type.DataTypes.TEXT;
+import static com.datastax.oss.driver.api.querybuilder.SchemaBuilder.RowsPerPartition.rows;
import org.apache.james.backends.cassandra.components.CassandraModule;
import org.apache.james.backends.cassandra.utils.CassandraConstants;
-import com.datastax.driver.core.schemabuilder.SchemaBuilder;
-
public interface CassandraMailRepositoryUrlModule {
CassandraModule MODULE = CassandraModule.table(UrlsTable.TABLE_NAME)
.comment("Holds the list of available mail repository")
.options(options -> options
- .caching(SchemaBuilder.KeyCaching.ALL,
- SchemaBuilder.rows(CassandraConstants.DEFAULT_CACHED_ROW_PER_PARTITION)))
- .statement(statement -> statement
- .addPartitionKey(UrlsTable.URL, text()))
+ .withCaching(true, rows(CassandraConstants.DEFAULT_CACHED_ROW_PER_PARTITION)))
+ .statement(statement -> types -> statement
+ .withPartitionKey(UrlsTable.URL, TEXT))
.build();
}
diff --git a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/UrlsDao.java b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/UrlsDao.java
index 8600357401..b76116610e 100644
--- a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/UrlsDao.java
+++ b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/UrlsDao.java
@@ -19,10 +19,10 @@
package org.apache.james.mailrepository.cassandra;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.insertInto;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.selectFrom;
+import static com.datastax.oss.driver.api.querybuilder.relation.Relation.column;
import static org.apache.james.mailrepository.cassandra.UrlsTable.TABLE_NAME;
import static org.apache.james.mailrepository.cassandra.UrlsTable.URL;
@@ -33,9 +33,9 @@ import javax.inject.Inject;
import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
import org.apache.james.mailrepository.api.MailRepositoryUrl;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.cql.Row;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -47,7 +47,7 @@ public class UrlsDao {
private final PreparedStatement select;
@Inject
- public UrlsDao(Session session) {
+ public UrlsDao(CqlSession session) {
this.executor = new CassandraAsyncExecutor(session);
this.insert = prepareInsert(session);
@@ -55,20 +55,23 @@ public class UrlsDao {
this.select = prepareSelect(session);
}
- private PreparedStatement prepareSelect(Session session) {
- return session.prepare(select(URL)
- .from(TABLE_NAME)
- .where(eq(URL, bindMarker(URL))));
+ private PreparedStatement prepareSelect(CqlSession session) {
+ return session.prepare(selectFrom(TABLE_NAME)
+ .column(URL)
+ .where(column(URL).isEqualTo(bindMarker(URL)))
+ .build());
}
- private PreparedStatement prepareSelectAll(Session session) {
- return session.prepare(select(URL)
- .from(TABLE_NAME));
+ private PreparedStatement prepareSelectAll(CqlSession session) {
+ return session.prepare(selectFrom(TABLE_NAME)
+ .column(URL)
+ .build());
}
- private PreparedStatement prepareInsert(Session session) {
+ private PreparedStatement prepareInsert(CqlSession session) {
return session.prepare(insertInto(TABLE_NAME)
- .value(URL, bindMarker(URL)));
+ .value(URL, bindMarker(URL))
+ .build());
}
public Mono<Void> addUrl(MailRepositoryUrl url) {
diff --git a/server/mailrepository/mailrepository-cassandra/src/test/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryTest.java b/server/mailrepository/mailrepository-cassandra/src/test/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryTest.java
index 10a3c03e4a..d4ee40632b 100644
--- a/server/mailrepository/mailrepository-cassandra/src/test/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryTest.java
+++ b/server/mailrepository/mailrepository-cassandra/src/test/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryTest.java
@@ -19,7 +19,7 @@
package org.apache.james.mailrepository.cassandra;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.selectFrom;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
@@ -103,7 +103,7 @@ class CassandraMailRepositoryTest {
testee.remove(key1);
- assertThat(cassandra.getConf().execute(select().from(BlobTables.DefaultBucketBlobTable.TABLE_NAME)))
+ assertThat(cassandra.getConf().execute(selectFrom(BlobTables.DefaultBucketBlobTable.TABLE_NAME).all().build()))
.isEmpty();
}
}
diff --git a/server/mailrepository/mailrepository-cassandra/src/test/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryWithFakeImplementationsTest.java b/server/mailrepository/mailrepository-cassandra/src/test/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryWithFakeImplementationsTest.java
index 714da2dffd..2ae770cf51 100644
--- a/server/mailrepository/mailrepository-cassandra/src/test/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryWithFakeImplementationsTest.java
+++ b/server/mailrepository/mailrepository-cassandra/src/test/java/org/apache/james/mailrepository/cassandra/CassandraMailRepositoryWithFakeImplementationsTest.java
@@ -19,7 +19,7 @@
package org.apache.james.mailrepository.cassandra;
-import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.selectFrom;
import static org.apache.james.backends.cassandra.Scenario.Builder.fail;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -44,7 +44,7 @@ import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import com.datastax.driver.core.ResultSet;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
class CassandraMailRepositoryWithFakeImplementationsTest {
@RegisterExtension
@@ -81,7 +81,7 @@ class CassandraMailRepositoryWithFakeImplementationsTest {
cassandra.getConf()
.registerScenario(fail()
.forever()
- .whenQueryStartsWith("INSERT INTO blobParts (id,chunkNumber,data) VALUES (:id,:chunkNumber,:data);"));
+ .whenQueryStartsWith("INSERT INTO blobparts"));
MailImpl mail = MailImpl.builder()
.name("mymail")
@@ -109,7 +109,7 @@ class CassandraMailRepositoryWithFakeImplementationsTest {
cassandra.getConf()
.registerScenario(fail()
.forever()
- .whenQueryStartsWith("INSERT INTO mailRepositoryContent"));
+ .whenQueryStartsWith("INSERT INTO mailrepositorycontent"));
MailImpl mail = MailImpl.builder()
.name("mymail")
@@ -134,7 +134,7 @@ class CassandraMailRepositoryWithFakeImplementationsTest {
cassandra.getConf()
.registerScenario(fail()
.forever()
- .whenQueryStartsWith("INSERT INTO mailRepositoryContent"));
+ .whenQueryStartsWith("INSERT INTO mailrepositorycontent"));
MailImpl mail = MailImpl.builder()
.name("mymail")
@@ -151,8 +151,7 @@ class CassandraMailRepositoryWithFakeImplementationsTest {
assertThatThrownBy(() -> cassandraMailRepository.store(mail))
.isInstanceOf(RuntimeException.class);
- ResultSet resultSet = cassandra.getConf().execute(select()
- .from(BlobTables.DefaultBucketBlobTable.TABLE_NAME));
+ ResultSet resultSet = cassandra.getConf().execute(selectFrom(BlobTables.DefaultBucketBlobTable.TABLE_NAME).all().build());
assertThat(resultSet.all()).hasSize(2);
}
}
@@ -164,7 +163,7 @@ class CassandraMailRepositoryWithFakeImplementationsTest {
cassandra.getConf()
.registerScenario(fail()
.forever()
- .whenQueryStartsWith("INSERT INTO mailRepositoryKeys (name,mailKey)"));
+ .whenQueryStartsWith("INSERT INTO mailrepositorykeys"));
MailImpl mail = MailImpl.builder()
.name("mymail")
@@ -189,7 +188,7 @@ class CassandraMailRepositoryWithFakeImplementationsTest {
cassandra.getConf()
.registerScenario(fail()
.forever()
- .whenQueryStartsWith("INSERT INTO mailRepositoryKeys (name,mailKey)"));
+ .whenQueryStartsWith("INSERT INTO mailrepositorykeys"));
MailImpl mail = MailImpl.builder()
.name("mymail")
@@ -206,8 +205,7 @@ class CassandraMailRepositoryWithFakeImplementationsTest {
assertThatThrownBy(() -> cassandraMailRepository.store(mail))
.isInstanceOf(RuntimeException.class);
- ResultSet resultSet = cassandra.getConf().execute(select()
- .from(BlobTables.DefaultBucketBlobTable.TABLE_NAME));
+ ResultSet resultSet = cassandra.getConf().execute(selectFrom(BlobTables.DefaultBucketBlobTable.TABLE_NAME).all().build());
assertThat(resultSet.all()).hasSize(2);
}
@@ -216,7 +214,7 @@ class CassandraMailRepositoryWithFakeImplementationsTest {
cassandra.getConf()
.registerScenario(fail()
.forever()
- .whenQueryStartsWith("INSERT INTO mailRepositoryKeys (name,mailKey)"));
+ .whenQueryStartsWith("INSERT INTO mailrepositorykeys"));
MailImpl mail = MailImpl.builder()
.name("mymail")
@@ -233,8 +231,7 @@ class CassandraMailRepositoryWithFakeImplementationsTest {
assertThatThrownBy(() -> cassandraMailRepository.store(mail))
.isInstanceOf(RuntimeException.class);
- ResultSet resultSet = cassandra.getConf().execute(select()
- .from(MailRepositoryTableV2.CONTENT_TABLE_NAME));
+ ResultSet resultSet = cassandra.getConf().execute(selectFrom(MailRepositoryTableV2.CONTENT_TABLE_NAME).all().build());
assertThat(resultSet.all()).hasSize(1);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org