You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@nifi.apache.org by bbende <gi...@git.apache.org> on 2016/01/19 23:59:57 UTC
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
GitHub user bbende opened a pull request:
https://github.com/apache/nifi/pull/179
NIFI-1273 Adding ListenRELP Processor
This pull request adds a ListenRELP processor which includes refactoring code that was previously part of ListenSyslog into a reusable framework for implementing listener processors, as well as back-end code for handling the RELP protocol.
## Overivew
* The listen package in nifi-processor-utils contains the generic dispatcher/handler code along with an AbstractListenEventProcessor which is the base class for future processors that want to act as a server reading data off a channel.
* The relp package in nifi-standard-processors contains the RELP specific implementations for the "listen" classes, along with other back-end code for the RELP protocol.
* ListenRELP extends AbstractListenEventProcessor
## Testing
I tested this against rsyslog 8.x putting the following in /etc/rsyslog.conf:
module(load="omrelp")
action(type="omrelp" target="centralserv" port="2514")
With TLS:
action(type="omrelp" target="192.168.233.153″ port="20514″ tls="on"
tls.caCert="/home/test/cert/ca.pem"
tls.myCert="/home/test/cert/ubuntu1-cert.pem"
tls.myPrivKey="/home/test/cert/ubuntu1-key.pem"
tls.authmode="name"
tls.permittedpeer=["ubuntu-server"]
)
RELP Reference:
http://www.rsyslog.com/doc/relp.html
http://www.rsyslog.com/using-tls-with-relp/
RELPFrameProducer in nifi-standard-processors src/test/java/org.apache.nifi.processors.standard.relp can also be used to act as an rsyslog server sending messages to ListenRELP.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/bbende/nifi NIFI-1273
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/179.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #179
----
commit ed23c57d3a1f9c07f91f67d0ed42c10f91558f40
Author: Bryan Bende <bb...@apache.org>
Date: 2016-01-05T22:41:33Z
NIFI-1273 Adding ListenRELP processor which includes refactoring code that was previously part of ListenSyslog into a reusable framework for implementing listener processors, as well as back-end code for handling the RELP protocol.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on the pull request:
https://github.com/apache/nifi/pull/179#issuecomment-173830226
@bbende I have been running some tests that I usually use around reliable destinations (disconnects, repeated messages, unexpected restarts) and all worked to expected (except for the issue already noted by @mcgilman).
Amazing code and it certainly enable NiFi to have a greater number of inbound protocols. I will port my Lumberjack spaghetti into the new and cleaner code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on a diff in the pull request:
https://github.com/apache/nifi/pull/179#discussion_r50489858
--- Diff: nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/listen/dispatcher/ChannelDispatcher.java ---
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.nifi.processor.util.listen.dispatcher;
+
+import java.io.IOException;
+import java.nio.channels.SelectionKey;
+
+/**
+ * Dispatches handlers for a given channel.
+ */
+public interface ChannelDispatcher extends Runnable {
--- End diff --
For extensiblity, it would be very helpful to have javadoc for this interface.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on a diff in the pull request:
https://github.com/apache/nifi/pull/179#discussion_r50490353
--- Diff: nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/listen/dispatcher/DatagramChannelDispatcher.java ---
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.nifi.processor.util.listen.dispatcher;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.logging.ProcessorLog;
+import org.apache.nifi.processor.util.listen.event.Event;
+import org.apache.nifi.processor.util.listen.event.EventFactory;
+import org.apache.nifi.processor.util.listen.event.EventFactoryUtil;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.StandardSocketOptions;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Reads from the Datagram channel into an available buffer. If data is read then the buffer is queued for
+ * processing, otherwise the buffer is returned to the buffer pool.
+ */
+public class DatagramChannelDispatcher<E extends Event<DatagramChannel>> implements ChannelDispatcher {
+
+ private final EventFactory<E> eventFactory;
+ private final BlockingQueue<ByteBuffer> bufferPool;
+ private final BlockingQueue<E> events;
+ private final ProcessorLog logger;
+
+ private Selector selector;
+ private DatagramChannel datagramChannel;
+ private volatile boolean stopped = false;
+
+ public DatagramChannelDispatcher(final EventFactory<E> eventFactory,
+ final BlockingQueue<ByteBuffer> bufferPool,
+ final BlockingQueue<E> events,
+ final ProcessorLog logger) {
+ this.eventFactory = eventFactory;
+ this.bufferPool = bufferPool;
+ this.events = events;
+ this.logger = logger;
+
+ if (bufferPool == null || bufferPool.size() == 0) {
+ throw new IllegalArgumentException("A pool of available ByteBuffers is required");
+ }
+ }
+
+ @Override
+ public void open(final int port, int maxBufferSize) throws IOException {
+ datagramChannel = DatagramChannel.open();
+ datagramChannel.configureBlocking(false);
+ if (maxBufferSize > 0) {
+ datagramChannel.setOption(StandardSocketOptions.SO_RCVBUF, maxBufferSize);
+ final int actualReceiveBufSize = datagramChannel.getOption(StandardSocketOptions.SO_RCVBUF);
+ if (actualReceiveBufSize < maxBufferSize) {
+ logger.warn("Attempted to set Socket Buffer Size to " + maxBufferSize + " bytes but could only set to "
+ + actualReceiveBufSize + "bytes. You may want to consider changing the Operating System's "
+ + "maximum receive buffer");
+ }
+ }
+ datagramChannel.socket().bind(new InetSocketAddress(port));
+ selector = Selector.open();
+ datagramChannel.register(selector, SelectionKey.OP_READ);
+ }
+
+ @Override
+ public void run() {
+ final ByteBuffer buffer = bufferPool.poll();
+ while (!stopped) {
+ try {
+ int selected = selector.select();
+ if (selected > 0){
+ Iterator<SelectionKey> selectorKeys = selector.selectedKeys().iterator();
+ while (selectorKeys.hasNext()) {
+ SelectionKey key = selectorKeys.next();
+ selectorKeys.remove();
+ if (!key.isValid()) {
+ continue;
+ }
+ DatagramChannel channel = (DatagramChannel) key.channel();
+ SocketAddress socketAddress;
+ buffer.clear();
+ while (!stopped && (socketAddress = channel.receive(buffer)) != null) {
+ String sender = "";
+ if (socketAddress instanceof InetSocketAddress) {
+ sender = ((InetSocketAddress) socketAddress).getAddress().toString();
+ }
+
+ // create a byte array from the buffer
+ buffer.flip();
+ byte bytes[] = new byte[buffer.limit()];
+ buffer.get(bytes, 0, buffer.limit());
+
+ final Map<String,String> metadata = EventFactoryUtil.createMapWithSender(sender);
+ final E event = eventFactory.create(bytes, metadata, null);
+
+ // queue the raw message with the sender, block until space is available
+ events.put(event);
+ buffer.clear();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ stopped = true;
+ } catch (IOException e) {
+ logger.error("Error reading from DatagramChannel", e);
+ }
+ }
+
+ if (buffer != null) {
+ try {
+ bufferPool.put(buffer);
+ } catch (InterruptedException e) {
+ // nothing to do here
+ }
+ }
+ }
+
+ @Override
+ public int getPort() {
+ return datagramChannel == null ? 0 : datagramChannel.socket().getLocalPort();
+ }
+
+ @Override
+ public void stop() {
+ selector.wakeup();
+ stopped = true;
+ }
+
+ @Override
+ public void close() {
+ IOUtils.closeQuietly(selector);
+ IOUtils.closeQuietly(datagramChannel);
+ }
+
+ @Override
+ public void completeConnection(SelectionKey key) {
--- End diff --
since these two are not useful for this, should the interface ChannelDispatcher include these? Or should there be an AsyncChannelDispatcher interface that extends ChannelDispatcher?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on a diff in the pull request:
https://github.com/apache/nifi/pull/179#discussion_r50490153
--- Diff: nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/listen/dispatcher/DatagramChannelDispatcher.java ---
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.nifi.processor.util.listen.dispatcher;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.logging.ProcessorLog;
+import org.apache.nifi.processor.util.listen.event.Event;
+import org.apache.nifi.processor.util.listen.event.EventFactory;
+import org.apache.nifi.processor.util.listen.event.EventFactoryUtil;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.StandardSocketOptions;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Reads from the Datagram channel into an available buffer. If data is read then the buffer is queued for
+ * processing, otherwise the buffer is returned to the buffer pool.
+ */
+public class DatagramChannelDispatcher<E extends Event<DatagramChannel>> implements ChannelDispatcher {
+
+ private final EventFactory<E> eventFactory;
+ private final BlockingQueue<ByteBuffer> bufferPool;
+ private final BlockingQueue<E> events;
+ private final ProcessorLog logger;
+
+ private Selector selector;
+ private DatagramChannel datagramChannel;
+ private volatile boolean stopped = false;
+
+ public DatagramChannelDispatcher(final EventFactory<E> eventFactory,
+ final BlockingQueue<ByteBuffer> bufferPool,
+ final BlockingQueue<E> events,
+ final ProcessorLog logger) {
+ this.eventFactory = eventFactory;
+ this.bufferPool = bufferPool;
+ this.events = events;
+ this.logger = logger;
+
+ if (bufferPool == null || bufferPool.size() == 0) {
+ throw new IllegalArgumentException("A pool of available ByteBuffers is required");
+ }
+ }
+
+ @Override
+ public void open(final int port, int maxBufferSize) throws IOException {
+ datagramChannel = DatagramChannel.open();
+ datagramChannel.configureBlocking(false);
+ if (maxBufferSize > 0) {
+ datagramChannel.setOption(StandardSocketOptions.SO_RCVBUF, maxBufferSize);
+ final int actualReceiveBufSize = datagramChannel.getOption(StandardSocketOptions.SO_RCVBUF);
+ if (actualReceiveBufSize < maxBufferSize) {
+ logger.warn("Attempted to set Socket Buffer Size to " + maxBufferSize + " bytes but could only set to "
+ + actualReceiveBufSize + "bytes. You may want to consider changing the Operating System's "
+ + "maximum receive buffer");
+ }
+ }
+ datagramChannel.socket().bind(new InetSocketAddress(port));
+ selector = Selector.open();
+ datagramChannel.register(selector, SelectionKey.OP_READ);
+ }
+
+ @Override
+ public void run() {
+ final ByteBuffer buffer = bufferPool.poll();
+ while (!stopped) {
+ try {
+ int selected = selector.select();
+ if (selected > 0){
+ Iterator<SelectionKey> selectorKeys = selector.selectedKeys().iterator();
+ while (selectorKeys.hasNext()) {
+ SelectionKey key = selectorKeys.next();
+ selectorKeys.remove();
+ if (!key.isValid()) {
+ continue;
+ }
+ DatagramChannel channel = (DatagramChannel) key.channel();
+ SocketAddress socketAddress;
+ buffer.clear();
+ while (!stopped && (socketAddress = channel.receive(buffer)) != null) {
+ String sender = "";
+ if (socketAddress instanceof InetSocketAddress) {
+ sender = ((InetSocketAddress) socketAddress).getAddress().toString();
+ }
+
+ // create a byte array from the buffer
+ buffer.flip();
+ byte bytes[] = new byte[buffer.limit()];
+ buffer.get(bytes, 0, buffer.limit());
+
+ final Map<String,String> metadata = EventFactoryUtil.createMapWithSender(sender);
+ final E event = eventFactory.create(bytes, metadata, null);
+
+ // queue the raw message with the sender, block until space is available
+ events.put(event);
+ buffer.clear();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ stopped = true;
+ } catch (IOException e) {
+ logger.error("Error reading from DatagramChannel", e);
+ }
+ }
+
+ if (buffer != null) {
+ try {
+ bufferPool.put(buffer);
+ } catch (InterruptedException e) {
+ // nothing to do here
--- End diff --
when reviewing one of @olegz's patches, I thought this was a another place where adding a Thread.currentThread().interrupt() made sense
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on a diff in the pull request:
https://github.com/apache/nifi/pull/179#discussion_r50489401
--- Diff: nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/listen/AbstractListenEventProcessor.java ---
@@ -0,0 +1,356 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.nifi.processor.util.listen;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processor.util.listen.dispatcher.ChannelDispatcher;
+import org.apache.nifi.processor.util.listen.event.Event;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * An abstract processor to extend from when listening for events over a channel. This processor
+ * will start a ChannelDispatcher, and optionally a ChannelResponseDispatcher, in a background
+ * thread which will end up placing events on a queue to polled by the onTrigger method. Sub-classes
+ * are responsible for providing the dispatcher implementations.
+ *
+ * @param <E> the type of events being produced
+ */
+public abstract class AbstractListenEventProcessor<E extends Event> extends AbstractProcessor {
+
+ public static final PropertyDescriptor PORT = new PropertyDescriptor
+ .Builder().name("Port")
+ .description("The port to listen on for communication.")
+ .required(true)
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
+ .name("Character Set")
+ .description("Specifies the character set of the received data.")
+ .required(true)
+ .defaultValue("UTF-8")
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor RECV_BUFFER_SIZE = new PropertyDescriptor.Builder()
+ .name("Receive Buffer Size")
+ .description("The size of each buffer used to receive messages. Adjust this value appropriately based on the expected size of the " +
+ "incoming messages.")
+ .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
+ .defaultValue("65507 B")
+ .required(true)
+ .build();
+ public static final PropertyDescriptor MAX_SOCKET_BUFFER_SIZE = new PropertyDescriptor.Builder()
+ .name("Max Size of Socket Buffer")
+ .description("The maximum size of the socket buffer that should be used. This is a suggestion to the Operating System " +
+ "to indicate how big the socket buffer should be. If this value is set too low, the buffer may fill up before " +
+ "the data can be read, and incoming data will be dropped.")
+ .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
+ .defaultValue("1 MB")
+ .required(true)
+ .build();
+
+ // Putting these properties here so sub-classes don't have to redefine them, but they are
+ // not added to the properties by default since not all processors may need them
+
+ public static final PropertyDescriptor MAX_CONNECTIONS = new PropertyDescriptor.Builder()
+ .name("Max Number of TCP Connections")
+ .description("The maximum number of concurrent TCP connections to accept.")
+ .addValidator(StandardValidators.createLongValidator(1, 65535, true))
+ .defaultValue("2")
+ .required(true)
+ .build();
+ public static final PropertyDescriptor MAX_BATCH_SIZE = new PropertyDescriptor.Builder()
+ .name("Max Batch Size")
+ .description(
+ "The maximum number of messages to add to a single FlowFile. If multiple messages are available, they will be concatenated along with "
+ + "the <Message Delimiter> up to this configured maximum number of messages")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(false)
+ .defaultValue("1")
+ .required(true)
+ .build();
+ public static final PropertyDescriptor MESSAGE_DELIMITER = new PropertyDescriptor.Builder()
+ .name("Message Delimiter")
+ .description("Specifies the delimiter to place between messages when multiple messages are bundled together (see <Max Batch Size> property).")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .defaultValue("\\n")
+ .required(true)
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Messages received successfully will be sent out this relationship.")
+ .build();
+
+ private Set<Relationship> relationships;
+ private List<PropertyDescriptor> descriptors;
+
+ protected volatile int port;
+ protected volatile Charset charset;
+ protected volatile ChannelDispatcher dispatcher;
+ protected volatile BlockingQueue<E> events = new LinkedBlockingQueue<>(10);
+ protected volatile BlockingQueue<E> errorEvents = new LinkedBlockingQueue<>();
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(PORT);
+ descriptors.add(RECV_BUFFER_SIZE);
+ descriptors.add(MAX_SOCKET_BUFFER_SIZE);
+ descriptors.add(CHARSET);
+ descriptors.addAll(getAdditionalProperties());
+ this.descriptors = Collections.unmodifiableList(descriptors);
+
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.addAll(getAdditionalRelationships());
+ this.relationships = Collections.unmodifiableSet(relationships);
+ }
+
+ /**
+ * Override to provide additional relationships for the processor.
+ *
+ * @return a list of relationships
+ */
+ protected List<Relationship> getAdditionalRelationships() {
+ return Collections.EMPTY_LIST;
+ }
+
+ /**
+ * Override to provide additional properties for the processor.
+ *
+ * @return a list of properties
+ */
+ protected List<PropertyDescriptor> getAdditionalProperties() {
+ return Collections.EMPTY_LIST;
+ }
+
+ @Override
+ public final Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) throws IOException {
+ charset = Charset.forName(context.getProperty(CHARSET).getValue());
+ port = context.getProperty(PORT).asInteger();
+
+ final int maxChannelBufferSize = context.getProperty(MAX_SOCKET_BUFFER_SIZE).asDataSize(DataUnit.B).intValue();
+
+ // create the dispatcher and call open() to bind to the given port
+ dispatcher = createDispatcher(context, events);
+ dispatcher.open(port, maxChannelBufferSize);
+
+ // start a thread to run the dispatcher
+ final Thread readerThread = new Thread(dispatcher);
+ readerThread.setName(getClass().getName() + " [" + getIdentifier() + "]");
+ readerThread.setDaemon(true);
+ readerThread.start();
+ }
+
+ /**
+ * @param context the ProcessContext to retrieve property values from
+ * @return a ChannelDispatcher to handle incoming connections
+ *
+ * @throws IOException if unable to listen on the requested port
+ */
+ protected abstract ChannelDispatcher createDispatcher(final ProcessContext context, final BlockingQueue<E> events) throws IOException;
+
+ // used for testing to access the random port that was selected
+ public final int getDispatcherPort() {
+ return dispatcher == null ? 0 : dispatcher.getPort();
+ }
+
+ public int getErrorQueueSize() {
+ return errorEvents.size();
+ }
+
+ @OnUnscheduled
+ public void onUnscheduled() {
+ if (dispatcher != null) {
+ dispatcher.stop();
+ dispatcher.close();
+ }
+ }
+
+ /**
+ * If pollErrorQueue is true, the error queue will be checked first and event will be
+ * returned from the error queue if available.
+ *
+ * If pollErrorQueue is false, or no data is in the error queue, the regular queue is polled.
+ *
+ * If longPoll is true, the regular queue will be polled with a short timeout, otherwise it will
+ * poll with no timeout which will return immediately.
+ *
+ * @param longPoll whether or not to poll the main queue with a small timeout
+ * @param pollErrorQueue whether or not to poll the error queue first
+ *
+ * @return an event from one of the queues, or null if none are available
+ */
+ protected E getMessage(final boolean longPoll, final boolean pollErrorQueue) {
+ E event = null;
+ if (pollErrorQueue) {
+ event = errorEvents.poll();
+ }
+
+ if (event == null) {
+ try {
+ if (longPoll) {
+ event = events.poll(100, TimeUnit.MILLISECONDS);
--- End diff --
next chance we get, should replace this 100 with a constant.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by mcgilman <gi...@git.apache.org>.
Github user mcgilman commented on a diff in the pull request:
https://github.com/apache/nifi/pull/179#discussion_r50451076
--- Diff: nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/listen/dispatcher/DatagramChannelDispatcher.java ---
@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.nifi.processor.util.listen.dispatcher;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.logging.ProcessorLog;
+import org.apache.nifi.processor.util.listen.event.Event;
+import org.apache.nifi.processor.util.listen.event.EventFactory;
+import org.apache.nifi.processor.util.listen.event.EventFactoryUtil;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.StandardSocketOptions;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Reads from the Datagram channel into an available buffer. If data is read then the buffer is queued for
+ * processing, otherwise the buffer is returned to the buffer pool.
+ */
+public class DatagramChannelDispatcher<E extends Event<DatagramChannel>> implements ChannelDispatcher {
+
+ private final EventFactory<E> eventFactory;
+ private final BlockingQueue<ByteBuffer> bufferPool;
+ private final BlockingQueue<E> events;
+ private final ProcessorLog logger;
+
+ private Selector selector;
+ private DatagramChannel datagramChannel;
+ private volatile boolean stopped = false;
+
+ public DatagramChannelDispatcher(final EventFactory<E> eventFactory,
+ final BlockingQueue<ByteBuffer> bufferPool,
+ final BlockingQueue<E> events,
+ final ProcessorLog logger) {
+ this.eventFactory = eventFactory;
+ this.bufferPool = bufferPool;
+ this.events = events;
+ this.logger = logger;
+ }
+
+ @Override
+ public void open(final int port, int maxBufferSize) throws IOException {
+ datagramChannel = DatagramChannel.open();
+ datagramChannel.configureBlocking(false);
+ if (maxBufferSize > 0) {
+ datagramChannel.setOption(StandardSocketOptions.SO_RCVBUF, maxBufferSize);
+ final int actualReceiveBufSize = datagramChannel.getOption(StandardSocketOptions.SO_RCVBUF);
+ if (actualReceiveBufSize < maxBufferSize) {
+ logger.warn("Attempted to set Socket Buffer Size to " + maxBufferSize + " bytes but could only set to "
+ + actualReceiveBufSize + "bytes. You may want to consider changing the Operating System's "
+ + "maximum receive buffer");
+ }
+ }
+ datagramChannel.socket().bind(new InetSocketAddress(port));
+ selector = Selector.open();
+ datagramChannel.register(selector, SelectionKey.OP_READ);
+ }
+
+ @Override
+ public void run() {
+ final ByteBuffer buffer = bufferPool.poll();
--- End diff --
Protect when buffer is null or ensure bufferPool is not empty initially?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on a diff in the pull request:
https://github.com/apache/nifi/pull/179#discussion_r50490991
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListenRELP.java ---
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.nifi.processors.standard;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.flowfile.attributes.FlowFileAttributeKey;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.listen.AbstractListenEventProcessor;
+import org.apache.nifi.processor.util.listen.dispatcher.ChannelDispatcher;
+import org.apache.nifi.processor.util.listen.dispatcher.SocketChannelDispatcher;
+import org.apache.nifi.processor.util.listen.event.EventFactory;
+import org.apache.nifi.processor.util.listen.handler.ChannelHandlerFactory;
+import org.apache.nifi.processor.util.listen.response.ChannelResponder;
+import org.apache.nifi.processor.util.listen.response.ChannelResponse;
+import org.apache.nifi.processors.standard.relp.event.RELPEvent;
+import org.apache.nifi.processors.standard.relp.event.RELPEventFactory;
+import org.apache.nifi.processors.standard.relp.frame.RELPEncoder;
+import org.apache.nifi.processors.standard.relp.handler.RELPSocketChannelHandlerFactory;
+import org.apache.nifi.processors.standard.relp.response.RELPChannelResponse;
+import org.apache.nifi.processors.standard.relp.response.RELPResponse;
+import org.apache.nifi.ssl.SSLContextService;
+
+import javax.net.ssl.SSLContext;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@Tags({"listen", "relp", "tcp", "logs"})
+@CapabilityDescription("Listens for RELP messages being sent to a given port over TCP. Each message will be " +
+ "acknowledged after successfully writing the message to a FlowFile. Each FlowFile will contain data " +
+ "portion of one or more RELP frames. In the case where the RELP frames contain syslog messages, the " +
+ "output of this processor can be sent to a ParseSyslog processor for further processing.")
+@WritesAttributes({
+ @WritesAttribute(attribute="relp.comand", description="The command of the RELP frames."),
--- End diff --
s/comand/command/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the pull request:
https://github.com/apache/nifi/pull/179#issuecomment-173726265
Thanks for reviewing @mcgilman and @trkurc !
I pushed up a small commit that fixed a couple of minor logging issues and addressed the comment Matt made about the pool of buffers.
Will wait to see any feedback from Tony and can address anything else as needed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by mcgilman <gi...@git.apache.org>.
Github user mcgilman commented on the pull request:
https://github.com/apache/nifi/pull/179#issuecomment-173707708
Spent some time reviewing the patch. Overall, I think the code looks good. Additional nit-picky item outside of my previous comment.
- I could see possibly wanting to extract the RELP utility classes into a separate artifact. Maybe we wait to shuffle this around until we have a case for it. Maybe introduce a RELP NAR if we ever get to that point.
When running ListenRELP, there were times (especially right after stopping and starting the Processor) that new messages were not picked up for some time. At some point later, a backlog of messages flooded in. It's not clear what triggered it and the backlog appeared to produce some duplicate messages.
@trkurc did you want to check this out before we proceed with merging?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on the pull request:
https://github.com/apache/nifi/pull/179#issuecomment-173770289
@bbende - this is an awesome contribution. most of my comments are minor and of the ilk of we should fix next chance we get, with the exception of the 'comand' typo
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on the pull request:
https://github.com/apache/nifi/pull/179#issuecomment-173044611
I like the docs to help with reviewing! I good amount here to review and test!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the pull request:
https://github.com/apache/nifi/pull/179#issuecomment-173968940
@trkurc thanks for the review, i liked your idea about the AsyncChannelDispatcher extending ChannelDispatcher so I went ahead and did that, and addressed the other stuff as well.
@trixpan thanks for testing this out!
As far as I can tell the scenario with the messages being resent seems to be something happening on the rsyslog side of things. I can't really come up with a pattern of what it is doing as it is not always resending the same amount of messages.
I think we are good to merge this in.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on a diff in the pull request:
https://github.com/apache/nifi/pull/179#discussion_r50490135
--- Diff: nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/listen/dispatcher/DatagramChannelDispatcher.java ---
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.nifi.processor.util.listen.dispatcher;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.logging.ProcessorLog;
+import org.apache.nifi.processor.util.listen.event.Event;
+import org.apache.nifi.processor.util.listen.event.EventFactory;
+import org.apache.nifi.processor.util.listen.event.EventFactoryUtil;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.StandardSocketOptions;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Reads from the Datagram channel into an available buffer. If data is read then the buffer is queued for
+ * processing, otherwise the buffer is returned to the buffer pool.
+ */
+public class DatagramChannelDispatcher<E extends Event<DatagramChannel>> implements ChannelDispatcher {
+
+ private final EventFactory<E> eventFactory;
+ private final BlockingQueue<ByteBuffer> bufferPool;
+ private final BlockingQueue<E> events;
+ private final ProcessorLog logger;
+
+ private Selector selector;
+ private DatagramChannel datagramChannel;
+ private volatile boolean stopped = false;
+
+ public DatagramChannelDispatcher(final EventFactory<E> eventFactory,
+ final BlockingQueue<ByteBuffer> bufferPool,
+ final BlockingQueue<E> events,
+ final ProcessorLog logger) {
+ this.eventFactory = eventFactory;
+ this.bufferPool = bufferPool;
+ this.events = events;
+ this.logger = logger;
+
+ if (bufferPool == null || bufferPool.size() == 0) {
+ throw new IllegalArgumentException("A pool of available ByteBuffers is required");
+ }
+ }
+
+ @Override
+ public void open(final int port, int maxBufferSize) throws IOException {
+ datagramChannel = DatagramChannel.open();
+ datagramChannel.configureBlocking(false);
+ if (maxBufferSize > 0) {
+ datagramChannel.setOption(StandardSocketOptions.SO_RCVBUF, maxBufferSize);
+ final int actualReceiveBufSize = datagramChannel.getOption(StandardSocketOptions.SO_RCVBUF);
+ if (actualReceiveBufSize < maxBufferSize) {
+ logger.warn("Attempted to set Socket Buffer Size to " + maxBufferSize + " bytes but could only set to "
+ + actualReceiveBufSize + "bytes. You may want to consider changing the Operating System's "
+ + "maximum receive buffer");
+ }
+ }
+ datagramChannel.socket().bind(new InetSocketAddress(port));
+ selector = Selector.open();
+ datagramChannel.register(selector, SelectionKey.OP_READ);
+ }
+
+ @Override
+ public void run() {
+ final ByteBuffer buffer = bufferPool.poll();
+ while (!stopped) {
+ try {
+ int selected = selector.select();
+ if (selected > 0){
+ Iterator<SelectionKey> selectorKeys = selector.selectedKeys().iterator();
+ while (selectorKeys.hasNext()) {
+ SelectionKey key = selectorKeys.next();
+ selectorKeys.remove();
+ if (!key.isValid()) {
+ continue;
+ }
+ DatagramChannel channel = (DatagramChannel) key.channel();
+ SocketAddress socketAddress;
+ buffer.clear();
+ while (!stopped && (socketAddress = channel.receive(buffer)) != null) {
+ String sender = "";
+ if (socketAddress instanceof InetSocketAddress) {
+ sender = ((InetSocketAddress) socketAddress).getAddress().toString();
+ }
+
+ // create a byte array from the buffer
+ buffer.flip();
+ byte bytes[] = new byte[buffer.limit()];
+ buffer.get(bytes, 0, buffer.limit());
+
+ final Map<String,String> metadata = EventFactoryUtil.createMapWithSender(sender);
+ final E event = eventFactory.create(bytes, metadata, null);
+
+ // queue the raw message with the sender, block until space is available
+ events.put(event);
+ buffer.clear();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ stopped = true;
--- End diff --
when reviewing one of @olegz's patches, I thought this was a place where adding a Thread.currentThread().interrupt() made sense
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on the pull request:
https://github.com/apache/nifi/pull/179#issuecomment-173712640
@mcgilman - I have been reviewing. I should be complete tonight. Haven't found any show stoppers yet.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by mcgilman <gi...@git.apache.org>.
Github user mcgilman commented on the pull request:
https://github.com/apache/nifi/pull/179#issuecomment-174050331
Awesome @bbende great work! I'll go ahead and merge this in now. Thanks everyone!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on a diff in the pull request:
https://github.com/apache/nifi/pull/179#discussion_r50490685
--- Diff: nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/listen/dispatcher/SocketChannelDispatcher.java ---
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.nifi.processor.util.listen.dispatcher;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.logging.ProcessorLog;
+import org.apache.nifi.processor.util.listen.event.Event;
+import org.apache.nifi.processor.util.listen.event.EventFactory;
+import org.apache.nifi.processor.util.listen.handler.ChannelHandlerFactory;
+import org.apache.nifi.remote.io.socket.ssl.SSLSocketChannel;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.StandardSocketOptions;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.nio.charset.Charset;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Accepts Socket connections on the given port and creates a handler for each connection to
+ * be executed by a thread pool.
+ */
+public class SocketChannelDispatcher<E extends Event<SocketChannel>> implements ChannelDispatcher {
+
+ private final EventFactory<E> eventFactory;
+ private final ChannelHandlerFactory<E> handlerFactory;
+ private final BlockingQueue<ByteBuffer> bufferPool;
+ private final BlockingQueue<E> events;
+ private final ProcessorLog logger;
+ private final int maxConnections;
+ private final SSLContext sslContext;
+ private final Charset charset;
+
+ private ExecutorService executor;
+ private volatile boolean stopped = false;
+ private Selector selector;
+ private final BlockingQueue<SelectionKey> keyQueue;
+ private final AtomicInteger currentConnections = new AtomicInteger(0);
+
+
+ public SocketChannelDispatcher(final EventFactory<E> eventFactory,
+ final ChannelHandlerFactory<E> handlerFactory,
+ final BlockingQueue<ByteBuffer> bufferPool,
+ final BlockingQueue<E> events,
+ final ProcessorLog logger,
+ final int maxConnections,
+ final SSLContext sslContext,
+ final Charset charset) {
+ this.eventFactory = eventFactory;
+ this.handlerFactory = handlerFactory;
+ this.bufferPool = bufferPool;
+ this.events = events;
+ this.logger = logger;
+ this.maxConnections = maxConnections;
+ this.keyQueue = new LinkedBlockingQueue<>(maxConnections);
+ this.sslContext = sslContext;
+ this.charset = charset;
+
+ if (bufferPool == null || bufferPool.size() == 0 || bufferPool.size() != maxConnections) {
+ throw new IllegalArgumentException(
+ "A pool of available ByteBuffers equal to the maximum number of connections is required");
+ }
+ }
+
+ @Override
+ public void open(final int port, int maxBufferSize) throws IOException {
+ this.executor = Executors.newFixedThreadPool(maxConnections);
+
+ final ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
+ serverSocketChannel.configureBlocking(false);
+ if (maxBufferSize > 0) {
+ serverSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, maxBufferSize);
+ final int actualReceiveBufSize = serverSocketChannel.getOption(StandardSocketOptions.SO_RCVBUF);
+ if (actualReceiveBufSize < maxBufferSize) {
+ logger.warn("Attempted to set Socket Buffer Size to " + maxBufferSize + " bytes but could only set to "
+ + actualReceiveBufSize + "bytes. You may want to consider changing the Operating System's "
+ + "maximum receive buffer");
+ }
+ }
+ serverSocketChannel.socket().bind(new InetSocketAddress(port));
+ selector = Selector.open();
+ serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
+ }
+
+ @Override
+ public void run() {
+ while (!stopped) {
+ try {
+ int selected = selector.select();
+ if (selected > 0){
+ Iterator<SelectionKey> selectorKeys = selector.selectedKeys().iterator();
+ while (selectorKeys.hasNext()){
+ SelectionKey key = selectorKeys.next();
+ selectorKeys.remove();
+ if (!key.isValid()){
+ continue;
+ }
+ if (key.isAcceptable()) {
+ // Handle new connections coming in
+ final ServerSocketChannel channel = (ServerSocketChannel) key.channel();
+ final SocketChannel socketChannel = channel.accept();
+ // Check for available connections
+ if (currentConnections.incrementAndGet() > maxConnections){
+ currentConnections.decrementAndGet();
+ logger.warn("Rejecting connection from {} because max connections has been met",
+ new Object[]{ socketChannel.getRemoteAddress().toString() });
+ IOUtils.closeQuietly(socketChannel);
+ continue;
+ }
+ logger.debug("Accepted incoming connection from {}",
+ new Object[]{socketChannel.getRemoteAddress().toString()});
+ // Set socket to non-blocking, and register with selector
+ socketChannel.configureBlocking(false);
+ SelectionKey readKey = socketChannel.register(selector, SelectionKey.OP_READ);
+
+ // Prepare the byte buffer for the reads, clear it out
+ ByteBuffer buffer = bufferPool.poll();
+ buffer.clear();
+ buffer.mark();
+
+ // If we have an SSLContext then create an SSLEngine for the channel
+ SSLSocketChannel sslSocketChannel = null;
+ if (sslContext != null) {
+ final SSLEngine sslEngine = sslContext.createSSLEngine();
+ sslSocketChannel = new SSLSocketChannel(sslEngine, socketChannel, false);
+ }
+
+ // Attach the buffer and SSLSocketChannel to the key
+ SocketChannelAttachment attachment = new SocketChannelAttachment(buffer, sslSocketChannel);
+ readKey.attach(attachment);
+ } else if (key.isReadable()) {
+ // Clear out the operations the select is interested in until done reading
+ key.interestOps(0);
+ // Create a handler based on the protocol and whether an SSLEngine was provided or not
+ final Runnable handler;
+ if (sslContext != null) {
+ handler = handlerFactory.createSSLHandler(key, this, charset, eventFactory, events, logger);
+ } else {
+ handler = handlerFactory.createHandler(key, this, charset, eventFactory, events, logger);
+ }
+
+ // run the handler
+ executor.execute(handler);
+ }
+ }
+ }
+ // Add back all idle sockets to the select
+ SelectionKey key;
+ while((key = keyQueue.poll()) != null){
+ key.interestOps(SelectionKey.OP_READ);
+ }
+ } catch (IOException e) {
+ logger.error("Error accepting connection from SocketChannel", e);
+ }
+ }
+ }
+
+ @Override
+ public int getPort() {
+ // Return the port for the key listening for accepts
+ for(SelectionKey key : selector.keys()){
+ if (key.isValid()) {
+ final Channel channel = key.channel();
+ if (channel instanceof ServerSocketChannel) {
+ return ((ServerSocketChannel)channel).socket().getLocalPort();
+ }
+ }
+ }
+ return 0;
+ }
+
+ @Override
+ public void stop() {
+ stopped = true;
+ selector.wakeup();
+ }
+
+ @Override
+ public void close() {
+ executor.shutdown();
+ try {
+ // Wait a while for existing tasks to terminate
+ if (!executor.awaitTermination(1000L, TimeUnit.MILLISECONDS)) {
+ executor.shutdownNow();
+ }
+ } catch (InterruptedException ie) {
+ // (Re-)Cancel if current thread also interrupted
+ executor.shutdownNow();
+ // Preserve interrupt status
+ Thread.currentThread().interrupt();
+ }
+ for(SelectionKey key : selector.keys()){
+ IOUtils.closeQuietly(key.channel());
+ }
+ IOUtils.closeQuietly(selector);
+ }
+
+ @Override
+ public void completeConnection(SelectionKey key) {
+ // connection is done. Return the buffer to the pool
+ SocketChannelAttachment attachment = (SocketChannelAttachment) key.attachment();
+ try {
+ bufferPool.put(attachment.getByteBuffer());
+ } catch (InterruptedException e) {
+ // nothing to do here
--- End diff --
like some of the above, may want to add a Thread.currentThread().interrupt();
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/179
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1273 Adding ListenRELP Processor
Posted by bbende <gi...@git.apache.org>.
Github user bbende commented on the pull request:
https://github.com/apache/nifi/pull/179#issuecomment-174049864
Just a quick update on the scenario where some messages get resent...
I turned on rsyslog debug and I noticed that rsyslog seemed to be queueing up responses until 128 responses were received at which point it processed all the responses and started back over, this was consistent with a long running test. So if you stop and start the processor before that, say when it has received 50 responses, it hasn't processed those yet so it thinks those 50 messages need to be resent.
There may be some advanced config in rsyslog to control this, but I am convinced we are sending the correct responses and believe we are good to merge this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---