HCX-971: Implement simple email service (SES) for sending emails

hcx-pipeline-jobs/message-service/src/main/java/org/swasth/dp/message/service/functions/EmailDispatcher.java

@@ -20,7 +20,7 @@ public EmailDispatcher(MessageServiceConfig config) {
     }
 
     @Override
-    public void processElement(Map<String, Object> event, ProcessFunction<Map<String, Object>, Map<String, Object>>.Context context, Collector<Map<String, Object>> collector) {
+    public void processElement(Map<String, Object> event, ProcessFunction<Map<String, Object>, Map<String, Object>>.Context context, Collector<Map<String, Object>> collector) throws MessagingException {
         try{
             System.out.println("Processing Email Event :: Mid: " + event.get("mid"));
             Map<String,Object> recipients = (Map<String,Object>) event.getOrDefault("recipients", new HashMap<>());
@@ -38,42 +38,43 @@ public void processElement(Map<String, Object> event, ProcessFunction<Map<String
     }
 
 
-    public Boolean sendMail(List<String> to, List<String> cc, List<String> bcc, String subject, String message){
-        //compose message
+    public Boolean sendMail(List<String> to, List<String> cc, List<String> bcc, String subject, String message) throws MessagingException {
+        Session session = Session.getDefaultInstance(getMailProperties());
+        MimeMessage mimeMessage = new MimeMessage(session);
+        for(String id: to){
+            mimeMessage.addRecipient(Message.RecipientType.TO,new InternetAddress(id));
+        }
+        for(String id: cc){
+            mimeMessage.addRecipient(Message.RecipientType.CC,new InternetAddress(id));
+        }
+        for(String id: bcc){
+            mimeMessage.addRecipient(Message.RecipientType.BCC,new InternetAddress(id));
+        }
+        mimeMessage.setFrom(new InternetAddress(config.fromEmail));
+        mimeMessage.setSubject(subject);
+        mimeMessage.setContent(message,"text/html");
+        Transport transport = session.getTransport();
         try {
-            MimeMessage mimeMessage = new MimeMessage(getSession());
-            for(String id: to){
-                mimeMessage.addRecipient(Message.RecipientType.TO,new InternetAddress(id));
-            }
-            for(String id: cc){
-                mimeMessage.addRecipient(Message.RecipientType.CC,new InternetAddress(id));
-            }
-            for(String id: bcc){
-                mimeMessage.addRecipient(Message.RecipientType.BCC,new InternetAddress(id));
-            }
-            mimeMessage.setSubject(subject);
-            mimeMessage.setContent(message, "text/html");
-            //send message
-            Transport.send(mimeMessage);
-            return true;
-        } catch (MessagingException e) {throw new RuntimeException(e);}
-    }
-
-    private Session getSession() {
-        return Session.getDefaultInstance(getMailProperties(),
-                new javax.mail.Authenticator() {
-                    protected PasswordAuthentication getPasswordAuthentication() {
-                        return new PasswordAuthentication(config.emailId, config.emailPwd);
-                    }
-                });
+            transport.connect(config.emailServerHost, config.emailServerUsername, config.emailServerPassword);
+            transport.sendMessage(mimeMessage, mimeMessage.getAllRecipients());
+            System.out.println("Email sent!");
+        }
+        catch (Exception ex) {
+            System.out.println("The email was not sent.");
+            System.out.println("Error message: " + ex.getMessage());
+        }
+        finally {
+            transport.close();
+        }
+        return true;
     }
 
     private Properties getMailProperties(){
         Properties properties = new Properties();
-        properties.put("mail.smtp.host", "smtp.gmail.com");
-        properties.put("mail.smtp.port", "465");
-        properties.put("mail.smtp.auth", "true");
+        properties.put("mail.transport.protocol", "smtp");
+        properties.put("mail.smtp.port", config.emailServerPort);
         properties.put("mail.smtp.starttls.enable", "true");
+        properties.put("mail.smtp.auth", "true");
         properties.put("mail.smtp.starttls.required", "true");
         properties.put("mail.smtp.ssl.protocols", "TLSv1.2");
         properties.put("mail.smtp.socketFactory.class", "javax.net.ssl.SSLSocketFactory");

hcx-pipeline-jobs/message-service/src/main/java/org/swasth/dp/message/service/task/MessageServiceConfig.java

@@ -22,8 +22,11 @@ public class MessageServiceConfig extends BaseJobConfig {
     public String awsAccessKey;
     public String awsAccessSecret;
     public String awsRegion;
-    public String emailId;
-    public String emailPwd;
+    public String emailServerHost;
+    public String emailServerPassword;
+    public int emailServerPort;
+    public String emailServerUsername;
+    public String fromEmail;
 
     public String onboardIndex;
 
@@ -49,9 +52,12 @@ public void initValues(){
         awsAccessKey = config.getString("aws.access-key");
         awsAccessSecret = config.getString("aws.access-secret");
         awsRegion = config.getString("aws.region");
-        emailId = config.getString("email.id");
-        emailPwd = config.getString("email.pwd");
         onboardIndex = config.getString("audit.onboard.index");
         onboardIndexAlias = config.getString("audit.onboard.alias");
+        fromEmail = config.getString("email.server.from");
+        emailServerUsername = config.getString("email.server.username");
+        emailServerPassword = config.getString("email.server.password");
+        emailServerHost = config.getString("email.server.host");
+        emailServerPort = config.getInt("email.server.port");
     }
 }

hcx-pipeline-jobs/message-service/src/main/resources/message-service.conf

@@ -19,6 +19,9 @@ aws {
 }
 
 email {
-  id = ""
-  pwd = ""
+  server.from = ""
+  server.username = ""
+  server.password = ""
+  server.host = ""
+  server.port = ""
 }