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

Make WebSocket modules as auto-configuration

Browse Source 
* Fix all the Checkstyle violations and compiler warnings
* Also fix Checkstyle violations for `spring-time-supplier`
This commit is contained in:
Artem Bilan
2024-01-08 14:17:29 -05:00
parent 373fbdd8a2
commit 80e0b80a30
26 changed files with 143 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
consumer/spring-websocket-consumer/README.adoc
View File

@@ -4,7 +4,7 @@ A consumer that allows you to send messages using websocket.
## Beans for injection
You can import `WebsocketConsumerConfiguration` in the application and then inject the following bean.
The `WebsocketConsumerConfiguration` auto-configuration provides the following bean:
`Consumer<Message<?>> websocketConsumer`

22
consumer/spring-websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-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.
@@ -28,6 +28,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.consumer.websocket.actuator.WebsocketConsumerTraceEndpoint;
@@ -39,12 +40,14 @@ import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
/**
* Auto-configuration for WebSocket consumer.
*
* @author Oliver Moser
* @author Gary Russell
* @author Artem Bilan
* @author Chris Bono
*/
@Configuration
@AutoConfiguration
@EnableConfigurationProperties(WebsocketConsumerProperties.class)
public class WebsocketConsumerConfiguration {
@@ -58,7 +61,7 @@ public class WebsocketConsumerConfiguration {
@PostConstruct
public void init() throws InterruptedException {
websocketConsumerServer.run();
this.websocketConsumerServer.run();
}
@Bean
@@ -69,14 +72,14 @@ public class WebsocketConsumerConfiguration {
@Bean
public Consumer<Message<?>> websocketConsumer(InMemoryTraceRepository websocketTraceRepository) {
return message -> {
return (message) -> {
if (logger.isTraceEnabled()) {
logger.trace("Handling message: " + message);
}
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
headers.setMessageTypeIfNotSet(SimpMessageType.MESSAGE);
String messagePayload = message.getPayload().toString();
for (Channel channel : WebsocketConsumerServer.channels) {
for (Channel channel : WebsocketConsumerServer.CHANNELS) {
if (logger.isTraceEnabled()) {
logger.trace(
String.format("Writing message %s to channel %s", messagePayload, channel.localAddress()));
@@ -101,22 +104,23 @@ public class WebsocketConsumerConfiguration {
websocketTraceRepository.add(trace);
}
@Configuration
@Configuration(proxyBeanMethods = false)
static class WebsocketConsumerServerConfiguration {
@Bean
public InMemoryTraceRepository websocketTraceRepository() {
InMemoryTraceRepository websocketTraceRepository() {
return new InMemoryTraceRepository();
}
@Bean
public WebsocketConsumerServer server(WebsocketConsumerProperties properties,
WebsocketConsumerServer server(WebsocketConsumerProperties properties,
WebsocketConsumerServerInitializer initializer) {
return new WebsocketConsumerServer(properties, initializer);
}
@Bean
public WebsocketConsumerServerInitializer initializer(InMemoryTraceRepository websocketTraceRepository) {
WebsocketConsumerServerInitializer initializer(InMemoryTraceRepository websocketTraceRepository) {
return new WebsocketConsumerServerInitializer(websocketTraceRepository);
}

24
consumer/spring-websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2020 the original author or authors.
* Copyright 2014-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.
@@ -21,6 +21,8 @@ import io.netty.handler.logging.LogLevel;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* Configuration properties for WebSocket consumer.
*
* @author Oliver Moser
* @author Gary Russell
*/
@@ -48,34 +50,34 @@ public class WebsocketConsumerProperties {
public static final int DEFAULT_PORT = 9292;
/**
* whether or not to create a {@link io.netty.handler.ssl.SslContext}.
* Whether to create a {@link io.netty.handler.ssl.SslContext}.
*/
boolean ssl;
/**
* the port on which the Netty server listens. Default is <tt>9292</tt>
* The port on which the Netty server listens. Default is <tt>9292</tt>
*/
int port = DEFAULT_PORT;
/**
* the number of threads for the Netty {@link io.netty.channel.EventLoopGroup}.
* The number of threads for the Netty {@link io.netty.channel.EventLoopGroup}.
* Default is <tt>1</tt>
*/
int threads = DEFAULT_THREADS;
/**
* the logLevel for netty channels. Default is <tt>WARN</tt>
* The logLevel for netty channels. Default is <tt>WARN</tt>
*/
String logLevel = DEFAULT_LOGLEVEL;
/**
* the path on which a WebsocketSink consumer needs to connect. Default is
* The path on which a WebsocketSink consumer needs to connect. Default is
* <tt>/websocket</tt>
*/
String path = DEFAULT_PATH;
public boolean isSsl() {
return ssl;
return this.ssl;
}
public void setSsl(boolean ssl) {
@@ -83,7 +85,7 @@ public class WebsocketConsumerProperties {
}
public int getPort() {
return port;
return this.port;
}
public void setPort(int port) {
@@ -91,7 +93,7 @@ public class WebsocketConsumerProperties {
}
public int getThreads() {
return threads;
return this.threads;
}
public void setThreads(int threads) {
@@ -99,7 +101,7 @@ public class WebsocketConsumerProperties {
}
public String getLogLevel() {
return logLevel;
return this.logLevel;
}
public void setLogLevel(String logLevel) {
@@ -107,7 +109,7 @@ public class WebsocketConsumerProperties {
}
public String getPath() {
return path;
return this.path;
}
public void setPath(String path) {

48
consumer/spring-websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-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.
@@ -43,13 +43,13 @@ import org.apache.commons.logging.LogFactory;
*/
public class WebsocketConsumerServer {
private static final Log logger = LogFactory.getLog(WebsocketConsumerServer.class);
private static final Log LOGGER = LogFactory.getLog(WebsocketConsumerServer.class);
static final List<Channel> channels = Collections.synchronizedList(new ArrayList<Channel>());
static final List<Channel> CHANNELS = Collections.synchronizedList(new ArrayList<>());
private WebsocketConsumerProperties properties;
private final WebsocketConsumerProperties properties;
private WebsocketConsumerServerInitializer initializer;
private final WebsocketConsumerServerInitializer initializer;
private EventLoopGroup bossGroup;
@@ -59,6 +59,7 @@ public class WebsocketConsumerServer {
public WebsocketConsumerServer(WebsocketConsumerProperties properties,
WebsocketConsumerServerInitializer initializer) {
this.properties = properties;
this.initializer = initializer;
}
@@ -69,22 +70,23 @@ public class WebsocketConsumerServer {
@PostConstruct
public void init() {
bossGroup = new NioEventLoopGroup(properties.getThreads());
workerGroup = new NioEventLoopGroup();
this.bossGroup = new NioEventLoopGroup(this.properties.getThreads());
this.workerGroup = new NioEventLoopGroup();
}
@PreDestroy
public void shutdown() {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
this.bossGroup.shutdownGracefully();
this.workerGroup.shutdownGracefully();
}
public void run() throws InterruptedException {
NioServerSocketChannel channel = (NioServerSocketChannel) new ServerBootstrap().group(bossGroup, workerGroup)
NioServerSocketChannel channel = (NioServerSocketChannel) new ServerBootstrap()
.group(this.bossGroup, this.workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(nettyLogLevel()))
.childHandler(initializer)
.bind(properties.getPort())
.childHandler(this.initializer)
.bind(this.properties.getPort())
.sync()
.channel();
this.port = channel.localAddress().getPort();
@@ -92,23 +94,23 @@ public class WebsocketConsumerServer {
}
private void dumpProperties() {
logger.info("███████████████████████████████████████████████████████████");
logger.info(" >> websocket-sink config << ");
logger.info("");
logger.info(String.format("port: %s", this.port));
logger.info(String.format("ssl: %s", this.properties.isSsl()));
logger.info(String.format("path: %s", this.properties.getPath()));
logger.info(String.format("logLevel: %s", this.properties.getLogLevel()));
logger.info(String.format("threads: %s", this.properties.getThreads()));
logger.info("");
logger.info("████████████████████████████████████████████████████████████");
LOGGER.info("███████████████████████████████████████████████████████████");
LOGGER.info(" >> websocket-sink config << ");
LOGGER.info("");
LOGGER.info(String.format("port: %s", this.port));
LOGGER.info(String.format("ssl: %s", this.properties.isSsl()));
LOGGER.info(String.format("path: %s", this.properties.getPath()));
LOGGER.info(String.format("logLevel: %s", this.properties.getLogLevel()));
LOGGER.info(String.format("threads: %s", this.properties.getThreads()));
LOGGER.info("");
LOGGER.info("████████████████████████████████████████████████████████████");
}
//
// HELPERS
//
private LogLevel nettyLogLevel() {
return LogLevel.valueOf(properties.getLogLevel().toUpperCase());
return LogLevel.valueOf(this.properties.getLogLevel().toUpperCase());
}
}

36
consumer/spring-websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServerHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2020 the original author or authors.
* Copyright 2014-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.
@@ -28,7 +28,11 @@ import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
@@ -42,12 +46,6 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.fn.consumer.websocket.trace.InMemoryTraceRepository;
import static io.netty.handler.codec.http.HttpHeaderNames.HOST;
import static io.netty.handler.codec.http.HttpMethod.GET;
import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
/**
* Handles handshakes and messages. Based on the Netty
* <a href="https://bit.ly/1jVBj5T">websocket examples</a>.
@@ -59,7 +57,7 @@ import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
*/
public class WebsocketConsumerServerHandler extends SimpleChannelInboundHandler<Object> {
private static final Log logger = LogFactory.getLog(WebsocketConsumerServerHandler.class);
private static final Log LOGGER = LogFactory.getLog(WebsocketConsumerServerHandler.class);
private final boolean traceEnabled;
@@ -95,15 +93,16 @@ public class WebsocketConsumerServerHandler extends SimpleChannelInboundHandler<
private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
// Handle a bad request.
if (!req.decoderResult().isSuccess()) {
logger.warn(String.format("Bad request: %s", req.uri()));
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
LOGGER.warn(String.format("Bad request: %s", req.uri()));
sendHttpResponse(ctx, req,
new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
return;
}
// Allow only GET methods.
if (req.method() != GET) {
logger.warn(String.format("Unsupported HTTP method: %s", req.method()));
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
if (req.method() != HttpMethod.GET) {
LOGGER.warn(String.format("Unsupported HTTP method: %s", req.method()));
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.FORBIDDEN));
return;
}
@@ -122,7 +121,7 @@ public class WebsocketConsumerServerHandler extends SimpleChannelInboundHandler<
}
else {
this.handshaker.handshake(ctx.channel(), req);
WebsocketConsumerServer.channels.add(ctx.channel());
WebsocketConsumerServer.CHANNELS.add(ctx.channel());
}
}
@@ -155,8 +154,8 @@ public class WebsocketConsumerServerHandler extends SimpleChannelInboundHandler<
// simple echo implementation
private void handleTextWebSocketFrameInternal(TextWebSocketFrame frame, ChannelHandlerContext ctx) {
if (logger.isTraceEnabled()) {
logger.trace(String.format("%s received %s", ctx.channel(), frame.text()));
if (LOGGER.isTraceEnabled()) {
LOGGER.trace(String.format("%s received %s", ctx.channel(), frame.text()));
}
addTraceForFrame(frame, "text");
@@ -195,13 +194,12 @@ public class WebsocketConsumerServerHandler extends SimpleChannelInboundHandler<
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
logger.error("Websocket error", cause);
cause.printStackTrace();
LOGGER.error("Websocket error", cause);
ctx.close();
}
private String getWebSocketLocation(FullHttpRequest req) {
String location = req.headers().get(HOST) + this.properties.getPath();
String location = req.headers().get(HttpHeaderNames.HOST) + this.properties.getPath();
if (this.properties.isSsl()) {
return "wss://" + location;
}

6
consumer/spring-websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/actuator/WebsocketConsumerTraceEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2020 the original author or authors.
* Copyright 2014-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.
@@ -39,7 +39,7 @@ import org.springframework.cloud.fn.consumer.websocket.trace.Trace;
@Endpoint(id = "websocketconsumertrace")
public class WebsocketConsumerTraceEndpoint {
private static final Log logger = LogFactory.getLog(WebsocketConsumerTraceEndpoint.class);
private static final Log LOGGER = LogFactory.getLog(WebsocketConsumerTraceEndpoint.class);
private boolean enabled;
@@ -47,7 +47,7 @@ public class WebsocketConsumerTraceEndpoint {
public WebsocketConsumerTraceEndpoint(InMemoryTraceRepository repository) {
this.repository = repository;
logger.info(String.format("/websocketsinktrace enabled: %b", this.enabled));
LOGGER.info(String.format("/websocketsinktrace enabled: %b", this.enabled));
}
@PostConstruct

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

@@ -0,0 +1,4 @@
/**
* The WebSocket consumer actuator support.
*/
package org.springframework.cloud.fn.consumer.websocket.actuator;

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

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

4
consumer/spring-websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/InMemoryTraceRepository.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-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.
@@ -24,7 +24,7 @@ import java.util.Map;
/**
* A repository for {@link Trace}s.
*
* <p>
* It is a copy of {@code InMemoryTraceRepository} from Spring Boot 1.5.x. Since Spring
* Boot 2.0 traces are only available for HTTP.
*

27
consumer/spring-websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/Trace.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-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.
@@ -19,8 +19,6 @@ package org.springframework.cloud.fn.consumer.websocket.trace;
import java.util.Date;
import java.util.Map;
import org.springframework.util.Assert;
/**
* A value object representing a trace event: at a particular time with a simple (map)
* information. Can be used for analyzing contextual information such as HTTP headers.
@@ -29,29 +27,12 @@ import org.springframework.util.Assert;
* It is a copy of {@code InMemoryTraceRepository} from Spring Boot 1.5.x. Since Spring
* Boot 2.0 traces are only available for HTTP.
*
* @param timestamp the time for trace.
* @param info the map of that tags for trace.
* @author Dave Syer
* @author Artem Bilan
* @since 2.0
*/
public class Trace {
private final Date timestamp;
private final Map<String, Object> info;
public Trace(Date timestamp, Map<String, Object> info) {
Assert.notNull(timestamp, "Timestamp must not be null");
Assert.notNull(info, "Info must not be null");
this.timestamp = timestamp;
this.info = info;
}
public Date getTimestamp() {
return this.timestamp;
}
public Map<String, Object> getInfo() {
return this.info;
}
public record Trace(Date timestamp, Map<String, Object> info) {
}

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

@@ -0,0 +1,4 @@
/**
* The WebSocket consumer tracing support.
*/
package org.springframework.cloud.fn.consumer.websocket.trace;

1
consumer/spring-websocket-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.websocket.WebsocketConsumerConfiguration

10
consumer/spring-websocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-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.
@@ -83,7 +83,7 @@ public class WebsocketConsumerTests {
handler.await();
assertThat(handler.getReceivedMessages().size()).isEqualTo(MESSAGE_COUNT);
messagesToSend.forEach(s -> assertThat(handler.getReceivedMessages().contains(s)).isTrue());
messagesToSend.forEach((s) -> assertThat(handler.getReceivedMessages().contains(s)).isTrue());
}
@Test
@@ -112,7 +112,7 @@ public class WebsocketConsumerTests {
// create multiple handlers
List<WebsocketConsumerClientHandler> handlers = createHandlerList(CLIENT_COUNT, MESSAGE_COUNT);
// submit mulitple message
// submit multiple message
List<String> messagesReceived = submitMultipleMessages(MESSAGE_COUNT);
// wait on each handle
@@ -126,12 +126,12 @@ public class WebsocketConsumerTests {
private WebSocketSession doHandshake(WebsocketConsumerClientHandler handler)
throws InterruptedException, ExecutionException {
String wsEndpoint = "ws://localhost:" + this.consumerServer.getPort() + this.properties.getPath();
return new StandardWebSocketClient().doHandshake(handler, wsEndpoint).get();
return new StandardWebSocketClient().execute(handler, wsEndpoint).get();
}
private List<String> submitMultipleMessages(int messageCount) {
List<String> messagesToSend = new ArrayList<>(messageCount);
synchronized (websocketConsumer) {
synchronized (this) {
for (int i = 0; i < messageCount; i++) {
String message = "message_" + i;
messagesToSend.add(message);

2
supplier/spring-time-supplier/README.adoc
View File

@@ -6,7 +6,7 @@ The `Supplier` uses the `FastDateFormat` from Apache Commons library.
## Beans for injection
You can import the `TimeSupplierConfiguration` in the application and then inject the following bean.
The `TimeSupplierConfiguration` auto-configuration provides the following bean:
`timeSupplier`

15
supplier/spring-time-supplier/src/main/java/org/springframework/cloud/fn/supplier/time/DateFormat.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-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.
@@ -41,12 +41,7 @@ import jakarta.validation.Payload;
@Constraint(validatedBy = { DateFormat.DateFormatValidator.class })
public @interface DateFormat {
/**
* Default message for validation.
*/
String DEFAULT_MESSAGE = "";
String message() default DEFAULT_MESSAGE;
String message() default "";
Class<?>[] groups() default {};
@@ -69,10 +64,10 @@ public @interface DateFormat {
try {
new SimpleDateFormat(value.toString());
}
catch (IllegalArgumentException e) {
if (DEFAULT_MESSAGE.equals(this.message)) {
catch (IllegalArgumentException ex) {
if ("".equals(this.message)) {
context.disableDefaultConstraintViolation();
context.buildConstraintViolationWithTemplate(e.getMessage()).addConstraintViolation();
context.buildConstraintViolationWithTemplate(ex.getMessage()).addConstraintViolation();
}
return false;
}

4
supplier/spring-time-supplier/src/main/java/org/springframework/cloud/fn/supplier/time/TimeSupplierConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2023 the original author or authors.
* Copyright 2020-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.boot.context.properties.EnableConfigurationProperties
import org.springframework.context.annotation.Bean;
/**
* Auto-configuration for time supplier.
*
* @author Soby Chacko
* @author Artem Bilan
*/

4
supplier/spring-time-supplier/src/main/java/org/springframework/cloud/fn/supplier/time/TimeSupplierProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-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.
@@ -20,6 +20,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;
/**
* The time supplier properties.
*
* @author Soby Chacko
*/
@ConfigurationProperties("time")

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

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

10
supplier/spring-time-supplier/src/test/java/org/springframework/cloud/fn/supplier/time/SimpleTimeSupplierTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-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.
@@ -17,12 +17,11 @@
package org.springframework.cloud.fn.supplier.time;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatNoException;
/**
* @author Soby Chacko
@@ -34,10 +33,7 @@ public class SimpleTimeSupplierTests extends TimeSupplierApplicationTests {
public void testTimeSupplier() {
final String time = timeSupplier.get();
SimpleDateFormat dateFormat = new SimpleDateFormat(new TimeSupplierProperties().getDateFormat());
assertThatCode(() -> {
Date date = dateFormat.parse(time);
assertThat(date).isNotNull();
}).doesNotThrowAnyException();
assertThatNoException().isThrownBy(() -> assertThat(dateFormat.parse(time)).isNotNull());
}
}

10
supplier/spring-time-supplier/src/test/java/org/springframework/cloud/fn/supplier/time/VariationToSimpleTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-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.
@@ -17,15 +17,14 @@
package org.springframework.cloud.fn.supplier.time;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.assertj.core.api.Assertions.assertThatNoException;
/**
* @author Soby Chacko
@@ -38,10 +37,7 @@ public class VariationToSimpleTests extends TimeSupplierApplicationTests {
public void testTimeSupplier() {
final String time = timeSupplier.get();
SimpleDateFormat dateFormat = new SimpleDateFormat(timeSupplierProperties.getDateFormat());
assertThatCode(() -> {
Date date = dateFormat.parse(time);
assertThat(date).isNotNull();
}).doesNotThrowAnyException();
assertThatNoException().isThrownBy(() -> assertThat(dateFormat.parse(time)).isNotNull());
}
@Test

2
supplier/spring-websocket-supplier/README.adoc
View File

@@ -7,7 +7,7 @@ Users have to subscribe to this `Flux` and receive the data.
## Beans for injection
You can import the `WebsocketSupplierConfiguration` in the application and then inject the following bean.
The `WebsocketSupplierConfiguration` auto-configuration provides the following bean:
`websocketSupplier`

2
supplier/spring-websocket-supplier/build.gradle
View File

@@ -1,7 +1,7 @@
dependencies {
api 'org.springframework.integration:spring-integration-websocket'
api 'org.springframework.boot:spring-boot-starter-web'
testImplementation 'org.springframework.boot:spring-boot-starter-web'
testImplementation 'org.springframework.boot:spring-boot-starter-security'
testImplementation project(':spring-function-test-support')
}

18
supplier/spring-websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2023 the original author or authors.
* Copyright 2018-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.
@@ -23,11 +23,13 @@ import reactor.core.publisher.Flux;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration;
import org.springframework.boot.autoconfigure.websocket.servlet.WebSocketServletAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.websocket.IntegrationWebSocketContainer;
import org.springframework.integration.websocket.ServerWebSocketContainer;
@@ -35,13 +37,13 @@ import org.springframework.integration.websocket.inbound.WebSocketInboundChannel
import org.springframework.messaging.Message;
/**
* A supplier that receives data over WebSocket.
* Auto-configuration for supplier that receives data over WebSocket.
*
* @author Krishnaprasad A S
* @author Artem Bilan
*
*/
@Configuration
@AutoConfiguration(before = { WebSocketServletAutoConfiguration.class, SecurityAutoConfiguration.class })
@EnableConfigurationProperties(WebsocketSupplierProperties.class)
public class WebsocketSupplierConfiguration {
@@ -51,8 +53,9 @@ public class WebsocketSupplierConfiguration {
@Bean
public Supplier<Flux<Message<?>>> websocketSupplier(Publisher<Message<?>> websocketPublisher,
WebSocketInboundChannelAdapter webSocketInboundChannelAdapter) {
return () -> Flux.from(websocketPublisher)
.doOnSubscribe(subscription -> webSocketInboundChannelAdapter.start())
.doOnSubscribe((subscription) -> webSocketInboundChannelAdapter.start())
.doOnTerminate(webSocketInboundChannelAdapter::stop);
}
@@ -63,6 +66,7 @@ public class WebsocketSupplierConfiguration {
private WebSocketInboundChannelAdapter webSocketInboundChannelAdapter(
IntegrationWebSocketContainer serverWebSocketContainer) {
WebSocketInboundChannelAdapter webSocketInboundChannelAdapter = new WebSocketInboundChannelAdapter(
serverWebSocketContainer);
webSocketInboundChannelAdapter.setAutoStartup(false);
@@ -80,7 +84,9 @@ public class WebsocketSupplierConfiguration {
@Bean
public IntegrationWebSocketContainer serverWebSocketContainer(
ObjectProvider<ServerWebSocketContainer.SockJsServiceOptions> sockJsServiceOptions) {
return new ServerWebSocketContainer(properties.getPath()).setAllowedOrigins(properties.getAllowedOrigins())
return new ServerWebSocketContainer(this.properties.getPath())
.setAllowedOrigins(this.properties.getAllowedOrigins())
.withSockJs(sockJsServiceOptions.getIfAvailable());
}

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

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

1
supplier/spring-websocket-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.websocket.WebsocketSupplierConfiguration

11
supplier/spring-websocket-supplier/src/test/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-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.
@@ -18,6 +18,7 @@ package org.springframework.cloud.fn.supplier.websocket;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.function.Supplier;
import org.junit.jupiter.api.Test;
@@ -33,7 +34,6 @@ import org.springframework.http.HttpHeaders;
import org.springframework.integration.websocket.ClientWebSocketContainer;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.util.Base64Utils;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
@@ -77,9 +77,10 @@ public class WebsocketSupplierTests {
"ws://localhost:{port}/{path}", this.port, this.properties.getPath());
HttpHeaders httpHeaders = new HttpHeaders();
String token = Base64Utils.encodeToString(
(this.securityProperties.getUser().getName() + ":" + this.securityProperties.getUser().getPassword())
.getBytes(StandardCharsets.UTF_8));
String token = Base64.getEncoder()
.encodeToString((this.securityProperties.getUser().getName() + ":"
+ this.securityProperties.getUser().getPassword())
.getBytes(StandardCharsets.UTF_8));
httpHeaders.set(HttpHeaders.AUTHORIZATION, "Basic " + token);
clientWebSocketContainer.setHeaders(httpHeaders);
clientWebSocketContainer.start();