Skip to content

Commit b3d87c8

Browse files
rujcherujche
committed
Create new project: spring-cloud-azure-testcontainers/event-hubs/spring-cloud-stream-binders
1 parent d2d64da commit b3d87c8

4 files changed

Lines changed: 233 additions & 0 deletions

File tree

‎pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,7 @@
8787
<module>spring-cloud-azure-docker-compose/storage-queue</module>
8888
<module>spring-cloud-azure-testcontainers/cosmos</module>
8989
<module>spring-cloud-azure-testcontainers/event-hubs/client</module>
90+
<module>spring-cloud-azure-testcontainers/event-hubs/spring-cloud-stream-binders</module>
9091
<module>spring-cloud-azure-testcontainers/service-bus/spring-cloud-stream-binders</module>
9192
<module>spring-cloud-azure-testcontainers/service-bus/spring-messaging</module>
9293
<module>spring-cloud-azure-testcontainers/storage-blob</module>

‎spring-cloud-azure-testcontainers/event-hubs/spring-cloud-stream-binders/pom.xml‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
3+
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xmlns="http://maven.apache.org/POM/4.0.0"
5+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
6+
<modelVersion>4.0.0</modelVersion>
7+
8+
<parent>
9+
<groupId>org.springframework.boot</groupId>
10+
<artifactId>spring-boot-starter-parent</artifactId>
11+
<version>4.0.3</version>
12+
<relativePath/> <!-- lookup parent from repository -->
13+
</parent>
14+
15+
<groupId>com.azure.spring</groupId>
16+
<artifactId>spring-cloud-azure-testcontainers-for-event-hubs-spring-cloud-stream-sample</artifactId>
17+
<version>1.0.0</version>
18+
<packaging>jar</packaging>
19+
20+
<properties>
21+
<version.spring.cloud.azure>7.1.0-beta.1</version.spring.cloud.azure> <!--TODO: Change this version-->
22+
</properties>
23+
24+
<dependencyManagement>
25+
<dependencies>
26+
<dependency>
27+
<groupId>com.azure.spring</groupId>
28+
<artifactId>spring-cloud-azure-dependencies</artifactId>
29+
<version>${version.spring.cloud.azure}</version>
30+
<type>pom</type>
31+
<scope>import</scope>
32+
</dependency>
33+
</dependencies>
34+
</dependencyManagement>
35+
36+
<dependencies>
37+
<dependency>
38+
<groupId>com.azure.spring</groupId>
39+
<artifactId>spring-cloud-azure-stream-binder-eventhubs</artifactId>
40+
</dependency>
41+
<dependency>
42+
<groupId>org.springframework.boot</groupId>
43+
<artifactId>spring-boot-starter-test</artifactId>
44+
<scope>test</scope>
45+
</dependency>
46+
<dependency>
47+
<groupId>org.springframework.boot</groupId>
48+
<artifactId>spring-boot-testcontainers</artifactId>
49+
<scope>test</scope>
50+
</dependency>
51+
<dependency>
52+
<groupId>org.testcontainers</groupId>
53+
<artifactId>azure</artifactId>
54+
<version>1.21.3</version>
55+
<scope>test</scope>
56+
</dependency>
57+
<dependency>
58+
<groupId>org.testcontainers</groupId>
59+
<artifactId>junit-jupiter</artifactId>
60+
<version>1.21.3</version>
61+
<scope>test</scope>
62+
</dependency>
63+
<dependency>
64+
<groupId>com.azure.spring</groupId>
65+
<artifactId>spring-cloud-azure-testcontainers</artifactId>
66+
<scope>test</scope>
67+
</dependency>
68+
<dependency>
69+
<groupId>org.awaitility</groupId>
70+
<artifactId>awaitility</artifactId>
71+
<scope>test</scope>
72+
</dependency>
73+
</dependencies>
74+
75+
</project>

