SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Syslog supplier/source

Browse Source 
- Provide TCP/UDP based syslog supplier and then generate apps based on that
This commit is contained in:
Soby Chacko
2020-08-18 15:37:50 -04:00
committed by GitHub
parent 19353bf7d9
commit 71ec62b58e
21 changed files with 1125 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
applications/source/pom.xml
View File

@@ -29,5 +29,6 @@
<module>twitter-message-source</module>
<module>websocket-source</module>
<module>cdc-debezium-source</module>
<module>syslog-source</module>
</modules>
</project>

18
applications/source/syslog-source/README.adoc Normal file
View File

@@ -0,0 +1,18 @@
//tag::ref-doc[]
= SYSLOG
The syslog source receives SYSLOG packets over UDP, TCP, or both. RFC3164 (BSD) and RFC5424 formats are supported.
== Options
//tag::configuration-properties[]
$$syslog.supplier.buffer-size$$:: $$the buffer size used when decoding messages; larger messages will be rejected.$$ *($$Integer$$, default: `$$2048$$`)*
$$syslog.supplier.nio$$:: $$whether or not to use NIO (when supporting a large number of connections).$$ *($$Boolean$$, default: `$$false$$`)*
$$syslog.supplier.port$$:: $$The port to listen on.$$ *($$Integer$$, default: `$$1514$$`)*
$$syslog.supplier.protocol$$:: $$Protocol used for SYSLOG (tcp or udp).$$ *($$Protocol$$, default: `$$<none>$$`, possible values: `tcp`,`udp`,`both`)*
$$syslog.supplier.reverse-lookup$$:: $$whether or not to perform a reverse lookup on the incoming socket.$$ *($$Boolean$$, default: `$$false$$`)*
$$syslog.supplier.rfc$$:: $$'5424' or '3164' - the syslog format according the the RFC; 3164 is aka 'BSD' format.$$ *($$String$$, default: `$$3164$$`)*
$$syslog.supplier.socket-timeout$$:: $$the socket timeout.$$ *($$Integer$$, default: `$$0$$`)*
//end::configuration-properties[]
//end::ref-doc[]

128
applications/source/syslog-source/pom.xml Normal file
View File

@@ -0,0 +1,128 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>syslog-source</artifactId>
<version>3.0.0-SNAPSHOT</version>
<name>syslog-source</name>
<description>syslog source apps</description>
<packaging>jar</packaging>
<parent>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-core</artifactId>
<version>3.0.0-SNAPSHOT</version>
<relativePath/>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>syslog-supplier</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-app-starter-doc-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.springframework.cloud.stream.app.plugin</groupId>
<artifactId>spring-cloud-stream-app-maven-plugin</artifactId>
<configuration>
<generatedApp>
<name>syslog</name>
<type>source</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.supplier.syslog.SyslogSupplierConfiguration.class</configClass>
</generatedApp>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>syslog-supplier</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-composite-function-support</artifactId>
<version>${stream-apps-core.version}</version>
</dependency>
</dependencies>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/release</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-libs-release</id>
<name>Spring Libs Release</name>
<url>https://repo.spring.io/libs-release</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestone-release</id>
<name>Spring Milestone Release</name>
<url>https://repo.spring.io/libs-milestone</url>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release</url>
</pluginRepository>
<pluginRepository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</pluginRepository>
<pluginRepository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</pluginRepository>
</pluginRepositories>
</project>

2
applications/source/syslog-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties Normal file
View File

@@ -0,0 +1,2 @@
configuration-properties.classes=org.springframework.cloud.fn.supplier.syslog.SyslogSupplierProperties

2
applications/source/syslog-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties Normal file
View File

@@ -0,0 +1,2 @@
configuration-properties.classes=org.springframework.cloud.fn.supplier.syslog.SyslogSupplierProperties

87
applications/source/syslog-source/src/test/java/org/springframework/cloud/stream/app/source/syslog/SyslogSourceTests.java Normal file
View File

