forked from Azure/azure-event-hubs-java
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix Interop issue: Amqp body section can be AmqpValue or AmqpSequence (…
…Azure#66) * support interop with 2 other body types - amqp-value & amqp-sequence * receivedEvent.getBody() should throw and communicate - in case of interop issues * fix flaky CIT issues * Change exception name from Illegal to Unexpected * implement reSend path for other Body section types
- Loading branch information
1 parent
180bc11
commit ad9fddb
Showing
6 changed files
with
260 additions
and
17 deletions.
There are no files selected for viewing
This file contains 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
39 changes: 39 additions & 0 deletions
39
...enthubs/src/main/java/com/microsoft/azure/eventhubs/UnexpectedEventDataBodyException.java
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| /* | ||
| * Copyright (c) Microsoft. All rights reserved. | ||
| * Licensed under the MIT license. See LICENSE file in the project root for full license information. | ||
| */ | ||
| package com.microsoft.azure.eventhubs; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| import org.apache.qpid.proton.amqp.messaging.AmqpSequence; | ||
| import org.apache.qpid.proton.amqp.messaging.AmqpValue; | ||
| import org.apache.qpid.proton.amqp.messaging.Data; | ||
|
|
||
| import com.microsoft.azure.servicebus.amqp.AmqpConstants; | ||
|
|
||
| public class UnexpectedEventDataBodyException extends RuntimeException { | ||
|
|
||
| private static final Map<Class, String> KNOWN_SECTIONS = new HashMap<Class, String>() {{ | ||
| put(AmqpValue.class, AmqpConstants.AMQP_VALUE); | ||
| put(AmqpSequence.class, AmqpConstants.AMQP_SEQUENCE); | ||
| }}; | ||
|
|
||
| private final Class bodySection; | ||
|
|
||
| public UnexpectedEventDataBodyException(final Class actualBodySection) { | ||
| super(KNOWN_SECTIONS.containsKey(actualBodySection) | ||
| ? String.format("AmqpMessage Body Section will be available in %s.getBody() only if it is of type: %s. " + | ||
| "If AmqpMessage has any other type as part of Body Section - it will be added to %s.getSystemProperties()." + | ||
| " Use '%s' as Key to fetch this from %s.getSystemProperties().", | ||
| EventData.class, Data.class, EventData.class, KNOWN_SECTIONS.get(actualBodySection), EventData.class) | ||
| : "AmqpMessage Body Section cannot be mapped to any EventData section."); | ||
| this.bodySection = actualBodySection; | ||
| } | ||
|
|
||
| // used for testing | ||
| public String getSystemPropertyName() { | ||
| return KNOWN_SECTIONS.get(this.bodySection); | ||
| } | ||
| } |
This file contains 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
This file contains 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
150 changes: 150 additions & 0 deletions
150
...eventhubs/src/test/java/com/microsoft/azure/eventhubs/eventdata/InteropEventBodyTest.java
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,150 @@ | ||
| /* | ||
| * Copyright (c) Microsoft. All rights reserved. | ||
| * Licensed under the MIT license. See LICENSE file in the project root for full license information. | ||
| */ | ||
| package com.microsoft.azure.eventhubs.eventdata; | ||
|
|
||
| import java.io.IOException; | ||
| import java.time.Duration; | ||
| import java.time.Instant; | ||
| import java.util.concurrent.ExecutionException; | ||
| import java.util.LinkedList; | ||
| import java.util.List; | ||
|
|
||
| import org.apache.qpid.proton.amqp.Binary; | ||
| import org.apache.qpid.proton.amqp.messaging.AmqpSequence; | ||
| import org.apache.qpid.proton.amqp.messaging.AmqpValue; | ||
| import org.apache.qpid.proton.amqp.messaging.Data; | ||
| import org.apache.qpid.proton.Proton; | ||
| import org.apache.qpid.proton.message.Message; | ||
|
|
||
| import org.junit.AfterClass; | ||
| import org.junit.Assert; | ||
| import org.junit.BeforeClass; | ||
| import org.junit.Test; | ||
|
|
||
| import com.microsoft.azure.eventhubs.EventData; | ||
| import com.microsoft.azure.eventhubs.EventHubClient; | ||
| import com.microsoft.azure.eventhubs.UnexpectedEventDataBodyException; | ||
| import com.microsoft.azure.eventhubs.PartitionReceiver; | ||
| import com.microsoft.azure.eventhubs.PartitionSender; | ||
| import com.microsoft.azure.eventhubs.lib.ApiTestBase; | ||
| import com.microsoft.azure.eventhubs.lib.TestContext; | ||
| import com.microsoft.azure.servicebus.ConnectionStringBuilder; | ||
| import com.microsoft.azure.servicebus.MessageSender; | ||
| import com.microsoft.azure.servicebus.MessagingFactory; | ||
| import com.microsoft.azure.servicebus.ServiceBusException; | ||
| import com.microsoft.azure.servicebus.amqp.AmqpConstants; | ||
|
|
||
| public class InteropEventBodyTest extends ApiTestBase { | ||
|
|
||
| static EventHubClient ehClient; | ||
| static MessagingFactory msgFactory; | ||
| static PartitionReceiver receiver; | ||
| static MessageSender partitionMsgSender; | ||
| static PartitionSender partitionSender; | ||
|
|
||
| static final String partitionId = "0"; | ||
| static EventData receivedEvent; | ||
| static EventData reSentAndReceivedEvent; | ||
| static Message reSendAndReceivedMessage; | ||
|
|
||
| @BeforeClass | ||
| public static void initialize() throws ServiceBusException, IOException, InterruptedException, ExecutionException | ||
| { | ||
| final ConnectionStringBuilder connStrBuilder = TestContext.getConnectionString(); | ||
| final String connectionString = connStrBuilder.toString(); | ||
|
|
||
| ehClient = EventHubClient.createFromConnectionStringSync(connectionString); | ||
| msgFactory = MessagingFactory.createFromConnectionString(connectionString).get(); | ||
| receiver = ehClient.createReceiverSync(TestContext.getConsumerGroupName(), partitionId, Instant.now()); | ||
| partitionSender = ehClient.createPartitionSenderSync(partitionId); | ||
| partitionMsgSender = MessageSender.create(msgFactory, "link1", connStrBuilder.getEntityPath() + "/partitions/" + partitionId).get(); | ||
|
|
||
| // run out of messages in that specific partition - to account for clock-skew with Instant.now() on test machine vs eventhubs service | ||
| receiver.setReceiveTimeout(Duration.ofSeconds(5)); | ||
| Iterable<EventData> clockSkewEvents; | ||
| do { | ||
| clockSkewEvents = receiver.receiveSync(100); | ||
| } while (clockSkewEvents != null && clockSkewEvents.iterator().hasNext()); | ||
| } | ||
|
|
||
| @Test | ||
| public void interopWithProtonAmqpMessageBodyAsAmqpValue() throws ServiceBusException, InterruptedException, ExecutionException | ||
| { | ||
| Message originalMessage = Proton.message(); | ||
| String payload = "testmsg"; | ||
| originalMessage.setBody(new AmqpValue(payload)); | ||
| partitionMsgSender.send(originalMessage).get(); | ||
| receivedEvent = receiver.receiveSync(10).iterator().next(); | ||
|
|
||
| Assert.assertEquals(payload, receivedEvent.getSystemProperties().get(AmqpConstants.AMQP_VALUE)); | ||
|
|
||
| try { | ||
| receivedEvent.getBody(); | ||
| Assert.assertTrue(false); // this line shouldn't be reachable | ||
| } catch (UnexpectedEventDataBodyException exception) { | ||
| Assert.assertEquals(AmqpConstants.AMQP_VALUE, exception.getSystemPropertyName()); | ||
| } | ||
|
|
||
| partitionSender.sendSync(receivedEvent); | ||
| reSentAndReceivedEvent = receiver.receiveSync(10).iterator().next(); | ||
| Assert.assertEquals(payload, reSentAndReceivedEvent.getSystemProperties().get(AmqpConstants.AMQP_VALUE)); | ||
|
|
||
| try { | ||
| reSentAndReceivedEvent.getBody(); | ||
| Assert.assertTrue(false); // this line shouldn't be reachable | ||
| } catch (UnexpectedEventDataBodyException exception) { | ||
| Assert.assertEquals(AmqpConstants.AMQP_VALUE, exception.getSystemPropertyName()); | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| public void interopWithProtonAmqpMessageBodyAsAmqpSequence() throws ServiceBusException, InterruptedException, ExecutionException | ||
| { | ||
| Message originalMessage = Proton.message(); | ||
| String payload = "testmsg"; | ||
| LinkedList<Data> datas = new LinkedList<>(); | ||
| datas.add(new Data(new Binary(payload.getBytes()))); | ||
| originalMessage.setBody(new AmqpSequence(datas)); | ||
|
|
||
| partitionMsgSender.send(originalMessage).get(); | ||
| receivedEvent = receiver.receiveSync(10).iterator().next(); | ||
|
|
||
| Assert.assertEquals(payload, new String(((List<Data>)(receivedEvent.getSystemProperties().get(AmqpConstants.AMQP_SEQUENCE))).get(0).getValue().getArray())); | ||
|
|
||
| try { | ||
| receivedEvent.getBody(); | ||
| Assert.assertTrue(false); // getBody() should throw; this line shouldn't be reachable | ||
| } catch (UnexpectedEventDataBodyException exception) { | ||
| Assert.assertEquals(AmqpConstants.AMQP_SEQUENCE, exception.getSystemPropertyName()); | ||
| } | ||
|
|
||
| partitionSender.sendSync(receivedEvent); | ||
| reSentAndReceivedEvent = receiver.receiveSync(10).iterator().next(); | ||
| Assert.assertEquals(payload, new String(((List<Data>)(reSentAndReceivedEvent.getSystemProperties().get(AmqpConstants.AMQP_SEQUENCE))).get(0).getValue().getArray())); | ||
|
|
||
| try { | ||
| reSentAndReceivedEvent.getBody(); | ||
| Assert.assertTrue(false); // getBody() should throw; this line shouldn't be reachable | ||
| } catch (UnexpectedEventDataBodyException exception) { | ||
| Assert.assertEquals(AmqpConstants.AMQP_SEQUENCE, exception.getSystemPropertyName()); | ||
| } | ||
| } | ||
|
|
||
| @AfterClass | ||
| public static void cleanup() throws ServiceBusException | ||
| { | ||
| if (partitionMsgSender != null) | ||
| partitionMsgSender.closeSync(); | ||
|
|
||
| if (receiver != null) | ||
| receiver.closeSync(); | ||
|
|
||
| if (ehClient != null) | ||
| ehClient.closeSync(); | ||
|
|
||
| if (msgFactory != null) | ||
| msgFactory.closeSync(); | ||
| } | ||
| } |
Oops, something went wrong.