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, 17 May 2016 18:33:12 GMT

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

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

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

    https://github.com/apache/nifi/pull/392#discussion_r63578415
  
    --- Diff: nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/PublishMQTT.java
---
    @@ -0,0 +1,239 @@
    +/*
    + * 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.annotation.behavior.SupportsBatching;
    +import org.apache.nifi.annotation.lifecycle.OnStopped;
    +import org.apache.nifi.components.PropertyDescriptor;
    +import org.apache.nifi.expression.AttributeExpression;
    +import org.apache.nifi.flowfile.FlowFile;
    +import org.apache.nifi.annotation.lifecycle.OnScheduled;
    +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.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.InputStreamCallback;
    +import org.apache.nifi.annotation.behavior.InputRequirement;
    +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
    +
    +import org.apache.nifi.processor.util.StandardValidators;
    +import org.apache.nifi.processors.mqtt.common.AbstractMQTTProcessor;
    +import org.apache.nifi.stream.io.StreamUtils;
    +import org.eclipse.paho.client.mqttv3.MqttCallback;
    +import org.eclipse.paho.client.mqttv3.MqttException;
    +import org.eclipse.paho.client.mqttv3.MqttMessage;
    +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    +
    +import java.util.Arrays;
    +import java.util.Collections;
    +import java.util.HashSet;
    +import java.util.List;
    +import java.util.Set;
    +import java.io.InputStream;
    +import java.io.IOException;
    +
    +@SupportsBatching
    +@InputRequirement(Requirement.INPUT_REQUIRED)
    +@Tags({"publish", "MQTT", "IOT"})
    +@CapabilityDescription("Publishes a message to an MQTT topic")
    +@SeeAlso({ConsumeMQTT.class})
    +public class PublishMQTT extends AbstractMQTTProcessor {
    +
    +    public static final PropertyDescriptor PROP_TOPIC = new PropertyDescriptor.Builder()
    +            .name("Topic")
    +            .description("The topic to publish the message to.")
    +            .expressionLanguageSupported(true)
    +            .required(true)
    +            .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING,
true))
    +            .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 send the message with. Accepts
three values '0', '1' and '2'; '0' for 'at most once', '1' for 'at least once', '2' for 'exactly
once'. " +
    +                    "Expression language is allowed in order to support publishing messages
with different QoS but the end value of the property must be either '0', '1' or '2'. ")
    +            .required(true)
    +            .expressionLanguageSupported(true)
    +            .addValidator(QoSValidator)
    +            .build();
    +
    +    public static final PropertyDescriptor PROP_RETAIN = new PropertyDescriptor.Builder()
    +            .name("Retain Message")
    +            .description("Whether or not the retain flag should be set on the MQTT message.")
    +            .required(true)
    +            .expressionLanguageSupported(true)
    +            .addValidator(RetainValidator)
    +            .build();
    +
    +    public static final Relationship REL_SUCCESS = new Relationship.Builder()
    +            .name("success")
    +            .description("FlowFiles that are sent successfully to the destination are
sent out this relationship.")
    +            .build();
    +    public static final Relationship REL_FAILURE = new Relationship.Builder()
    +            .name("failure")
    +            .description("FlowFiles that failed to send to the destination are sent out
this relationship.")
    +            .build();
    +
    +    private static final List<PropertyDescriptor> descriptors;
    +    private static final Set<Relationship> relationships;
    +
    +    static {
    +        final List<PropertyDescriptor> innerDescriptorsList = getAbstractPropertyDescriptors();
    +        innerDescriptorsList.add(PROP_TOPIC);
    +        innerDescriptorsList.add(PROP_QOS);
    +        innerDescriptorsList.add(PROP_RETAIN);
    +        descriptors = Collections.unmodifiableList(innerDescriptorsList);
    +
    +        final Set<Relationship> innerRelationshipsSet = new HashSet<>();
    +        innerRelationshipsSet.add(REL_SUCCESS);
    +        innerRelationshipsSet.add(REL_FAILURE);
    +        relationships = Collections.unmodifiableSet(innerRelationshipsSet);
    +    }
    +
    +    @Override
    +    protected void init(final ProcessorInitializationContext context) {
    +        logger = getLogger();
    +    }
    +
    +    @Override
    +    public Set<Relationship> getRelationships() {
    +        return relationships;
    +    }
    +
    +    @Override
    +    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
    +        return descriptors;
    +    }
    +
    +    @OnScheduled
    +    public void onScheduled(final ProcessContext context) {
    +        buildClient(context);
    +    }
    +
    +    @OnStopped
    +    public void onStop(final ProcessContext context) {
    +        try {
    +            synchronized (mqttClientConnectLock) {
    +                if (mqttClient != null && mqttClient.isConnected()) {
    +                    mqttClient.disconnect();
    +                    logger.info("Disconnected the MQTT client.");
    +                }
    +            }
    +        } catch(MqttException me) {
    +            logger.error("Failed when disconnecting the MQTT client.", me);
    +        }
    +    }
    +
    +    @Override
    +    public void onTrigger(final ProcessContext context, final ProcessSession session)
throws ProcessException {
    +        FlowFile flowfile = session.get();
    +        if (flowfile == null) {
    +            return;
    +        }
    +
    +        if(mqttClient == null || !mqttClient.isConnected()){
    +            logger.info("Was disconnected from client or was never connected, attempting
to connect.");
    +            synchronized (mqttClientConnectLock){
    +                try {
    +                    reconnect();
    +                } catch (MqttException e) {
    +                    context.yield();
    +                    session.transfer(flowfile, REL_FAILURE);
    +                    logger.error("MQTT client is disconnected and re-connecting failed.
Transferring FlowFile to fail and yielding", e);
    +                    return;
    +                }
    +            }
    +        }
    +
    +        // get the MQTT topic
    +        String topic = context.getProperty(PROP_TOPIC).evaluateAttributeExpressions(flowfile).getValue();
    +
    +        if (topic == null || topic.isEmpty()) {
    +            logger.warn("Evaluation of the topic property returned null or evaluated
to be empty, routing to failure");
    +            session.transfer(flowfile, REL_FAILURE);
    +            return;
    +        }
    +
    +        // do the read
    +        final byte[] messageContent = new byte[(int) flowfile.getSize()];
    +        session.read(flowfile, new InputStreamCallback() {
    +            @Override
    +            public void process(final InputStream in) throws IOException {
    +                StreamUtils.fillBuffer(in, messageContent, true);
    +            }
    +        });
    +
    +        int qos = context.getProperty(PROP_QOS).evaluateAttributeExpressions(flowfile).asInteger();
    --- End diff --
    
    Same null check comment


> 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