SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make ZeroMQ modules as auto-configuration

Browse Source 
* Fix all the Checkstyle violations in these modules
This commit is contained in:
Artem Bilan
2024-01-08 10:21:20 -05:00
parent dceec3250b
commit 985a753819
10 changed files with 60 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
consumer/spring-zeromq-consumer/src/main/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -25,18 +25,20 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.zeromq.outbound.ZeroMqMessageHandler;
import org.springframework.messaging.Message;
/**
* The auto-configuration for ZeroMQ consumer.
*
* @author Daniel Frey
* @since 3.1.0
*/
@Configuration
@AutoConfiguration
@EnableConfigurationProperties(ZeroMqConsumerProperties.class)
public class ZeroMqConsumerConfiguration {
@@ -49,6 +51,7 @@ public class ZeroMqConsumerConfiguration {
public ZeroMqMessageHandler zeromqMessageHandler(ZeroMqConsumerProperties properties, ZContext zContext,
@Autowired(required = false) Consumer<ZMQ.Socket> socketConfigurer,
@Autowired(required = false) OutboundMessageMapper<byte[]> messageMapper) {
ZeroMqMessageHandler zeroMqMessageHandler = new ZeroMqMessageHandler(zContext, properties.getConnectUrl(),
properties.getSocketType());
@@ -69,7 +72,7 @@ public class ZeroMqConsumerConfiguration {
@Bean
public Function<Flux<Message<?>>, Mono<Void>> zeromqConsumer(ZeroMqMessageHandler zeromqMessageHandler) {
return input -> input.flatMap(zeromqMessageHandler::handleMessage).ignoreElements();
return (input) -> input.flatMap(zeromqMessageHandler::handleMessage).ignoreElements();
}
}

19
consumer/spring-zeromq-consumer/src/main/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -25,6 +25,8 @@ import org.springframework.expression.Expression;
import org.springframework.validation.annotation.Validated;
/**
* The configuration properties for ZeroMQ consumer.
*
* @author Daniel Frey
* @since 3.1.0
*/
@@ -49,35 +51,26 @@ public class ZeroMqConsumerProperties {
@NotNull(message = "'socketType' is required")
public SocketType getSocketType() {
return socketType;
return this.socketType;
}
/**
* @param socketType the {@link SocketType} to establish.
*/
public void setSocketType(SocketType socketType) {
this.socketType = socketType;
}
@NotEmpty(message = "connectUrl is required like protocol://server:port")
public String getConnectUrl() {
return connectUrl;
return this.connectUrl;
}
/**
* @param connectUrl The ZeroMQ socket to expose
*/
public void setConnectUrl(String connectUrl) {
this.connectUrl = connectUrl;
}
public Expression getTopic() {
return topic;
return this.topic;
}
/**
* @param topic The 'topic' SpEL expression to set
*/
public void setTopic(Expression topic) {
this.topic = topic;
}

4
consumer/spring-zeromq-consumer/src/main/java/org/springframework/cloud/fn/consumer/zeromq/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The ZeroMQ consumer auto-configuration support.
*/
package org.springframework.cloud.fn.consumer.zeromq;

1
consumer/spring-zeromq-consumer/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.consumer.zeromq.ZeroMqConsumerConfiguration

15
consumer/spring-zeromq-consumer/src/test/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerConfigurationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -34,6 +34,8 @@ import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
@@ -51,16 +53,21 @@ public class ZeroMqConsumerConfigurationTests {
private static ZMQ.Socket socket;
private static int bindPort;
@Autowired
Function<Flux<Message<?>>, Mono<Void>> subject;
@BeforeAll
static void setup() {
socket = CONTEXT.createSocket(SocketType.SUB);
socket.setReceiveTimeOut(10_000);
int bindPort = socket.bindToRandomPort("tcp://*");
System.setProperty("zeromq.consumer.connectUrl", "tcp://localhost:" + bindPort);
bindPort = socket.bindToRandomPort("tcp://*");
}
@DynamicPropertySource
static void zeromqConnectUrl(DynamicPropertyRegistry dynamicPropertyRegistry) {
dynamicPropertyRegistry.add("zeromq.consumer.connectUrl", () -> "tcp://localhost:" + bindPort);
}
@AfterAll

12
supplier/spring-zeromq-supplier/src/main/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -25,21 +25,21 @@ import org.zeromq.ZMQ;
import reactor.core.publisher.Flux;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
/**
* A source module that receives data from ZeroMQ.
* A supplier auto-configuration that receives data from ZeroMQ.
*
* @author Daniel Frey
* @since 3.1.0
*/
@Configuration
@AutoConfiguration
@EnableConfigurationProperties(ZeroMqSupplierProperties.class)
public class ZeroMqSupplierConfiguration {
@@ -70,7 +70,7 @@ public class ZeroMqSupplierConfiguration {
if (socketConfigurer != null) {
zeroMqMessageProducer.setSocketConfigurer(socketConfigurer);
}
zeroMqMessageProducer.setOutputChannel(output);
zeroMqMessageProducer.setOutputChannel(this.output);
zeroMqMessageProducer.setAutoStartup(false);
return zeroMqMessageProducer;
@@ -78,7 +78,7 @@ public class ZeroMqSupplierConfiguration {
@Bean
public Supplier<Flux<Message<?>>> zeromqSupplier(ZeroMqMessageProducer adapter) {
return () -> Flux.from(output).doOnSubscribe(subscription -> adapter.start());
return () -> Flux.from(this.output).doOnSubscribe((subscription) -> adapter.start());
}
}

36
supplier/spring-zeromq-supplier/src/main/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -27,6 +27,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;
/**
* The ZeroMQ supplier configuration properties.
*
* @author Daniel Frey
* @since 3.1.0
*/
@@ -59,68 +61,46 @@ public class ZeroMqSupplierProperties {
*/
private String[] topics = { "" };
/**
* @param socketType the {@link SocketType} to establish.
*/
public void setSocketType(SocketType socketType) {
this.socketType = socketType;
}
@NotNull(message = "'socketType' is required")
public SocketType getSocketType() {
return socketType;
return this.socketType;
}
@NotEmpty(message = "connectUrl is required like tcp://server:port")
public String getConnectUrl() {
return connectUrl;
return this.connectUrl;
}
/**
* @param connectUrl The ZeroMQ server connect url
*
* @see org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer#setConnectUrl(String)
*/
public void setConnectUrl(String connectUrl) {
this.connectUrl = connectUrl;
}
@Range(min = 0, message = "'bindPort' must not be negative")
public int getBindPort() {
return bindPort;
return this.bindPort;
}
/**
* @param bindPort The Port to bind to on all interfaces
*
* @see org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer#setBindPort(int)
*/
public void setBindPort(int bindPort) {
this.bindPort = bindPort;
}
@NotNull(message = "'consumeDelay' is required")
public Duration getConsumeDelay() {
return consumeDelay;
return this.consumeDelay;
}
/**
* Specify a {@link Duration} to delay consumption when no data received.
* @param consumeDelay the {@link Duration} to delay consumption when empty.
*/
public void setConsumeDelay(Duration consumeDelay) {
this.consumeDelay = consumeDelay;
}
public String[] getTopics() {
return topics;
return this.topics;
}
/**
* @param topics The ZeroMQ Topics to subscribe to
*
* @see org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer#setTopics(String...)
*/
public void setTopics(String... topics) {
this.topics = topics;
}

4
supplier/spring-zeromq-supplier/src/main/java/org/springframework/cloud/fn/supplier/zeromq/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The ZeroMQ supplier auto-configuration support.
*/
package org.springframework.cloud.fn.supplier.zeromq;

1
supplier/spring-zeromq-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.supplier.zeromq.ZeroMqSupplierConfiguration

18
supplier/spring-zeromq-supplier/src/test/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierConfigurationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -36,11 +36,14 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Daniel Frey since 3.1.0
* @author Daniel Frey
* @since 3.1.0
*/
@SpringBootTest(properties = { "zeromq.supplier.topics=test-topic" })
@DirtiesContext
@@ -50,6 +53,8 @@ public class ZeroMqSupplierConfigurationTests {
private static ZMQ.Socket socket;
private static int bindPort;
@Autowired
Supplier<Flux<Message<?>>> subject;
@@ -58,10 +63,12 @@ public class ZeroMqSupplierConfigurationTests {
String socketAddress = "tcp://*";
socket = CONTEXT.createSocket(SocketType.PUB);
int bindPort = socket.bindToRandomPort(socketAddress);
System.setProperty("zeromq.supplier.connectUrl", "tcp://localhost:" + bindPort);
bindPort = socket.bindToRandomPort(socketAddress);
}
@DynamicPropertySource
static void zeromqConnectUrl(DynamicPropertyRegistry dynamicPropertyRegistry) {
dynamicPropertyRegistry.add("zeromq.supplier.connectUrl", () -> "tcp://localhost:" + bindPort);
}
@AfterAll
@@ -72,7 +79,6 @@ public class ZeroMqSupplierConfigurationTests {
@Test
void testSubscriptionConfiguration() throws InterruptedException {
StepVerifier stepVerifier = StepVerifier.create(subject.get())
.assertNext((message) -> assertThat(message.getPayload())
.asInstanceOf(InstanceOfAssertFactories.type(byte[].class))