This is an automated email from the ASF dual-hosted git repository.
pvillard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 67ccdf6 NIFI-9326 Added Socket Keep Alive property to ListenSyslog
67ccdf6 is described below
commit 67ccdf615923b9bd7dced04ddf5ce8de46d6ae11
Author: exceptionfactory <[email protected]>
AuthorDate: Mon Oct 25 15:11:00 2021 -0500
NIFI-9326 Added Socket Keep Alive property to ListenSyslog
Signed-off-by: Pierre Villard <[email protected]>
This closes #5479.
---
.../event/transport/netty/NettyEventServerFactory.java | 18 ++++++++++++++++--
.../apache/nifi/processors/standard/ListenSyslog.java | 14 ++++++++++++++
.../nifi/processors/standard/TestListenSyslog.java | 1 +
3 files changed, 31 insertions(+), 2 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventServerFactory.java
b/nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventServerFactory.java
index 4e99010..753844f 100644
---
a/nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventServerFactory.java
+++
b/nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventServerFactory.java
@@ -58,6 +58,8 @@ public class NettyEventServerFactory extends
EventLoopGroupFactory implements Ev
private Integer socketReceiveBuffer;
+ private Boolean socketKeepAlive;
+
private SSLContext sslContext;
private ClientAuth clientAuth = ClientAuth.NONE;
@@ -82,6 +84,15 @@ public class NettyEventServerFactory extends
EventLoopGroupFactory implements Ev
}
/**
+ * Set Socket Keep Alive for TCP Sockets
+ *
+ * @param socketKeepAlive Keep Alive can be null to use default setting
+ */
+ public void setSocketKeepAlive(final Boolean socketKeepAlive) {
+ this.socketKeepAlive = socketKeepAlive;
+ }
+
+ /**
* Set Socket Receive Buffer Size for TCP Sockets
*
* @param socketReceiveBuffer Receive Buffer size can be null to use
default setting
@@ -134,17 +145,20 @@ public class NettyEventServerFactory extends
EventLoopGroupFactory implements Ev
@Override
public EventServer getEventServer() {
final AbstractBootstrap<?, ?> bootstrap = getBootstrap();
- setBufferSize(bootstrap);
+ setChannelOptions(bootstrap);
final EventLoopGroup group = getEventLoopGroup();
bootstrap.group(group);
return getBoundEventServer(bootstrap, group);
}
- private void setBufferSize(AbstractBootstrap<?, ?> bootstrap) {
+ private void setChannelOptions(final AbstractBootstrap<?, ?> bootstrap) {
if (socketReceiveBuffer != null) {
bootstrap.option(ChannelOption.SO_RCVBUF, socketReceiveBuffer);
bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new
FixedRecvByteBufAllocator(socketReceiveBuffer));
}
+ if (socketKeepAlive != null) {
+ bootstrap.option(ChannelOption.SO_KEEPALIVE, socketKeepAlive);
+ }
}
private AbstractBootstrap<?, ?> getBootstrap() {
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java
index b3b6688..1927b9a 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenSyslog.java
@@ -183,6 +183,16 @@ public class ListenSyslog extends AbstractSyslogProcessor {
.defaultValue(ClientAuth.REQUIRED.name())
.dependsOn(SSL_CONTEXT_SERVICE)
.build();
+ public static final PropertyDescriptor SOCKET_KEEP_ALIVE = new
PropertyDescriptor.Builder()
+ .name("socket-keep-alive")
+ .displayName("Socket Keep Alive")
+ .description("Whether or not to have TCP socket keep alive turned
on. Timing details depend on operating system properties.")
+ .required(true)
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .allowableValues(Boolean.TRUE.toString(), Boolean.FALSE.toString())
+ .defaultValue(Boolean.FALSE.toString())
+ .dependsOn(PROTOCOL, TCP_VALUE)
+ .build();
public static final Relationship REL_SUCCESS = new Relationship.Builder()
.name("success")
@@ -211,6 +221,7 @@ public class ListenSyslog extends AbstractSyslogProcessor {
descriptors.add(PROTOCOL);
descriptors.add(PORT);
descriptors.add(NETWORK_INTF_NAME);
+ descriptors.add(SOCKET_KEEP_ALIVE);
descriptors.add(SSL_CONTEXT_SERVICE);
descriptors.add(CLIENT_AUTH);
descriptors.add(RECV_BUFFER_SIZE);
@@ -290,6 +301,9 @@ public class ListenSyslog extends AbstractSyslogProcessor {
factory.setWorkerThreads(maxConnections);
factory.setSocketReceiveBuffer(maxSocketBufferSize);
+ final Boolean socketKeepAlive =
context.getProperty(SOCKET_KEEP_ALIVE).asBoolean();
+ factory.setSocketKeepAlive(socketKeepAlive);
+
final SSLContextService sslContextService =
context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);
if (sslContextService != null) {
final SSLContext sslContext = sslContextService.createContext();
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java
index 55d083f..57f494d 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java
@@ -98,6 +98,7 @@ public class TestListenSyslog {
final TransportProtocol protocol = TransportProtocol.TCP;
runner.setProperty(ListenSyslog.PROTOCOL, protocol.toString());
runner.setProperty(ListenSyslog.PORT, Integer.toString(port));
+ runner.setProperty(ListenSyslog.SOCKET_KEEP_ALIVE,
Boolean.FALSE.toString());
assertSendSuccess(protocol, port);
}
