[azure-service-bus] Add possibility to receive messages from Azur… #4876
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
bf22998 to
3ebce8c
Compare
vkepin
commented
Mar 12, 2024
...ure-service-bus/src/main/java/org/vividus/azure/servicebus/model/ServiceBusClientEntity.java
Outdated
Show resolved
Hide resolved
valfirst
requested changes
Mar 13, 2024
| @@ -42,4 +60,80 @@ public void sendMessageToServiceBus(ChannelType type, String name, String namesp | |||
| { | |||
| serviceBusService.send(type, name, namespaceName, payload); | |||
| } | |||
|
|
|||
| @When("I send message to service bus by client `$clientKey` with payload:`$payload`") | |||
There was a problem hiding this comment.
Suggested change
| @When("I send message to service bus by client `$clientKey` with payload:`$payload`") | |
| @When("I send message to `$serviceBusKey` service bus with payload:`$payload`") |
There was a problem hiding this comment.
and I guess we need to align step sending messages as well
| serviceBusService.send(clientKey, payload); | ||
| } | ||
|
|
8000
||
| @When("I $action consuming messages from service bus by client `$clientKey`") |
There was a problem hiding this comment.
to have better autocompletion it's recommended to have 2 separate steps: start, stop
| serviceBusService.send(clientKey, payload); | ||
| } | ||
|
|
||
| @When("I $action consuming messages from service bus by client `$clientKey`") |
There was a problem hiding this comment.
service bus by client `$clientKey` -> `$serviceBusKey` service bus (everywhere)
| public void send(String clientKey, String message) | ||
| { | ||
| ServiceBusClientEntity serviceBusClientEntity = clientConfigs.get(clientKey, | ||
| "No info provided for Service Bus client `%s`", clientKey); |
There was a problem hiding this comment.
Suggested change
| "No info provided for Service Bus client `%s`", clientKey); | |
| "No Service Bus client with key `%s` is configured", clientKey); |
| BlockingQueue<ServiceBusReceivedMessage> messages = new LinkedBlockingDeque<>(); | ||
| testContext.get(ServiceBusReceivedMessage.class, HashMap::new).put(clientKey, messages); | ||
| ServiceBusClientEntity serviceBusClientEntity = clientConfigs.get(clientKey, | ||
| "No info provided for Service Bus client `%s`", clientKey); |
There was a problem hiding this comment.
make sure to avoid duplications
| ServiceBusProcessorClient client = processorBuilder.buildProcessorClient(); | ||
|
|
||
| clients.put(clientKey, client); | ||
| LOGGER.info("START the client processor '{}' to consume messages from the {} '{}' in namespace {}", clientKey, |
There was a problem hiding this comment.
Suggested change
| LOGGER.info("START the client processor '{}' to consume messages from the {} '{}' in namespace {}", clientKey, | |
| LOGGER.info("'{}' Azure Service Bus Consumer of messages from the {} '{}' in namespace {} is started", clientKey, |
There was a problem hiding this comment.
in similar way for stop message
| } | ||
| else | ||
| { | ||
| throw new IllegalArgumentException("There are no running Service Bus clients associated with the key: " |
There was a problem hiding this comment.
Suggested change
| throw new IllegalArgumentException("There are no running Service Bus clients associated with the key: " | |
| throw new IllegalArgumentException("There are no running Azure Service Bus consumers associated with the key: " |
5852d6d to
494b55c
Compare
494b55c to
b814c82
Compare
Codecov ReportAll modified and coverable lines are covered by tests ✅
Additional details and impacted files@@ Coverage Diff @@
## master #4876 +/- ##
============================================
+ Coverage 97.38% 97.40% +0.01%
- Complexity 6598 6624 +26
============================================
Files 921 922 +1
Lines 19071 19157 +86
Branches 1269 1271 +2
============================================
+ Hits 18573 18659 +86
Misses 390 390
Partials 108 108 ☔ View full report in Codecov by Sentry. |
valfirst
requested changes
Mar 15, 2024
Comment on lines
16
to
18
| === Azure Client Configuration | ||
|
|
||
| Azure clients are configured with a set of properties with the following format: |
There was a problem hiding this comment.
pleas avoid using client, it's hardly understandable by users
|
|
||
| [source,properties] | ||
| ---- | ||
| service-bus.{service-bus-key}.{property-name}=property value |
There was a problem hiding this comment.
Suggested change
| service-bus.{service-bus-key}.{property-name}=property value | |
| azure.service-bus.{service-bus-key}.{property-name}=property value |
| service-bus.{service-bus-key}.{property-name}=property value | ||
| ---- | ||
|
|
||
| Where: |
There was a problem hiding this comment.
Suggested change
| Where: | |
| where: |
| When I start consuming messages from `myProjectTopic` service bus | ||
| ---- | ||
|
|
||
| === Stop consume messages |
There was a problem hiding this comment.
Suggested change
| === Stop consume messages | |
| === Stop consuming messages |
|
|
||
| [source,gherkin] | ||
| ---- | ||
| When I $queueOperation consumed `$serviceBusKey` service bus messages to $scopes variable `$variableName` |
There was a problem hiding this comment.
I don't think it's for queue only
| .Peek messages from the consumer and save to variable | ||
| [source,gherkin] | ||
| ---- | ||
| When I PEEK consumed `myProjectQueue` service bus messages to STORY variable `queueMessages` |
There was a problem hiding this comment.
please add a set of complete examples to get understanding how to use these steps (in the same way we have for Kafka)
uarlouski
reviewed
Mar 18, 2024
| ---- | ||
|
|
||
| * `$queueOperation` - `DRAIN` - saves the messages consumed since the last drain or from the consumption start and moves the consumer cursor to the position after the last consumed message; | ||
| `PEEK` - saves the messages consumed since the last drain or from the consumption start and doesn't change the consumer cursor position. |
There was a problem hiding this comment.
does it makes sense to represent operations as sublist of queueOperation?
uarlouski
reviewed
Mar 18, 2024
| } | ||
| catch (ServiceBusException e) | ||
| { | ||
| throw new ServiceBusSendMessageException("Unable to send message to Azure Service Bus", e); | ||
| } | ||
| } | ||
|
|
||
| public void startConsuming(String clientKey) |
There was a problem hiding this comment.
do we need to handle cases when end user invokes startConsuming several times with the same key?
uarlouski
reviewed
Mar 18, 2024
|
|
||
| public void stopAll() | ||
| { | ||
| clients.keySet().stream().toList().forEach(this::stopConsuming); |
There was a problem hiding this comment.
forEach can be invoked on stream
There was a problem hiding this comment.
I use remove method over the clients map. Usage of forEach will lead to ConcurrentModificationException
b814c82 to
3ba8e34
Compare
valfirst
requested changes
Mar 19, 2024
| where: | ||
|
|
||
| . `service-bus-key` - The key associated with the Azure Service Bus configuration, which will be used as a step parameter. | ||
| . `property-name` - The name of the topic property. One of: |
There was a problem hiding this comment.
Suggested change
| . `property-name` - The name of the topic property. One of: | |
| . `property-name` - The name of Service Bus property. One of: |
|
|
||
| . `service-bus-key` - The key associated with the Azure Service Bus configuration, which will be used as a step parameter. | ||
| . `property-name` - The name of the topic property. One of: | ||
| .. `channel-type` - The type of service bus messaging components: either |
There was a problem hiding this comment.
Suggested change
| .. `channel-type` - The type of service bus messaging components: either | |
| .. `channel-type` - The type of Service Bus messaging channel: either |
| . `property-name` - The name of the topic property. One of: | ||
| .. `channel-type` - The type of service bus messaging components: either | ||
| https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-overview#concepts[QUEUE or TOPIC]. | ||
| .. `namespace` - The name of the namespace the service bus belongs to. |
There was a problem hiding this comment.
Suggested change
| .. `namespace` - The name of the namespace the service bus belongs to. | |
| .. `namespace` - The name of the namespace Service Bus belongs to. |
| https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-overview#concepts[QUEUE or TOPIC]. | ||
| .. `namespace` - The name of the namespace the service bus belongs to. | ||
| .. `name` - The queue or topic name. | ||
| .. `subscription-name` - The name of the topic subscription. Only for TOPIC channel-type. |
There was a problem hiding this comment.
Suggested change
| .. `subscription-name` - The name of the topic subscription. Only for TOPIC channel-type. | |
| .. `subscription-name` - The name of the topic subscription. Only for `TOPIC` channel type. |
|
|
||
| === Start consuming messages | ||
|
|
||
| Starts the Azure Service Bus consumer with the provided configuration to listen the specified |
There was a problem hiding this comment.
Suggested change
| Starts the Azure Service Bus consumer with the provided configuration to listen the specified | |
| Starts Azure Service Bus consumer with the provided configuration to listen the specified |
| * <li><b>name</b> - the queue or topic name</li> | ||
| * <li><b>subscription-name</b> - the name of the topic subscription. Only for TOPIC channel-type</li> | ||
| * </ul> | ||
| * @param serviceBusKey Identifying the Service Bus connection details. |
There was a problem hiding this comment.
make sure to keep docs and javadocs aligned
| Matcher<Integer> countMatcher = comparisonRule.getComparisonRule(expectedCount); | ||
| Integer result = new DurationBasedWaiter(timeout, Duration.ofSeconds(1)).wait( | ||
| () -> serviceBusService.getMessagesForClient(serviceBusKey).size(), countMatcher::matches); | ||
| softAssert.assertThat("Total count of messages for Service Bus client with key: " + serviceBusKey, |
There was a problem hiding this comment.
please avoid using client naming
|
|
||
| package org.vividus.azure.servicebus.model; | ||
|
|
||
| public class ServiceBusClientEntity |
There was a problem hiding this comment.
Suggested change
| public class ServiceBusClientEntity | |
| public class ServiceBusConnectionParameters |
| else | ||
| { | ||
| processorBuilder.topicName(name); | ||
| processorBuilder.subscriptionName(serviceBusClientEntity.getSubscriptionName()); |
There was a problem hiding this comment.
we need to throw an error if subscription name is configured for queue
|
|
||
| private ServiceBusClientEntity getServiceBusClientConfig(String clientKey) | ||
| { | ||
| return clientConfigs.get(clientKey, "No Service Bus client with key `%s` is configured", clientKey); |
There was a problem hiding this comment.
Suggested change
| return clientConfigs.get(clientKey, "No Service Bus client with key `%s` is configured", clientKey); | |
| return clientConfigs.get(clientKey, "No Service Bus connection with key `%s` is configured", clientKey); |
3029a80 to
2fd5919
Compare
2fd5919 to
1b631ba
Compare
1b631ba to
a0b34c1
Compare
valfirst
approved these changes
Mar 20, 2024
uarlouski
approved these changes
Mar 21, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
…e Service Bus