DefaultMessageDistributor implements MessageDistributor, Serializable { @Override public void distribute(MessageRoute route, Context context) { try { ProfileBuilder builder = getBuilder(route, context); builder.apply(route.getMessage(), route.getTimestamp()); } catch(ExecutionException e) { LOG.error("Unexpected error", e); throw new RuntimeException(e); } } DefaultMessageDistributor(
          long periodDurationMillis,
          long profileTimeToLiveMillis,
          long maxNumberOfRoutes);  DefaultMessageDistributor(
          long periodDurationMillis,
          long profileTimeToLiveMillis,
          long maxNumberOfRoutes,
          Ticker ticker); @Override void distribute(MessageRoute route, Context context); @Override List<ProfileMeasurement> flush(); @Override List<ProfileMeasurement> flushExpired(); ProfileBuilder getBuilder(MessageRoute route, Context context); DefaultMessageDistributor withPeriodDurationMillis(long periodDurationMillis); DefaultMessageDistributor withPeriodDuration(int duration, TimeUnit units);  }
@Test public void testDistribute() throws Exception { long timestamp = 100; ProfileConfig definition = createDefinition(profileOne); String entity = (String) messageOne.get("ip_src_addr"); MessageRoute route = new MessageRoute(definition, entity, messageOne, timestamp); distributor.distribute(route, context); List<ProfileMeasurement> measurements = distributor.flush(); assertEquals(1, measurements.size()); ProfileMeasurement m = measurements.get(0); assertEquals(definition.getProfile(), m.getProfileName()); assertEquals(entity, m.getEntity()); }