diff --git a/applications/source/pom.xml b/applications/source/pom.xml index 38bdcfc0..ab76b0fd 100644 --- a/applications/source/pom.xml +++ b/applications/source/pom.xml @@ -29,5 +29,6 @@ twitter-message-source websocket-source cdc-debezium-source + syslog-source diff --git a/applications/source/syslog-source/README.adoc b/applications/source/syslog-source/README.adoc new file mode 100644 index 00000000..685fb4c9 --- /dev/null +++ b/applications/source/syslog-source/README.adoc @@ -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: `$$$$`, 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[] diff --git a/applications/source/syslog-source/pom.xml b/applications/source/syslog-source/pom.xml new file mode 100644 index 00000000..2a8e8335 --- /dev/null +++ b/applications/source/syslog-source/pom.xml @@ -0,0 +1,128 @@ + + + 4.0.0 + syslog-source + 3.0.0-SNAPSHOT + syslog-source + syslog source apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.cloud.fn + syslog-supplier + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + syslog + source + ${project.version} + org.springframework.cloud.fn.supplier.syslog.SyslogSupplierConfiguration.class + + + + org.springframework.cloud.fn + syslog-supplier + + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${stream-apps-core.version} + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + false + + spring-releases + Spring Releases + https://repo.spring.io/release + + + + false + + spring-libs-release + Spring Libs Release + https://repo.spring.io/libs-release + + + + false + + spring-milestone-release + Spring Milestone Release + https://repo.spring.io/libs-milestone + + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/source/syslog-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/syslog-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..60491dd5 --- /dev/null +++ b/applications/source/syslog-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,2 @@ +configuration-properties.classes=org.springframework.cloud.fn.supplier.syslog.SyslogSupplierProperties + diff --git a/applications/source/syslog-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/source/syslog-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties new file mode 100644 index 00000000..60491dd5 --- /dev/null +++ b/applications/source/syslog-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -0,0 +1,2 @@ +configuration-properties.classes=org.springframework.cloud.fn.supplier.syslog.SyslogSupplierProperties + diff --git a/applications/source/syslog-source/src/test/java/org/springframework/cloud/stream/app/source/syslog/SyslogSourceTests.java b/applications/source/syslog-source/src/test/java/org/springframework/cloud/stream/app/source/syslog/SyslogSourceTests.java new file mode 100644 index 00000000..f70e61ca --- /dev/null +++ b/applications/source/syslog-source/src/test/java/org/springframework/cloud/stream/app/source/syslog/SyslogSourceTests.java @@ -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 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 { + } +} diff --git a/functions/function-dependencies/pom.xml b/functions/function-dependencies/pom.xml index 6b4a0d6b..cceac754 100644 --- a/functions/function-dependencies/pom.xml +++ b/functions/function-dependencies/pom.xml @@ -15,6 +15,11 @@ + + org.springframework.cloud.fn + syslog-supplier + ${project.version} + org.springframework.cloud.fn cdc-debezium-supplier diff --git a/functions/pom.xml b/functions/pom.xml index b53fcb19..a90e612a 100644 --- a/functions/pom.xml +++ b/functions/pom.xml @@ -101,6 +101,7 @@ supplier/s3-supplier supplier/twitter-supplier supplier/cdc-debezium-supplier + supplier/syslog-supplier spring-functions-parent function-dependencies diff --git a/functions/supplier/syslog-supplier/README.adoc b/functions/supplier/syslog-supplier/README.adoc new file mode 100644 index 00000000..85cd3b0e --- /dev/null +++ b/functions/supplier/syslog-supplier/README.adoc @@ -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>>`. +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>>`. + +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. \ No newline at end of file diff --git a/functions/supplier/syslog-supplier/pom.xml b/functions/supplier/syslog-supplier/pom.xml new file mode 100644 index 00000000..f3653f73 --- /dev/null +++ b/functions/supplier/syslog-supplier/pom.xml @@ -0,0 +1,50 @@ + + + 4.0.0 + syslog-supplier + 1.0.0-SNAPSHOT + syslog-supplier + syslog supplier + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.integration + spring-integration-syslog + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.integration + spring-integration-test-support + + + io.projectreactor + reactor-test + test + + + + diff --git a/functions/supplier/syslog-supplier/src/main/java/org/springframework/cloud/fn/supplier/syslog/SyslogSupplierConfiguration.java b/functions/supplier/syslog-supplier/src/main/java/org/springframework/cloud/fn/supplier/syslog/SyslogSupplierConfiguration.java new file mode 100644 index 00000000..af81e090 --- /dev/null +++ b/functions/supplier/syslog-supplier/src/main/java/org/springframework/cloud/fn/supplier/syslog/SyslogSupplierConfiguration.java @@ -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>> syslogSupplier(ObjectProvider udpApapterProvider, + ObjectProvider 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 { + + } + +} diff --git a/functions/supplier/syslog-supplier/src/main/java/org/springframework/cloud/fn/supplier/syslog/SyslogSupplierProperties.java b/functions/supplier/syslog-supplier/src/main/java/org/springframework/cloud/fn/supplier/syslog/SyslogSupplierProperties.java new file mode 100644 index 00000000..fa085111 --- /dev/null +++ b/functions/supplier/syslog-supplier/src/main/java/org/springframework/cloud/fn/supplier/syslog/SyslogSupplierProperties.java @@ -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; + } +} diff --git a/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/AbstractSyslogSupplierTests.java b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/AbstractSyslogSupplierTests.java new file mode 100644 index 00000000..599f1581 --- /dev/null +++ b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/AbstractSyslogSupplierTests.java @@ -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>> 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 { + + } + +} diff --git a/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/NotNioTests.java b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/NotNioTests.java new file mode 100644 index 00000000..fd485d96 --- /dev/null +++ b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/NotNioTests.java @@ -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); + } +} diff --git a/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/PropertiesPopulatedTests.java b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/PropertiesPopulatedTests.java new file mode 100644 index 00000000..5e2c7dbb --- /dev/null +++ b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/PropertiesPopulatedTests.java @@ -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); + } +} diff --git a/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Tcp3164Tests.java b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Tcp3164Tests.java new file mode 100644 index 00000000..648a4130 --- /dev/null +++ b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Tcp3164Tests.java @@ -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> 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(); + } +} diff --git a/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Tcp5424Tests.java b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Tcp5424Tests.java new file mode 100644 index 00000000..a0934956 --- /dev/null +++ b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Tcp5424Tests.java @@ -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> 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(); + } +} diff --git a/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/TcpAndUdp3164Tests.java b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/TcpAndUdp3164Tests.java new file mode 100644 index 00000000..da19ce97 --- /dev/null +++ b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/TcpAndUdp3164Tests.java @@ -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> 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(); + } +} diff --git a/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/TcpAndUdp5424Tests.java b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/TcpAndUdp5424Tests.java new file mode 100644 index 00000000..394ce58c --- /dev/null +++ b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/TcpAndUdp5424Tests.java @@ -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> 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(); + } + +} diff --git a/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Udp3164Tests.java b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Udp3164Tests.java new file mode 100644 index 00000000..ab1ccf6a --- /dev/null +++ b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Udp3164Tests.java @@ -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> 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(); + + + } + +} diff --git a/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Udp5424Tests.java b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Udp5424Tests.java new file mode 100644 index 00000000..c64cf307 --- /dev/null +++ b/functions/supplier/syslog-supplier/src/test/java/org/springframework/cloud/fn/supplier/syslog/Udp5424Tests.java @@ -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> 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(); + } +}