SmtpOutputOperator extends BaseOperator { public Map<String, String> getRecipients() { return recipients; }  String getSubject(); void setSubject(String subject); String getContent(); void setContent(String content); String getFrom(); void setFrom(String from); int getSmtpPort(); void setSmtpPort(int smtpPort); String getSmtpHost(); void setSmtpHost(String smtpHost); String getSmtpUserName(); void setSmtpUserName(String smtpUserName); String getSmtpPassword(); void setSmtpPassword(String smtpPassword); String getContentType(); void setContentType(String contentType); boolean isUseSsl(); void setUseSsl(boolean useSsl); @Override void setup(OperatorContext context); Map<String, String> getRecipients(); void setRecipients(Map<String, String> recipients); final transient DefaultInputPort<Object> input; }
@Test public void testProperties() throws Exception { Configuration conf = new Configuration(false); conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.subject", subject); conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.content", content); conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.from", from); conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.smtpHost", "127.0.0.1"); conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.smtpUserName", from); conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.smtpPassword", "<password>"); conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.recipients.TO", to + "," + cc); conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.recipients.CC", cc); final AtomicReference<SmtpOutputOperator> o1 = new AtomicReference<SmtpOutputOperator>(); StreamingApplication app = new StreamingApplication() { @Override public void populateDAG(DAG dag, Configuration conf) { o1.set(dag.addOperator("o1", new SmtpOutputOperator())); } }; LocalMode lma = LocalMode.newInstance(); lma.prepareDAG(app, conf); Assert.assertEquals("checking TO list", to + "," + cc, o1.get().getRecipients().get("TO")); Assert.assertEquals("checking CC list", cc, o1.get().getRecipients().get("CC")); }