@@ -0,0 +1,87 @@
/*
* Copyright 2020-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.app.source.syslog;
import java.net.Socket;
import java.util.Map;
import javax.net.SocketFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.fn.supplier.syslog.SyslogSupplierConfiguration;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Import;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.messaging.Message;
import static org.assertj.core.api.Assertions.assertThat;
public class SyslogSourceTests {
private static final String RFC3164_PACKET = "<157>JUL 26 22:08:35 WEBERN TESTING[70729]: TEST SYSLOG MESSAGE";
@Test
public void testBasicSyslogSourceWithBinder() throws Exception {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration
.getCompleteConfiguration(SyslogSourceTestApplication.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.function.definition=syslogSupplier", "--syslog.supplier.port=0")) {
AbstractServerConnectionFactory connectionFactory = context.getBean(AbstractServerConnectionFactory.class);
sendTcp(RFC3164_PACKET + "\n", connectionFactory);
OutputDestination target = context.getBean(OutputDestination.class);
Message<byte[]> sourceMessage = target.receive(10000);
String actual = new String(sourceMessage.getPayload());
final Map map = new ObjectMapper().readValue(actual, Map.class);
assertThat(map.get("HOST")).isEqualTo("WEBERN");
}
}
private void sendTcp(String syslog, AbstractServerConnectionFactory connectionFactory) throws Exception {
int port = getPort(connectionFactory);
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write(syslog.getBytes());
socket.close();
}
private int getPort(AbstractServerConnectionFactory connectionFactory) throws Exception {
int n = 0;
while (n++ < 100 && !connectionFactory.isListening()) {
Thread.sleep(100);
}
assertThat(connectionFactory.isListening()).isTrue();
int port = connectionFactory.getPort();
assertThat(port > 0).isTrue();
return port;
}
@SpringBootApplication
@Import(SyslogSupplierConfiguration.class)
public static class SyslogSourceTestApplication {
}
}

5
functions/function-dependencies/pom.xml
View File

@@ -15,6 +15,11 @@
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>syslog-supplier</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>cdc-debezium-supplier</artifactId>

1
functions/pom.xml
View File

@@ -101,6 +101,7 @@
<module>supplier/s3-supplier</module>
<module>supplier/twitter-supplier</module>
<module>supplier/cdc-debezium-supplier</module>
<module>supplier/syslog-supplier</module>
<module>spring-functions-parent</module>
<module>function-dependencies</module>

32
functions/supplier/syslog-supplier/README.adoc Normal file
View File

@@ -0,0 +1,32 @@
# Syslog Supplier
Syslog supplier that produces both TCP and UDP based syslog events.
The `Supplier` uses the `TcpSyslogReceivingChannelAdapter` and `UdpSyslogReceivingChannelAdapter` from Spring Integration.
This supplier gives you a reactive stream of messages and the supplier has a signature of `Supplier<Flux<Message<?>>>`.
Users have to subscribe to this `Flux` and receive the data.
## Beans for injection
You can import the `SyslogSupplierConfiguration` in the application and then inject the following bean.
`syslogSupplier`
You need to inject this as `Supplier<Flux<Message<?>>>`.
You can use `syslogSupplier` as a qualifier when injecting.
Once injected, you can use the `get` method of the `Supplier` to invoke it and then subscribe to the returned `Flux`.
## Configuration Options
All configuration properties are prefixed with `syslog.supplier`.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/syslog/SyslogSupplierProperties.java[SyslogSupplierProperties].
## Tests
See this link:src/test/java/org/springframework/cloud/fn/supplier/syslog[test suite] for the various ways, this supplier is used.
## Other usage
See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/syslog-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application where it makes a Syslog Source.

50
functions/supplier/syslog-supplier/pom.xml Normal file
View File

@@ -0,0 +1,50 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>syslog-supplier</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>syslog-supplier</name>
<description>syslog supplier</description>
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>1.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent</relativePath>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-syslog</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-test-support</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

177
functions/supplier/syslog-supplier/src/main/java/org/springframework/cloud/fn/supplier/syslog/SyslogSupplierConfiguration.java Normal file
View File

@@ -0,0 +1,177 @@
/*
* Copyright 2020-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import java.util.function.Supplier;
import reactor.core.publisher.Flux;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.serializer.Deserializer;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
import org.springframework.integration.ip.tcp.serializer.ByteArrayLfSerializer;
import org.springframework.integration.syslog.DefaultMessageConverter;
import org.springframework.integration.syslog.MessageConverter;
import org.springframework.integration.syslog.RFC5424MessageConverter;
import org.springframework.integration.syslog.inbound.RFC6587SyslogDeserializer;
import org.springframework.integration.syslog.inbound.SyslogReceivingChannelAdapterSupport;
import org.springframework.integration.syslog.inbound.TcpSyslogReceivingChannelAdapter;
import org.springframework.integration.syslog.inbound.UdpSyslogReceivingChannelAdapter;
import org.springframework.messaging.Message;
/**
* Configuration class for SYSLOG Supplier.
*
* @author Soby Chacko
*/
@Configuration
@EnableConfigurationProperties(SyslogSupplierProperties.class)
public class SyslogSupplierConfiguration {
@Autowired
private SyslogSupplierProperties properties;
@Bean
public FluxMessageChannel output() {
return new FluxMessageChannel();
}
@Bean
public Supplier<Flux<Message<?>>> syslogSupplier(ObjectProvider<UdpSyslogReceivingChannelAdapter> udpApapterProvider,
ObjectProvider<TcpSyslogReceivingChannelAdapter> tcpAdapterProvider) {
return () -> Flux.from(output())
.doOnSubscribe(subscription -> {
final UdpSyslogReceivingChannelAdapter udpAdapter = udpApapterProvider.getIfAvailable();
final TcpSyslogReceivingChannelAdapter tcpAdapter = tcpAdapterProvider.getIfAvailable();
if (udpAdapter != null) {
udpAdapter.start();
}
if (tcpAdapter != null) {
tcpAdapter.start();
}
});
}
@Bean
@ConditionalOnProperty(name = "syslog.supplier.protocol", havingValue = "udp")
public UdpSyslogReceivingChannelAdapter udpAdapter() {
return createUdpAdapter();
}
@Bean
@ConditionalOnProperty(name = "syslog.supplier.protocol", havingValue = "both")
public UdpSyslogReceivingChannelAdapter udpBothAdapter() {
return createUdpAdapter();
}
private UdpSyslogReceivingChannelAdapter createUdpAdapter() {
UdpSyslogReceivingChannelAdapter adapter = new UdpSyslogReceivingChannelAdapter();
setAdapterProperties(adapter);
return adapter;
}
@Bean
@ConditionalOnProperty(name = "syslog.supplier.protocol", havingValue = "tcp", matchIfMissing = true)
public TcpSyslogReceivingChannelAdapter tcpAdapter(
@Qualifier("syslogSupplierConnectionFactory") AbstractServerConnectionFactory connectionFactory) {
return createTcpAdapter(connectionFactory);
}
@Bean
@ConditionalOnProperty(name = "syslog.supplier.protocol", havingValue = "both")
public TcpSyslogReceivingChannelAdapter tcpBothAdapter(
@Qualifier("syslogSupplierConnectionFactory") AbstractServerConnectionFactory connectionFactory) {
return createTcpAdapter(connectionFactory);
}
@Bean
public MessageConverter syslogConverter() {
if (this.properties.getRfc().equals("5424")) {
return new RFC5424MessageConverter();
}
else {
return new DefaultMessageConverter();
}
}
private TcpSyslogReceivingChannelAdapter createTcpAdapter(AbstractServerConnectionFactory connectionFactory) {
TcpSyslogReceivingChannelAdapter adapter = new TcpSyslogReceivingChannelAdapter();
adapter.setConnectionFactory(connectionFactory);
setAdapterProperties(adapter);
return adapter;
}
private void setAdapterProperties(SyslogReceivingChannelAdapterSupport adapter) {
adapter.setPort(this.properties.getPort());
adapter.setConverter(syslogConverter());
adapter.setOutputChannel(output());
adapter.setAutoStartup(false);
}
@Configuration
@ConditionalOnProperty(name = "syslog.supplier.protocol", havingValue = "tcp", matchIfMissing = true)
protected static class TcpBits {
@Autowired
private SyslogSupplierProperties properties;
@Bean
public AbstractServerConnectionFactory syslogSupplierConnectionFactory(
@Qualifier("syslogSupplierDecoder") Deserializer<?> decoder) throws Exception {
AbstractServerConnectionFactory factory;
if (this.properties.isNio()) {
factory = new TcpNioServerConnectionFactory(this.properties.getPort());
}
else {
factory = new TcpNetServerConnectionFactory(this.properties.getPort());
}
factory.setLookupHost(this.properties.isReverseLookup());
factory.setDeserializer(decoder);
factory.setSoTimeout(this.properties.getSocketTimeout());
return factory;
}
@Bean
public Deserializer<?> syslogSupplierDecoder() {
ByteArrayLfSerializer decoder = new ByteArrayLfSerializer();
decoder.setMaxMessageSize(this.properties.getBufferSize());
if (this.properties.getRfc().equals("5424")) {
return new RFC6587SyslogDeserializer(decoder);
}
else {
return decoder;
}
}
}
@Configuration
@ConditionalOnProperty(name = "syslog.supplier.protocol", havingValue = "both")
protected static class BothBits extends TcpBits {
}
}

