[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user asfgit closed the pull request at: https://github.com/apache/nifi/pull/483 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70723234
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70721418
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70714800
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,159 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+ByteArrayOutputStream messageData;
+
+private CountDownLatch latch;
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+this.latch = new CountDownLatch(1);
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(true);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+final long serverTimeout =
TimeUnit.MILLISECONDS.convert(messageContext.getSMTPServer().getConnectionTimeout(),
TimeUnit.MILLISECONDS);
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+
+int rd;
+
+try {
+while ((rd = inputStream.read(buffer, 0, buffer.length))
!= -1 ) {
+baos.write(buffer, 0, rd);
+if (baos.getBufferLength() >
server.getMaxMessageSize() ) {
+throw new TooMuchDataException("Data exceeds the
amount allowed.");
+}
+}
+baos.flush();
+} catch (IOException e) {
+throw new DropConnectionException(450, "Unexpected error
processing your message. ");
+}
+
+this.messageData = baos;
+
+X509Certificate[
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70714764
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,159 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+ByteArrayOutputStream messageData;
+
+private CountDownLatch latch;
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+this.latch = new CountDownLatch(1);
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(true);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+final long serverTimeout =
TimeUnit.MILLISECONDS.convert(messageContext.getSMTPServer().getConnectionTimeout(),
TimeUnit.MILLISECONDS);
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+
+int rd;
+
+try {
+while ((rd = inputStream.read(buffer, 0, buffer.length))
!= -1 ) {
+baos.write(buffer, 0, rd);
+if (baos.getBufferLength() >
server.getMaxMessageSize() ) {
+throw new TooMuchDataException("Data exceeds the
amount allowed.");
+}
+}
+baos.flush();
+} catch (IOException e) {
+throw new DropConnectionException(450, "Unexpected error
processing your message. ");
+}
+
+this.messageData = baos;
+
+X509Certificate[
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70639371
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70635447
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,159 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+ByteArrayOutputStream messageData;
+
+private CountDownLatch latch;
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+this.latch = new CountDownLatch(1);
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(true);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+final long serverTimeout =
TimeUnit.MILLISECONDS.convert(messageContext.getSMTPServer().getConnectionTimeout(),
TimeUnit.MILLISECONDS);
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+
+int rd;
+
+try {
+while ((rd = inputStream.read(buffer, 0, buffer.length))
!= -1 ) {
+baos.write(buffer, 0, rd);
+if (baos.getBufferLength() >
server.getMaxMessageSize() ) {
+throw new TooMuchDataException("Data exceeds the
amount allowed.");
+}
+}
+baos.flush();
+} catch (IOException e) {
+throw new DropConnectionException(450, "Unexpected error
processing your message. ");
+}
+
+this.messageData = baos;
+
+X509Certifica
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70634756
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,159 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+ByteArrayOutputStream messageData;
+
+private CountDownLatch latch;
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+this.latch = new CountDownLatch(1);
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(true);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+final long serverTimeout =
TimeUnit.MILLISECONDS.convert(messageContext.getSMTPServer().getConnectionTimeout(),
TimeUnit.MILLISECONDS);
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+
+int rd;
+
+try {
+while ((rd = inputStream.read(buffer, 0, buffer.length))
!= -1 ) {
+baos.write(buffer, 0, rd);
+if (baos.getBufferLength() >
server.getMaxMessageSize() ) {
+throw new TooMuchDataException("Data exceeds the
amount allowed.");
+}
+}
+baos.flush();
+} catch (IOException e) {
+throw new DropConnectionException(450, "Unexpected error
processing your message. ");
+}
+
+this.messageData = baos;
+
+X509Certifica
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70633207
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
--- End diff --
You're right you can't tell how much data will be sent without reading it.
The current implementation is probably fine but if you wanted to improve it
more, you could add a flag in SmtpEvent that allows the onTrigger to tell
data() that something went wrong.
This would allow you to read the data in the onTrigger, determine if it
went over and then properly handle it in the data() method. It could
potentially open up the onTrigger to let data() know of other errors but I'm
not sure there is anything the sender of the email really care
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70615355
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
--- End diff --
I understood that, however, from what I gather I cannot determine the
amount of data an InputStream contains or will contain until I read it?
Since we must tell the client he exceeded the limit, I chose to perform the
reading loop within the handler rather than sending it down the queue to the
`onTrigger` thread for reading and then returning error from there.
Having said that, I am open to suggestions.
---
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
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70612776
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70612760
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70612563
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70574245
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,155 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+private CountDownLatch latch;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+this.latch = new CountDownLatch(1);
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+ByteArrayOutputStream messageData;
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(true);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+final long serverTimeout =
TimeUnit.MILLISECONDS.convert(messageContext.getSMTPServer().getConnectionTimeout(),
TimeUnit.MILLISECONDS);
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+
+int rd;
+
+try {
+while ((rd = inputStream.read(buffer, 0, buffer.length))
!= -1 ) {
+baos.write(buffer, 0, rd);
+if (baos.getBufferLength() >
server.getMaxMessageSize() ) {
+throw new TooMuchDataException("Data exceeds the
amount allowed.");
+}
+}
+baos.flush();
+} catch (IOException e) {
+throw new DropConnectionException(450, "Unexpected error
processing your message. ");
+}
+
+this.messageData = baos;
+
+X509Certificate[] certificates = new X509C
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70574121
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,155 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+private CountDownLatch latch;
--- End diff --
ouch. addressed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70565378
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70562091
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70561694
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,155 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+private CountDownLatch latch;
--- End diff --
Shouldn't this be moved to the Handler class? If I understand it correctly,
a handler gets created for each message. By having this latch in
SMTPMessageHandlerFactory, the only the first message will properly wait.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70561569
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,155 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+private CountDownLatch latch;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+this.latch = new CountDownLatch(1);
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+ByteArrayOutputStream messageData;
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(true);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+final long serverTimeout =
TimeUnit.MILLISECONDS.convert(messageContext.getSMTPServer().getConnectionTimeout(),
TimeUnit.MILLISECONDS);
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+
+int rd;
+
+try {
+while ((rd = inputStream.read(buffer, 0, buffer.length))
!= -1 ) {
+baos.write(buffer, 0, rd);
+if (baos.getBufferLength() >
server.getMaxMessageSize() ) {
+throw new TooMuchDataException("Data exceeds the
amount allowed.");
+}
+}
+baos.flush();
+} catch (IOException e) {
+throw new DropConnectionException(450, "Unexpected error
processing your message. ");
+}
+
+this.messageData = baos;
+
+X509Certificate[] certificates = new X5
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70561030
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
--- End diff --
This doesn't address passing a stream instead of bytes. The
ByteArrayOutputStream is backed by a byte array[1] so it's getting copied into
memory anyway. I'm suggesting to pass the InputStream itself so that in the
onTrigger you can just stream it from input to the content.
[1]
https://docs.oracle.com/javase/7/docs/api/java/io/ByteArrayOutputStream.html
---
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
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70457549
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
--- End diff --
addressed on both
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70457237
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
--- End diff --
partially addressed. Instead of `messageBody `has been replaced with
`messageContent `and passed as a `ByteBufferOutputStream `back to `ontrigger`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70456956
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70366557
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70366486
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
--- End diff --
I was aware of the reason just the implementation was flawed. I need to
revisit this habit of coding late in the evening. :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70288235
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
--- End diff --
sounds likes a great idea. will have a look at 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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70287734
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70286727
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,209 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.FlowFileHandlingException;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+@SupportsBatching
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failure")
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70283561
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,209 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.FlowFileHandlingException;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+@SupportsBatching
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failure")
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70265687
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
--- End diff --
This should be done while the data is being read in. One of the purposes of
this is to prevent enormous amounts of information from being read into memory
(streaming straight into the flowfile also prevents this). By doing the check
like this, there is the chance for an exceedingly large email to come in and
wreck the JVM.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70265074
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70264229
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,209 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.FlowFileHandlingException;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+@SupportsBatching
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failur
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70264008
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,209 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.FlowFileHandlingException;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+@SupportsBatching
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failur
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70263147
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,209 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.FlowFileHandlingException;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+@SupportsBatching
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failur
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70262587
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,209 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.FlowFileHandlingException;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+@SupportsBatching
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failur
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70262348
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70262049
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
--- End diff --
Now that you are processing the message in the "data" section, why not pass
the "inputStream" itself instead of the message body to the ontrigger? That way
you don't have to read the whole message into memory and can just write the
data straight to the flowfile content.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70261777
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70259118
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70258684
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70258169
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70257596
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
--- End diff --
I really like the idea to move the processing to data() and be able to tell
the client to retry. This allows the processor to handle back-pressure and
failures in a way that aligns with the transmitting protocol. Very nice!
That said, this new flow of logic requires us to think through the new
scenarios that may pop up. I will comment in the places that require attention.
FYI, in the future there may be an improvement ticket opened to always
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70246717
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
--- End diff --
@trixpan yes it would, good call
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70206885
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70196911
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70195092
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70194879
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70182626
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,147 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.stream.io.ByteArrayOutputStream;
+import org.apache.nifi.util.StopWatch;
+import org.subethamail.smtp.DropConnectionException;
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue incomingMessages;
+final ComponentLog logger;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
incomingMessages, ComponentLog logger) {
+this.incomingMessages = incomingMessages;
+this.logger = logger;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, incomingMessages, logger);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue incomingMessages, ComponentLog logger){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+// Start counting the timer...
+
+StopWatch watch = new StopWatch(false);
+
+SMTPServer server = messageContext.getSMTPServer();
+
+ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+byte [] buffer = new byte[1024];
+int rd;
+
+while ((rd = inputStream.read(buffer, 0, buffer.length)) !=
-1) {
+baos.write(buffer, 0, rd);
+}
+if (baos.getBufferLength() > server.getMaxMessageSize()) {
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+}
+
+baos.flush();
+this.messageBody = baos.toByteArray();
+
+
+X509Certificate[] certificates = new X509Certificate[]{};
--- End diff --
@JPercivall
Great feedback. Learned lots from it. Truly thankful!
Following your feedback I changed a bit the approach I took with ListenSMTP.
Instead of processing messages at `done()` (i.e. after the acknowledgment
back to client) I moved most of the processing of the message into the `data()`
method which gets called immediately after the sending client sends the end of
data command (i.e. the ".") but before the SMTP server acknowle
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70181608
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final Strin
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70180543
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failure")
+.description("Flowfiles that could not be parsed")
+.build();
+private Set relationships;
+private List desc
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70180515
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
--- End diff --
@JPercivall - would that also apply to ExtractEmailHeaders as well? If not,
minding explaining why not?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70180484
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
--- End diff --
I didn't know that! Thanks for the insight.
addressed
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70180458
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,106 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.commons.io.IOUtils;
+
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue messages;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
messages) {
+this.messages = messages;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, messages);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+boolean failed;
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue messages){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+SMTPServer server = messageContext.getSMTPServer();
+
+if (inputStream.available() > server.getMaxMessageSize()) {
+failed = true;
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+} else {
+this.messageBody = IOUtils.toByteArray(inputStream);
+}
+}
+
+@Override
+public void done() {
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+ certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+if (!failed) {
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+messages.put(message);
+} catch (InterruptedException e) {
+// Perhaps this should be logged?
--- End diff --
correct. I added log but also added a` throw new DropConnectionException`
so that the client knows it failed and try to resend again 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 infrastruct...@apache.org or file a JIRA ticke
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70180468
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final Strin
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70180440
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final Strin
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70169919
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
--- End diff --
This annotation gives the Data Flow Manager to the option to batch session
commits together in order to increase throughput (the slider in the "schedule"
tab). By adding it you aren't making it less safe, you're merely saying "this
processor is safe to roll back multiple sessions if necessary in order to give
it higher throughput overall". Since this processor merely extracts information
within NiFi and doesn't touch another system, it is safe to support batching.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request: https://github.com/apache/nifi/pull/483#discussion_r70158696 --- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/test/java/org/apache/nifi/processors/TestExtractEmailAttachments.java --- @@ -0,0 +1,84 @@ +/* + * 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; --- End diff -- fixed --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70158262
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/event/SmtpEvent.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.smtp.event;
+
+
+import java.security.cert.X509Certificate;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A Smtp event which adds the transaction number and command to the
StandardEvent.
+ */
+
+public class SmtpEvent{
+
+private final String remoteIP;
+private final String helo;
+private final String from;
+private final String to;
+private final byte[] bodyData;
+private List> certificatesDetails;
+
+public SmtpEvent(final String remoteIP, final String helo, final
String from, final String to, final X509Certificate[] certificates, final
byte[] bodyData) {
+this.remoteIP = remoteIP;
+this.helo = helo;
+this.from = from;
+this.to = to;
+this.bodyData = bodyData;
+
+this.certificatesDetails = new ArrayList<>();
+
+for (int c = 0; c < certificates.length; c++) {
+X509Certificate cert = certificates[c];
+if (cert.getSerialNumber() != null && cert.getSubjectDN() !=
null) {
+Map certificate = new HashMap<>();
+
+String certSerialNumber =
cert.getSerialNumber().toString();
+String certSubjectDN = cert.getSubjectDN().getName();
+
+
+certificate.put("SerialNumber", certSerialNumber);
+certificate.put("SubjectName", certSubjectDN);
+
+certificatesDetails.add(certificate);
+
+}
+}
+}
+
+public List getCertifcateDetails() {
--- End diff --
addressed?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70157808
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailHeaders.java
---
@@ -0,0 +1,233 @@
+/*
+ * 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.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+import javax.mail.Address;
+import javax.mail.Header;
+import javax.mail.Message;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Array;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Using the flowfile content as source of data,
extract header from an RFC compliant email file adding the relevant
attributes to the flowfile. " +
+"This processor does not perform extensive RFC validation but
still requires a bare minimum compliance with RFC 2822")
+@WritesAttributes({
+@WritesAttribute(attribute = "email.headers.bcc.*", description =
"Each individual BCC recipient (if available)"),
+@WritesAttribute(attribute = "email.headers.cc.*", description =
"Each individual CC recipient (if available)"),
+@WritesAttribute(attribute = "email.headers.from.*", description =
"Each individual mailbox contained in the From of the Email (array as per
RFC-2822)"),
+@WritesAttribute(attribute = "email.headers.message-id",
description = "The value of the Message-ID header (if available)"),
+@WritesAttribute(attribute = "email.headers.received_date",
description = "The Received-Date of the message (if available)"),
+@WritesAttribute(attribute = "email.headers.sent_date",
description = "Date the message was sent"),
+@WritesAttribute(attribute = "email.headers.subject", description
= "Subject of the message (if available)"),
+@WritesAttribute(attribute = "email.headers.to.*", description =
"Each individual TO recipient (if available)"),
+@WritesAttribute(attribute = "email.attachment_count", description
= "Number of attachments of the message" )})
+
+public class ExtractEmailHeaders extends AbstractProcessor {
+public static final String
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70157622
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailHeaders.java
---
@@ -0,0 +1,233 @@
+/*
+ * 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.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+import javax.mail.Address;
+import javax.mail.Header;
+import javax.mail.Message;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Array;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Using the flowfile content as source of data,
extract header from an RFC compliant email file adding the relevant
attributes to the flowfile. " +
+"This processor does not perform extensive RFC validation but
still requires a bare minimum compliance with RFC 2822")
+@WritesAttributes({
+@WritesAttribute(attribute = "email.headers.bcc.*", description =
"Each individual BCC recipient (if available)"),
+@WritesAttribute(attribute = "email.headers.cc.*", description =
"Each individual CC recipient (if available)"),
+@WritesAttribute(attribute = "email.headers.from.*", description =
"Each individual mailbox contained in the From of the Email (array as per
RFC-2822)"),
+@WritesAttribute(attribute = "email.headers.message-id",
description = "The value of the Message-ID header (if available)"),
+@WritesAttribute(attribute = "email.headers.received_date",
description = "The Received-Date of the message (if available)"),
+@WritesAttribute(attribute = "email.headers.sent_date",
description = "Date the message was sent"),
+@WritesAttribute(attribute = "email.headers.subject", description
= "Subject of the message (if available)"),
+@WritesAttribute(attribute = "email.headers.to.*", description =
"Each individual TO recipient (if available)"),
+@WritesAttribute(attribute = "email.attachment_count", description
= "Number of attachments of the message" )})
+
+public class ExtractEmailHeaders extends AbstractProcessor {
+public static final String
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70157601
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failure")
+.description("Flowfiles that could not be parsed")
+.build();
+private Set relationships;
+private List desc
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70157551
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
--- End diff --
I'm fine with the idea however I haven't considered any batching during the
development of the processor... Would assume it is safer to leave as it is?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request: https://github.com/apache/nifi/pull/483#discussion_r70157262 --- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml --- @@ -0,0 +1,75 @@ + + +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";> +4.0.0 + + +org.apache.nifi +nifi-email-bundle +1.0.0-SNAPSHOT + + +nifi-email-processors +jar + + + +org.apache.nifi +nifi-api + + +org.apache.nifi +nifi-processor-utils + + +javax.mail +mail +1.4.7 --- End diff -- addressed --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request: https://github.com/apache/nifi/pull/483#discussion_r70157266 --- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml --- @@ -0,0 +1,75 @@ + + +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";> +4.0.0 + + +org.apache.nifi +nifi-email-bundle +1.0.0-SNAPSHOT + + +nifi-email-processors +jar + + + +org.apache.nifi +nifi-api + + +org.apache.nifi +nifi-processor-utils + + +javax.mail +mail +1.4.7 + + +org.apache.commons +commons-email +1.4 + + +org.subethamail +subethasmtp +3.1.7 + + +org.apache.nifi +nifi-ssl-context-service-api + + +org.apache.nifi +nifi-ssl-context-service +test + + +org.apache.nifi +nifi-mock +1.0.0-SNAPSHOT --- End diff -- addressed --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user trixpan commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70156995
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final Strin
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70131169
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final St
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70130923
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/handler/SMTPMessageHandlerFactory.java
---
@@ -0,0 +1,106 @@
+/*
+ * 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.smtp.handler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.cert.X509Certificate;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.commons.io.IOUtils;
+
+import org.subethamail.smtp.MessageContext;
+import org.subethamail.smtp.MessageHandler;
+import org.subethamail.smtp.MessageHandlerFactory;
+import org.subethamail.smtp.RejectException;
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+
+
+public class SMTPMessageHandlerFactory implements MessageHandlerFactory {
+final LinkedBlockingQueue messages;
+
+public SMTPMessageHandlerFactory(LinkedBlockingQueue
messages) {
+this.messages = messages;
+}
+
+@Override
+public MessageHandler create(MessageContext messageContext) {
+return new Handler(messageContext, messages);
+}
+
+class Handler implements MessageHandler {
+final MessageContext messageContext;
+String from;
+String recipient;
+byte [] messageBody;
+
+
+boolean failed;
+
+public Handler(MessageContext messageContext,
LinkedBlockingQueue messages){
+this.messageContext = messageContext;
+}
+
+@Override
+public void from(String from) throws RejectException {
+// TODO: possibly whitelist senders?
+this.from = from;
+}
+
+@Override
+public void recipient(String recipient) throws RejectException {
+// TODO: possibly whitelist receivers?
+this.recipient = recipient;
+}
+
+@Override
+public void data(InputStream inputStream) throws RejectException,
TooMuchDataException, IOException {
+SMTPServer server = messageContext.getSMTPServer();
+
+if (inputStream.available() > server.getMaxMessageSize()) {
+failed = true;
+throw new TooMuchDataException("Data exceeds the amount
allowed.");
+} else {
+this.messageBody = IOUtils.toByteArray(inputStream);
+}
+}
+
+@Override
+public void done() {
+
+X509Certificate[] certificates = new X509Certificate[]{};
+
+String remoteIP = messageContext.getRemoteAddress().toString();
+String helo = messageContext.getHelo();
+if (messageContext.getTlsPeerCertificates() != null ){
+ certificates = (X509Certificate[])
messageContext.getTlsPeerCertificates().clone();
+}
+if (!failed) {
+SmtpEvent message = new SmtpEvent(remoteIP, helo, from,
recipient, certificates, messageBody);
+try {
+messages.put(message);
+} catch (InterruptedException e) {
+// Perhaps this should be logged?
--- End diff --
This is data loss and should definitely be logged as an error. The email
server has acknowledged (as far I as I understand it) and received the message
but we're failing to store/process 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 infrastru
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request: https://github.com/apache/nifi/pull/483#discussion_r70129106 --- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/test/java/org/apache/nifi/processors/TestExtractEmailAttachments.java --- @@ -0,0 +1,84 @@ +/* + * 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; --- End diff -- This, TestExtractEmailHeaders and GenerateAttachment are in the "org.apache.nifi.processors" package, not the "org.apache.nifi.processors.email" everything else is in. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70128468
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final St
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70128375
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/smtp/event/SmtpEvent.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.smtp.event;
+
+
+import java.security.cert.X509Certificate;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A Smtp event which adds the transaction number and command to the
StandardEvent.
+ */
+
+public class SmtpEvent{
+
+private final String remoteIP;
+private final String helo;
+private final String from;
+private final String to;
+private final byte[] bodyData;
+private List> certificatesDetails;
+
+public SmtpEvent(final String remoteIP, final String helo, final
String from, final String to, final X509Certificate[] certificates, final
byte[] bodyData) {
+this.remoteIP = remoteIP;
+this.helo = helo;
+this.from = from;
+this.to = to;
+this.bodyData = bodyData;
+
+this.certificatesDetails = new ArrayList<>();
+
+for (int c = 0; c < certificates.length; c++) {
+X509Certificate cert = certificates[c];
+if (cert.getSerialNumber() != null && cert.getSubjectDN() !=
null) {
+Map certificate = new HashMap<>();
+
+String certSerialNumber =
cert.getSerialNumber().toString();
+String certSubjectDN = cert.getSubjectDN().getName();
+
+
+certificate.put("SerialNumber", certSerialNumber);
+certificate.put("SubjectName", certSubjectDN);
+
+certificatesDetails.add(certificate);
+
+}
+}
+}
+
+public List getCertifcateDetails() {
--- End diff --
Bit of a nit pick but, it is best practice to keep the explicit type here
(List>) that way you don't have the "unchecked assignment"
warnings in ListenSMTP.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70128118
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final St
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70127060
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final St
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70127425
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final St
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70126772
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final St
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70126308
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ListenSMTP.java
---
@@ -0,0 +1,385 @@
+/*
+ * 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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processors.email.smtp.event.SmtpEvent;
+import
org.apache.nifi.processors.email.smtp.handler.SMTPMessageHandlerFactory;
+import org.subethamail.smtp.server.SMTPServer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.ssl.SSLContextService;
+
+@Tags({"listen", "email", "smtp"})
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@CapabilityDescription("This processor implements a lightweight SMTP
server to an arbitrary port, " +
+"allowing nifi to listen for incoming email. " +
+"" +
+"Note this server does not perform any email validation. If direct
exposure to the internet is sought," +
+"it may be a better idea to use the combination of NiFi and an
industrial scale MTA (e.g. Postfix)")
+@WritesAttributes({
+@WritesAttribute(attribute = "mime.type", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.helo", description = "The value
used during HELO"),
+@WritesAttribute(attribute = "smtp.certificates.*.serial",
description = "The serial numbers for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.certificates.*.principal",
description = "The principal for each of the " +
+"certificates used by an TLS peer"),
+@WritesAttribute(attribute = "smtp.from", description = "The value
used during MAIL FROM (i.e. envelope)"),
+@WritesAttribute(attribute = "smtp.to", description = "The value
used during RCPT TO (i.e. envelope)")})
+
+public class ListenSMTP extends AbstractProcessor {
+public static final St
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70124420
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailHeaders.java
---
@@ -0,0 +1,233 @@
+/*
+ * 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.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+import javax.mail.Address;
+import javax.mail.Header;
+import javax.mail.Message;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Array;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Using the flowfile content as source of data,
extract header from an RFC compliant email file adding the relevant
attributes to the flowfile. " +
+"This processor does not perform extensive RFC validation but
still requires a bare minimum compliance with RFC 2822")
+@WritesAttributes({
+@WritesAttribute(attribute = "email.headers.bcc.*", description =
"Each individual BCC recipient (if available)"),
+@WritesAttribute(attribute = "email.headers.cc.*", description =
"Each individual CC recipient (if available)"),
+@WritesAttribute(attribute = "email.headers.from.*", description =
"Each individual mailbox contained in the From of the Email (array as per
RFC-2822)"),
+@WritesAttribute(attribute = "email.headers.message-id",
description = "The value of the Message-ID header (if available)"),
+@WritesAttribute(attribute = "email.headers.received_date",
description = "The Received-Date of the message (if available)"),
+@WritesAttribute(attribute = "email.headers.sent_date",
description = "Date the message was sent"),
+@WritesAttribute(attribute = "email.headers.subject", description
= "Subject of the message (if available)"),
+@WritesAttribute(attribute = "email.headers.to.*", description =
"Each individual TO recipient (if available)"),
+@WritesAttribute(attribute = "email.attachment_count", description
= "Number of attachments of the message" )})
+
+public class ExtractEmailHeaders extends AbstractProcessor {
+public static final Str
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70123655
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailHeaders.java
---
@@ -0,0 +1,233 @@
+/*
+ * 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.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+import javax.mail.Address;
+import javax.mail.Header;
+import javax.mail.Message;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Array;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Using the flowfile content as source of data,
extract header from an RFC compliant email file adding the relevant
attributes to the flowfile. " +
+"This processor does not perform extensive RFC validation but
still requires a bare minimum compliance with RFC 2822")
+@WritesAttributes({
+@WritesAttribute(attribute = "email.headers.bcc.*", description =
"Each individual BCC recipient (if available)"),
+@WritesAttribute(attribute = "email.headers.cc.*", description =
"Each individual CC recipient (if available)"),
+@WritesAttribute(attribute = "email.headers.from.*", description =
"Each individual mailbox contained in the From of the Email (array as per
RFC-2822)"),
+@WritesAttribute(attribute = "email.headers.message-id",
description = "The value of the Message-ID header (if available)"),
+@WritesAttribute(attribute = "email.headers.received_date",
description = "The Received-Date of the message (if available)"),
+@WritesAttribute(attribute = "email.headers.sent_date",
description = "Date the message was sent"),
+@WritesAttribute(attribute = "email.headers.subject", description
= "Subject of the message (if available)"),
+@WritesAttribute(attribute = "email.headers.to.*", description =
"Each individual TO recipient (if available)"),
+@WritesAttribute(attribute = "email.attachment_count", description
= "Number of attachments of the message" )})
+
+public class ExtractEmailHeaders extends AbstractProcessor {
+public static final Str
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request: https://github.com/apache/nifi/pull/483#discussion_r70121265 --- Diff: nifi-assembly/NOTICE --- @@ -818,6 +818,15 @@ The following binary components are provided under the Apache Software License v The code for the t-digest was originally authored by Ted Dunning A number of small but very helpful changes have been contributed by Adrien Grand (https://github.com/jpountz) +(ASLv2) subethasmtp + The following NOTICE information applies: + + Copyright (C) 2006-2007 SubEthaMail.org --- End diff -- @joewitt, in regards to your comments on the NOTICE, I can't seem to find this NOTICE information in the version of the dependency that is being used[1] [1] https://github.com/voodoodyne/subethasmtp/tree/3.1.7 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70097725
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failure")
+.description("Flowfiles that could not be parsed")
+.build();
+private Set relationships;
+private List d
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70096131
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
+public class ExtractEmailAttachments extends AbstractProcessor {
+public static final String ATTACHMENT_ORIGINAL_FILENAME =
"email.attachment.parent.filename";
+public static final String ATTACHMENT_ORIGINAL_UUID =
"email.attachment.parent.uuid";
+
+public static final Relationship REL_ATTACHMENTS = new
Relationship.Builder()
+.name("attachments")
+.description("Each individual attachment will be routed to the
attachments relationship")
+.build();
+public static final Relationship REL_ORIGINAL = new
Relationship.Builder()
+.name("original")
+.description("The original file")
+.build();
+public static final Relationship REL_FAILURE = new
Relationship.Builder()
+.name("failure")
+.description("Flowfiles that could not be parsed")
+.build();
+private Set relationships;
+private List d
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/483#discussion_r70094915
--- Diff:
nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/src/main/java/org/apache/nifi/processors/email/ExtractEmailAttachments.java
---
@@ -0,0 +1,201 @@
+/*
+ * 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.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.Date;
+
+import javax.activation.DataSource;
+import javax.mail.Address;
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.mail.util.MimeMessageParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.stream.io.BufferedInputStream;
+
+
+
+
+@EventDriven
+@SideEffectFree
+@Tags({"split", "email"})
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("Extract attachments from a mime formatted email
file, splitting them into individual flowfiles.")
+@WritesAttributes({
+@WritesAttribute(attribute = "filename ", description = "The
filename of the attachment"),
+@WritesAttribute(attribute = "email.attachment.parent.filename ",
description = "The filename of the parent FlowFile"),
+@WritesAttribute(attribute = "email.attachment.parent.uuid",
description = "The UUID of the original FlowFile."),
+@WritesAttribute(attribute = "mime.type", description = "The mime
type of the attachment.")})
+
--- End diff --
I believe this processor should also have the "SupportsBatching"
annotation[1]
[1]
https://github.com/apache/nifi/blob/a83ed34f91432de22552b8ab1abfd7791fbb4412/nifi-api/src/main/java/org/apache/nifi/annotation/behavior/SupportsBatching.java#L49-L49
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request: https://github.com/apache/nifi/pull/483#discussion_r70093399 --- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml --- @@ -0,0 +1,75 @@ + + +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";> +4.0.0 + + +org.apache.nifi +nifi-email-bundle +1.0.0-SNAPSHOT + + +nifi-email-processors +jar + + + +org.apache.nifi +nifi-api + + +org.apache.nifi +nifi-processor-utils + + +javax.mail +mail +1.4.7 --- End diff -- javax.mail mail is already defined in the dependency management section of the root pom[1] so the "version" here can be removed. [1] https://github.com/apache/nifi/blob/2c95abbe20876d30ccd14fe80e766fd509a67d1a/pom.xml#L270-L270 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request: https://github.com/apache/nifi/pull/483#discussion_r70093109 --- Diff: nifi-nar-bundles/nifi-email-bundle/nifi-email-processors/pom.xml --- @@ -0,0 +1,75 @@ + + +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";> +4.0.0 + + +org.apache.nifi +nifi-email-bundle +1.0.0-SNAPSHOT + + +nifi-email-processors +jar + + + +org.apache.nifi +nifi-api + + +org.apache.nifi +nifi-processor-utils + + +javax.mail +mail +1.4.7 + + +org.apache.commons +commons-email +1.4 + + +org.subethamail +subethasmtp +3.1.7 + + +org.apache.nifi +nifi-ssl-context-service-api + + +org.apache.nifi +nifi-ssl-context-service +test + + +org.apache.nifi +nifi-mock +1.0.0-SNAPSHOT --- End diff -- The version here should be removed. It will use the version set in the dependency management section of the root pom. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] nifi pull request #483: NIFI-1899 - Introduce ExtractEmailAttachments and Ex...
Github user JPercivall commented on a diff in the pull request: https://github.com/apache/nifi/pull/483#discussion_r70092692 --- Diff: nifi-assembly/NOTICE --- @@ -818,6 +818,15 @@ The following binary components are provided under the Apache Software License v The code for the t-digest was originally authored by Ted Dunning A number of small but very helpful changes have been contributed by Adrien Grand (https://github.com/jpountz) +(ASLv2) subethasmtp + The following NOTICE information applies: + + Copyright (C) 2006-2007 SubEthaMail.org --- End diff -- This project is Licensed under ASLv2[1] without any notice[2] so it can be removed from our NOTICE file. [1] https://github.com/voodoodyne/subethasmtp/blob/3.1.7/LICENSE.txt [2] https://github.com/voodoodyne/subethasmtp/tree/3.1.7 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
