You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by olegz <gi...@git.apache.org> on 2016/07/23 14:44:07 UTC
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
GitHub user olegz opened a pull request:
https://github.com/apache/nifi/pull/710
NIFI-1148 added IMAP/POP3 support
added initial set of processors to support consumption of Email via IMAP/IMAPS and POP3 protocols
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/olegz/nifi NIFI-1148
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/710.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #710
----
commit 2110809eed5dd3046fea1404ac29947f998d8c25
Author: Oleg Zhurakousky <ol...@suitcase.io>
Date: 2016-07-23T14:42:40Z
NIFI-1148 added IMAP/POP3 support
added initial set of processors to support consumption of Email via IMAP/IMAPS and POP3 protocols
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73710841
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/AbstractEmailProcessor.java ---
@@ -0,0 +1,400 @@
+/*
+ * 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.Flags;
+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")
--- End diff --
I can do that, especially since I am the one advocating it ;)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/710
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on the issue:
https://github.com/apache/nifi/pull/710
Getting errors when trying to use mailtrap.io for POP3 testing:
```
2016-07-25 10:37:27,706 ERROR [Timer-Driven Process Thread-6] o.a.nifi.processors.email.ConsumePOP3
org.apache.nifi.processor.exception.ProcessException: Failed to receive messages from Email server: [javax.mail.Mes
sagingException - error getting size
at org.apache.nifi.processors.email.AbstractEmailProcessor.fillMessageQueueIfNecessary(AbstractEmailProcess
or.java:311) ~[na:na]
at org.apache.nifi.processors.email.AbstractEmailProcessor.receiveMessage(AbstractEmailProcessor.java:373)
~[na:na]
at org.apache.nifi.processors.email.AbstractEmailProcessor.onTrigger(AbstractEmailProcessor.java:182) ~[na:
na]
at org.apache.nifi.processors.email.ConsumePOP3.onTrigger(ConsumePOP3.java:31) ~[na:na]
at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27) ~[nifi-api-1.0.0-SNAPSH
OT.jar:1.0.0-SNAPSHOT]
at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1077) [nifi-framew
ork-core-1.0.0-SNAPSHOT.jar:1.0.0-SNAPSHOT]
at org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:136)
[nifi-framework-core-1.0.0-SNAPSHOT.jar:1.0.0-SNAPSHOT]
at org.apache.nifi.controller.tasks.ContinuallyRunProcessorTask.call(ContinuallyRunProcessorTask.java:47) [
nifi-framework-core-1.0.0-SNAPSHOT.jar:1.0.0-SNAPSHOT]
at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:1
23) [nifi-framework-core-1.0.0-SNAPSHOT.jar:1.0.0-SNAPSHOT]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_91]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) [na:1.8.0_91]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecu
tor.java:180) [na:1.8.0_91]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.jav
a:294) [na:1.8.0_91]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_91]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_91]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_91]
Caused by: javax.mail.MessagingException: error getting size
at com.sun.mail.pop3.POP3Message.getSize(POP3Message.java:145) ~[mail-1.4.7.jar:1.4.7]
at javax.mail.internet.MimeMessage.<init>(MimeMessage.java:238) ~[mail-1.4.7.jar:1.4.7]
at org.springframework.integration.mail.AbstractMailReceiver$IntegrationMimeMessage.<init>(AbstractMailRece
iver.java:557) ~[na:na]
at org.springframework.integration.mail.AbstractMailReceiver$IntegrationMimeMessage.<init>(AbstractMailRece
iver.java:552) ~[na:na]
at org.springframework.integration.mail.AbstractMailReceiver.postProcessFilteredMessages(AbstractMailReceiv
er.java:415) ~[na:na]
at org.springframework.integration.mail.AbstractMailReceiver.receive(AbstractMailReceiver.java:342) ~[na:na
]
at org.apache.nifi.processors.email.AbstractEmailProcessor.fillMessageQueueIfNecessary(AbstractEmailProcess
or.java:306) ~[na:na]
... 15 common frames omitted
Caused by: java.io.IOException: Unexpected response: 1 95
at com.sun.mail.pop3.Protocol.readResponse(Protocol.java:752) ~[mail-1.4.7.jar:1.4.7]
at com.sun.mail.pop3.Protocol.simpleCommand(Protocol.java:692) ~[mail-1.4.7.jar:1.4.7]
at com.sun.mail.pop3.Protocol.list(Protocol.java:391) ~[mail-1.4.7.jar:1.4.7]
at com.sun.mail.pop3.POP3Message.getSize(POP3Message.java:138) ~[mail-1.4.7.jar:1.4.7]
... 21 common frames omitted
```
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73769916
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/docs/org.apache.nifi.processors.email.ConsumeIMAP/additionalDetails.html ---
@@ -0,0 +1,60 @@
+<!DOCTYPE html>
+<html lang="en">
+<!--
+ 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.
+ -->
+<head>
+<meta charset="utf-8" />
+<title>ConsumeIMAP</title>
+<link rel="stylesheet" href="../../css/component-usage.css"
+ type="text/css" />
+</head>
+
+<body>
+ <!-- Processor Documentation ================================================== -->
+ <h2>Description:</h2>
+ <p>This Processors consumes email messages via IMAP protocol and
+ sends the content of an email message as content of the FlowFile.
+ Content of the incoming email message is written as raw bytes to the
+ content of the outgoing Flow File.
+ </p>
+
+ <p>Since different serves may require different Java Mail
+ properties such properties could be provided via dynamic properties.
--- End diff --
that's a lot of "properties" in the same sentence ;)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the issue:
https://github.com/apache/nifi/pull/710
@mattyb149 I was just trying to reproduce the issue with the hanging thread you described above in relation to ConsumePop3, but I can't. Are you sure that what you saw is related to this processor or NiFo in general?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the issue:
https://github.com/apache/nifi/pull/710
@mattyb149 your comments are addressed in the last commit. Thanks for reviewing
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on the issue:
https://github.com/apache/nifi/pull/710
+1 LGTM, ran tests and verified comments were incorporated, merging to master
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71973534
--- 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 --
That is correct. What do you have in mind? I am open to suggestions. Basically my idea was "final disposition", but if it is not clear I am definitely open to change.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71975032
--- 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) {
+ if (this.messageReceiver == null) {
+ this.processSession = processSession;
+ this.messageReceiver = this.buildMessageReceiver(context);
+
+ boolean shouldDelete = context.getProperty(SHOULD_DELETE_MESSAGES).asBoolean();
+ int fetchSize = context.getProperty(FETCH_SIZE).evaluateAttributeExpressions().asInteger();
+
+ this.messageReceiver.setShouldDeleteMessages(shouldDelete);
--- End diff --
True(wrote significant portion of that code back in the days), but with regard to NiFi that is the configuration exposed to the user which they can chose to set to _true_ as it defaults to _false_.
To your point though it may be wise to set the delete flag on the message after we sent FF. Basically keep configuration the same as it is but instead of setting it as receiver's property do this: ```message.setFlag(Flags.Flag.DELETED, true);```.
I like!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the issue:
https://github.com/apache/nifi/pull/710
@trixpan Well, I am not sure what that is (will be looking) and how reliable mailtrap.io is for such testing giving it's description "Fake smtp testing server. Dummy smtp email testing". I've tested it with few real servers (gmail, yahoo etc) and all is good. I am not dismissing that there may be a problem but let's make sure we are not chasing some dummy error in mailtrap.io.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73805385
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/docs/org.apache.nifi.processors.email.ConsumeIMAP/additionalDetails.html ---
@@ -0,0 +1,60 @@
+<!DOCTYPE html>
+<html lang="en">
+<!--
+ 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.
+ -->
+<head>
+<meta charset="utf-8" />
+<title>ConsumeIMAP</title>
+<link rel="stylesheet" href="../../css/component-usage.css"
+ type="text/css" />
+</head>
+
+<body>
+ <!-- Processor Documentation ================================================== -->
+ <h2>Description:</h2>
+ <p>This Processors consumes email messages via IMAP protocol and
+ sends the content of an email message as content of the FlowFile.
+ Content of the incoming email message is written as raw bytes to the
+ content of the outgoing Flow File.
+ </p>
+
+ <p>Since different serves may require different Java Mail
+ properties such properties could be provided via dynamic properties.
+ For example, below is a sample configuration for GMail:
+ </p>
+ <p>
+ <b>Processor's static properties:</b>
+ <ul>
+ <li><b>Host Name</b> - imap.gmail.com</li>
+ <li><b>Port</b> - 993</li>
+ <li><b>User Name</b> - <i>[your user name]</i></li>
+ <li><b>Password</b> - <i>[your password]</i></li>
+ <li><b>Folder</b> - INBOX</li>
+ </ul>
+ <b>Processor's dynamic properties:</b>
+ <ul>
+ <li><b>mail.imap.socketFactory.class</b> - javax.net.ssl.SSLSocketFactory</li>
+ <li><b>mail.imap.socketFactory.fallback</b> - false</li>
+ <li><b>mail.store.protocol</b> - imaps</li>
--- End diff --
Unfortunately we can't do that, for reasons that there are no sensible defaults for protocols, only for providers. So if we ever get to something like ConsumeGmailViaIMAP, then we could eliminate most of not all dynamic properties, but unless we get into provider specific processors we have to stay pure to protocols and APIs.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on the issue:
https://github.com/apache/nifi/pull/710
I ran ConsumePop3 against GMail but hadn't seen your comment on all the dynamic properties that need to be set. I started and stopped the processor, but there is a thread continuing to run. Stopped NiFi but the thread kept going, had to be killed after 20 seconds. Maybe there's a timeout that can be set?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on the issue:
https://github.com/apache/nifi/pull/710
@olegz, seems like an option. How about adding an additional help file withbthe basics of what you already found?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71973578
--- 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) {
+ if (this.messageReceiver == null) {
+ this.processSession = processSession;
+ this.messageReceiver = this.buildMessageReceiver(context);
+
+ boolean shouldDelete = context.getProperty(SHOULD_DELETE_MESSAGES).asBoolean();
+ int fetchSize = context.getProperty(FETCH_SIZE).evaluateAttributeExpressions().asInteger();
+
+ this.messageReceiver.setShouldDeleteMessages(shouldDelete);
--- End diff --
That is not correct. This is an instruction to email server (similar to mark messages as read in IMAP).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on the issue:
https://github.com/apache/nifi/pull/710
@olegz agreed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73710769
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml ---
@@ -1,73 +1,84 @@
<?xml version="1.0" encoding="UTF-8"?>
-<!-- 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. -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
+<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor
--- End diff --
@mattyb149 I spoke with @JPercivall on this. This file was merged with formatting messed up so we agree that this PR will fix it.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the issue:
https://github.com/apache/nifi/pull/710
One advise though is that there are additional JavaMail properties that one may need to provide to accomodate specific servers. For example in Gmail you need to specify ```mail.pop3.socketFactory.class=javax.net.ssl.SSLSocketFactory``` and possibly few more (depending on your server settings). Those should be set as Dynamic properties since there is no common set. One property you should definitely set to get more info as to what is happening is ```mail.debug=true```. That will spit out all kinds of protocol messages which could explain what is going on.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73515235
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml ---
@@ -1,73 +1,84 @@
<?xml version="1.0" encoding="UTF-8"?>
-<!-- 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. -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
+<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor
--- End diff --
Lots of whitespace changes here but only a small number of added lines, maybe restore the original version and paste in the addition?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on the issue:
https://github.com/apache/nifi/pull/710
+1 to the "Additional Details" file for all the dynamic properties, etc.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73769714
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/docs/org.apache.nifi.processors.email.ConsumeIMAP/additionalDetails.html ---
@@ -0,0 +1,60 @@
+<!DOCTYPE html>
+<html lang="en">
+<!--
+ 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.
+ -->
+<head>
+<meta charset="utf-8" />
+<title>ConsumeIMAP</title>
+<link rel="stylesheet" href="../../css/component-usage.css"
+ type="text/css" />
+</head>
+
+<body>
+ <!-- Processor Documentation ================================================== -->
+ <h2>Description:</h2>
+ <p>This Processors consumes email messages via IMAP protocol and
+ sends the content of an email message as content of the FlowFile.
+ Content of the incoming email message is written as raw bytes to the
+ content of the outgoing Flow File.
+ </p>
+
+ <p>Since different serves may require different Java Mail
+ properties such properties could be provided via dynamic properties.
+ For example, below is a sample configuration for GMail:
+ </p>
+ <p>
+ <b>Processor's static properties:</b>
+ <ul>
+ <li><b>Host Name</b> - imap.gmail.com</li>
+ <li><b>Port</b> - 993</li>
+ <li><b>User Name</b> - <i>[your user name]</i></li>
+ <li><b>Password</b> - <i>[your password]</i></li>
+ <li><b>Folder</b> - INBOX</li>
+ </ul>
+ <b>Processor's dynamic properties:</b>
+ <ul>
+ <li><b>mail.imap.socketFactory.class</b> - javax.net.ssl.SSLSocketFactory</li>
+ <li><b>mail.imap.socketFactory.fallback</b> - false</li>
+ <li><b>mail.store.protocol</b> - imaps</li>
--- End diff --
I get that the properties are often different for different services, servers, etc. But is there something we can do with setting property defaults, such as socket factory or protocol, for example if "Use secure" is true? If dynamic properties are provided we'd use those, but if they were inadvertantly left out then the defaults would be populated.
The idea here is that there's a "Use Secure" property, but then you have to set the socket factory class to a secure factory, set the protocol to the secure version, etc. If there aren't widely used services that *require* these properties to be un-set, then I would think a better user experience would be for the "Use Secure" to take care of as much of the Java Mail property config as possible. Thoughts?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71975044
--- 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);
--- End diff --
This as well as other ideas that I have could be nice features for the future, so please do raise JIRA for that.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the issue:
https://github.com/apache/nifi/pull/710
@trixpan I can't say I am in favor of it given the fact that JavaMail Properties is a distinct part of Java Mail API and there are quite a number of them, so the question is how do you choose which once are more essential then others and what should their default values be if exposed as top level properties of a Processor? These are all tough questions to answer especially in the early stages of the processor development when we didn't really have a chance to analyze how it's been used and what are the common suggestions from users.
So what I would propose is to add a note to CapabilityDescription that "additional JavaMail properties should be set as Dynamic Properties". What do you think?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71975039
--- 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 --
+1
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71973483
--- 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 --
good one! thx
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71973968
--- 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) {
+ if (this.messageReceiver == null) {
+ this.processSession = processSession;
+ this.messageReceiver = this.buildMessageReceiver(context);
+
+ boolean shouldDelete = context.getProperty(SHOULD_DELETE_MESSAGES).asBoolean();
+ int fetchSize = context.getProperty(FETCH_SIZE).evaluateAttributeExpressions().asInteger();
+
+ this.messageReceiver.setShouldDeleteMessages(shouldDelete);
--- End diff --
Are you referring to the POP3 command `"DELE xxxxx"` ? If so, do not assume Trash as POP3 is notorious for lacking support of folders. Messages marked with DELE may be deleted by the server side after disconnect.
Spring seems to transparently handle that [here](https://github.com/spring-projects/spring-integration/blob/master/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java#L409):
```
private void postProcessFilteredMessages(Message[] filteredMessages) throws MessagingException {
setMessageFlags(filteredMessages);
if (shouldDeleteMessages()) {
deleteMessages(filteredMessages);
}
if (this.headerMapper == null) {
// Copy messages to cause an eager fetch
for (int i = 0; i < filteredMessages.length; i++) {
MimeMessage mimeMessage = new IntegrationMimeMessage((MimeMessage) filteredMessages[i]);
filteredMessages[i] = mimeMessage;
}
}
}
```
I am happy to test tomorrow morning but if Spring follows the traditional POP3 model it will perform a RETR followed by a DELE.
That could mean the returned message exists solely within the NiFi instance retrieving it. If by any chance the processor disposeMessage method hits an issue, the message would be lost.
I will double check it tomorrow morning (AEST)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
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?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73712664
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ConsumeIMAP.java ---
@@ -0,0 +1,91 @@
+/*
+ * 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.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.springframework.integration.mail.ImapMailReceiver;
+
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("Consumes messages from Email Server using IMAP protocol. "
+ + "The raw-bytes of each received email message are written as contents of the FlowFile")
+@Tags({ "Email", "Imap", "Get", "Ingest", "Ingress", "Message", "Consume" })
+public class ConsumeIMAP extends AbstractEmailProcessor<ImapMailReceiver> {
+
+ public static final PropertyDescriptor SHOULD_MARK_READ = new PropertyDescriptor.Builder()
+ .name("Mark Messages as Read")
+ .description("Specify if messages should be marked as read after retrieval.")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor USE_SSL = new PropertyDescriptor.Builder()
+ .name("Use SSL")
+ .description("Specifies if IMAP connection must be obtained via SSL encypted connection (i.e., IMAPS)")
--- End diff --
@mattyb149 Actually you're not missing anything and there is no extra configuration on the cline side (NIFI). This is all derived form the email server capabilities and how it's configured. For example if you want to test it with your existing GMail you would have to use "imaps". That is because Google has set it up as such. Does that help?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73515360
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/AbstractEmailProcessor.java ---
@@ -0,0 +1,400 @@
+/*
+ * 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.Flags;
+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")
--- End diff --
Not a requirement but the latest conventional wisdom says to put the friendly name in displayName() and something more machine-friendly in name(), this will help with internationalization later.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on the issue:
https://github.com/apache/nifi/pull/710
@olegz
I am not experienced enough in JAVA to vouch the quality of the code but I appreciate the simplicity of this processor.
Except for the small comments (and the action item on my side to validate my suspicion around RETR / DELE ) the work is looking great.
Thanks!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on the issue:
https://github.com/apache/nifi/pull/710
@olegz what about using the same strategy as PutEmail where the user can configure the socket on runtime?
https://github.com/apache/nifi/blob/master/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutEmail.java#L122
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71973547
--- 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) {
+ if (this.messageReceiver == null) {
+ this.processSession = processSession;
+ this.messageReceiver = this.buildMessageReceiver(context);
+
+ boolean shouldDelete = context.getProperty(SHOULD_DELETE_MESSAGES).asBoolean();
+ int fetchSize = context.getProperty(FETCH_SIZE).evaluateAttributeExpressions().asInteger();
+
+ this.messageReceiver.setShouldDeleteMessages(shouldDelete);
--- End diff --
I haven't tested fully, but am I correct to assume that Spring agent will delete the message upon retrieval, independent of the flow file being generated or not?
If that is the case, wouldn't this lead to message loss?
Wouldn't be safer to use the fetchMessages method followed by a deleteMessages one the message is sent to success?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71973640
--- 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);
--- End diff --
what do you think about the idea of adding a shared state + a property controlling the number of incorrect login attempts? This would prevent a processor with the incorrect password causing an account lockup against IMAP or POP3.
That would mimic the behavior used by both Mozilla and Outlook will not blindly try to login once a password fails.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by trixpan <gi...@git.apache.org>.
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r71973519
--- 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 --
Nit picking
it may be just me, but I found this method name confusing. If my understanding is correct, the method will transfer the message into success ?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73805537
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/docs/org.apache.nifi.processors.email.ConsumeIMAP/additionalDetails.html ---
@@ -0,0 +1,60 @@
+<!DOCTYPE html>
+<html lang="en">
+<!--
+ 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.
+ -->
+<head>
+<meta charset="utf-8" />
+<title>ConsumeIMAP</title>
+<link rel="stylesheet" href="../../css/component-usage.css"
+ type="text/css" />
+</head>
+
+<body>
+ <!-- Processor Documentation ================================================== -->
+ <h2>Description:</h2>
+ <p>This Processors consumes email messages via IMAP protocol and
+ sends the content of an email message as content of the FlowFile.
+ Content of the incoming email message is written as raw bytes to the
+ content of the outgoing Flow File.
+ </p>
+
+ <p>Since different serves may require different Java Mail
+ properties such properties could be provided via dynamic properties.
+ For example, below is a sample configuration for GMail:
+ </p>
+ <p>
+ <b>Processor's static properties:</b>
+ <ul>
+ <li><b>Host Name</b> - imap.gmail.com</li>
+ <li><b>Port</b> - 993</li>
+ <li><b>User Name</b> - <i>[your user name]</i></li>
+ <li><b>Password</b> - <i>[your password]</i></li>
+ <li><b>Folder</b> - INBOX</li>
+ </ul>
+ <b>Processor's dynamic properties:</b>
+ <ul>
+ <li><b>mail.imap.socketFactory.class</b> - javax.net.ssl.SSLSocketFactory</li>
+ <li><b>mail.imap.socketFactory.fallback</b> - false</li>
+ <li><b>mail.store.protocol</b> - imaps</li>
--- End diff --
Additionally as you have learned there is some provider work that needs to happen as well. For example with GMail you first have to enable IMAP and POP3 in your settings and then set additional security preferences to enable it.
Basically my point is that for these processors to work most of the information should be supplied by the email server administrator.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73518325
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ConsumeIMAP.java ---
@@ -0,0 +1,91 @@
+/*
+ * 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.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.springframework.integration.mail.ImapMailReceiver;
+
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("Consumes messages from Email Server using IMAP protocol. "
+ + "The raw-bytes of each received email message are written as contents of the FlowFile")
+@Tags({ "Email", "Imap", "Get", "Ingest", "Ingress", "Message", "Consume" })
+public class ConsumeIMAP extends AbstractEmailProcessor<ImapMailReceiver> {
+
+ public static final PropertyDescriptor SHOULD_MARK_READ = new PropertyDescriptor.Builder()
+ .name("Mark Messages as Read")
+ .description("Specify if messages should be marked as read after retrieval.")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor USE_SSL = new PropertyDescriptor.Builder()
+ .name("Use SSL")
+ .description("Specifies if IMAP connection must be obtained via SSL encypted connection (i.e., IMAPS)")
--- End diff --
Typo in "encrypted". Also the description should include any instructions/guidance on how to configure NiFi and/or the processor in order to use IMAPS, especially since the default value is true. When I tested this, I got the following error:
Failed to receive messages from Email server: [javax.mail.MessagingException - sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
I assume I'm missing something like a truststore but not sure where to set that up. Is it possible instead of a "Use SSL" property to use an SSLContext Controller Service?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73770089
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/docs/org.apache.nifi.processors.email.ConsumePOP3/additionalDetails.html ---
@@ -0,0 +1,59 @@
+<!DOCTYPE html>
+<html lang="en">
+<!--
+ 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.
+ -->
+<head>
+<meta charset="utf-8" />
+<title>ConsumePOP3</title>
+<link rel="stylesheet" href="../../css/component-usage.css"
+ type="text/css" />
+</head>
+
+<body>
+ <!-- Processor Documentation ================================================== -->
+ <h2>Description:</h2>
+ <p>This Processors consumes email messages via POP3 protocol and
+ sends the content of an email message as content of the FlowFile.
+ Content of the incoming email message is written as raw bytes to the
+ content of the outgoing Flow File.
+ </p>
+
+ <p>Since different serves may require different Java Mail
+ properties such properties could be provided via dynamic properties.
+ For example, below is a sample configuration for GMail:
+ </p>
+ <p>
+ <b>Processor's static properties:</b>
+ <ul>
+ <li><b>Host Name</b> - pop.gmail.com</li>
+ <li><b>Port</b> - 995</li>
+ <li><b>User Name</b> - <i>[your user name]</i></li>
+ <li><b>Password</b> - <i>[your password]</i></li>
+ <li><b>Folder</b> - INBOX</li>
+ </ul>
+ <b>Processor's dynamic properties:</b>
+ <ul>
+ <li><b>mail.pop3.socketFactory.class</b> - javax.net.ssl.SSLSocketFactory</li>
--- End diff --
Same comments about populating, typos, etc. as for ConsumeIMAP above.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the issue:
https://github.com/apache/nifi/pull/710
@mattyb149 all the suggested changes have been addressed in the last commit. I've also included in the additional documentation sample configuration for GMail, so you can use it to play around/test.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #710: NIFI-1148 added IMAP/POP3 support
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/710#discussion_r73769802
--- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/docs/org.apache.nifi.processors.email.ConsumeIMAP/additionalDetails.html ---
@@ -0,0 +1,60 @@
+<!DOCTYPE html>
+<html lang="en">
+<!--
+ 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.
+ -->
+<head>
+<meta charset="utf-8" />
+<title>ConsumeIMAP</title>
+<link rel="stylesheet" href="../../css/component-usage.css"
+ type="text/css" />
+</head>
+
+<body>
+ <!-- Processor Documentation ================================================== -->
+ <h2>Description:</h2>
+ <p>This Processors consumes email messages via IMAP protocol and
+ sends the content of an email message as content of the FlowFile.
+ Content of the incoming email message is written as raw bytes to the
+ content of the outgoing Flow File.
+ </p>
+
+ <p>Since different serves may require different Java Mail
+ properties such properties could be provided via dynamic properties.
+ For example, below is a sample configuration for GMail:
+ </p>
+ <p>
+ <b>Processor's static properties:</b>
+ <ul>
+ <li><b>Host Name</b> - imap.gmail.com</li>
+ <li><b>Port</b> - 993</li>
+ <li><b>User Name</b> - <i>[your user name]</i></li>
+ <li><b>Password</b> - <i>[your password]</i></li>
+ <li><b>Folder</b> - INBOX</li>
+ </ul>
+ <b>Processor's dynamic properties:</b>
+ <ul>
+ <li><b>mail.imap.socketFactory.class</b> - javax.net.ssl.SSLSocketFactory</li>
+ <li><b>mail.imap.socketFactory.fallback</b> - false</li>
+ <li><b>mail.store.protocol</b> - imaps</li>
+ </ul>
+ </p>
+ <p>
+ Another useful property to <b>mail.debug</b> which allows Java Mail API to print protocol messages to the console helping you to both understand what's going on as well as debug issues.
--- End diff --
I think this should be "Another useful property *is*" rather than *to*.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #710: NIFI-1148 added IMAP/POP3 support
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on the issue:
https://github.com/apache/nifi/pull/710
Digging into the source code for it looks like something is out of compliance with POP3 protocol
```java
if (line.startsWith("+OK"))
r.ok = true;
else if (line.startsWith("-ERR"))
r.ok = false;
else
throw new IOException("Unexpected response: " + line);
int i;
```
So, it appears that the received message is not really compliant with POP3, but will dig some more in the AM
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---