|
|
@@ -0,0 +1,111 @@
|
|
|
+/*
|
|
|
+ * 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 fr.chickenkiller.nifi.processors.xmpp;
|
|
|
+
|
|
|
+import java.util.ArrayList;
|
|
|
+import java.util.Collections;
|
|
|
+import java.util.HashSet;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Set;
|
|
|
+
|
|
|
+import org.apache.nifi.annotation.behavior.InputRequirement;
|
|
|
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
|
|
|
+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.processor.AbstractProcessor;
|
|
|
+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.util.StandardValidators;
|
|
|
+
|
|
|
+@Tags({ "xmpp", "conversation", "chat.postMessage", "social media", "team", "text", "unstructured", "read" })
|
|
|
+@CapabilityDescription("Consumes messages from a XMPP server")
|
|
|
+@SeeAlso({})
|
|
|
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
|
|
|
+@WritesAttributes({ @WritesAttribute(attribute = "xmpp.sender.jid", description = "Message Sender JID"),
|
|
|
+ @WritesAttribute(attribute = "xmpp.sending.date", description = "Sending date of the message") })
|
|
|
+public class ConsumeXMPP extends AbstractProcessor {
|
|
|
+
|
|
|
+ public static final PropertyDescriptor SERVER = new PropertyDescriptor.Builder().name("XMPP Server")
|
|
|
+ .description("jabber.org").required(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
|
|
|
+
|
|
|
+ public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder().name("PORT")
|
|
|
+ .displayName("XMPP Server port").description("5222").defaultValue("5222")
|
|
|
+ .addValidator(StandardValidators.PORT_VALIDATOR).build();
|
|
|
+
|
|
|
+ public static final PropertyDescriptor JID = new PropertyDescriptor.Builder().name("Receiving JID").required(true)
|
|
|
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
|
|
|
+
|
|
|
+ public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder().name("JID Password")
|
|
|
+ .required(true).sensitive(true).build();
|
|
|
+
|
|
|
+ public static final Relationship SUCCESS = new Relationship.Builder().name("success")
|
|
|
+ .description("Message received and successfully parsed").build();
|
|
|
+
|
|
|
+ public static final Relationship FAILURE = new Relationship.Builder().name("failure")
|
|
|
+ .description("Problem when receiving message").build();
|
|
|
+
|
|
|
+ private List<PropertyDescriptor> descriptors;
|
|
|
+
|
|
|
+ private Set<Relationship> relationships;
|
|
|
+
|
|
|
+ @Override
|
|
|
+ protected void init(final ProcessorInitializationContext context) {
|
|
|
+ descriptors = new ArrayList<>();
|
|
|
+ descriptors.add(SERVER);
|
|
|
+ descriptors.add(PORT);
|
|
|
+ descriptors.add(JID);
|
|
|
+ descriptors.add(PASSWORD);
|
|
|
+ descriptors = Collections.unmodifiableList(descriptors);
|
|
|
+
|
|
|
+ relationships = new HashSet<>();
|
|
|
+ relationships.add(SUCCESS);
|
|
|
+ relationships.add(FAILURE);
|
|
|
+ relationships = Collections.unmodifiableSet(relationships);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public Set<Relationship> getRelationships() {
|
|
|
+ return this.relationships;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
|
|
|
+ return descriptors;
|
|
|
+ }
|
|
|
+
|
|
|
+ @OnScheduled
|
|
|
+ public void onScheduled(final ProcessContext context) {
|
|
|
+
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void onTrigger(final ProcessContext context, final ProcessSession session) {
|
|
|
+ FlowFile flowFile = session.get();
|
|
|
+ if (flowFile == null) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ session.transfer(flowFile, SUCCESS);
|
|
|
+ }
|
|
|
+}
|
