ProfileBuilderFunction implements MapGroupsFunction<String, MessageRoute, ProfileMeasurement> { @Override public ProfileMeasurement call(String group, Iterator<MessageRoute> iterator) { int maxRoutes = Integer.MAX_VALUE; long profileTTLMillis = Long.MAX_VALUE; MessageDistributor distributor = new DefaultMessageDistributor(periodDurationMillis, profileTTLMillis, maxRoutes); Context context = TaskUtils.getContext(globals); List<MessageRoute> routes = toStream(iterator) .sorted(comparing(rt -> rt.getTimestamp())) .collect(Collectors.toList()); LOG.debug("Building a profile for group '{}' from {} message(s)", group, routes.size()); for(MessageRoute route: routes) { distributor.distribute(route, context); } ProfileMeasurement result; List<ProfileMeasurement> measurements = distributor.flush(); if(measurements.size() == 1) { result = measurements.get(0); LOG.debug("Profile measurement created; profile={}, entity={}, period={}, value={}", result.getProfileName(), result.getEntity(), result.getPeriod().getPeriod(), result.getProfileValue()); } else if(measurements.size() == 0) { String msg = format("No profile measurement can be calculated. Review the profile for bugs. profile=%s, entity=%s, period=%s", profileFromKey(group), entityFromKey(group), periodFromKey(group)); LOG.error(msg); throw new IllegalStateException(msg); } else { String msg = format("Expected 1 profile measurement, but got %d. profile=%s, entity=%s, period=%s", measurements.size(), profileFromKey(group), entityFromKey(group), periodFromKey(group)); LOG.error(msg); throw new IllegalStateException(msg); } return result; } ProfileBuilderFunction(Properties properties, Map<String, String> globals); @Override ProfileMeasurement call(String group, Iterator<MessageRoute> iterator);  }
@Test public void shouldBuildProfileMeasurement() throws Exception { JSONObject message = getMessage(); String entity = "192.168.1.1"; long timestamp = (Long) message.get("timestamp"); ProfileConfig profile = ProfileConfig.fromJSON(profileJSON); MessageRoute route = new MessageRoute(profile, entity, message, timestamp); List<MessageRoute> routes = new ArrayList<>(); routes.add(route); routes.add(route); routes.add(route); Properties profilerProperties = getProfilerProperties(); int periodDuration = PERIOD_DURATION.get(profilerProperties, Integer.class); TimeUnit periodDurationUnits = TimeUnit.valueOf(PERIOD_DURATION_UNITS.get(profilerProperties, String.class)); ProfilePeriod expectedPeriod = ProfilePeriod.fromTimestamp(timestamp, periodDuration, periodDurationUnits); ProfileBuilderFunction function = new ProfileBuilderFunction(profilerProperties, getGlobals()); ProfileMeasurement measurement = function.call("profile1-192.168.1.1-0", routes.iterator()); assertEquals(entity, measurement.getEntity()); assertEquals(profile.getProfile(), measurement.getProfileName()); assertEquals(routes.size(), measurement.getProfileValue()); assertEquals(expectedPeriod.getPeriod(), (long) measurement.getPeriod().getPeriod()); }
@Test public void shouldThrowExceptionIfInvalidProfile() throws Exception { JSONObject message = getMessage(); String entity = "192.168.1.1"; long timestamp = (Long) message.get("timestamp"); ProfileConfig profile = ProfileConfig.fromJSON(invalidProfileJson); MessageRoute route = new MessageRoute(profile, entity, message, timestamp); List<MessageRoute> routes = new ArrayList<>(); routes.add(route); routes.add(route); routes.add(route); Properties profilerProperties = getProfilerProperties(); ProfileBuilderFunction function = new ProfileBuilderFunction(profilerProperties, getGlobals()); assertThrows(IllegalStateException.class, () -> function.call("profile1-192.168.1.1-0", routes.iterator())); }