142
functions/supplier/syslog-supplier/src/main/java/org/springframework/cloud/fn/supplier/syslog/SyslogSupplierProperties.java Normal file
View File

@@ -0,0 +1,142 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import javax.validation.constraints.AssertTrue;
import javax.validation.constraints.NotNull;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;
@ConfigurationProperties("syslog.supplier")
@Validated
public class SyslogSupplierProperties {
/**
* the buffer size used when decoding messages; larger messages will be rejected.
*/
private int bufferSize = 2048;
/**
* Protocol used for SYSLOG (tcp or udp).
*/
private Protocol protocol = Protocol.tcp;
/**
* The port to listen on.
*/
private int port = 1514;
/**
* whether or not to use NIO (when supporting a large number of connections).
*/
private boolean nio = false;
/**
* whether or not to perform a reverse lookup on the incoming socket.
*/
private boolean reverseLookup;
/**
* the socket timeout.
*/
private int socketTimeout;
/**
* '5424' or '3164' - the syslog format according to the RFC; 3164 is aka 'BSD' format.
*/
private String rfc = "3164";
public int getBufferSize() {
return bufferSize;
}
public void setBufferSize(int bufferSize) {
this.bufferSize = bufferSize;
}
public Protocol getProtocol() {
return protocol;
}
public void setProtocol(Protocol protocol) {
this.protocol = protocol;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
public boolean isNio() {
return nio;
}
public void setNio(boolean nio) {
this.nio = nio;
}
public boolean isReverseLookup() {
return reverseLookup;
}
public void setReverseLookup(boolean reverseLookup) {
this.reverseLookup = reverseLookup;
}
public int getSocketTimeout() {
return socketTimeout;
}
public void setSocketTimeout(int socketTimeout) {
this.socketTimeout = socketTimeout;
}
@NotNull
public String getRfc() {
return rfc;
}
public void setRfc(String rfc) {
this.rfc = rfc;
}
@AssertTrue(message = "rfc must be 5424 or 3164")
public boolean isSupportedRfc() {
return "5424".equals(this.rfc) || "3164".equals(this.rfc);
}
public enum Protocol {
/**
* TCP protocol.
*/
tcp,
/**
* UDP protocol.
*/
udp,
/**
* Represents both TCP and UDP.
*/
both;
}
}

107
functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/AbstractSyslogSupplierTests.java Normal file
View File

@@ -0,0 +1,107 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.function.Supplier;
import javax.net.SocketFactory;
import reactor.core.publisher.Flux;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
import org.springframework.integration.syslog.inbound.UdpSyslogReceivingChannelAdapter;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
import static org.assertj.core.api.Assertions.assertThat;
@SpringBootTest(properties = "syslog.supplier.port = 0")
@DirtiesContext
public class AbstractSyslogSupplierTests {
protected static final String RFC3164_PACKET = "<157>JUL 26 22:08:35 WEBERN TESTING[70729]: TEST SYSLOG MESSAGE";
protected static final String RFC5424_PACKET =
"<14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - "
+ "[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"]"
+ "[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] "
+ "Removing instance";
@Autowired
Supplier<Flux<Message<?>>> syslogSupplier;
@Autowired
protected SyslogSupplierProperties properties;
@Autowired(required = false)
protected AbstractServerConnectionFactory connectionFactory;
@Autowired(required = false)
protected UdpSyslogReceivingChannelAdapter udpAdapter;
protected void sendTcp(String syslog) throws Exception {
int port = getPort();
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write(syslog.getBytes());
socket.close();
}
private int getPort() throws Exception {
int n = 0;
while (n++ < 100 && !this.connectionFactory.isListening()) {
Thread.sleep(100);
}
assertThat(this.connectionFactory.isListening()).isTrue();
int port = this.connectionFactory.getPort();
assertThat(port > 0).isTrue();
return port;
}
protected void sendUdp(String syslog) throws Exception {
int port = waitUdp();
DatagramSocket socket = new DatagramSocket();
DatagramPacket packet = new DatagramPacket(syslog.getBytes(), syslog.length());
packet.setSocketAddress(new InetSocketAddress("localhost", port));
socket.send(packet);
socket.close();
}
private int waitUdp() throws Exception {
int n = 0;
DirectFieldAccessor dfa = new DirectFieldAccessor(this.udpAdapter);
while (n++ < 100 && !((UnicastReceivingChannelAdapter) dfa.getPropertyValue("udpAdapter")).isListening()) {
Thread.sleep(100);
}
return ((UnicastReceivingChannelAdapter) dfa.getPropertyValue("udpAdapter")).getPort();
}
@SpringBootApplication
public static class SyslogSupplierTestApplication {
}
}

