I got the following set-up: Producer actor -> Akka stream -> alpakka-amqp -> Consumer akka stream

When I restart either stream (which implies draining the stream, canceling upstream and completing downstream, and then creating a new stream with the same AmqpConnectionProvider and configs), the mock stops working with the new stream.

Using messageCount and consumerCount on the used channel shows the correct number of messages waiting to be read and the new consumer in there (for example, starts at one, goes to 0 when the stream is shutdown and back to 1 when it is recreated), but nothing is delivered to the stream.

Using the same set-up on a real RMQ server instead of the mock seems to be working. Any idea?

Hi Team ,

On application exception i'm Dead lettering the message .( channel.basicNack with requeue false)

When i try to consume the message from DLQ, the messages does not have the x-death headers .

Given a queue contains a message that is being nacked by the consumer, we need to break the consumption loop upon connection shutdown, otherwise the consumer will continue to process messages.

This behavior resembles the behavior of a real rabbitmq connection that cannot receive more messages after a shutdown.

Hello,

I'm using spring boot version 2.1.2.RELEASE, amqp-client:5.4.3 and rabbitmq-mock:1.0.9. When I try to mock the RabbitMQ Connection Factory like the showed example:

    @Bean
    public ConnectionFactory connectionFactory() {
        return new CachingConnectionFactory(MockConnectionFactoryFactory.build());
    }

I'm receiving error org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$DeclarationException: Failed to declare queue(s):[****(name of queue)]

I was using this library with spring boot version 2.0.6.RELEASE it was working perfect.

Can you please check it :)

For a spring boot test case i do send multiple messages to an exchange routing to an single queue in a test setup. While verifying the messages in the queue only the first message can be consumed. Any further rabbitTemplate.receive() calls for the remaining messages deliver null (with a sufficient timeout). The tests do work though if i immediately consume messages after sending before publishing the next one. Regardless of this behavior being intended... it might be useful to be able to queue multiple messages at once

The messages queue is accessed and mutated concurrently by both consumer and producer threads.

Quoting the PriorityQueue javadoc documentation:

Multiple threads should not access a PriorityQueue instance concurrently if any of the threads modifies the queue. Instead, use the thread-safe PriorityBlockingQueue class.

I'm actually facing this situation in my project where once in a while a message gets lost causing the tests to fail.

Thank you for this great open source project!

I didn't add a test for this as nothing functionally changed and is hard to test threading issues.

Let me know if you think that is acceptable or not.

Suffice to say, LinkedHashMap is not safe to use by multiple threads (PriorityBlockingQueue, the other structure used in MockQueue, is though)

Fix for #125.

When trying to use the MockConnectionFactory for integration tests for a Spring Boot application I noticed that metrics were not complete when using a transaction manager. Upon further investigation I noticed MockQueue stores the MockChannel used to create it. It subsequently uses this in deliverToConsumerIfPossible to register consumption.

This means that acknowledgements are not recorded in AbstractMetricsCollector.updateChannelStateAfterAckReject for single messages because channelState.unackedMessageDeliveryTags does not contain the consumption, instead it was recorded against the channel used to declare the queue.

I've provided a PR with two commits. The first provides a failing test, the second commit introduces a fix which stores the MockChannel in the ConsumerAndTag when basicConsume is called so it can be used when delivering to the consumer.

I came across this whilst creating a Spring Boot integration test since one channel is created by the RabbitAdmin and the other by the RabbitTransactionManager. I use the metrics to check for acks before testing the output of the application.

Issue summary:

Methods in MockChannel which override Channel interface methods do not declare the same exceptions. For example, basicAck does not declare an IOException:

Channel:

    void basicAck(long deliveryTag, boolean multiple) throws IOException;

MockChannel:

    @Override
    public void basicAck(long deliveryTag, boolean multiple) {
        getTransactionOrNode().basicAck(deliveryTag, multiple);
        metricsCollectorWrapper.basicAck(this, deliveryTag, multiple);
    }

Why it matters:

Without the declared exceptions, it is not possible to mock throwing of the checked exception. Example / use-case:

try(Channel channel = Mockito.spy(conn.createChannel())) {
  assertThat(channel).isInstanceOf(MockChannel.class);
  
  Mockito.doThrow(new IOException("ACK failed"))
    .when(channel)
    .basicAck(anyLong(), anyBoolean());
  
  // If the ACK fails (i.e. throws an IOException, my app will try 3 times.
  // This is the behaviour I want to test
  verify(channel, times(3)).basicAck(anyLong(), anyBoolean());
}

Due to the undeclared exception, this test results in an error: Checked exception is invalid for this method!

Is this a valid use case?

Is this a valid use-case, or am I using rabbitmq-mock incorrectly?

If its a valid issue, I'm happy to create a PR to fix it if that's OK?

