nifi-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (NIFI-1808) Create General MQTT Processors
Date Tue, 03 May 2016 20:28:13 GMT

    [ https://issues.apache.org/jira/browse/NIFI-1808?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15269518#comment-15269518
] 

ASF GitHub Bot commented on NIFI-1808:
--------------------------------------

Github user JPercivall commented on a diff in the pull request:

    https://github.com/apache/nifi/pull/392#discussion_r61949787
  
    --- Diff: nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/SubscribeMQTT.java
---
    @@ -0,0 +1,301 @@
    +/*
    + * 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.mqtt;
    +
    +import org.apache.nifi.components.PropertyDescriptor;
    +import org.apache.nifi.components.PropertyValue;
    +import org.apache.nifi.flowfile.FlowFile;
    +import org.apache.nifi.annotation.behavior.WritesAttribute;
    +import org.apache.nifi.annotation.behavior.WritesAttributes;
    +import org.apache.nifi.annotation.lifecycle.OnScheduled;
    +import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
    +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.logging.ProcessorLog;
    +import org.apache.nifi.processor.ProcessContext;
    +import org.apache.nifi.processor.ProcessorInitializationContext;
    +import org.apache.nifi.processor.Relationship;
    +import org.apache.nifi.processor.exception.ProcessException;
    +import org.apache.nifi.processor.ProcessSession;
    +import org.apache.nifi.processor.io.OutputStreamCallback;
    +
    +import org.apache.nifi.processor.util.StandardValidators;
    +import org.apache.nifi.processors.mqtt.common.AbstractMQTTProcessor;
    +import org.apache.nifi.processors.mqtt.common.MQTTQueueMessage;
    +import org.apache.nifi.ssl.SSLContextService;
    +import org.eclipse.paho.client.mqttv3.MqttCallback;
    +import org.eclipse.paho.client.mqttv3.MqttClient;
    +import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    +import org.eclipse.paho.client.mqttv3.MqttException;
    +import org.eclipse.paho.client.mqttv3.MqttMessage;
    +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
    +
    +import java.util.Collections;
    +import java.util.HashSet;
    +import java.util.List;
    +import java.util.Properties;
    +import java.util.Set;
    +import java.util.concurrent.LinkedBlockingQueue;
    +import java.io.OutputStream;
    +import java.io.IOException;
    +import java.util.concurrent.ScheduledExecutorService;
    +import java.util.concurrent.ScheduledFuture;
    +import java.util.concurrent.ScheduledThreadPoolExecutor;
    +import java.util.concurrent.TimeUnit;
    +
    +
    +@Tags({"subscribe", "MQTT", "IOT"})
    +@CapabilityDescription("Subscribes to a topic and receives messages from an MQTT broker")
    +@SeeAlso({PublishMQTT.class})
    +@WritesAttributes({@WritesAttribute(attribute="broker", description="MQTT broker that
was the message source"),
    +    @WritesAttribute(attribute="topic", description="MQTT topic on which message was
received"),
    +    @WritesAttribute(attribute="qos", description="The quality of service for this message."),
    +    @WritesAttribute(attribute="isDuplicate", description="Whether or not this message
might be a duplicate of one which has already been received."),
    +    @WritesAttribute(attribute="isRetained", description="Whether or not this message
was from a current publisher, or was \"retained\" by the server as the last message published
on the topic.")})
    +public class SubscribeMQTT extends AbstractMQTTProcessor {
    +
    +    public static final PropertyDescriptor PROP_TOPIC_FILTER = new PropertyDescriptor.Builder()
    +            .name("Topic Filter")
    +            .description("The MQTT topic filter to designate the topics to subscribe
to.")
    +            .required(true)
    +            .expressionLanguageSupported(false)
    +            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
    +            .build();
    +
    +    public static final PropertyDescriptor PROP_QOS = new PropertyDescriptor.Builder()
    +            .name("Quality of Service(QoS)")
    +            .description("The Quality of Service(QoS) to receive the message with. Accepts
three values '0', '1' and '2'; '0' for 'at most once', '1' for 'at least once', '2' for 'exactly
once'.")
    +            .required(true)
    +            .allowableValues("0", "1", "2")
    +            .build();
    +
    +
    +    public static final PropertyDescriptor PROP_MAX_QUEUE_SIZE = new PropertyDescriptor.Builder()
    +            .name("Max Queue Size")
    +            .description("The MQTT messages are always being sent to subscribers on a
topic. 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.")
    +            .required(true)
    +            .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
    +            .build();
    +
    +    private String broker;
    +    private ProcessorLog logger;
    +    private MqttConnectOptions connOpts;
    +    private MqttClient mqttClient;
    +    private final Object mqttClientLock = new Object();
    +    private ScheduledExecutorService autoReconnectExecutor;
    +    private final Object autoReconnectLock = new Object();
    +    private ScheduledFuture<?> autoReconnectScheduledFuture;
    +    private static int AUTO_RECONNECT_PERIOD = 5;
    +    private static int DISCONNECT_TIMEOUT = 5000;
    +    private long maxQueueSize;
    +
    +    private MemoryPersistence persistence = new MemoryPersistence();
    +
    +    LinkedBlockingQueue<MQTTQueueMessage> mqttQueue = new LinkedBlockingQueue<>();
    +
    +    public static final Relationship REL_MESSAGE = new Relationship.Builder()
    +            .name("Message")
    +            .description("The MQTT message output")
    +            .build();
    +
    +     private List<PropertyDescriptor> descriptors;
    --- End diff --
    
    Good catch, will fix


> Create General MQTT Processors
> ------------------------------
>
>                 Key: NIFI-1808
>                 URL: https://issues.apache.org/jira/browse/NIFI-1808
>             Project: Apache NiFi
>          Issue Type: Improvement
>            Reporter: Joseph Percivall
>            Assignee: Joseph Percivall
>
> MQTT[1] is a great "Internet of Things" (IoT) connectivity protocol that implementing
processors for would allow NiFi to continue expanding into the IoT domain. A prime opportunity
would be to use in conjunction with Apache NiFi - MiNiFi.
> [1] http://mqtt.org/



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message