‎spring-cloud-azure-testcontainers/event-hubs/spring-cloud-stream-binders/src/test/java/EventHubsTestContainerTest.java‎

Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
2+
import com.azure.spring.cloud.autoconfigure.implementation.context.AzureGlobalPropertiesAutoConfiguration;
3+
import com.azure.spring.cloud.autoconfigure.implementation.eventhubs.AzureEventHubsAutoConfiguration;
4+
import com.azure.spring.cloud.autoconfigure.implementation.eventhubs.AzureEventHubsMessagingAutoConfiguration;
5+
import com.azure.spring.messaging.checkpoint.Checkpointer;
6+
import com.azure.storage.blob.BlobContainerAsyncClient;
7+
import com.azure.storage.blob.BlobServiceAsyncClient;
8+
import com.azure.storage.blob.BlobServiceClientBuilder;
9+
import com.azure.storage.blob.BlobServiceVersion;
10+
import org.junit.jupiter.api.Test;
11+
import org.slf4j.Logger;
12+
import org.slf4j.LoggerFactory;
13+
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
14+
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
15+
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
16+
import org.springframework.context.annotation.Bean;
17+
import org.springframework.context.annotation.Configuration;
18+
import org.springframework.messaging.Message;
19+
import org.springframework.messaging.support.MessageBuilder;
20+
import org.springframework.test.context.TestPropertySource;
21+
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
22+
import org.testcontainers.azure.AzuriteContainer;
23+
import org.testcontainers.azure.EventHubsEmulatorContainer;
24+
import org.testcontainers.containers.Network;
25+
import org.testcontainers.junit.jupiter.Container;
26+
import org.testcontainers.junit.jupiter.Testcontainers;
27+
import org.testcontainers.utility.MountableFile;
28+
29+
import java.time.Duration;
30+
import java.util.Set;
31+
import java.util.concurrent.ConcurrentHashMap;
32+
import java.util.concurrent.atomic.AtomicInteger;
33+
import java.util.function.Consumer;
34+
import java.util.function.Supplier;
35+
36+
import static com.azure.spring.messaging.AzureHeaders.CHECKPOINTER;
37+
import static org.assertj.core.api.Assertions.assertThat;
38+
import static org.awaitility.Awaitility.waitAtMost;
39+
40+
@SpringJUnitConfig
41+
@TestPropertySource(properties = {
42+
"spring.cloud.function.definition=consume;supply",
43+
"spring.cloud.stream.bindings.consume-in-0.destination=eh1",
44+
"spring.cloud.stream.bindings.consume-in-0.group=$Default",
45+
"spring.cloud.stream.bindings.supply-out-0.destination=eh1",
46+
"spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode=MANUAL",
47+
"spring.cloud.stream.poller.fixed-delay=1000",
48+
"spring.cloud.stream.poller.initial-delay=0"})
49+
@Testcontainers
50+
class EventHubsTestContainerTest {
51+
52+
private static final Network NETWORK = Network.newNetwork();
53+
54+
private static final AzuriteContainer AZURITE = new AzuriteContainer(
55+
"mcr.microsoft.com/azure-storage/azurite:latest")
56+
.withCommand("azurite", "--blobHost", "0.0.0.0", "--queueHost", "0.0.0.0", "--tableHost", "0.0.0.0",
57+
"--skipApiVersionCheck")
58+
.withNetwork(NETWORK)
59+
.withNetworkAliases("azurite");
60+
61+
@Container
62+
@ServiceConnection
63+
private static final EventHubsEmulatorContainer EVENT_HUBS = new EventHubsEmulatorContainer(
64+
"mcr.microsoft.com/azure-messaging/eventhubs-emulator:latest")
65+
.acceptLicense()
66+
.withCopyFileToContainer(MountableFile.forClasspathResource("Config.json"),
67+
"/Eventhubs_Emulator/ConfigFiles/Config.json")
68+
.withNetwork(NETWORK)
69+
.withAzuriteContainer(AZURITE);
70+
71+
private static final Logger LOGGER = LoggerFactory.getLogger(EventHubsTestContainerTest.class);
72+
private static final Set<String> RECEIVED_MESSAGES = ConcurrentHashMap.newKeySet();
73+
private static final AtomicInteger messageSequence = new AtomicInteger(0);
74+
75+
@Test
76+
void supplierAndConsumerShouldWorkThroughEventHubs() {
77+
waitAtMost(Duration.ofSeconds(120))
78+
.pollDelay(Duration.ofSeconds(2))
79+
.pollInterval(Duration.ofSeconds(2))
80+
.untilAsserted(() -> {
81+
assertThat(RECEIVED_MESSAGES).isNotEmpty();
82+
LOGGER.info("✓ Test passed - Consumer received {} message(s)", RECEIVED_MESSAGES.size());
83+
});
84+
}
85+
86+
@Configuration(proxyBeanMethods = false)
87+
@EnableAutoConfiguration
88+
@ImportAutoConfiguration(classes = {
89+
AzureGlobalPropertiesAutoConfiguration.class,
90+
AzureEventHubsAutoConfiguration.class,
91+
AzureEventHubsMessagingAutoConfiguration.class})
92+
static class Config {
93+
94+
private static final String CHECKPOINT_CONTAINER_NAME = "eventhubs-checkpoint";
95+
96+
@Bean
97+
public BlobCheckpointStore blobCheckpointStore() {
98+
BlobServiceAsyncClient blobServiceAsyncClient = new BlobServiceClientBuilder()
99+
.connectionString(AZURITE.getConnectionString())
100+
.serviceVersion(BlobServiceVersion.V2025_01_05)
101+
.buildAsyncClient();
102+
BlobContainerAsyncClient containerAsyncClient = blobServiceAsyncClient
103+
.getBlobContainerAsyncClient(CHECKPOINT_CONTAINER_NAME);
104+
if (Boolean.FALSE.equals(containerAsyncClient.exists().block(Duration.ofSeconds(3)))) {
105+
containerAsyncClient.create().block(Duration.ofSeconds(3));
106+
}
107+
return new BlobCheckpointStore(containerAsyncClient);
108+
}
109+
110+
@Bean
111+
public Supplier<Message<String>> supply() {
112+
return () -> {
113+
int sequence = messageSequence.getAndIncrement();
114+
String payload = "Hello world, " + sequence;
115+
LOGGER.info("[Supplier] Invoked - message sequence: {}", sequence);
116+
return MessageBuilder.withPayload(payload).build();
117+
};
118+
}
119+
120+
@Bean
121+
public Consumer<Message<String>> consume() {
122+
return message -> {
123+
String payload = message.getPayload();
124+
RECEIVED_MESSAGES.add(payload);
125+
LOGGER.info("[Consumer] Received message: {}", payload);
126+
127+
Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
128+
if (checkpointer != null) {
129+
checkpointer.success()
130+
.doOnSuccess(s -> LOGGER.info("[Consumer] Message checkpointed"))
131+
.doOnError(e -> LOGGER.error("[Consumer] Checkpoint failed", e))
132+
.block();
133+
}
134+
};
135+
}
136+
}
137+
}

‎spring-cloud-azure-testcontainers/event-hubs/spring-cloud-stream-binders/src/test/resources/Config.json‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
{
2+
"UserConfig": {
3+
"NamespaceConfig": [
4+
{
5+
"Type": "EventHub",
6+
"Name": "emulatorns1",
7+
"Entities": [
8+
{
9+
"Name": "eh1",
10+
"PartitionCount": "2",
11+
"ConsumerGroups": []
12+
}
13+
]
14+
}
15+
],
16+
"LoggingConfig": {
17+
"Type": "File"
18+
}
19+
}
20+
}

0 commit comments

Comments
 (0)