MergeTransformRuntime implements TransformRuntime { @Override public List<StreamlineEvent> execute(StreamlineEvent input) { StreamlineEventImpl.Builder builder = StreamlineEventImpl.builder(); builder.putAll(input); for (Map.Entry<String, ?> entry : mergeTransform.getDefaults().entrySet()) { if (!input.containsKey(entry.getKey())) { builder.put(entry.getKey(), entry.getValue()); } } return Collections.<StreamlineEvent>singletonList(builder.dataSourceId(input.getDataSourceId()).build()); } MergeTransformRuntime(MergeTransform mergeTransform); @Override List<StreamlineEvent> execute(StreamlineEvent input); @Override String toString();  }
@Test public void testExecute() throws Exception { Map<String, Object> fieldsAndValues = new HashMap<>(); fieldsAndValues.put("1", "one"); fieldsAndValues.put("2", "two"); Map<String, String> defaults = new HashMap<>(); defaults.put("2", "TWO"); defaults.put("3", "THREE"); StreamlineEvent event = StreamlineEventImpl.builder().fieldsAndValues(fieldsAndValues).dataSourceId("dsrcid").build(); TransformRuntime transformRuntime = new MergeTransformRuntime(new MergeTransform(defaults)); List<StreamlineEvent> result = transformRuntime.execute(event); System.out.println(result); assertEquals(1, result.size()); assertEquals("two", result.get(0).get("2")); assertEquals("THREE", result.get(0).get("3")); }