NIFI-1002: Added WebSocket support. by ijokarumawak · Pull Request #1184 · apache/nifi

ijokarumawak · 2016-11-04T15:05:26Z

Thank you for submitting a contribution to Apache NiFi.

In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:

For all changes:

Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
Does your PR title start with NIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
Has your PR been rebased against the latest commit within the target branch (typically master)?
Is your initial contribution a single, squashed commit?

For code changes:

Have you ensured that the full suite of tests is executed via mvn -Pcontrib-check clean install at the root nifi folder?
Have you written or updated unit tests to verify your changes?
If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
If applicable, have you updated the LICENSE file, including the main LICENSE file under nifi-assembly?
If applicable, have you updated the NOTICE file, including the main NOTICE file found under nifi-assembly?
If adding new Properties, have you added .displayName in addition to .name (programmatic access) for each of the new properties?

For documentation related changes:

Have you ensured that format looks appropriate for the output in which it is rendered?

Note:

Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible.

ijokarumawak · 2016-11-04T15:27:55Z

Hello NiFi developers,

This PR contains a new set of WebSocket support components.

At least there're two WebSocket processors available on GitHub already, nifi-websocket by @xmlking , and nifi-websockets-bundle by @acesir. I appreciate these projects providing WebSocket support for NiFi, and wanted to merge the capability to Apache NiFi code base. I referred these existing projects, and add few functionalities and Unit Test codes.

Features:

As a WebSocket client to connect a remote WebSocket server
As a WebSocket server to receive connection from remote WebSocket clients
Receive and send text messages as well as binary messages
Secure communication (using SslContextService) is supported

It may look more complex, so for the description purpose, I wrote a blog post about how it works, and how it's designed. Please check this post as well, NiFi WebSocket Support.

Any comment or advice would be appreciated, thanks in advance!

olegz · 2016-11-09T20:08:31Z

...rocessors/src/main/java/org/apache/nifi/processors/websocket/AbstractWebSocketProcessor.java

