[
https://issues.apache.org/jira/browse/NIFI-1148?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15390712#comment-15390712
]
ASF GitHub Bot commented on NIFI-1148:
--------------------------------------
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71973345
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/AbstractEmailProcessor.java
---
@@ -0,0 +1,391 @@
+/*
+ * 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.email;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import javax.mail.Address;
+import javax.mail.Message;
+import javax.mail.MessagingException;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+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.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.support.StaticListableBeanFactory;
+import org.springframework.integration.mail.AbstractMailReceiver;
+import org.springframework.util.Assert;
+import org.springframework.util.StreamUtils;
+
+/**
+ * Base processor for implementing processors to consume messages from
Email
+ * servers using Spring Integration libraries.
+ *
+ * @param <T>
+ * the type of {@link AbstractMailReceiver}.
+ */
+abstract class AbstractEmailProcessor<T extends AbstractMailReceiver>
extends AbstractProcessor {
+
+ public static final PropertyDescriptor HOST = new
PropertyDescriptor.Builder()
+ .name("Host Name")
+ .description("Network address of Email server (e.g.,
pop.gmail.com, imap.gmail.com . . .)")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor PORT = new
PropertyDescriptor.Builder()
+ .name("Port")
+ .description("Numeric value identifying Port of Email server
(e.g., 993)")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor USER = new
PropertyDescriptor.Builder()
+ .name("User Name")
+ .description("User Name used for authentication and
authorization with Email server.")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor PASSWORD = new
PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password used for authentication and
authorization with Email server.")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .sensitive(true)
+ .build();
+ public static final PropertyDescriptor FOLDER = new
PropertyDescriptor.Builder()
+ .name("Folder")
+ .description("Email folder to retrieve messages from (e.g.,
INBOX)")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .defaultValue("INBOX")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor FETCH_SIZE = new
PropertyDescriptor.Builder()
+ .name("Fetch Size")
+ .description("Specify the maximum number of Messages to fetch
per call to Email Server.")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .defaultValue("10")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor SHOULD_DELETE_MESSAGES = new
PropertyDescriptor.Builder()
+ .name("Delete Messages")
+ .description("Specify whether mail messages should be deleted
after retrieval.")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .build();
+
+ static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("All messages that are the are successfully
received from Email server and converted to FlowFiles are routed to this
relationship")
+ .build();
+
+ static List<PropertyDescriptor> SHARED_DESCRIPTORS = new ArrayList<>();
+
+ static Set<Relationship> SHARED_RELATIONSHIPS = new HashSet<>();
+
+ /*
+ * Will ensure that list of PropertyDescriptors is build only once,
since
+ * all other lifecycle methods are invoked multiple times.
+ */
+ static {
+ SHARED_DESCRIPTORS.add(HOST);
+ SHARED_DESCRIPTORS.add(PORT);
+ SHARED_DESCRIPTORS.add(USER);
+ SHARED_DESCRIPTORS.add(PASSWORD);
+ SHARED_DESCRIPTORS.add(FOLDER);
+ SHARED_DESCRIPTORS.add(FETCH_SIZE);
+ SHARED_DESCRIPTORS.add(SHOULD_DELETE_MESSAGES);
+
+ SHARED_RELATIONSHIPS.add(REL_SUCCESS);
+ }
+
+ protected final Logger logger =
LoggerFactory.getLogger(this.getClass());
+
+ protected volatile T messageReceiver;
+
+ private volatile BlockingQueue<Message> messageQueue;
+
+ private volatile String displayUrl;
+
+ private volatile ProcessSession processSession;
+
+ @OnStopped
+ public void stop(ProcessContext processContext) {
+ this.flushRemainingMessages(processContext);
+ try {
+ this.messageReceiver.destroy();
+ this.messageReceiver = null;
+ } catch (Exception e) {
+ this.logger.warn("Failure while closing processor", e);
+ }
+ }
+
+ /**
+ *
+ */
+ @Override
+ public Set<Relationship> getRelationships() {
+ return SHARED_RELATIONSHIPS;
+ }
+
+ /**
+ *
+ */
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession
processSession) throws ProcessException {
+ this.initializefNecessary(context, processSession);
+
+ Message emailMessage = this.receiveMessage();
+ if (emailMessage != null) {
+ this.disposeMessage(emailMessage, context, processSession);
+ }
+ }
+
+ /**
+ *
+ */
+ @Override
+ protected PropertyDescriptor
getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .description("Specifies the value for '" +
propertyDescriptorName + "' Java Mail property.")
+
.name(propertyDescriptorName).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).dynamic(true)
+ .build();
+ }
+
+ /**
+ * Delegates to sub-classes to build the target receiver as
+ * {@link AbstractMailReceiver}
+ *
+ * @param context
+ * instance of {@link ProcessContext}
+ * @return new instance of {@link AbstractMailReceiver}
+ */
+ protected abstract T buildMessageReceiver(ProcessContext context);
+
+ /**
+ * Return the target receivere's mail protocol (e.g., imap, pop etc.)
+ */
+ protected abstract String getProtocol(ProcessContext processContext);
+
+ /**
+ * Builds the url used to connect to the email server.
+ */
+ String buildUrl(ProcessContext processContext) {
+ String host =
processContext.getProperty(HOST).evaluateAttributeExpressions().getValue();
+ String port =
processContext.getProperty(PORT).evaluateAttributeExpressions().getValue();
+ String user =
processContext.getProperty(USER).evaluateAttributeExpressions().getValue();
+ String password =
processContext.getProperty(PASSWORD).evaluateAttributeExpressions().getValue();
+ String folder =
processContext.getProperty(FOLDER).evaluateAttributeExpressions().getValue();
+
+ StringBuilder urlBuilder = new StringBuilder();
+ try {
+ urlBuilder.append(URLEncoder.encode(user, "UTF-8"));
+ } catch (UnsupportedEncodingException e) {
+ throw new ProcessException(e);
+ }
+ urlBuilder.append(":");
+ try {
+ urlBuilder.append(URLEncoder.encode(password, "UTF-8"));
+ } catch (UnsupportedEncodingException e) {
+ throw new ProcessException(e);
+ }
+ urlBuilder.append("@");
+ urlBuilder.append(host);
+ urlBuilder.append(":");
+ urlBuilder.append(port);
+ urlBuilder.append("/");
+ urlBuilder.append(folder);
+
+ String protocol = this.getProtocol(processContext);
+ String finalUrl = protocol + "://" + urlBuilder.toString();
+
+ // build display-safe URL
+ int passwordStartIndex = urlBuilder.indexOf(":") + 1;
+ int passwordEndIndex = urlBuilder.indexOf("@");
+ urlBuilder.replace(passwordStartIndex, passwordEndIndex,
"[password]");
+ this.displayUrl = protocol + "://" + urlBuilder.toString();
+ if (this.logger.isInfoEnabled()) {
+ this.logger.info("Connecting to Email server at the following
URL: " + this.displayUrl);
+ }
+
+ return finalUrl;
+ }
+
+ /**
+ * Builds and initializes the target message receiver if necessary (if
it's
+ * null). Upon execution of this operation the receiver is fully
functional
+ * and is ready to receive messages.
+ */
+ private synchronized void initializefNecessary(ProcessContext context,
ProcessSession processSession) {
--- End diff --
method name typo?
> Add processor to GetEmail supporting IMAP and POP3
> --------------------------------------------------
>
> Key: NIFI-1148
> URL: https://issues.apache.org/jira/browse/NIFI-1148
> Project: Apache NiFi
> Issue Type: New Feature
> Components: Extensions
> Reporter: Joseph Witt
> Assignee: Oleg Zhurakousky
> Fix For: 1.0.0
>
>
> It is fairly common that users want to be able to acquire data via email.
> This means both IMAP and POP3. POP3 is easier as it is a sort of fire/forget
> model whereas IMAP involves more state handling. But in any event both modes
> are important to support.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
