You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by bt...@apache.org on 2019/02/25 03:16:29 UTC
[james-project] 12/16: MAILBOX-380 Create
CassandraEventDeadLettersModule With DAO and Table
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 5967f8fed6e22de9bd7df6b88498c5777098cbc3
Author: datph <dp...@linagora.com>
AuthorDate: Fri Feb 22 09:41:46 2019 +0700
MAILBOX-380 Create CassandraEventDeadLettersModule With DAO and Table
---
mailbox/event/event-cassandra/pom.xml | 51 ++++++++
.../events/CassandraEventDeadLettersDAO.java | 131 +++++++++++++++++++++
.../events/CassandraEventDeadLettersModule.java | 41 +++++++
.../tables/CassandraEventDeadLettersTable.java | 29 +++++
.../events/CassandraEventDeadLettersDAOTest.java | 124 +++++++++++++++++++
mailbox/pom.xml | 1 +
6 files changed, 377 insertions(+)
diff --git a/mailbox/event/event-cassandra/pom.xml b/mailbox/event/event-cassandra/pom.xml
new file mode 100644
index 0000000..1bc6222
--- /dev/null
+++ b/mailbox/event/event-cassandra/pom.xml
@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>apache-james-mailbox</artifactId>
+ <groupId>org.apache.james</groupId>
+ <version>3.4.0-SNAPSHOT</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>apache-james-mailbox-event-cassandra</artifactId>
+ <name>Apache James :: Mailbox :: Event :: In Cassandra implementation</name>
+ <description>In Cassandra implementation for the eventDeadLetter API</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>apache-james-backends-cassandra</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>apache-james-mailbox-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>apache-james-mailbox-event-json</artifactId>
+ </dependency>
+ </dependencies>
+
+
+</project>
\ No newline at end of file
diff --git a/mailbox/event/event-cassandra/src/main/java/org/apache/james/mailbox/events/CassandraEventDeadLettersDAO.java b/mailbox/event/event-cassandra/src/main/java/org/apache/james/mailbox/events/CassandraEventDeadLettersDAO.java
new file mode 100644
index 0000000..0eab33b
--- /dev/null
+++ b/mailbox/event/event-cassandra/src/main/java/org/apache/james/mailbox/events/CassandraEventDeadLettersDAO.java
@@ -0,0 +1,131 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.mailbox.events;
+
+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 org.apache.james.mailbox.events.tables.CassandraEventDeadLettersTable.EVENT;
+import static org.apache.james.mailbox.events.tables.CassandraEventDeadLettersTable.EVENT_ID;
+import static org.apache.james.mailbox.events.tables.CassandraEventDeadLettersTable.GROUP;
+import static org.apache.james.mailbox.events.tables.CassandraEventDeadLettersTable.TABLE_NAME;
+
+import javax.inject.Inject;
+
+import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
+import org.apache.james.event.json.EventSerializer;
+
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class CassandraEventDeadLettersDAO {
+ private final CassandraAsyncExecutor executor;
+ private final EventSerializer eventSerializer;
+ private final PreparedStatement insertStatement;
+ private final PreparedStatement deleteStatement;
+ private final PreparedStatement selectAllGroupStatement;
+ private final PreparedStatement selectEventStatement;
+ private final PreparedStatement selectEventIdsWithGroupStatement;
+
+ @Inject
+ CassandraEventDeadLettersDAO(Session session, EventSerializer eventSerializer) {
+ this.executor = new CassandraAsyncExecutor(session);
+ this.eventSerializer = eventSerializer;
+ this.insertStatement = prepareInsertStatement(session);
+ this.deleteStatement = prepareDeleteStatement(session);
+ this.selectAllGroupStatement = prepareSelectAllGroupStatement(session);
+ this.selectEventStatement = prepareSelectEventStatement(session);
+ this.selectEventIdsWithGroupStatement = prepareSelectEventIdsWithGroupStatement(session);
+ }
+
+ private PreparedStatement prepareInsertStatement(Session session) {
+ return session.prepare(insertInto(TABLE_NAME)
+ .value(GROUP, bindMarker(GROUP))
+ .value(EVENT_ID, bindMarker(EVENT_ID))
+ .value(EVENT, bindMarker(EVENT)));
+ }
+
+ private PreparedStatement prepareDeleteStatement(Session session) {
+ return session.prepare(delete()
+ .from(TABLE_NAME)
+ .where(eq(GROUP, bindMarker(GROUP)))
+ .and(eq(EVENT_ID, bindMarker(EVENT_ID))));
+ }
+
+ private PreparedStatement prepareSelectAllGroupStatement(Session session) {
+ return session.prepare(select(GROUP)
+ .from(TABLE_NAME));
+ }
+
+ private PreparedStatement prepareSelectEventStatement(Session session) {
+ return session.prepare(select(EVENT)
+ .from(TABLE_NAME)
+ .where(eq(GROUP, bindMarker(GROUP)))
+ .and(eq(EVENT_ID, bindMarker(EVENT_ID))));
+ }
+
+ private PreparedStatement prepareSelectEventIdsWithGroupStatement(Session session) {
+ return session.prepare(select(EVENT_ID)
+ .from(TABLE_NAME)
+ .where(eq(GROUP, bindMarker(GROUP))));
+ }
+
+ Mono<Void> store(Group group, Event failedEvent) {
+ return executor.executeVoid(insertStatement.bind()
+ .setString(GROUP, group.asString())
+ .setUUID(EVENT_ID, failedEvent.getEventId().getId())
+ .setString(EVENT, eventSerializer.toJson(failedEvent)));
+ }
+
+ Mono<Void> removeEvent(Group group, Event.EventId failedEventId) {
+ return executor.executeVoid(deleteStatement.bind()
+ .setString(GROUP, group.asString())
+ .setUUID(EVENT_ID, failedEventId.getId()));
+ }
+
+ Mono<Event> retrieveFailedEvent(Group group, Event.EventId failedEventId) {
+ return executor.executeSingleRow(selectEventStatement.bind()
+ .setString(GROUP, group.asString())
+ .setUUID(EVENT_ID, failedEventId.getId()))
+ .map(row -> deserializeEvent(row.getString(EVENT)));
+ }
+
+ Flux<Event.EventId> retrieveEventIdsWithGroup(Group group) {
+ return executor.executeRows(selectEventIdsWithGroupStatement.bind()
+ .setString(GROUP, group.asString()))
+ .map(row -> Event.EventId.of(row.getUUID(EVENT_ID)));
+ }
+
+ Flux<Group> retrieveAllGroups() {
+ return executor.executeRows(selectAllGroupStatement.bind())
+ .map(Throwing.function(row -> Group.deserialize(row.getString(GROUP))))
+ .distinct();
+ }
+
+ private Event deserializeEvent(String serializedEvent) {
+ return eventSerializer.fromJson(serializedEvent).get();
+ }
+}
diff --git a/mailbox/event/event-cassandra/src/main/java/org/apache/james/mailbox/events/CassandraEventDeadLettersModule.java b/mailbox/event/event-cassandra/src/main/java/org/apache/james/mailbox/events/CassandraEventDeadLettersModule.java
new file mode 100644
index 0000000..f8b1e30
--- /dev/null
+++ b/mailbox/event/event-cassandra/src/main/java/org/apache/james/mailbox/events/CassandraEventDeadLettersModule.java
@@ -0,0 +1,41 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.mailbox.events;
+
+import org.apache.james.backends.cassandra.components.CassandraModule;
+import org.apache.james.backends.cassandra.utils.CassandraConstants;
+import org.apache.james.mailbox.events.tables.CassandraEventDeadLettersTable;
+
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.schemabuilder.SchemaBuilder;
+
+public interface CassandraEventDeadLettersModule {
+ CassandraModule MODULE = CassandraModule.builder()
+ .table(CassandraEventDeadLettersTable.TABLE_NAME)
+ .comment("Holds event dead letter")
+ .options(options -> options
+ .caching(SchemaBuilder.KeyCaching.ALL,
+ SchemaBuilder.rows(CassandraConstants.DEFAULT_CACHED_ROW_PER_PARTITION)))
+ .statement(statement -> statement
+ .addPartitionKey(CassandraEventDeadLettersTable.GROUP, DataType.text())
+ .addClusteringColumn(CassandraEventDeadLettersTable.EVENT_ID, DataType.uuid())
+ .addColumn(CassandraEventDeadLettersTable.EVENT, DataType.text()))
+ .build();
+}
diff --git a/mailbox/event/event-cassandra/src/main/java/org/apache/james/mailbox/events/tables/CassandraEventDeadLettersTable.java b/mailbox/event/event-cassandra/src/main/java/org/apache/james/mailbox/events/tables/CassandraEventDeadLettersTable.java
new file mode 100644
index 0000000..2418ffe
--- /dev/null
+++ b/mailbox/event/event-cassandra/src/main/java/org/apache/james/mailbox/events/tables/CassandraEventDeadLettersTable.java
@@ -0,0 +1,29 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.mailbox.events.tables;
+
+public interface CassandraEventDeadLettersTable {
+
+ String TABLE_NAME = "event_dead_letters";
+
+ String GROUP = "group";
+ String EVENT_ID = "eventId";
+ String EVENT = "event";
+}
diff --git a/mailbox/event/event-cassandra/src/test/java/org/apache/james/mailbox/events/CassandraEventDeadLettersDAOTest.java b/mailbox/event/event-cassandra/src/test/java/org/apache/james/mailbox/events/CassandraEventDeadLettersDAOTest.java
new file mode 100644
index 0000000..7bc66a7
--- /dev/null
+++ b/mailbox/event/event-cassandra/src/test/java/org/apache/james/mailbox/events/CassandraEventDeadLettersDAOTest.java
@@ -0,0 +1,124 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.mailbox.events;
+
+import static org.apache.james.mailbox.events.EventDeadLettersContract.EVENT_1;
+import static org.apache.james.mailbox.events.EventDeadLettersContract.EVENT_2;
+import static org.apache.james.mailbox.events.EventDeadLettersContract.EVENT_3;
+import static org.apache.james.mailbox.events.EventDeadLettersContract.EVENT_ID_1;
+import static org.apache.james.mailbox.events.EventDeadLettersContract.EVENT_ID_2;
+import static org.apache.james.mailbox.events.EventDeadLettersContract.EVENT_ID_3;
+import static org.apache.james.mailbox.events.EventDeadLettersContract.GROUP_A;
+import static org.apache.james.mailbox.events.EventDeadLettersContract.GROUP_B;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.james.backends.cassandra.CassandraCluster;
+import org.apache.james.backends.cassandra.CassandraClusterExtension;
+import org.apache.james.event.json.EventSerializer;
+import org.apache.james.mailbox.model.TestId;
+import org.apache.james.mailbox.model.TestMessageId;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class CassandraEventDeadLettersDAOTest {
+
+ @RegisterExtension
+ static CassandraClusterExtension cassandraClusterExtension = new CassandraClusterExtension(CassandraEventDeadLettersModule.MODULE);
+
+ private CassandraEventDeadLettersDAO cassandraEventDeadLettersDAO;
+
+ @BeforeEach
+ void setUp(CassandraCluster cassandraCluster) {
+ EventSerializer eventSerializer = new EventSerializer(new TestId.Factory(), new TestMessageId.Factory());
+ cassandraEventDeadLettersDAO = new CassandraEventDeadLettersDAO(cassandraCluster.getConf(), eventSerializer);
+ }
+
+ @Test
+ void removeEventShouldSucceededWhenRemoveStoredEvent() {
+ cassandraEventDeadLettersDAO.store(GROUP_A, EVENT_1).block();
+
+ cassandraEventDeadLettersDAO.removeEvent(GROUP_A, EVENT_ID_1).block();
+
+ assertThat(cassandraEventDeadLettersDAO
+ .retrieveAllGroups()
+ .collectList().block())
+ .isEmpty();
+ }
+
+ @Test
+ void retrieveFailedEventShouldReturnEmptyWhenDefault() {
+ assertThat(cassandraEventDeadLettersDAO
+ .retrieveFailedEvent(GROUP_A, EVENT_ID_1)
+ .blockOptional().isPresent())
+ .isFalse();
+ }
+
+ @Test
+ void retrieveFailedEventShouldReturnStoredEvent() {
+ cassandraEventDeadLettersDAO.store(GROUP_A, EVENT_1).block();
+ cassandraEventDeadLettersDAO.store(GROUP_B, EVENT_2).block();
+
+ assertThat(cassandraEventDeadLettersDAO
+ .retrieveFailedEvent(GROUP_B, EVENT_ID_2)
+ .blockOptional().get())
+ .isEqualTo(EVENT_2);
+ }
+
+ @Test
+ void retrieveEventIdsWithGroupShouldReturnEmptyWhenDefault() {
+ assertThat(cassandraEventDeadLettersDAO
+ .retrieveEventIdsWithGroup(GROUP_A)
+ .collectList().block())
+ .isEmpty();
+ }
+
+ @Test
+ void retrieveEventIdsWithGroupShouldReturnStoredEventId() {
+ cassandraEventDeadLettersDAO.store(GROUP_B, EVENT_1).block();
+ cassandraEventDeadLettersDAO.store(GROUP_B, EVENT_2).block();
+ cassandraEventDeadLettersDAO.store(GROUP_B, EVENT_3).block();
+
+ assertThat(cassandraEventDeadLettersDAO
+ .retrieveEventIdsWithGroup(GROUP_B)
+ .collectList().block())
+ .containsOnly(EVENT_ID_1, EVENT_ID_2, EVENT_ID_3);
+ }
+
+ @Test
+ void retrieveAllGroupsShouldReturnEmptyWhenDefault() {
+ assertThat(cassandraEventDeadLettersDAO
+ .retrieveAllGroups()
+ .collectList().block())
+ .isEmpty();
+ }
+
+ @Test
+ void retrieveAllGroupsShouldReturnStoredGroups() {
+ cassandraEventDeadLettersDAO.store(GROUP_A, EVENT_1).block();
+ cassandraEventDeadLettersDAO.store(GROUP_B, EVENT_2).block();
+ cassandraEventDeadLettersDAO.store(GROUP_B, EVENT_3).block();
+
+ assertThat(cassandraEventDeadLettersDAO
+ .retrieveAllGroups()
+ .collectList().block())
+ .containsOnly(GROUP_A, GROUP_B);
+ }
+}
diff --git a/mailbox/pom.xml b/mailbox/pom.xml
index 08102cc..55fa3ab 100644
--- a/mailbox/pom.xml
+++ b/mailbox/pom.xml
@@ -41,6 +41,7 @@
<module>cassandra</module>
<module>elasticsearch</module>
+ <module>event/event-cassandra</module>
<module>event/event-memory</module>
<module>event/event-rabbitmq</module>
<module>event/json</module>
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org