35
functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/NotNioTests.java Normal file
View File

@@ -0,0 +1,35 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import org.junit.jupiter.api.Test;
import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
import org.springframework.integration.test.util.TestUtils;
import static org.assertj.core.api.Assertions.assertThat;
public class NotNioTests extends AbstractSyslogSupplierTests {
@Test
public void test() throws Exception {
assertThat(this.connectionFactory).isInstanceOf(TcpNetServerConnectionFactory.class);
assertThat(TestUtils.getPropertyValue(this.connectionFactory, "lookupHost", Boolean.class)).isFalse();
assertThat(TestUtils.getPropertyValue(this.connectionFactory, "soTimeout")).isEqualTo(0);
assertThat(TestUtils.getPropertyValue(this.connectionFactory, "deserializer.maxMessageSize")).isEqualTo(2048);
}
}

38
functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/PropertiesPopulatedTests.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import org.junit.jupiter.api.Test;
import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.TestPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
@TestPropertySource(properties = { "syslog.supplier.port = 0", "syslog.supplier.nio = true", "syslog.supplier.reverseLookup = true",
"syslog.supplier.socketTimeout = 123", "syslog.supplier.bufferSize = 5" })
public class PropertiesPopulatedTests extends AbstractSyslogSupplierTests {
@Test
public void test() throws Exception {
assertThat(this.connectionFactory).isInstanceOf(TcpNioServerConnectionFactory.class);
assertThat(TestUtils.getPropertyValue(this.connectionFactory, "lookupHost", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(this.connectionFactory, "soTimeout")).isEqualTo(123);
assertThat(TestUtils.getPropertyValue(this.connectionFactory, "deserializer.maxMessageSize")).isEqualTo(5);
}
}

47
functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Tcp3164Tests.java Normal file
View File

@@ -0,0 +1,47 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import java.util.Map;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.messaging.Message;
import static org.assertj.core.api.Assertions.assertThat;
public class Tcp3164Tests extends AbstractSyslogSupplierTests {
@Test
public void test() throws Exception {
final Flux<Message<?>> messageFlux = syslogSupplier.get();
final StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext((message) -> {
assertThat(((Map) message.getPayload()).get("HOST")).isEqualTo("WEBERN");
}
)
.thenCancel()
.verifyLater();
sendTcp(AbstractSyslogSupplierTests.RFC3164_PACKET + "\n");
stepVerifier.verify();
}
}

49
functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Tcp5424Tests.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import java.util.Map;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.messaging.Message;
import org.springframework.test.context.TestPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
@TestPropertySource(properties = { "syslog.supplier.port = 0", "syslog.supplier.rfc = 5424" })
public class Tcp5424Tests extends AbstractSyslogSupplierTests {
@Test
public void test() throws Exception {
final Flux<Message<?>> messageFlux = syslogSupplier.get();
final StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext((message) -> {
assertThat(((Map) message.getPayload()).get("syslog_HOST")).isEqualTo("loggregator");
}
)
.thenCancel()
.verifyLater();
sendTcp("253 " + RFC5424_PACKET);
stepVerifier.verify();
}
}

52
functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/TcpAndUdp3164Tests.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import java.util.Map;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.messaging.Message;
import org.springframework.test.context.TestPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
@TestPropertySource(properties = "syslog.supplier.protocol = both")
public class TcpAndUdp3164Tests extends AbstractSyslogSupplierTests {
@Test
public void test() throws Exception {
final Flux<Message<?>> messageFlux = syslogSupplier.get();
final StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext((message) -> {
assertThat(((Map) message.getPayload()).get("HOST")).isEqualTo("WEBERN");
}
)
.assertNext((message) -> {
assertThat(((Map) message.getPayload()).get("HOST")).isEqualTo("WEBERN");
}
)
.thenCancel()
.verifyLater();
sendTcp(RFC3164_PACKET + "\n");
sendUdp(RFC3164_PACKET);
stepVerifier.verify();
}
}

53
functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/TcpAndUdp5424Tests.java Normal file
View File

@@ -0,0 +1,53 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import java.util.Map;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.messaging.Message;
import org.springframework.test.context.TestPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
@TestPropertySource(properties = { "syslog.supplier.protocol = both", "syslog.supplier.rfc = 5424" })
public class TcpAndUdp5424Tests extends AbstractSyslogSupplierTests {
@Test
public void test() throws Exception {
final Flux<Message<?>> messageFlux = syslogSupplier.get();
final StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext((message) -> {
assertThat(((Map) message.getPayload()).get("syslog_HOST")).isEqualTo("loggregator");
}
)
.assertNext((message) -> {
assertThat(((Map) message.getPayload()).get("syslog_HOST")).isEqualTo("loggregator");
}
)
.thenCancel()
.verifyLater();
sendTcp("253 " + RFC5424_PACKET);
sendUdp(RFC5424_PACKET);
stepVerifier.verify();
}
}

52
functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Udp3164Tests.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import java.util.Map;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.messaging.Message;
import org.springframework.test.context.TestPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
@TestPropertySource(properties = "syslog.supplier.protocol = udp")
public class Udp3164Tests extends AbstractSyslogSupplierTests {
@Test
public void test() throws Exception {
final Flux<Message<?>> messageFlux = syslogSupplier.get();
final StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext((message) -> {
assertThat(((Map) message.getPayload()).get("HOST")).isEqualTo("WEBERN");
}
)
.thenCancel()
.verifyLater();
sendUdp(RFC3164_PACKET);
stepVerifier.verify();
}
}

47
functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Udp5424Tests.java Normal file
View File

@@ -0,0 +1,47 @@
/*
* Copyright 2016-2020 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.syslog;
import java.util.Map;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.messaging.Message;
import org.springframework.test.context.TestPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
@TestPropertySource(properties = { "syslog.supplier.protocol = udp", "syslog.supplier.rfc = 5424" })
public class Udp5424Tests extends AbstractSyslogSupplierTests {
@Test
public void test() throws Exception {
final Flux<Message<?>> messageFlux = syslogSupplier.get();
final StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext((message) -> {
assertThat(((Map) message.getPayload()).get("syslog_HOST")).isEqualTo("loggregator");
}
)
.thenCancel()
.verifyLater();
sendUdp(RFC5424_PACKET);
stepVerifier.verify();
}
}