This is an automated email from the ASF dual-hosted git repository.
pvillard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new ba788d6e39 NIFI-11236 Removed ConsumeEWS Processor
ba788d6e39 is described below
commit ba788d6e399b8af0cbf563d46c9eb86524fda410
Author: exceptionfactory <[email protected]>
AuthorDate: Wed Mar 1 16:24:32 2023 -0600
NIFI-11236 Removed ConsumeEWS Processor
- Removed ews-java-api dependency
- Added commons-lang3 dependency to maintain current usage in other
Processors
Signed-off-by: Pierre Villard <[email protected]>
This closes #7001.
---
nifi-assembly/LICENSE | 24 -
.../src/main/resources/META-INF/LICENSE | 23 -
.../nifi-email-processors/pom.xml | 25 +-
.../apache/nifi/processors/email/ConsumeEWS.java | 581 ---------------------
.../services/org.apache.nifi.processor.Processor | 1 -
.../nifi/processors/email/ConsumeEWSTest.java | 54 --
6 files changed, 4 insertions(+), 704 deletions(-)
diff --git a/nifi-assembly/LICENSE b/nifi-assembly/LICENSE
index 3160381c1f..8f1a18e86c 100644
--- a/nifi-assembly/LICENSE
+++ b/nifi-assembly/LICENSE
@@ -2092,30 +2092,6 @@ The binary distribution of this product bundles
'HdrHistogram' which is availabl
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
THE POSSIBILITY OF SUCH DAMAGE.
-
-The binary distribution of this product bundles 'EWS JAVA API' which is
available
-under an MIT style license. For details see
https://github.com/OfficeDev/ews-java-api.
-
- Copyright (c) 2012 Microsoft Corporation
-
- Permission is hereby granted, free of charge, to any person obtaining a
copy
- of this software and associated documentation files (the "Software"), to
deal
- in the Software without restriction, including without limitation the
rights
- to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- copies of the Software, and to permit persons to whom the Software is
- furnished to do so, subject to the following conditions:
-
- The above copyright notice and this permission notice shall be included in
- all copies or substantial portions of the Software.
-
- THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM,
- OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
- THE SOFTWARE.
-
This product bundles 'libffi' which is available under an MIT style license.
libffi - Copyright (c) 1996-2014 Anthony Green, Red Hat, Inc and others.
see
https://github.com/java-native-access/jna/blob/master/native/libffi/LICENSE
diff --git
a/nifi-nar-bundles/nifi-email-bundle/nifi-email-nar/src/main/resources/META-INF/LICENSE
b/nifi-nar-bundles/nifi-email-bundle/nifi-email-nar/src/main/resources/META-INF/LICENSE
index 1b39d77a8c..73bf70d792 100644
---
a/nifi-nar-bundles/nifi-email-bundle/nifi-email-nar/src/main/resources/META-INF/LICENSE
+++
b/nifi-nar-bundles/nifi-email-bundle/nifi-email-nar/src/main/resources/META-INF/LICENSE
@@ -254,26 +254,3 @@
LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
-
-The binary distribution of this product bundles 'EWS JAVA API' which is
available
-under an MIT style license. For details see
https://github.com/OfficeDev/ews-java-api.
-
- Copyright (c) 2012 Microsoft Corporation
-
- Permission is hereby granted, free of charge, to any person obtaining a
copy
- of this software and associated documentation files (the "Software"), to
deal
- in the Software without restriction, including without limitation the
rights
- to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- copies of the Software, and to permit persons to whom the Software is
- furnished to do so, subject to the following conditions:
-
- The above copyright notice and this permission notice shall be included in
- all copies or substantial portions of the Software.
-
- THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM,
- OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
- THE SOFTWARE.
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml
b/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml
index b6fd7fa36a..1e9a54d454 100644
--- a/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml
+++ b/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml
@@ -64,6 +64,10 @@
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ </dependency>
<dependency>
<groupId>com.sun.mail</groupId>
<artifactId>javax.mail</artifactId>
@@ -89,27 +93,6 @@
<artifactId>findbugs-annotations</artifactId>
<version>1.3.9-1</version>
</dependency>
- <dependency>
- <groupId>com.microsoft.ews-java-api</groupId>
- <artifactId>ews-java-api</artifactId>
- <version>2.0</version>
- <exclusions>
- <exclusion>
- <groupId>commons-logging</groupId>
- <artifactId>commons-logging</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <!-- jaxws-api required for ews-java-api on Java 11 and later -->
- <dependency>
- <groupId>javax.xml.ws</groupId>
- <artifactId>jaxws-api</artifactId>
- <version>2.3.1</version>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>jcl-over-slf4j</artifactId>
- </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-ssl-context-service-api</artifactId>
diff --git
a/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ConsumeEWS.java
b/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ConsumeEWS.java
deleted file mode 100644
index ef676411bf..0000000000
---
a/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ConsumeEWS.java
+++ /dev/null
@@ -1,581 +0,0 @@
-/*
- * 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
microsoft.exchange.webservices.data.autodiscover.IAutodiscoverRedirectionUrl;
-import microsoft.exchange.webservices.data.core.ExchangeService;
-import microsoft.exchange.webservices.data.core.PropertySet;
-import
microsoft.exchange.webservices.data.core.enumeration.misc.ExchangeVersion;
-import microsoft.exchange.webservices.data.core.enumeration.property.BodyType;
-import
microsoft.exchange.webservices.data.core.enumeration.property.BasePropertySet;
-import
microsoft.exchange.webservices.data.core.enumeration.property.WellKnownFolderName;
-import
microsoft.exchange.webservices.data.core.enumeration.search.FolderTraversal;
-import
microsoft.exchange.webservices.data.core.enumeration.search.LogicalOperator;
-import
microsoft.exchange.webservices.data.core.enumeration.search.SortDirection;
-import
microsoft.exchange.webservices.data.core.enumeration.service.ConflictResolutionMode;
-import microsoft.exchange.webservices.data.core.enumeration.service.DeleteMode;
-import microsoft.exchange.webservices.data.core.service.folder.Folder;
-import microsoft.exchange.webservices.data.core.service.item.EmailMessage;
-import microsoft.exchange.webservices.data.core.service.item.Item;
-import
microsoft.exchange.webservices.data.core.service.schema.EmailMessageSchema;
-import microsoft.exchange.webservices.data.core.service.schema.FolderSchema;
-import microsoft.exchange.webservices.data.core.service.schema.ItemSchema;
-import microsoft.exchange.webservices.data.credential.ExchangeCredentials;
-import microsoft.exchange.webservices.data.credential.WebCredentials;
-import microsoft.exchange.webservices.data.property.complex.FileAttachment;
-import microsoft.exchange.webservices.data.property.complex.ItemAttachment;
-import microsoft.exchange.webservices.data.search.FindFoldersResults;
-import microsoft.exchange.webservices.data.search.FindItemsResults;
-import microsoft.exchange.webservices.data.search.FolderView;
-import microsoft.exchange.webservices.data.search.ItemView;
-import microsoft.exchange.webservices.data.search.filter.SearchFilter;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.mail.EmailAttachment;
-import org.apache.commons.mail.EmailException;
-import org.apache.commons.mail.HtmlEmail;
-import org.apache.commons.mail.MultiPartEmail;
-import org.apache.nifi.annotation.behavior.InputRequirement;
-import org.apache.nifi.annotation.documentation.CapabilityDescription;
-import org.apache.nifi.annotation.documentation.DeprecationNotice;
-import org.apache.nifi.annotation.documentation.Tags;
-import org.apache.nifi.annotation.lifecycle.OnStopped;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.components.Validator;
-import org.apache.nifi.expression.ExpressionLanguageScope;
-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 javax.mail.Address;
-import javax.mail.Flags;
-import javax.mail.Message;
-import javax.mail.MessagingException;
-import javax.mail.internet.MimeMessage;
-import javax.mail.util.ByteArrayDataSource;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.TimeUnit;
-
-@DeprecationNotice(
- reason = "The supporting ews-java-api library is no longer maintained"
-)
-@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
-@CapabilityDescription("Consumes messages from Microsoft Exchange using
Exchange Web Services. "
- + "The raw-bytes of each received email message are written as
contents of the FlowFile")
-@Tags({ "Email", "EWS", "Exchange", "Get", "Ingest", "Ingress", "Message",
"Consume" })
-public class ConsumeEWS extends AbstractProcessor {
- public static final PropertyDescriptor USER = new
PropertyDescriptor.Builder()
- .name("user")
- .displayName("User Name")
- .description("User Name used for authentication and authorization
with Email server.")
- .required(true)
-
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
- public static final PropertyDescriptor PASSWORD = new
PropertyDescriptor.Builder()
- .name("password")
- .displayName("Password")
- .description("Password used for authentication and authorization
with Email server.")
- .required(true)
-
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .sensitive(true)
- .build();
- public static final PropertyDescriptor FOLDER = new
PropertyDescriptor.Builder()
- .name("folder")
- .displayName("Folder")
- .description("Email folder to retrieve messages from (e.g.,
INBOX)")
- .required(true)
-
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .defaultValue("INBOX")
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
- public static final PropertyDescriptor FETCH_SIZE = new
PropertyDescriptor.Builder()
- .name("fetch.size")
- .displayName("Fetch Size")
- .description("Specify the maximum number of Messages to fetch per
call to Email Server.")
- .required(true)
-
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .defaultValue("10")
- .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .build();
- public static final PropertyDescriptor SHOULD_DELETE_MESSAGES = new
PropertyDescriptor.Builder()
- .name("delete.messages")
- .displayName("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 PropertyDescriptor CONNECTION_TIMEOUT = new
PropertyDescriptor.Builder()
- .name("connection.timeout")
- .displayName("Connection timeout")
- .description("The amount of time to wait to connect to Email
server")
- .required(true)
- .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
-
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .defaultValue("30 sec")
- .build();
- public static final PropertyDescriptor EXCHANGE_VERSION = new
PropertyDescriptor.Builder()
- .name("mail-ews-version")
- .displayName("Exchange Version")
- .description("What version of Exchange Server the server is
running.")
- .required(true)
- .allowableValues(ExchangeVersion.values())
- .defaultValue(ExchangeVersion.Exchange2010_SP2.name())
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor EWS_URL = new
PropertyDescriptor.Builder()
- .name("ews-url")
- .displayName("EWS URL")
- .description("URL of the EWS Endpoint. Required if Autodiscover is
false.")
- .required(false)
- .addValidator(StandardValidators.URL_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor USE_AUTODISCOVER = new
PropertyDescriptor.Builder()
- .name("ews-autodiscover")
- .displayName("Auto Discover URL")
- .description("Whether or not to use the Exchange email address to
Autodiscover the EWS endpoint URL.")
- .required(true)
- .allowableValues("true","false")
- .defaultValue("true")
- .build();
-
- public static final PropertyDescriptor SHOULD_MARK_READ = new
PropertyDescriptor.Builder()
- .name("ews-mark-as-read")
- .displayName("Mark Messages as Read")
- .description("Specify if messages should be marked as read after
retrieval.")
- .required(true)
- .allowableValues("true", "false")
- .defaultValue("true")
- .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor INCLUDE_EMAIL_HEADERS = new
PropertyDescriptor.Builder()
- .name("ews-include-headers")
- .displayName("Original Headers to Include")
- .description("Comma delimited list specifying which headers from
the original message to include in the exported email message. Blank means copy
all headers. " +
- "Some headers can cause problems with message parsing,
specifically the 'Content-Type' header.")
- .defaultValue("")
- .addValidator(Validator.VALID)
- .build();
-
- public static final PropertyDescriptor EXCLUDE_EMAIL_HEADERS = new
PropertyDescriptor.Builder()
- .name("ews-exclude-headers")
- .displayName("Original Headers to Exclude")
- .description("Comma delimited list specifying which headers from
the original message to exclude in the exported email message. Blank means
don't exclude any headers.")
- .defaultValue("")
- .addValidator(Validator.VALID)
- .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();
-
- final protected List<PropertyDescriptor> DESCRIPTORS;
-
- final protected Set<Relationship> RELATIONSHIPS;
-
- protected final Logger logger = LoggerFactory.getLogger(this.getClass());
-
- protected volatile BlockingQueue<Message> messageQueue;
-
- protected volatile String displayUrl;
-
- protected volatile ProcessSession processSession;
-
- protected volatile boolean shouldSetDeleteFlag;
-
- protected volatile String folderName;
-
- public ConsumeEWS(){
- final Set<Relationship> relationshipSet = new HashSet<>();
- relationshipSet.add(REL_SUCCESS);
-
- RELATIONSHIPS = relationshipSet;
-
- final List<PropertyDescriptor> descriptors = new ArrayList<>();
-
- descriptors.add(USER);
- descriptors.add(PASSWORD);
- descriptors.add(FOLDER);
- descriptors.add(FETCH_SIZE);
- descriptors.add(SHOULD_DELETE_MESSAGES);
- descriptors.add(CONNECTION_TIMEOUT);
- descriptors.add(EXCHANGE_VERSION);
- descriptors.add(EWS_URL);
- descriptors.add(USE_AUTODISCOVER);
- descriptors.add(SHOULD_MARK_READ);
- descriptors.add(INCLUDE_EMAIL_HEADERS);
- descriptors.add(EXCLUDE_EMAIL_HEADERS);
-
- DESCRIPTORS = descriptors;
- }
-
- @Override
- public Set<Relationship> getRelationships() {
- return RELATIONSHIPS;
- }
-
- @Override
- protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- return DESCRIPTORS;
- }
-
- @Override
- public void onTrigger(ProcessContext context, ProcessSession
processSession) throws ProcessException {
- if(this.messageQueue == null){
- int fetchSize =
context.getProperty(FETCH_SIZE).evaluateAttributeExpressions().asInteger();
- this.messageQueue = new ArrayBlockingQueue<>(fetchSize);
- }
-
- this.folderName =
context.getProperty(FOLDER).evaluateAttributeExpressions().getValue();
-
- Message emailMessage = this.receiveMessage(context);
- if (emailMessage != null) {
- this.transfer(emailMessage, context, processSession);
- } else {
- //No new messages found, yield the processor
- context.yield();
- }
- }
-
- protected ExchangeService initializeIfNecessary(ProcessContext context)
throws ProcessException {
- ExchangeVersion ver =
ExchangeVersion.valueOf(context.getProperty(EXCHANGE_VERSION).getValue());
- ExchangeService service = new ExchangeService(ver);
-
- final String timeoutInMillis =
String.valueOf(context.getProperty(CONNECTION_TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS));
- service.setTimeout(Integer.parseInt(timeoutInMillis));
-
- String userEmail =
context.getProperty(USER).evaluateAttributeExpressions().getValue();
- String password =
context.getProperty(PASSWORD).evaluateAttributeExpressions().getValue();
-
- ExchangeCredentials credentials = new WebCredentials(userEmail,
password);
- service.setCredentials(credentials);
-
- Boolean useAutodiscover =
context.getProperty(USE_AUTODISCOVER).asBoolean();
- if(useAutodiscover){
- try {
- service.autodiscoverUrl(userEmail, new
RedirectionUrlCallback());
- } catch (Exception e) {
- throw new ProcessException("Failure setting Autodiscover URL
from email address.", e);
- }
- } else {
- String ewsURL = context.getProperty(EWS_URL).getValue();
- try {
- service.setUrl(new URI(ewsURL));
- } catch (URISyntaxException e) {
- throw new ProcessException("Failure setting EWS URL.", e);
- }
- }
-
- return service;
- }
-
- // TODO remove this at next major release as dynamic properties are not
used by this processor
- @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();
- }
-
- /**
- * Return the target receivere's mail protocol (e.g., imap, pop etc.)
- */
- protected String getProtocol(ProcessContext processContext) {
- return "ews";
- }
-
- /**
- * Fills the internal message queue if such queue is empty. This is due to
- * the fact that per single session there may be multiple messages
retrieved
- * from the email server (see FETCH_SIZE).
- */
- protected void fillMessageQueueIfNecessary(ProcessContext context) throws
ProcessException {
- if (this.messageQueue.isEmpty()) {
- ExchangeService service = this.initializeIfNecessary(context);
- boolean deleteOnRead =
context.getProperty(SHOULD_DELETE_MESSAGES).getValue().equals("true");
- boolean markAsRead =
context.getProperty(SHOULD_MARK_READ).getValue().equals("true");
- String includeHeaders =
context.getProperty(INCLUDE_EMAIL_HEADERS).getValue();
- String excludeHeaders =
context.getProperty(EXCLUDE_EMAIL_HEADERS).getValue();
-
- List<String> includeHeadersList = null;
- List<String> excludeHeadersList = null;
-
- if (!StringUtils.isEmpty(includeHeaders)) {
- includeHeadersList = Arrays.asList(includeHeaders.split(","));
- }
-
- if (!StringUtils.isEmpty(excludeHeaders)) {
- excludeHeadersList = Arrays.asList(excludeHeaders.split(","));
- }
-
- try {
- //Get Folder
- Folder folder = getFolder(service);
-
- ItemView view = new ItemView(messageQueue.remainingCapacity());
- view.getOrderBy().add(ItemSchema.DateTimeReceived,
SortDirection.Ascending);
-
- SearchFilter sf = new
SearchFilter.SearchFilterCollection(LogicalOperator.And, new
SearchFilter.IsEqualTo(EmailMessageSchema.IsRead, false));
- FindItemsResults<Item> findResults =
service.findItems(folder.getId(), sf, view);
-
- if(findResults == null || findResults.getItems().size()== 0){
- return;
- }
-
- service.loadPropertiesForItems(findResults,
PropertySet.FirstClassProperties);
-
- for (Item item : findResults) {
- EmailMessage ewsMessage = (EmailMessage) item;
-
messageQueue.add(parseMessage(ewsMessage,includeHeadersList,excludeHeadersList));
-
- if(deleteOnRead){
- ewsMessage.delete(DeleteMode.HardDelete);
- } else if(markAsRead){
- ewsMessage.setIsRead(true);
-
ewsMessage.update(ConflictResolutionMode.AlwaysOverwrite);
- }
- }
-
- service.close();
- } catch (Exception e) {
- throw new ProcessException("Failed retrieving new messages
from EWS.", e);
- }
- }
- }
-
- protected Folder getFolder(ExchangeService service) {
- Folder folder;
- if(folderName.equals("INBOX")){
- try {
- folder = Folder.bind(service, WellKnownFolderName.Inbox);
- } catch (Exception e) {
- throw new ProcessException("Failed to bind Inbox Folder on EWS
Server", e);
- }
- } else {
- FolderView view = new FolderView(10);
- view.setTraversal(FolderTraversal.Deep);
- SearchFilter searchFilter = new
SearchFilter.IsEqualTo(FolderSchema.DisplayName, folderName);
- try {
- FindFoldersResults foldersResults =
service.findFolders(WellKnownFolderName.Root,searchFilter, view);
- ArrayList<Folder> folderIds = foldersResults.getFolders();
- if(folderIds.size() > 1){
- throw new ProcessException("More than 1 folder found with
the name " + folderName);
- }
-
- folder = Folder.bind(service, folderIds.get(0).getId());
- } catch (Exception e) {
- throw new ProcessException("Search for Inbox Subfolder
failed.", e);
- }
- }
- return folder;
- }
-
- public MimeMessage parseMessage(EmailMessage item, List<String>
hdrIncludeList, List<String> hdrExcludeList) throws Exception {
- EmailMessage ewsMessage = item;
- final String bodyText = ewsMessage.getBody().toString();
-
- MultiPartEmail mm;
-
- if(ewsMessage.getBody().getBodyType() == BodyType.HTML){
- mm = new HtmlEmail();
- if(!StringUtils.isEmpty(bodyText)){
- ((HtmlEmail)mm).setHtmlMsg(bodyText);
- }
- } else {
- mm = new MultiPartEmail();
- if(!StringUtils.isEmpty(bodyText)){
- mm.setMsg(bodyText);
- }
- }
- mm.setHostName("NiFi-EWS");
- //from
- mm.setFrom(ewsMessage.getFrom().getAddress());
- //to recipients
- ewsMessage.getToRecipients().forEach(x->{
- try {
- mm.addTo(x.getAddress());
- } catch (EmailException e) {
- throw new ProcessException("Failed to add TO recipient.", e);
- }
- });
- //cc recipients
- ewsMessage.getCcRecipients().forEach(x->{
- try {
- mm.addCc(x.getAddress());
- } catch (EmailException e) {
- throw new ProcessException("Failed to add CC recipient.", e);
- }
- });
- //subject
- mm.setSubject(ewsMessage.getSubject());
- //sent date
- mm.setSentDate(ewsMessage.getDateTimeSent());
- //add message headers
- ewsMessage.getInternetMessageHeaders().getItems().stream()
- .filter(x -> (hdrIncludeList == null ||
hdrIncludeList.isEmpty() || hdrIncludeList.contains(x.getName()))
- && (hdrExcludeList == null || hdrExcludeList.isEmpty()
|| !hdrExcludeList.contains(x.getName())))
- .forEach(x-> mm.addHeader(x.getName(), x.getValue()));
-
- //Any attachments
- if(ewsMessage.getHasAttachments()){
- ewsMessage.getAttachments().forEach(x->{
- try {
- if(x instanceof FileAttachment) {
- FileAttachment file = (FileAttachment) x;
- file.load();
-
- String type = file.getContentType() == null ?
"text/plain" : file.getContentType();
- ByteArrayDataSource bds = new
ByteArrayDataSource(file.getContent(), type);
-
- mm.attach(bds, file.getName(), "",
EmailAttachment.ATTACHMENT);
- } else { // x instanceof ItemAttachment
- ItemAttachment eml = (ItemAttachment) x;
- PropertySet oPropSetForBodyText = new
PropertySet(BasePropertySet.FirstClassProperties);
- oPropSetForBodyText.add(ItemSchema.MimeContent);
- eml.load(oPropSetForBodyText);
-
- Item it = eml.getItem();
- ByteArrayDataSource bds = new
ByteArrayDataSource(it.getMimeContent().getContent(), "text/plain");
-
- mm.attach(bds, eml.getName(), "",
EmailAttachment.ATTACHMENT);
- }
- } catch (MessagingException e) {
- e.printStackTrace();
- } catch (Exception e) {
- e.printStackTrace();
- }
- });
- }
-
- mm.buildMimeMessage();
- return mm.getMimeMessage();
- }
-
- /**
- * Disposes the message by converting it to a {@link FlowFile} transferring
- * it to the REL_SUCCESS relationship.
- */
- private void transfer(Message emailMessage, ProcessContext context,
ProcessSession processSession) {
- long start = System.nanoTime();
- FlowFile flowFile = processSession.create();
-
- flowFile = processSession.append(flowFile, new OutputStreamCallback() {
- @Override
- public void process(final OutputStream out) throws IOException {
- try {
- emailMessage.writeTo(out);
- } catch (MessagingException e) {
- throw new IOException(e);
- }
- }
- });
-
- long executionDuration =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
-
- String fromAddressesString = "";
- try {
- Address[] fromAddresses = emailMessage.getFrom();
- if (fromAddresses != null) {
- fromAddressesString = Arrays.asList(fromAddresses).toString();
- }
- } catch (MessagingException e) {
- this.logger.warn("Faild to retrieve 'From' attribute from
Message.");
- }
-
- processSession.getProvenanceReporter().receive(flowFile,
this.displayUrl, "Received message from " + fromAddressesString,
executionDuration);
- this.getLogger().info("Successfully received {} from {} in {} millis",
new Object[]{flowFile, fromAddressesString, executionDuration});
- processSession.transfer(flowFile, REL_SUCCESS);
-
- try {
- emailMessage.setFlag(Flags.Flag.DELETED, this.shouldSetDeleteFlag);
- } catch (MessagingException e) {
- this.logger.warn("Failed to set DELETE Flag on the message, data
duplication may occur.");
- }
- }
-
- /**
- * Receives message from the internal queue filling up the queue if
- * necessary.
- */
- protected Message receiveMessage(ProcessContext context) {
- Message emailMessage = null;
- try {
- this.fillMessageQueueIfNecessary(context);
- emailMessage = this.messageQueue.poll(1, TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- context.yield();
- this.logger.error("Failed retrieving messages from EWS.", e);
- Thread.currentThread().interrupt();
- this.logger.debug("Current thread is interrupted");
- }
- return emailMessage;
- }
-
-
-
- @OnStopped
- public void stop(ProcessContext processContext) {
- this.flushRemainingMessages(processContext);
- }
-
- /**
- * Will flush the remaining messages when this processor is stopped.
- */
- protected void flushRemainingMessages(ProcessContext processContext) {
- Message emailMessage;
- try {
- while ((emailMessage = this.messageQueue.poll(1,
TimeUnit.MILLISECONDS)) != null) {
- this.transfer(emailMessage, processContext,
this.processSession);
- this.processSession.commitAsync();
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- this.logger.debug("Current thread is interrupted");
- }
- }
-
- static class RedirectionUrlCallback implements IAutodiscoverRedirectionUrl
{
- public boolean autodiscoverRedirectionUrlValidationCallback(
- String redirectionUrl) {
- return redirectionUrl.toLowerCase().startsWith("https://";);
- }
- }
-}
diff --git
a/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
b/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index 8ff183b791..42187852c2 100644
---
a/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++
b/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -18,4 +18,3 @@ org.apache.nifi.processors.email.ExtractEmailHeaders
org.apache.nifi.processors.email.ListenSMTP
org.apache.nifi.processors.email.ConsumeIMAP
org.apache.nifi.processors.email.ConsumePOP3
-org.apache.nifi.processors.email.ConsumeEWS
\ No newline at end of file
diff --git
a/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/test/java/org/apache/nifi/processors/email/ConsumeEWSTest.java
b/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/test/java/org/apache/nifi/processors/email/ConsumeEWSTest.java
deleted file mode 100644
index b4312ae17d..0000000000
---
a/nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/test/java/org/apache/nifi/processors/email/ConsumeEWSTest.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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 org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-class ConsumeEWSTest {
-
- private static final String LOCALHOST_URL = "http://localhost:65000";;
-
- TestRunner runner;
-
- @BeforeEach
- void setRunner() {
- runner = TestRunners.newTestRunner(ConsumeEWS.class);
- }
-
- @Test
- void testRequiredProperties() {
- runner.setProperty(ConsumeEWS.USER, String.class.getSimpleName());
- runner.setProperty(ConsumeEWS.PASSWORD, String.class.getName());
-
- runner.assertValid();
- }
-
- @Test
- void testRunFailure() {
- runner.setProperty(ConsumeEWS.USER, String.class.getSimpleName());
- runner.setProperty(ConsumeEWS.PASSWORD, String.class.getName());
- runner.setProperty(ConsumeEWS.USE_AUTODISCOVER,
Boolean.FALSE.toString());
- runner.setProperty(ConsumeEWS.EWS_URL, LOCALHOST_URL);
-
- assertThrows(AssertionError.class, runner::run);
- }
-}
