[ https://issues.apache.org/jira/browse/NIFI-1002?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15666340#comment-15666340 ]
ASF GitHub Bot commented on NIFI-1002: -------------------------------------- Github user ijokarumawak commented on a diff in the pull request: https://github.com/apache/nifi/pull/1184#discussion_r87958885 --- Diff: nifi-nar-bundles/nifi-websocket-bundle/nifi-websocket-processors/src/main/java/org/apache/nifi/processors/websocket/AbstractWebSocketGatewayProcessor.java --- @@ -0,0 +1,314 @@ +/* + * 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.websocket; + +import org.apache.commons.lang3.StringUtils; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.annotation.lifecycle.OnUnscheduled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.flowfile.attributes.CoreAttributes; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.websocket.BinaryMessageConsumer; +import org.apache.nifi.websocket.ConnectedListener; +import org.apache.nifi.websocket.TextMessageConsumer; +import org.apache.nifi.websocket.WebSocketClientService; +import org.apache.nifi.websocket.WebSocketConfigurationException; +import org.apache.nifi.websocket.WebSocketConnectedMessage; +import org.apache.nifi.websocket.WebSocketMessage; +import org.apache.nifi.websocket.WebSocketService; +import org.apache.nifi.websocket.WebSocketSessionInfo; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.LinkedBlockingQueue; + +@TriggerSerially +public abstract class AbstractWebSocketGatewayProcessor extends AbstractWebSocketProcessor implements ConnectedListener, TextMessageConsumer, BinaryMessageConsumer { + + public static final PropertyDescriptor PROP_MAX_QUEUE_SIZE = new PropertyDescriptor.Builder() + .name("max-queue-size") + .displayName("Max Queue Size") + .description("The WebSocket messages are kept in an on-memory queue," + + " then transferred to relationships when this processor is triggered." + + " If the 'Run Schedule' is significantly behind the rate" + + " at which the messages are arriving to this processor then a back up can occur." + + " This property specifies the maximum number of messages this processor will hold in memory at one time." + + " CAUTION: Any incoming WebSocket message arrived while the queue being full" + + " will be discarded and a warning message will be logged.") + .required(true) + .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR) + .defaultValue("10000") + .build(); + + private volatile LinkedBlockingQueue<WebSocketMessage> incomingMessageQueue; --- End diff -- 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. > support for Listen WebSocket processor > --------------------------------------- > > Key: NIFI-1002 > URL: https://issues.apache.org/jira/browse/NIFI-1002 > Project: Apache NiFi > Issue Type: Improvement > Components: Extensions > Affects Versions: 0.4.0 > Reporter: sumanth chinthagunta > Priority: Minor > Labels: newbie > > A WebSocket listen processor will be helpful for IoT data ingestion. > I am playing with embedded Vert.X for WebSocket and also ability to put > FlowFiles back to WebSocket client via Vert.X EventBus. > https://github.com/xmlking/nifi-websocket > I am new to NiFi. any advise can be helpful. > PS: I feel forcing Interfaces for Controller Services is unnecessary as in > many cases Controller Services are only used by a set of Processors and > developers usually bundle them together. > -- This message was sent by Atlassian JIRA (v6.3.4#6332)