We discovered that when creating two queue bindings on the same exchange which only differ in binding arguments that rabbit mock will not deliver messages to the queue for all of the possible bindings (ie, one of the bindings simply does not work despite a message being delivered to the exchange with correct arguments).

This is because the BindConfigurations are at some point compared and its equals/hashCode does not take the binding arguments into account.

This PR shows a potential solution which fixes our test cases.

The test examples send and receive messages from the same queue. In practice often an application receives messages from one queue and sends them to another. Are there plans to implement direct access to the MockQueue and its messages ? Use case: application only sends messages to queue1. In the test run the code and check that the messages in MockQueue which mocks queue1 contain the expected content. (Scala, for instance, has a special backdoor TestActorRef which gives direct access to the underlying actor and its mailbox.)

Bumps spring-rabbit from 2.4.7 to 3.0.0.

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

The MockQueue class has many unsynchronized LinkedHashMaps. Any access, even read access like calling size() that may occur on different threads must be synchronized or the JVM is free to cache these values for the current thread. This can cause for example a message that is being received by a queue almost simultaneously with a consumer being added to be not delivered (and delivery can stop altogether even if more message are sent). It can also cause ConcurrentModificationExceptions; here's an example that I've seen:

java.util.ConcurrentModificationException: null
at java.base/java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:756)
at java.base/java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.java:778)
at com.github.fridujo.rabbitmq.mock.MockQueue.doWithUnackedUntil(MockQueue.java:323)
at com.github.fridujo.rabbitmq.mock.MockQueue.basicAck(MockQueue.java:200)
at com.github.fridujo.rabbitmq.mock.MockNode.lambda$basicAck$0(MockNode.java:120)
at java.base/java.util.concurrent.ConcurrentHashMap$ValuesView.forEach(ConcurrentHashMap.java:4780)
at com.github.fridujo.rabbitmq.mock.MockNode.basicAck(MockNode.java:120)
at com.github.fridujo.rabbitmq.mock.MockChannel.basicAck(MockChannel.java:407)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$CachedChannelInvocationHandler.invoke(CachingConnectionFactory.java:1162)
at jdk.proxy2/jdk.proxy2.$Proxy181.basicAck(Unknown Source)
at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.commitIfNecessary(BlockingQueueConsumer.java:876)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1048)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:940)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:84)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1317)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1223)
at java.base/java.lang.Thread.run(Thread.java:833)

I've fixed these problems locally, but I've already have two Pull Requests open on this project that have unfortunately not had any response, and I'm not sure if it is worth my time to make a 3rd PR to get this fixed.

The project seems somewhat abandoned -- are you looking for help? I wouldn't mind integrating the fix above, and my other two PR's and cutting a new release.

If it is abandoned, would you have any objection in my forking this project and releasing it to Maven central under my own groupId?

Snyk has created this PR to fix one or more vulnerable packages in the `maven` dependencies of this project.

Changes included in this PR

• Changes to the following files to upgrade the vulnerable dependencies to a fixed version:
  • src/it/alpakka/pom.xml

Vulnerabilities that will be fixed

With an upgrade:

Severity | Priority Score (*) | Issue | Upgrade | Breaking Change | Exploit Maturity :-------------------------:|-------------------------|:-------------------------|:-------------------------|:-------------------------|:------------------------- | 704/1000
Why? Has a fix available, CVSS 9.8 | Man-in-the-Middle (MitM)
SNYK-JAVA-COMRABBITMQ-598771 | com.lightbend.akka:akka-stream-alpakka-amqp_2.13:
2.0.1 -> 4.0.0
| Yes | No Known Exploit

(*) Note that the real score may have changed since the PR was raised.

Check the changes in this PR to ensure they won't cause issues with your project.

Note: You are seeing this because you or someone else with access to this repository has authorized Snyk to open fix PRs.

For more information: 🧐 View latest project report

🛠 Adjust project settings

📚 Read more about Snyk's upgrade and patch logic

Learn how to fix vulnerabilities with free interactive lessons:

🦉 Learn about vulnerability in an interactive lesson of Snyk Learn.

I have this simple test. As you can see, calling rabbitTemplate.receive() did not remove message from the queue, although the same message was received every time I called this method. This issue does not happen if I hook to a real rabbitmq

@TestConfiguration
public class TestConfig {

    @Bean
    @Primary
    ConnectionFactory connectionFactory() {
        return new CachingConnectionFactory(new MockConnectionFactory());
    }

    @Bean
    @Primary
    public RabbitAdmin getTestRabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    @Primary
    public RabbitTemplate getTestRabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }
}


