Skip to content

Refactor move servicebus message state key to azure-core-amqp #29309

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ZejiaJiang merged 5 commits into from
Jun 30, 2022
Merged

Refactor move servicebus message state key to azure-core-amqp #29309

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6980b8f
Move servicebus message state key to amqp core
ZejiaJiang Jun 8, 2022
d6fa920
update pom
ZejiaJiang Jun 8, 2022
879c498
Merge branch 'main' into refactor/move-msg-state
ZejiaJiang Jun 9, 2022
9420a56
update CHANGELOG
ZejiaJiang Jun 9, 2022
037a1cf
remove prefix
ZejiaJiang Jun 22, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions eng/versioning/version_client.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -360,6 +360,7 @@ com.azure.tools:azure-sdk-build-tool;1.0.0-beta.1;1.0.0-beta.2
# note: The unreleased dependencies will not be manipulated with the automatic PR creation code.
# In the pom, the version update tag after the version should name the unreleased package and the dependency version:
# <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
unreleased_com.azure:azure-core-amqp;2.6.0-beta.1

# Released Beta dependencies: Copy the entry from above, prepend "beta_", remove the current
# version and set the version to the released beta. Released beta dependencies are only valid
Expand Down
7 changes: 6 additions & 1 deletion sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/AmqpMessageConstant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,12 @@ public enum AmqpMessageConstant {
/**
* The identifier for deadletter reason.
*/
DEAD_LETTER_REASON_ANNOTATION_NAME("DeadLetterReason");
DEAD_LETTER_REASON_ANNOTATION_NAME("DeadLetterReason"),
/**
* The state of message.
*/
SERVICE_BUS_MESSAGE_STATE_ANNOTATION_NAME("x-opt-message-state");

@liukun-msft liukun-msft Jun 21, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Any reason for why we change the name from SERVICE_BUS_MESSAGE_STATE_KEY to SERVICE_BUS_MESSAGE_STATE_ANNOTATION_NAME?

Since we move it into amqp, is it ok to remove SERVICE_BUS prefix?

@ZejiaJiang ZejiaJiang Jun 22, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, the above keys have ANNOTATION_NAME suffix as well and we do get the key from annotation.

I'll check these keys usage and remove the prefix.



private static final Map<String, AmqpMessageConstant> RESERVED_CONSTANTS_MAP = new HashMap<>();
private final String constant;
Expand Down
1 change: 1 addition & 0 deletions sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
- Removed the thread hopping overhead with default max-concurrent-calls; this saves allocations incurred by the parallel operator and addresses the undesired prefetching when prefetch is disabled and max-concurrent-calls is set to default. ([#29011](https://github.com/Azure/azure-sdk-for-java/pull/29011))

### Other Changes
- Moved message state key to `azure-core-amqp` constants. ([#26898](https://github.com/Azure/azure-sdk-for-java/issues/26898))

## 7.9.0 (2022-05-18)

Expand Down
2 changes: 1 addition & 1 deletion sdk/servicebus/azure-messaging-servicebus/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@
<dependency>
<groupId>com.azure</groupId>
<artifactId>azure-core-amqp</artifactId>
<version>2.5.2</version> <!-- {x-version-update;com.azure:azure-core-amqp;dependency} -->
<version>2.6.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core-amqp;dependency} -->
</dependency>
<dependency>
<groupId>com.azure</groupId>
Expand Down
5 changes: 2 additions & 3 deletions ...ng-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceivedMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import static com.azure.core.amqp.AmqpMessageConstant.PARTITION_KEY_ANNOTATION_NAME;
import static com.azure.core.amqp.AmqpMessageConstant.SCHEDULED_ENQUEUE_UTC_TIME_NAME;
import static com.azure.core.amqp.AmqpMessageConstant.SEQUENCE_NUMBER_ANNOTATION_NAME;
import static com.azure.core.amqp.AmqpMessageConstant.SERVICE_BUS_MESSAGE_STATE_ANNOTATION_NAME;

/**
* The data structure encapsulating the message received from Service Bus. The message structure is discussed in detail
Expand All @@ -53,8 +54,6 @@ public final class ServiceBusReceivedMessage {
private boolean isSettled = false;
private Context context;

static final String SERVICE_BUS_MESSAGE_STATE_KEY = "x-opt-message-state";

ServiceBusReceivedMessage(BinaryData body) {
Objects.requireNonNull(body, "'body' cannot be null.");
amqpAnnotatedMessage = new AmqpAnnotatedMessage(AmqpMessageBody.fromData(body.toBytes()));
Expand Down Expand Up @@ -455,7 +454,7 @@ public String getSessionId() {
* @throws UnsupportedOperationException if the message state is an unknown value.
*/
public ServiceBusMessageState getState() {
final Object value = amqpAnnotatedMessage.getMessageAnnotations().get(SERVICE_BUS_MESSAGE_STATE_KEY);
final Object value = amqpAnnotatedMessage.getMessageAnnotations().get(SERVICE_BUS_MESSAGE_STATE_ANNOTATION_NAME.getValue());

if (value instanceof Integer) {
return ServiceBusMessageState.fromValue((Integer) value);
Expand Down
6 changes: 3 additions & 3 deletions ...ervicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceivedMessageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
import static com.azure.core.amqp.AmqpMessageConstant.ENQUEUED_TIME_UTC_ANNOTATION_NAME;
import static com.azure.core.amqp.AmqpMessageConstant.LOCKED_UNTIL_KEY_ANNOTATION_NAME;
import static com.azure.core.amqp.AmqpMessageConstant.SEQUENCE_NUMBER_ANNOTATION_NAME;
import static com.azure.messaging.servicebus.ServiceBusReceivedMessage.SERVICE_BUS_MESSAGE_STATE_KEY;
import static com.azure.core.amqp.AmqpMessageConstant.SERVICE_BUS_MESSAGE_STATE_ANNOTATION_NAME;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
Expand Down Expand Up @@ -167,7 +167,7 @@ public static Stream<Arguments> canGetMessageState() {
public void canGetMessageState(Integer value, ServiceBusMessageState expected) {
// Arrange
final ServiceBusReceivedMessage message = new ServiceBusReceivedMessage(PAYLOAD_BINARY);
message.getRawAmqpMessage().getMessageAnnotations().put(SERVICE_BUS_MESSAGE_STATE_KEY, value);
message.getRawAmqpMessage().getMessageAnnotations().put(SERVICE_BUS_MESSAGE_STATE_ANNOTATION_NAME.getValue(), value);

// Act
final ServiceBusMessageState actual = message.getState();
Expand All @@ -187,7 +187,7 @@ public void defaultMessageState() {
public void throwsOnInvalidMessageState() {
// Arrange
final ServiceBusReceivedMessage message = new ServiceBusReceivedMessage(PAYLOAD_BINARY);
message.getRawAmqpMessage().getMessageAnnotations().put(SERVICE_BUS_MESSAGE_STATE_KEY, 10);
message.getRawAmqpMessage().getMessageAnnotations().put(SERVICE_BUS_MESSAGE_STATE_ANNOTATION_NAME.getValue(), 10);

// Act & Assert
assertThrows(UnsupportedOperationException.class, () -> message.getState());
Expand Down