Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/193#discussion_r48641557
--- Diff:
engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
---
@@ -970,4 +969,48 @@ public void onClose(int closeCode, String message)
server.stop();
}
}
+
+ public static class TestMetricTransport implements AutoMetric.Transport,
Serializable
+ {
+ private String prefix;
+ private static List<String> messages = new ArrayList<>();
+
+ public TestMetricTransport(String prefix)
+ {
+ this.prefix = prefix;
+ }
+
+ @Override
+ public void push(String jsonData) throws IOException
+ {
+ messages.add(prefix + ":" + jsonData);
+ }
+
+ @Override
+ public long getSchemaResendInterval()
+ {
+ return 0;
+ }
+ }
+
+ @Test
+ public void testCustomMetricsTransport() throws Exception
+ {
+ TestGeneratorInputOperator o1 = dag.addOperator("o1",
TestGeneratorInputOperator.class);
+ GenericTestOperator o2 = dag.addOperator("o2",
GenericTestOperator.class);
+ dag.addStream("o1.outport", o1.outport, o2.inport1);
+ dag.setAttribute(LogicalPlan.METRICS_TRANSPORT, new
TestMetricTransport("xyz"));
+ StramLocalCluster lc = new StramLocalCluster(dag);
+ StreamingContainerManager dnmgr = lc.dnmgr;
+ StramAppContext appContext = new StramTestSupport.TestAppContext();
+
+ AppDataPushAgent pushAgent = new AppDataPushAgent(dnmgr, appContext);
+ pushAgent.init();
+ pushAgent.pushData();
+ Assert.assertTrue(TestMetricTransport.messages.size() > 0);
+ pushAgent.close();
+ String msg = TestMetricTransport.messages.get(0);
+ System.out.println("Got this message: " + msg);
--- End diff --
Not needed.
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