+import java.util.ArrayList;
+import java.util.List;
+
+public abstract class AbstractWebSocketProcessor extends AbstractSessionFactoryProcessor {


Is there a reason why this class extends from _ AbstractSessionFactoryProcessor_ and not AbstractProcessor? Just trying to figure out the need for _ onTrigger()_ implementation that is identical to the one in AbstractProcessor.

@olegz The reason why it extends AbstractSessionFactoryProcessor is to capture `processSessionFactory instance. So that it can create a session at onStopped() to flush incoming WebSocket messages those only exist on memory. I followed the pattern of AbstractMQTTProcessor.

olegz · 2016-11-09T20:14:14Z

...rocessors/src/main/java/org/apache/nifi/processors/websocket/AbstractWebSocketProcessor.java

+
+    static List<PropertyDescriptor> getAbstractPropertyDescriptors(){
+        final List<PropertyDescriptor> descriptors = new ArrayList<>();
+        return descriptors;


I am not sure I understand this code

For each call you create a new List

It's always empty
Could you please clarify?

@olegz As you suspected, it's not required, and it's there simply because I ported the basic code structure from MQTT processors. This method was used for sharing common property descriptors. However, since there's no such properties for now, it can be removed. I will remove this.

olegz · 2016-11-09T20:16:31Z

...rocessors/src/main/java/org/apache/nifi/processors/websocket/AbstractWebSocketProcessor.java

+    }
+
+    protected ComponentLog logger;
+    protected ProcessSessionFactory processSessionFactory;


Since there is a potential for the above variables to be accessed by different thread, there are thread visibility concerns. Consider making them volatile.

Added volatile keyword to these variables.

ijokarumawak · 2016-11-10T11:38:21Z

@olegz Thanks for taking time to review! I added a commit to address your feedback. Also, I've added displayName to the property descriptors those were added in this PR.

olegz · 2016-11-14T19:25:11Z

...ocket-services-jetty/src/main/java/org/apache/nifi/websocket/jetty/JettyWebSocketServer.java

+        servletHandler.addServletWithMapping(JettyWebSocketServlet.class, "/*");
+
+        // Need to specify classloader, otherwise since the callstack doesn't have any nifi specific class, so it can't use nar.
+        try (NarCloseable closeable = NarCloseable.withComponentNarLoader(WebSocketServerFactory.class)) {


This no longer compiles after recent work on dynamic class loading by @bbende

Thanks Oleg, I removed the use of NarCloseable and confirmed that JettyWebSocketServer works without this now.

olegz · 2016-11-14T19:46:02Z

...rs/src/main/java/org/apache/nifi/processors/websocket/AbstractWebSocketGatewayProcessor.java

+            .defaultValue("10000")
+            .build();
+
+    private volatile LinkedBlockingQueue<WebSocketMessage> incomingMessageQueue;


This particular issue keeps coming up. Basically the pattern of maintaining an internal queue for messages may lead to OOM exception as well as lost data if processor crashes while the queue still has data. This is due to the fact that you are essentially integrating with external scheduling mechanism
One way of mitigating this is to rely on such mechanism for distributing messages while essentially ignoring our schedule. You can look at the example of it in SMTP processor

nifi/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java

Line 175 in d838f61

public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {

While may not be ideal, it certainly paves a simpler path for future improvement when we get a chance to define an Event-driver Processor abstraction as a first class component of NiFi.

Thank you for bringing this up. I had been worrying about this queue, too. To make it simpler, I removed the intermediate queue, and let incoming messages to be transferred to NiFi relationships.

- Reflecting review comments - Added displayName to peroperty descriptors

- Reflecting review comments: - Removed unnecessary use of NarCloseable.withComponentNarLoader. - Removed intermediate on memory queue to make it simpler and more robust. Received messages in WebSocket layer now will be transferred to downstream relationships directly.

ijokarumawak · 2016-11-15T07:14:15Z

@olegz Thanks again for the great review! I rebased the PR and done additional refactoring based on your feedback. Please let me know if you prefer it to be squashed for further review process.

olegz · 2016-11-15T12:50:57Z

Koji, this is great and I'll be looking at it shortly with hopes of merging it some time today.

olegz · 2016-11-16T00:33:32Z

@ijokarumawak at this point I am going to say Great work! I am sure in month to come there will be things ti improve as people start using it , but LGTM for now. Will merge shortly!

ijokarumawak · 2016-11-16T00:59:42Z

@olegz Thank you very much! I hope it will be helpful for users and willing to improve it over time!

NIFI-1002: Added WebSocket support. - Reflecting review comments - Added displayName to peroperty descriptors NIFI-1002: Added WebSocket support. This closes apache#1184 - Reflecting review comments: - Removed unnecessary use of NarCloseable.withComponentNarLoader. - Removed intermediate on memory queue to make it simpler and more robust. Received messages in WebSocket layer now will be transferred to downstream relationships directly.

olegz reviewed Nov 9, 2016

View reviewed changes

olegz reviewed Nov 14, 2016

View reviewed changes

ijokarumawak added 2 commits November 15, 2016 16:02

NIFI-1002: Added WebSocket support.

2cc79ca

NIFI-1002: Added WebSocket support.

112fa70

- Reflecting review comments - Added displayName to peroperty descriptors

ijokarumawak force-pushed the nifi-1002 branch from 0794b0d to 112fa70 Compare November 15, 2016 07:03

asfgit closed this in 26a5881 Nov 16, 2016

Conversation

ijokarumawak commented Nov 4, 2016

For all changes:

For code changes:

For documentation related changes:

Note:

Uh oh!

ijokarumawak commented Nov 4, 2016

Uh oh!

olegz Nov 9, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijokarumawak commented Nov 10, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijokarumawak commented Nov 15, 2016

Uh oh!

olegz commented Nov 15, 2016

Uh oh!

olegz commented Nov 16, 2016

Uh oh!

ijokarumawak commented Nov 16, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

olegz Nov 9, 2016 •

edited

Loading