| /**************************************************************** |
| * 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.queue.rabbitmq; |
| |
| import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.selectFrom; |
| import static java.time.temporal.ChronoUnit.DAYS; |
| import static java.time.temporal.ChronoUnit.HOURS; |
| import static org.apache.james.backends.cassandra.Scenario.Builder.executeNormally; |
| import static org.apache.james.backends.cassandra.Scenario.Builder.fail; |
| import static org.apache.james.backends.cassandra.Scenario.Builder.returnEmpty; |
| import static org.apache.james.backends.cassandra.StatementRecorder.Selector.preparedStatementStartingWith; |
| import static org.apache.james.backends.rabbitmq.Constants.EMPTY_ROUTING_KEY; |
| import static org.apache.james.queue.api.Mails.defaultMail; |
| import static org.apache.james.queue.api.Mails.defaultMailNoRecipient; |
| import static org.apache.mailet.base.MailAddressFixture.RECIPIENT1; |
| import static org.apache.mailet.base.MailAddressFixture.SENDER; |
| import static org.assertj.core.api.Assertions.assertThat; |
| import static org.assertj.core.api.Assertions.assertThatCode; |
| import static org.awaitility.Awaitility.await; |
| import static org.awaitility.Durations.TEN_SECONDS; |
| import static org.mockito.ArgumentMatchers.any; |
| import static org.mockito.Mockito.never; |
| import static org.mockito.Mockito.verify; |
| |
| import java.nio.charset.StandardCharsets; |
| import java.time.Duration; |
| import java.time.Instant; |
| import java.time.temporal.ChronoUnit; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.concurrent.ConcurrentLinkedDeque; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.function.Function; |
| import java.util.stream.IntStream; |
| import java.util.stream.Stream; |
| |
| import org.apache.james.backends.cassandra.CassandraCluster; |
| import org.apache.james.backends.cassandra.CassandraClusterExtension; |
| import org.apache.james.backends.cassandra.StatementRecorder; |
| import org.apache.james.backends.cassandra.components.CassandraModule; |
| import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionModule; |
| import org.apache.james.backends.rabbitmq.RabbitMQExtension; |
| import org.apache.james.blob.api.BlobStore; |
| import org.apache.james.blob.api.HashBlobId; |
| import org.apache.james.blob.cassandra.BlobTables; |
| import org.apache.james.blob.cassandra.CassandraBlobModule; |
| import org.apache.james.blob.cassandra.CassandraBlobStoreFactory; |
| import org.apache.james.blob.mail.MimeMessageStore; |
| import org.apache.james.core.builder.MimeMessageBuilder; |
| import org.apache.james.eventsourcing.eventstore.cassandra.CassandraEventStoreModule; |
| import org.apache.james.lifecycle.api.LifecycleUtil; |
| import org.apache.james.metrics.api.Gauge; |
| import org.apache.james.metrics.tests.RecordingMetricFactory; |
| import org.apache.james.queue.api.MailQueue; |
| import org.apache.james.queue.api.MailQueueFactory; |
| import org.apache.james.queue.api.MailQueueMetricContract; |
| import org.apache.james.queue.api.MailQueueMetricExtension; |
| import org.apache.james.queue.api.ManageableMailQueue; |
| import org.apache.james.queue.api.ManageableMailQueueContract; |
| import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory; |
| import org.apache.james.queue.rabbitmq.view.RabbitMQMailQueueConfiguration; |
| import org.apache.james.queue.rabbitmq.view.api.MailQueueView; |
| import org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule; |
| import org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewTestFactory; |
| import org.apache.james.queue.rabbitmq.view.cassandra.EnqueuedMailsDAO; |
| import org.apache.james.queue.rabbitmq.view.cassandra.configuration.CassandraMailQueueViewConfiguration; |
| import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices; |
| import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice; |
| import org.apache.james.util.streams.Iterators; |
| import org.apache.james.utils.UpdatableTickingClock; |
| import org.apache.mailet.Mail; |
| import org.apache.mailet.base.test.FakeMail; |
| import org.assertj.core.api.SoftAssertions; |
| import org.junit.jupiter.api.AfterEach; |
| import org.junit.jupiter.api.BeforeEach; |
| import org.junit.jupiter.api.Disabled; |
| import org.junit.jupiter.api.Nested; |
| import org.junit.jupiter.api.RepeatedTest; |
| import org.junit.jupiter.api.Test; |
| import org.junit.jupiter.api.extension.RegisterExtension; |
| import org.mockito.ArgumentCaptor; |
| |
| import com.github.fge.lambdas.Throwing; |
| |
| import reactor.core.Disposable; |
| import reactor.core.publisher.Flux; |
| import reactor.core.publisher.Mono; |
| import reactor.core.scheduler.Schedulers; |
| import reactor.rabbitmq.BindingSpecification; |
| import reactor.rabbitmq.OutboundMessage; |
| import reactor.rabbitmq.Sender; |
| |
| class RabbitMQMailQueueTest { |
| private static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory(); |
| private static final int THREE_BUCKET_COUNT = 3; |
| private static final int UPDATE_BROWSE_START_PACE = 25; |
| private static final Duration ONE_HOUR_SLICE_WINDOW = Duration.ofHours(1); |
| private static final org.apache.james.queue.api.MailQueueName SPOOL = org.apache.james.queue.api.MailQueueName.of("spool"); |
| private static final Instant IN_SLICE_1 = Instant.now().minus(60, DAYS); |
| private static final Instant IN_SLICE_2 = IN_SLICE_1.plus(1, HOURS); |
| private static final Instant IN_SLICE_3 = IN_SLICE_1.plus(2, HOURS); |
| private static final Instant IN_SLICE_5 = IN_SLICE_1.plus(4, HOURS); |
| private static final Instant IN_SLICE_7 = IN_SLICE_1.plus(6, HOURS); |
| public static final ExecutorService EXECUTOR = Executors.newCachedThreadPool(); |
| |
| @RegisterExtension |
| static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(CassandraModule.aggregateModules( |
| CassandraBlobModule.MODULE, |
| CassandraMailQueueViewModule.MODULE, |
| CassandraEventStoreModule.MODULE(), |
| CassandraSchemaVersionModule.MODULE)); |
| |
| @RegisterExtension |
| static RabbitMQExtension rabbitMQExtension = RabbitMQExtension.singletonRabbitMQ() |
| .isolationPolicy(RabbitMQExtension.IsolationPolicy.WEAK); |
| |
| private RabbitMQMailQueueFactory mailQueueFactory; |
| private UpdatableTickingClock clock; |
| private RabbitMQMailQueue mailQueue; |
| private RabbitMQMailQueueManagement mqManagementApi; |
| |
| @AfterEach |
| void tearDown() { |
| mqManagementApi.deleteAllQueues(); |
| } |
| |
| @Nested |
| class MailQueueSizeMetricsEnabled implements ManageableMailQueueContract, MailQueueMetricContract { |
| @BeforeEach |
| void setup(CassandraCluster cassandra, |
| MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws Exception { |
| setUp(cassandra, |
| metricTestSystem, |
| RabbitMQMailQueueConfiguration.builder() |
| .sizeMetricsEnabled(true) |
| .build(), |
| CassandraBlobStoreFactory.forTesting(cassandra.getConf(), new RecordingMetricFactory()) |
| .passthrough(), |
| MailQueueFactory.prefetchCount(3)); |
| } |
| |
| @Override |
| public void enQueue(Mail mail) throws MailQueue.MailQueueException { |
| ManageableMailQueueContract.super.enQueue(mail); |
| clock.tick(); |
| } |
| |
| @Override |
| public RabbitMQMailQueue getMailQueue() { |
| return mailQueue; |
| } |
| |
| @Override |
| public ManageableMailQueue getManageableMailQueue() { |
| return mailQueue; |
| } |
| |
| @Test |
| void browseShouldReturnCurrentlyEnqueuedMailFromAllSlices() throws Exception { |
| ManageableMailQueue mailQueue = getManageableMailQueue(); |
| int emailCount = 5; |
| |
| clock.setInstant(IN_SLICE_1); |
| enqueueSomeMails(namePatternForSlice(1), emailCount); |
| |
| clock.setInstant(IN_SLICE_2); |
| enqueueSomeMails(namePatternForSlice(2), emailCount); |
| |
| clock.setInstant(IN_SLICE_3); |
| enqueueSomeMails(namePatternForSlice(3), emailCount); |
| |
| clock.setInstant(IN_SLICE_5); |
| enqueueSomeMails(namePatternForSlice(5), emailCount); |
| |
| clock.setInstant(IN_SLICE_7); |
| Stream<String> names = Iterators.toStream(mailQueue.browse()) |
| .map(ManageableMailQueue.MailQueueItemView::getMail) |
| .map(Mail::getName); |
| |
| assertThat(names).containsExactly( |
| "1-1", "1-2", "1-3", "1-4", "1-5", |
| "2-1", "2-2", "2-3", "2-4", "2-5", |
| "3-1", "3-2", "3-3", "3-4", "3-5", |
| "5-1", "5-2", "5-3", "5-4", "5-5"); |
| } |
| |
| @Test |
| void browseStartShouldBeUpdated(CassandraCluster cassandraCluster, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws Exception { |
| setUp(cassandraCluster, metricTestSystem, |
| RabbitMQMailQueueConfiguration.builder() |
| .sizeMetricsEnabled(true) |
| .build(), |
| CassandraBlobStoreFactory.forTesting(cassandraCluster.getConf(), new RecordingMetricFactory()) |
| .passthrough(), |
| MailQueueFactory.prefetchCount(1)); |
| |
| int emailCount = 250; |
| |
| StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE browsestart"); |
| StatementRecorder statementRecorder = cassandraCluster.getConf() |
| .recordStatements(selector); |
| |
| clock.setInstant(IN_SLICE_1); |
| enqueueSomeMails(namePatternForSlice(1), emailCount); |
| dequeueMails(emailCount); |
| |
| clock.setInstant(IN_SLICE_2); |
| enqueueSomeMails(namePatternForSlice(2), emailCount); |
| dequeueMails(emailCount); |
| |
| clock.setInstant(IN_SLICE_3); |
| enqueueSomeMails(namePatternForSlice(3), emailCount); |
| dequeueMails(emailCount); |
| |
| // The actual rate of update should actually be lower than the update probability. |
| assertThat(statementRecorder.listExecutedStatements(selector)) |
| .hasSizeBetween(2, 12); |
| } |
| |
| @Test |
| void contentStartShouldBeUpdated(CassandraCluster cassandraCluster, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws Exception { |
| setUp(cassandraCluster, metricTestSystem, |
| RabbitMQMailQueueConfiguration.builder() |
| .sizeMetricsEnabled(true) |
| .build(), |
| CassandraBlobStoreFactory.forTesting(cassandraCluster.getConf(), new RecordingMetricFactory()) |
| .passthrough(), |
| MailQueueFactory.prefetchCount(1)); |
| |
| int emailCount = 250; |
| |
| StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE contentstart"); |
| StatementRecorder statementRecorder = cassandraCluster.getConf().recordStatements(selector); |
| |
| clock.setInstant(IN_SLICE_1); |
| enqueueSomeMails(namePatternForSlice(1), emailCount); |
| dequeueMails(emailCount); |
| |
| clock.setInstant(IN_SLICE_2); |
| enqueueSomeMails(namePatternForSlice(2), emailCount); |
| dequeueMails(emailCount); |
| |
| clock.setInstant(IN_SLICE_3); |
| enqueueSomeMails(namePatternForSlice(3), emailCount); |
| dequeueMails(emailCount); |
| |
| // The actual rate of update should actually be lower than the update probability. |
| assertThat(statementRecorder.listExecutedStatements(selector)) |
| .hasSizeBetween(2, 12); |
| } |
| |
| @Test |
| void dequeueShouldDeleteBlobs(CassandraCluster cassandra) throws Exception { |
| String name1 = "myMail1"; |
| Flux<MailQueue.MailQueueItem> dequeueFlux = Flux.from(getMailQueue().deQueue()); |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| dequeueFlux.take(1) |
| .flatMap(mailQueueItem -> Mono.fromCallable(() -> { |
| mailQueueItem.done(MailQueue.MailQueueItem.CompletionStatus.SUCCESS); |
| return mailQueueItem; |
| }).subscribeOn(Schedulers.fromExecutor(EXECUTOR))).blockLast(Duration.ofSeconds(10)); |
| |
| assertThat(cassandra.getConf().execute(selectFrom(BlobTables.DefaultBucketBlobTable.TABLE_NAME) |
| .all().build())) |
| .isEmpty(); |
| } |
| |
| @Test |
| void clearShouldDeleteBlobs(CassandraCluster cassandra) throws Exception { |
| String name1 = "myMail1"; |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getManageableMailQueue().clear(); |
| |
| assertThat(cassandra.getConf().execute(selectFrom(BlobTables.DefaultBucketBlobTable.TABLE_NAME) |
| .all().build())) |
| .isEmpty(); |
| } |
| |
| @Test |
| void removeByNameShouldDeleteBlobs(CassandraCluster cassandra) throws Exception { |
| String name1 = "myMail1"; |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getManageableMailQueue().remove(ManageableMailQueue.Type.Name, name1); |
| |
| assertThat(cassandra.getConf().execute(selectFrom(BlobTables.DefaultBucketBlobTable.TABLE_NAME) |
| .all().build())) |
| .isEmpty(); |
| } |
| |
| @Test |
| void removeByRecipientShouldDeleteBlobs(CassandraCluster cassandra) throws Exception { |
| String name1 = "myMail1"; |
| getMailQueue().enQueue(defaultMailNoRecipient() |
| .name(name1) |
| .recipient(RECIPIENT1) |
| .build()); |
| |
| getManageableMailQueue().remove(ManageableMailQueue.Type.Recipient, RECIPIENT1.asString()); |
| |
| assertThat(cassandra.getConf().execute(selectFrom(BlobTables.DefaultBucketBlobTable.TABLE_NAME) |
| .all().build())) |
| .isEmpty(); |
| } |
| |
| @Test |
| void removeBySenderShouldDeleteBlobs(CassandraCluster cassandra) throws Exception { |
| String name1 = "myMail1"; |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .sender(SENDER) |
| .build()); |
| |
| getManageableMailQueue().remove(ManageableMailQueue.Type.Sender, SENDER.asString()); |
| |
| assertThat(cassandra.getConf().execute(selectFrom(BlobTables.DefaultBucketBlobTable.TABLE_NAME) |
| .all().build())) |
| .isEmpty(); |
| } |
| |
| @Test |
| void browseAndDequeueShouldCombineWellWhenDifferentSlices() throws Exception { |
| ManageableMailQueue mailQueue = getManageableMailQueue(); |
| int emailCount = 5; |
| |
| clock.setInstant(IN_SLICE_1); |
| enqueueSomeMails(namePatternForSlice(1), emailCount); |
| |
| clock.setInstant(IN_SLICE_2); |
| enqueueSomeMails(namePatternForSlice(2), emailCount); |
| |
| clock.setInstant(IN_SLICE_3); |
| enqueueSomeMails(namePatternForSlice(3), emailCount); |
| |
| clock.setInstant(IN_SLICE_5); |
| enqueueSomeMails(namePatternForSlice(5), emailCount); |
| |
| clock.setInstant(IN_SLICE_7); |
| dequeueMails(5); |
| dequeueMails(5); |
| dequeueMails(3); |
| |
| Stream<String> names = Iterators.toStream(mailQueue.browse()) |
| .map(ManageableMailQueue.MailQueueItemView::getMail) |
| .map(Mail::getName); |
| |
| assertThat(names) |
| .containsExactly("3-4", "3-5", "5-1", "5-2", "5-3", "5-4", "5-5"); |
| } |
| |
| @Test |
| void enqueuedEmailsShouldEventuallyBeCleaned() { |
| ManageableMailQueue mailQueue = getManageableMailQueue(); |
| int emailCount = 100; |
| |
| clock.setInstant(IN_SLICE_1); |
| enqueueSomeMails(namePatternForSlice(1), emailCount); |
| |
| clock.setInstant(IN_SLICE_2); |
| enqueueSomeMails(namePatternForSlice(2), emailCount); |
| |
| clock.setInstant(IN_SLICE_3); |
| enqueueSomeMails(namePatternForSlice(3), emailCount); |
| |
| clock.setInstant(IN_SLICE_5); |
| enqueueSomeMails(namePatternForSlice(5), emailCount); |
| |
| clock.setInstant(IN_SLICE_7); |
| dequeueMails(emailCount); |
| dequeueMails(emailCount); |
| dequeueMails(emailCount); |
| dequeueMails(emailCount); |
| |
| // ensure slice 1 was cleaned |
| EnqueuedMailsDAO mailsDAO = new EnqueuedMailsDAO(cassandraCluster.getCassandraCluster().getConf(), new HashBlobId.Factory()); |
| MailQueueName queueName = MailQueueName.fromString(mailQueue.getName().asString()); |
| Slice slice = Slice.of(currentSliceStartInstant(IN_SLICE_1)); |
| |
| SoftAssertions.assertSoftly(softly -> { |
| softly.assertThat(mailsDAO.selectEnqueuedMails(queueName, slice, BucketedSlices.BucketId.of(0)) |
| .collectList() |
| .block()) |
| .isEmpty(); |
| softly.assertThat(mailsDAO.selectEnqueuedMails(queueName, slice, BucketedSlices.BucketId.of(1)) |
| .collectList() |
| .block()) |
| .isEmpty(); |
| softly.assertThat(mailsDAO.selectEnqueuedMails(queueName, slice, BucketedSlices.BucketId.of(2)) |
| .collectList() |
| .block()) |
| .isEmpty(); |
| }); |
| } |
| |
| private Instant currentSliceStartInstant(Instant instant) { |
| long sliceSize = ONE_HOUR_SLICE_WINDOW.getSeconds(); |
| long sliceId = instant.getEpochSecond() / sliceSize; |
| return Instant.ofEpochSecond(sliceId * sliceSize); |
| } |
| |
| private Function<Integer, String> namePatternForSlice(int sliceId) { |
| return i -> sliceId + "-" + i; |
| } |
| |
| @Test |
| void mailQueueShouldBeInitializedWhenCreating(CassandraCluster cassandra) { |
| org.apache.james.queue.api.MailQueueName name = org.apache.james.queue.api.MailQueueName.of("myQueue"); |
| mailQueueFactory.createQueue(name); |
| |
| boolean initialized = CassandraMailQueueViewTestFactory.isInitialized(cassandra.getConf(), MailQueueName.fromString(name.asString())); |
| assertThat(initialized).isTrue(); |
| } |
| |
| @Test |
| void enQueueShouldNotThrowOnMailNameWithNegativeHash() { |
| String negativehashedString = "this sting will have a negative hash"; //hash value: -1256871313 |
| |
| assertThatCode(() -> getMailQueue().enQueue(defaultMail().name(negativehashedString).build())) |
| .doesNotThrowAnyException(); |
| } |
| |
| @Disabled("JAMES-2614 RabbitMQMailQueueTest::concurrentEnqueueDequeueShouldNotFail is unstable." + |
| "The related test is disabled, and need to be re-enabled after investigation and a fix.") |
| @Test |
| @Override |
| public void concurrentEnqueueDequeueShouldNotFail() { |
| |
| } |
| |
| @Test |
| void dequeueShouldWorkAfterNetworkOutage() throws Exception { |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| |
| List<MailQueue.MailQueueItem> receivedItem = new ArrayList<>(); |
| Flux.from(getMailQueue().deQueue()) |
| .doOnNext(receivedItem::add) |
| .subscribe(); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| rabbitMQExtension.getRabbitMQ().pause(); |
| Thread.sleep(2000); |
| |
| try { |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| } catch (Exception e) { |
| // Expected |
| } finally { |
| rabbitMQExtension.getRabbitMQ().unpause(); |
| Thread.sleep(100); |
| } |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| await().atMost(Duration.ofSeconds(10)) |
| .untilAsserted(() -> assertThat(receivedItem) |
| .extracting(item -> item.getMail().getName()) |
| .contains(name1, name3)); |
| } |
| |
| @Test |
| void enqueuedEmailsShouldNotBeLostDuringRabbitMQOutages() throws Exception { |
| String name = "myMail"; |
| |
| |
| try { |
| getMailQueue().enQueue(defaultMail() |
| .name(name) |
| .build()); |
| } catch (Exception e) { |
| // Ignore |
| } |
| rabbitMQExtension.managementAPI().purgeQueue(rabbitMQExtension.getRabbitMQ().getConfiguration().getVhost().orElse("/"), |
| "JamesMailQueue-workqueue-spool"); |
| |
| getMailQueue().republishNotProcessedMails(clock.instant().plus(30, ChronoUnit.MINUTES)).blockLast(); |
| |
| Flux<MailQueue.MailQueueItem> dequeueFlux = Flux.from(getMailQueue().deQueue()); |
| |
| List<MailQueue.MailQueueItem> items = dequeueFlux.take(1) |
| .collectList().block(Duration.ofSeconds(10)); |
| |
| assertThat(items) |
| .extracting(item -> item.getMail().getName()) |
| .containsOnly(name); |
| } |
| |
| @Test |
| void messagesShouldSurviveRabbitMQRestart() throws Exception { |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| Flux<MailQueue.MailQueueItem> dequeueFlux = Flux.from(getMailQueue().deQueue()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| rabbitMQExtension.getRabbitMQ().restart(); |
| |
| List<MailQueue.MailQueueItem> items = dequeueFlux.take(3).collectList().block(Duration.ofSeconds(10)); |
| |
| assertThat(items) |
| .extracting(item -> item.getMail().getName()) |
| .containsExactly(name1, name2, name3); |
| } |
| |
| @Test |
| void messagesShouldBeProcessedAfterNotPublishedMailsHaveBeenReprocessed() throws Exception { |
| clock.setInstant(Instant.now().minus(Duration.ofHours(2))); |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| Flux<MailQueue.MailQueueItem> dequeueFlux = Flux.from(getMailQueue().deQueue()); |
| |
| // Avoid early processing and prefetching |
| Sender sender = rabbitMQExtension.getSender(); |
| |
| suspendDequeuing(sender); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| resumeDequeuing(sender); |
| assertThat(getMailQueue() |
| .republishNotProcessedMails(Instant.now().minus(Duration.ofHours(1))) |
| .collectList() |
| .block()) |
| .containsExactlyInAnyOrder(name1, name2, name3); |
| |
| List<MailQueue.MailQueueItem> items = dequeueFlux.take(Duration.ofSeconds(10)).collectList().block(); |
| |
| assertThat(items) |
| .extracting(item -> item.getMail().getName()) |
| .containsExactlyInAnyOrder(name1, name2, name3); |
| } |
| |
| @Test |
| void onlyOldMessagesShouldBeProcessedAfterNotPublishedMailsHaveBeenReprocessed() throws Exception { |
| clock.setInstant(Instant.now().minus(Duration.ofHours(2))); |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| Flux<MailQueue.MailQueueItem> dequeueFlux = Flux.from(getMailQueue().deQueue()); |
| |
| // Avoid early processing and prefetching |
| Sender sender = rabbitMQExtension.getSender(); |
| |
| suspendDequeuing(sender); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| |
| clock.setInstant(Instant.now()); |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| resumeDequeuing(sender); |
| assertThat(getMailQueue() |
| .republishNotProcessedMails(Instant.now().minus(Duration.ofHours(1))) |
| .collectList() |
| .block()) |
| .containsExactlyInAnyOrder(name1, name2); |
| |
| List<MailQueue.MailQueueItem> items = dequeueFlux.take(Duration.ofSeconds(10)).collectList().block(); |
| |
| assertThat(items) |
| .extracting(item -> item.getMail().getName()) |
| .containsExactlyInAnyOrder(name1, name2); |
| } |
| |
| @Test |
| void messagesShouldBeProcessedAfterTwoMailsReprocessing() throws Exception { |
| clock.setInstant(Instant.now().minus(Duration.ofHours(2))); |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| Flux<MailQueue.MailQueueItem> dequeueFlux = Flux.from(getMailQueue().deQueue()); |
| |
| // Avoid early processing and prefetching |
| Sender sender = rabbitMQExtension.getSender(); |
| |
| suspendDequeuing(sender); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| assertThat(getMailQueue() |
| .republishNotProcessedMails(Instant.now().minus(Duration.ofHours(1))) |
| .collectList() |
| .block()) |
| .containsExactlyInAnyOrder(name1, name2, name3); |
| resumeDequeuing(sender); |
| assertThat(getMailQueue() |
| .republishNotProcessedMails(Instant.now().minus(Duration.ofHours(1))) |
| .collectList() |
| .block()) |
| .containsExactlyInAnyOrder(name1, name2, name3); |
| |
| List<MailQueue.MailQueueItem> items = dequeueFlux.take(Duration.ofSeconds(10)).collectList().block(); |
| |
| assertThat(items) |
| .extracting(item -> item.getMail().getName()) |
| .containsExactlyInAnyOrder(name1, name2, name3); |
| } |
| |
| @Test |
| void messagesShouldBeProcessedAfterNotPublishedMailsHaveBeenReprocessedAndNewMessagesShouldNotBeLost() throws Exception { |
| clock.setInstant(Instant.now().minus(Duration.ofHours(2))); |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| Flux<MailQueue.MailQueueItem> dequeueFlux = Flux.from(getMailQueue().deQueue()); |
| |
| // Avoid early processing and prefetching |
| Sender sender = rabbitMQExtension.getSender(); |
| |
| suspendDequeuing(sender); |
| //mail send when rabbit down |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| resumeDequeuing(sender); |
| |
| //mail send when rabbit is up again and before rebuild |
| clock.setInstant(Instant.now()); |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| Flux.merge( |
| Mono.fromCallable(() -> { |
| //mail send concurently with rebuild |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| return true; |
| |
| }).subscribeOn(Schedulers.fromExecutor(EXECUTOR)), |
| Mono.fromRunnable(() -> |
| assertThat(getMailQueue() |
| .republishNotProcessedMails(Instant.now().minus(Duration.ofHours(1))) |
| .collectList() |
| .block()) |
| .containsOnly(name1)).subscribeOn(Schedulers.fromExecutor(EXECUTOR))) |
| .then() |
| .block(Duration.ofSeconds(10)); |
| |
| List<MailQueue.MailQueueItem> items = dequeueFlux.take(Duration.ofSeconds(10)).collectList().block(); |
| |
| assertThat(items) |
| .extracting(item -> item.getMail().getName()) |
| .containsExactlyInAnyOrder(name1, name2, name3); |
| } |
| |
| private void enqueueSomeMails(Function<Integer, String> namePattern, int emailCount) { |
| IntStream.rangeClosed(1, emailCount) |
| .forEach(Throwing.intConsumer(i -> { |
| FakeMail mail = defaultMail() |
| .name(namePattern.apply(i)) |
| .build(); |
| enQueue(mail); |
| LifecycleUtil.dispose(mail); |
| })); |
| } |
| |
| private void dequeueMails(int times) { |
| AtomicInteger counter = new AtomicInteger(0); |
| |
| Disposable disposable = Flux.from(getManageableMailQueue() |
| .deQueue()) |
| .concatMap(mailQueueItem -> Mono.fromCallable(() -> { |
| if (counter.getAndIncrement() < times) { |
| mailQueueItem.done(MailQueue.MailQueueItem.CompletionStatus.SUCCESS); |
| return mailQueueItem; |
| } else { |
| mailQueueItem.done(MailQueue.MailQueueItem.CompletionStatus.RETRY); |
| return null; |
| } |
| }).subscribeOn(Schedulers.fromExecutor(EXECUTOR))) |
| .subscribe(); |
| |
| try { |
| await() |
| .atMost(Duration.ofMinutes(10)) |
| .untilAsserted(() -> assertThat(counter.get()).isGreaterThanOrEqualTo(times)); |
| } finally { |
| disposable.dispose(); |
| } |
| } |
| |
| @Test |
| void dequeueShouldRetryLoadingErrors(CassandraCluster cassandra) throws Exception { |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| cassandra.getConf().registerScenario(fail() |
| .times(1) |
| .whenQueryStartsWith("SELECT * FROM blobs WHERE id=:id;")); |
| |
| List<MailQueue.MailQueueItem> items = Flux.from(getMailQueue().deQueue()) |
| .take(3) |
| .collectList() |
| .block(Duration.ofSeconds(10)); |
| |
| assertThat(items) |
| .extracting(item -> item.getMail().getName()) |
| .containsOnly(name1, name2, name3); |
| } |
| |
| @Test |
| void dequeueShouldNotRetryWhenBlobIsMissing(CassandraCluster cassandra) throws Exception { |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| cassandra.getConf().registerScenario(returnEmpty() |
| .forever() |
| .whenQueryStartsWith("SELECT * FROM blobs WHERE id=:id")); |
| |
| ConcurrentLinkedDeque<String> dequeuedNames = new ConcurrentLinkedDeque<>(); |
| Flux.from(getMailQueue().deQueue()) |
| .take(3) |
| .doOnNext(item -> dequeuedNames.add(item.getMail().getName())) |
| .flatMap(item -> Mono.fromRunnable(Throwing.runnable(() -> item.done(MailQueue.MailQueueItem.CompletionStatus.SUCCESS))) |
| .subscribeOn(Schedulers.fromExecutor(EXECUTOR))) |
| .subscribeOn(Schedulers.fromExecutor(EXECUTOR)) |
| .subscribe(); |
| |
| // One second should be enough to attempt dequeues while we fail to load blobs |
| Thread.sleep(1000); |
| |
| // Restore normal behaviour |
| cassandra.getConf().registerScenario(executeNormally() |
| .forever() |
| .whenQueryStartsWith("SELECT * FROM blobs WHERE id=:id")); |
| |
| // Let one second to check if the queue is empty |
| Thread.sleep(1000); |
| |
| // We expect content missing blob references to be purged from the queue |
| assertThat(dequeuedNames).isEmpty(); |
| } |
| |
| @Test |
| void dequeueShouldNotAbortProcessingUponSerializationIssuesErrors() throws Exception { |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| |
| String emptyRoutingKey = ""; |
| rabbitMQExtension.getSender() |
| .send(Mono.just(new OutboundMessage("JamesMailQueue-exchange-spool", |
| emptyRoutingKey, |
| "BAD_PAYLOAD!".getBytes(StandardCharsets.UTF_8)))) |
| .block(); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| ConcurrentLinkedDeque<String> dequeuedMailNames = new ConcurrentLinkedDeque<>(); |
| |
| Flux.from(getMailQueue().deQueue()) |
| .doOnNext(item -> dequeuedMailNames.add(item.getMail().getName())) |
| .flatMap(item -> Mono.fromRunnable(Throwing.runnable(() -> item.done(MailQueue.MailQueueItem.CompletionStatus.SUCCESS))) |
| .subscribeOn(Schedulers.fromExecutor(EXECUTOR))) |
| .subscribe(); |
| |
| await().atMost(TEN_SECONDS) |
| .untilAsserted(() -> assertThat(dequeuedMailNames) |
| .containsExactly(name1, name2, name3)); |
| } |
| |
| @Test |
| void manyInvalidMessagesShouldNotAbortProcessing() throws Exception { |
| String name1 = "myMail1"; |
| String name2 = "myMail2"; |
| String name3 = "myMail3"; |
| |
| String emptyRoutingKey = ""; |
| |
| IntStream.range(0, 100) |
| .forEach(i -> rabbitMQExtension.getSender() |
| .send(Mono.just(new OutboundMessage("JamesMailQueue-exchange-spool", |
| emptyRoutingKey, |
| ("BAD_PAYLOAD " + i).getBytes(StandardCharsets.UTF_8)))) |
| .block()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name2) |
| .build()); |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name3) |
| .build()); |
| |
| ConcurrentLinkedDeque<String> dequeuedMailNames = new ConcurrentLinkedDeque<>(); |
| |
| Flux.from(getMailQueue().deQueue()) |
| .doOnNext(item -> dequeuedMailNames.add(item.getMail().getName())) |
| .flatMap(item -> Mono.fromRunnable(Throwing.runnable(() -> item.done(MailQueue.MailQueueItem.CompletionStatus.SUCCESS))) |
| .subscribeOn(Schedulers.fromExecutor(EXECUTOR))) |
| .subscribe(); |
| |
| await().atMost(TEN_SECONDS) |
| .untilAsserted(() -> assertThat(dequeuedMailNames) |
| .containsExactly(name1, name2, name3)); |
| } |
| |
| @Test |
| void invalidMessagesShouldBeDeadLettered() { |
| String emptyRoutingKey = ""; |
| rabbitMQExtension.getSender() |
| .send(Mono.just(new OutboundMessage("JamesMailQueue-exchange-spool", |
| emptyRoutingKey, |
| "BAD_PAYLOAD!".getBytes(StandardCharsets.UTF_8)))) |
| .block(); |
| |
| AtomicInteger deadLetteredCount = new AtomicInteger(); |
| rabbitMQExtension.getRabbitChannelPool() |
| .createReceiver() |
| .consumeAutoAck("JamesMailQueue-dead-letter-queue-spool") |
| .doOnNext(next -> deadLetteredCount.incrementAndGet()) |
| .subscribeOn(Schedulers.fromExecutor(EXECUTOR)) |
| .subscribe(); |
| |
| Flux.from(getMailQueue().deQueue()) |
| .doOnNext(Throwing.consumer(item -> item.done(MailQueue.MailQueueItem.CompletionStatus.SUCCESS))) |
| .subscribeOn(Schedulers.fromExecutor(EXECUTOR)) |
| .subscribe(); |
| |
| |
| await().atMost(TEN_SECONDS) |
| .untilAsserted(() -> assertThat(deadLetteredCount.get()).isEqualTo(1)); |
| } |
| |
| @Test |
| void rejectedMessagesShouldBeDeadLettered() throws Exception { |
| String name1 = "myMail1"; |
| |
| getMailQueue().enQueue(defaultMail() |
| .name(name1) |
| .build()); |
| |
| getMailQueue().deQueue() |
| .flatMap(item -> Mono.fromRunnable(Throwing.runnable(() -> item.done(MailQueue.MailQueueItem.CompletionStatus.REJECT))) |
| .subscribeOn(Schedulers.boundedElastic()) |
| .thenReturn(item)) |
| .blockFirst(); |
| |
| AtomicInteger deadLetteredCount = new AtomicInteger(); |
| rabbitMQExtension.getRabbitChannelPool() |
| .createReceiver() |
| .consumeAutoAck("JamesMailQueue-dead-letter-queue-spool") |
| .doOnNext(next -> deadLetteredCount.incrementAndGet()) |
| .subscribeOn(Schedulers.fromExecutor(EXECUTOR)) |
| .subscribe(); |
| |
| await().atMost(TEN_SECONDS) |
| .untilAsserted(() -> assertThat(deadLetteredCount.get()).isEqualTo(1)); |
| } |
| |
| private void resumeDequeuing(Sender sender) { |
| sender.bindQueue(getMailQueueBindingSpecification()).block(); |
| } |
| |
| private void suspendDequeuing(Sender sender) { |
| sender.unbindQueue(getMailQueueBindingSpecification()).block(); |
| } |
| |
| private BindingSpecification getMailQueueBindingSpecification() { |
| MailQueueName mailQueueName = MailQueueName.fromString(getMailQueue().getName().asString()); |
| return BindingSpecification.binding() |
| .exchange(mailQueueName.toRabbitExchangeName().asString()) |
| .queue(mailQueueName.toWorkQueueName().asString()) |
| .routingKey(EMPTY_ROUTING_KEY); |
| } |
| } |
| |
| @Nested |
| class MailQueueSizeMetricsDisabled { |
| @RegisterExtension |
| MailQueueMetricExtension mailQueueMetricExtension = new MailQueueMetricExtension(); |
| |
| @BeforeEach |
| void setup(CassandraCluster cassandra, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws Exception { |
| setUp(cassandra, |
| metricTestSystem, |
| RabbitMQMailQueueConfiguration.builder() |
| .sizeMetricsEnabled(false) |
| .build(), |
| CassandraBlobStoreFactory.forTesting(cassandra.getConf(), new RecordingMetricFactory()) |
| .passthrough(), |
| MailQueueFactory.prefetchCount(3)); |
| } |
| |
| @Test |
| void constructorShouldNotRegisterGetQueueSizeGaugeWhenSizeMetricsDisabled(MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) { |
| ArgumentCaptor<Gauge<?>> gaugeCaptor = ArgumentCaptor.forClass(Gauge.class); |
| verify(metricTestSystem.getSpyGaugeRegistry(), never()).register(any(), gaugeCaptor.capture()); |
| } |
| } |
| |
| @Nested |
| class DeDuplicationTest { |
| @RegisterExtension |
| MailQueueMetricExtension mailQueueMetricExtension = new MailQueueMetricExtension(); |
| |
| @BeforeEach |
| void setup(CassandraCluster cassandra, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws Exception { |
| setUp(cassandra, |
| metricTestSystem, |
| RabbitMQMailQueueConfiguration.builder() |
| .sizeMetricsEnabled(true) |
| .build(), |
| CassandraBlobStoreFactory.forTesting(cassandra.getConf(), new RecordingMetricFactory()) |
| .deduplication(), |
| MailQueueFactory.prefetchCount(3)); |
| } |
| |
| @Test |
| void dequeueShouldStillRetrieveAllBlobsWhenIdenticalContentAndDeduplication() throws Exception { |
| String identicalContent = "identical content"; |
| String identicalSubject = "identical subject"; |
| |
| mailQueue.enQueue(defaultMail() |
| .name("myMail1") |
| .mimeMessage(MimeMessageBuilder.mimeMessageBuilder() |
| .setSubject(identicalSubject) |
| .setText(identicalContent)) |
| .build()); |
| mailQueue.enQueue(defaultMail() |
| .name("myMail2") |
| .mimeMessage(MimeMessageBuilder.mimeMessageBuilder() |
| .setSubject(identicalSubject) |
| .setText(identicalContent)) |
| .build()); |
| |
| Flux.from(mailQueue.deQueue()) |
| .take(2) |
| .concatMap(mailQueueItem -> |
| Mono.fromCallable(() -> { |
| assertThat(mailQueueItem.getMail().getMessage().getContent()).isEqualTo(identicalContent); |
| mailQueueItem.done(MailQueue.MailQueueItem.CompletionStatus.SUCCESS); |
| return mailQueueItem; |
| }).subscribeOn(Schedulers.fromExecutor(EXECUTOR))) |
| .collectList() |
| .block(Duration.ofSeconds(10)); |
| } |
| } |
| |
| private void setUp(CassandraCluster cassandra, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem, RabbitMQMailQueueConfiguration configuration, |
| BlobStore blobStore, MailQueueFactory.PrefetchCount prefetchCount) throws Exception { |
| MimeMessageStore.Factory mimeMessageStoreFactory = MimeMessageStore.factory(blobStore); |
| clock = new UpdatableTickingClock(IN_SLICE_1); |
| |
| MailQueueView.Factory mailQueueViewFactory = CassandraMailQueueViewTestFactory.factory( |
| clock, |
| cassandra.getConf(), |
| CassandraMailQueueViewConfiguration.builder() |
| .bucketCount(THREE_BUCKET_COUNT) |
| .updateBrowseStartPace(UPDATE_BROWSE_START_PACE) |
| .sliceWindow(ONE_HOUR_SLICE_WINDOW) |
| .build(), |
| mimeMessageStoreFactory |
| ); |
| |
| RabbitMQMailQueueFactory.PrivateFactory factory = new RabbitMQMailQueueFactory.PrivateFactory( |
| metricTestSystem.getMetricFactory(), |
| metricTestSystem.getSpyGaugeRegistry(), |
| rabbitMQExtension.getSender(), rabbitMQExtension.getReceiverProvider(), |
| mimeMessageStoreFactory, |
| BLOB_ID_FACTORY, |
| mailQueueViewFactory, |
| clock, |
| new RawMailQueueItemDecoratorFactory(), |
| configuration); |
| mqManagementApi = new RabbitMQMailQueueManagement(rabbitMQExtension.managementAPI()); |
| mailQueueFactory = new RabbitMQMailQueueFactory(rabbitMQExtension.getSender(), mqManagementApi, factory, rabbitMQExtension.getRabbitMQ().getConfiguration()); |
| mailQueue = mailQueueFactory.createQueue(SPOOL, prefetchCount); |
| } |
| } |
