MessageRouterFunction implements FlatMapFunction<String, MessageRoute> { @Override public Iterator<MessageRoute> call(String jsonMessage) throws Exception { List<MessageRoute> routes = Collections.emptyList(); JSONParser parser = new JSONParser(); Context context = TaskUtils.getContext(globals); MessageRouter router = new DefaultMessageRouter(context); Optional<JSONObject> message = toMessage(jsonMessage, parser); if(message.isPresent()) { Optional<Long> timestampOpt = clock.currentTimeMillis(message.get()); if (timestampOpt.isPresent()) { Long timestamp = timestampOpt.get(); if(timestamp >= begin && timestamp <= end) { routes = router.route(message.get(), profilerConfig, context); LOG.trace("Found {} route(s) for a message", routes.size()); } else { LOG.trace("Ignoring message; timestamp={} not in [{},{}]", () -> timestamp, () -> prettyPrint(begin), () -> prettyPrint(end)); } } else { LOG.trace("No timestamp in message. Message will be ignored."); } } else { LOG.trace("Unable to parse message. Message will be ignored"); } return routes.iterator(); } MessageRouterFunction(ProfilerConfig profilerConfig, Map<String, String> globals); @Override Iterator<MessageRoute> call(String jsonMessage); MessageRouterFunction withBegin(Long begin); MessageRouterFunction withEnd(Long end); MessageRouterFunction withClockFactory(ClockFactory clockFactory);  }
@Test public void testFindRoutes() throws Exception { MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals()); Iterator<MessageRoute> iter = function.call(goodMessage); List<MessageRoute> routes = Lists.newArrayList(iter); assertEquals(1, routes.size()); assertEquals("profile1", routes.get(0).getProfileDefinition().getProfile()); }
@Test public void testWithBadMessage() throws Exception { MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals()); Iterator<MessageRoute> iter = function.call(badMessage); List<MessageRoute> routes = Lists.newArrayList(iter); assertEquals(0, routes.size()); }
@Test public void testFindMultipleRoutes() throws Exception { MessageRouterFunction function = new MessageRouterFunction(twoProfiles(), getGlobals()); Iterator<MessageRoute> iter = function.call(goodMessage); List<MessageRoute> routes = Lists.newArrayList(iter); assertEquals(2, routes.size()); assertEquals("profile1", routes.get(0).getProfileDefinition().getProfile()); assertEquals("profile2", routes.get(1).getProfileDefinition().getProfile()); }
@Test public void testWithNoTimestampInMessage() throws Exception { MessageRouterFunction function = new MessageRouterFunction(profile(), getGlobals()); Iterator<MessageRoute> iter = function.call(messageNoTimestamp); List<MessageRoute> routes = Lists.newArrayList(iter); assertEquals(0, routes.size()); }