Github user trixpan commented on a diff in the pull request: https://github.com/apache/nifi/pull/710#discussion_r71973688 --- 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); --- End diff -- I suspected that was the case, challenge is that disposal has a negative "to the bin" connotation that leads to confusion. (process|transfer)Message? In NiFi lingo we always transfer the message into a relationship so it may be a good candidate.
--- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---