@SpringBootTest
@Import(TestConfig.class)
class SampleTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitAdmin rabbitAdmin;

    @Test
    void test() throws Exception {
         rabbitTemplate.send("test-exchange", "test-key", new Message("test".getBytes(StandardCharsets.UTF_8)));

        Thread.sleep(500);
        QueueInformation info0 = rabbitAdmin.getQueueInfo("test-queue");  // QueueInformation [name=test-queue, messageCount=1, consumerCount=0]

        Message message1 = rabbitTemplate.receive("test-queue", -1);
        Thread.sleep(500);
        QueueInformation info1 = rabbitAdmin.getQueueInfo("test-queue");  // QueueInformation [name=test-queue, messageCount=1, consumerCount=1]

        Message message2 = rabbitTemplate.receive("test-queue", -1);
        Thread.sleep(500);
        QueueInformation info2 = rabbitAdmin.getQueueInfo("test-queue");  // QueueInformation [name=test-queue, messageCount=1, consumerCount=1]
    }
}

This contribution derives from issue (feature request) https://github.com/fridujo/rabbitmq-mock/issues/281

This PR is not ready to be merged. There are many things to look out for and the code style is not compliant.

Also, I cannot seem to pass a test for plugin com.github.fridujo:rabbitmq-mock-alpakka-amqp-integration

[ERROR] Tests run: 2, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 11.138 s <<< FAILURE! - in com.github.fridujo.rabbitmq.mock.integration.alpakka.AmqpConnectorsTest
[ERROR] publishAndConsumeRpcWithoutAutoAck  Time elapsed: 10.053 s  <<< FAILURE!
java.lang.AssertionError: assertion failed: timeout (4999314094 nanoseconds) during expectMsgClass waiting for interface akka.stream.testkit.TestSubscriber$SubscriberEvent
        at com.github.fridujo.rabbitmq.mock.integration.alpakka.AmqpConnectorsTest.publishAndConsumeRpcWithoutAutoAck(AmqpConnectorsTest.java:121)

[INFO] 
[INFO] Results:
[INFO] 
[ERROR] Failures: 
[ERROR]   AmqpConnectorsTest.publishAndConsumeRpcWithoutAutoAck:121 assertion failed: timeout (4999314094 nanoseconds) during expectMsgClass waiting for interface akka.stream.testkit.TestSubscriber$SubscriberEvent
[INFO] 
[ERROR] Tests run: 2, Failures: 1, Errors: 0, Skipped: 0

As of RabbitMQ 3.9 there is a new feature called Streams.

I was testing a small setup that I have with Apache Camel and RabbitMQ Streams until I found a configuration that works. I then chose to swap the ConnectionFactory with the Mock version. Almost everything works as expected but I will leave the few details that this mocking tool might want to add.

Regarding declaration, there is a new queue type: stream (the mock still works but it will become important for the behavior under testing). There are many other details but those are policy driven. I don't know if it these details would be interesting for this mock tool. Binding works exactly the same as any classic queue and the exchanges.

As far as the declaration goes, Streams

• must have autoAck set to false,
• must store all it's messages in an addressable fashion
  • by index (for offsets)
  • by timestamp (for dates)

Regarding the message persistence, Streams are hardcoded to persist everything and consumers cannot cause messages to disappear from the Stream. RabbitMQ does have a mechanism to shorten the Stream itself by way of policies (max-age, etc). Again, I don't know if these features are supposed to come in handy for a mock tool.

The queued messages must be accessible by a few methods

• "first" starts at the first available message;
• "last" starts at last written chunk (chunking is the storage segmentation mechanism, which would have to be incorporated for this feature to make sense, which I don't know if it is overkill for this tool);
• "next" starts at the end and will consume the first new offset written to the stream, this is the default if the offset is not declared;
• numerical offset (long) is self explanatory (0 based indexing);
• timestamp (java.util.Date) will start consuming from the closest message timestamp;
• an interval string (e.g.: "1D 20h 15m 22s") will start consuming from the date resulting in the subtraction of interval from the current date.

Regarding consumers, they will have to define some form of Stream offset to start reading from (the option x-stream-offset or "next" if undeclared) which can be populated by the different kinds of values described above.

In addition, Stream consumers

• must have qos (a prefetch size) defined on the channel,
• must manually acknowledge the delivery,
• must somehow store the offset in a durable medium to survive a reboot (this feature is supposedly handled by RabbitMQ but the AMQP0.9.1 spec and clients do not make use of this),
• must be provided the current offset of the message in the message.properties.headers["x-stream-offset"] so that they may store the offset to pick up after a reboot.

The RabbitMQ team is already working on a RabbitMQ Java Stream client library but the amqp-0.9.1 client is supposed to be able to provide the low level features so implementers can choose their own method.

In summary, a healthy minimum set of features could be

• a new queue type: stream
  • Streams keep all messages stored
  • Streams keep a numerical index
  • Streams keep a timestamp index
• stream consumers
  • must not autoAck (consumer should fail to set up),
  • must define qos (prefetch) (consumer should fail to set up),
  • should provide the option "x-stream-offset" ("first", "next", offset, timestamp)
  • must be provided the current message offset in the headers under "x-stream-offset" of each message.

More Info at the source

• https://www.rabbitmq.com/streams.html