@Test public void basicFunctionality() { map.put("hello", ValueFactory.create("boo")); map.put("hi", ValueFactory.create("hoo")); ValueMap map2 = ValueFactory.createMap(); map2.put("deep", ValueFactory.create("deeper")); map.put("nest", map2); bundle.setValue(bundle.getFormat().getField("foo"), map); bundle.setValue(bundle.getFormat().getField("bar"), ValueFactory.create("biz")); builder.process(bundle, emitter); ArgumentCaptor<Bundle> bundleCapture = ArgumentCaptor.forClass(Bundle.class); verify(emitter, times(3)).emit(bundleCapture.capture()); List<Bundle> newBundles = bundleCapture.getAllValues(); for (Bundle newBundle : newBundles) { assertEquals("biz", newBundle.getValue(newBundle.getFormat().getField("bar")).asString().asNative()); String key = newBundle.getValue(newBundle.getFormat().getField("key")).asString().asNative(); assertTrue(key.equals("hello") || key.equals("hi") || key.equals("nest")); if (key.equals("hello")) { assertEquals("boo", newBundle.getValue(newBundle.getFormat().getField("value")).asString().asNative()); } else if (key.equals("hi")) { assertEquals("hoo", newBundle.getValue(newBundle.getFormat().getField("value")).asString().asNative()); } else if (key.equals("nest")) { assertEquals(ValueObject.TYPE.MAP, newBundle.getValue(newBundle.getFormat().getField("value")).getObjectType()); } else { fail("An unexpected bundle was emitted"); } } }
@Override public void process(Bundle bundle, StreamEmitter emitter) { ValueObject obj = field.getValue(bundle); field.removeValue(bundle); if ((obj != null) && (obj.getObjectType() == ValueObject.TYPE.MAP)) { ValueMap map = obj.asMap(); for (ValueMapEntry entry : map) { Bundle newBundle = Bundles.deepCopyBundle(bundle); keyField.setValue(newBundle, ValueFactory.create(entry.getKey())); valueField.setValue(newBundle, entry.getValue()); if (emitFilter == null || emitFilter.filter(newBundle)) { emitter.emit(newBundle); } } } if (emitOriginal && (emitFilter == null || emitFilter.filter(bundle))) { emitter.emit(bundle); } }
StreamMapSplitBuilder extends StreamBuilder { @Override public void process(Bundle bundle, StreamEmitter emitter) { ValueObject obj = field.getValue(bundle); field.removeValue(bundle); if ((obj != null) && (obj.getObjectType() == ValueObject.TYPE.MAP)) { ValueMap map = obj.asMap(); for (ValueMapEntry entry : map) { Bundle newBundle = Bundles.deepCopyBundle(bundle); keyField.setValue(newBundle, ValueFactory.create(entry.getKey())); valueField.setValue(newBundle, entry.getValue()); if (emitFilter == null || emitFilter.filter(newBundle)) { emitter.emit(newBundle); } } } if (emitOriginal && (emitFilter == null || emitFilter.filter(bundle))) { emitter.emit(bundle); } } }
StreamMapSplitBuilder extends StreamBuilder { @Override public void process(Bundle bundle, StreamEmitter emitter) { ValueObject obj = field.getValue(bundle); field.removeValue(bundle); if ((obj != null) && (obj.getObjectType() == ValueObject.TYPE.MAP)) { ValueMap map = obj.asMap(); for (ValueMapEntry entry : map) { Bundle newBundle = Bundles.deepCopyBundle(bundle); keyField.setValue(newBundle, ValueFactory.create(entry.getKey())); valueField.setValue(newBundle, entry.getValue()); if (emitFilter == null || emitFilter.filter(newBundle)) { emitter.emit(newBundle); } } } if (emitOriginal && (emitFilter == null || emitFilter.filter(bundle))) { emitter.emit(bundle); } }  }
StreamMapSplitBuilder extends StreamBuilder { @Override public void process(Bundle bundle, StreamEmitter emitter) { ValueObject obj = field.getValue(bundle); field.removeValue(bundle); if ((obj != null) && (obj.getObjectType() == ValueObject.TYPE.MAP)) { ValueMap map = obj.asMap(); for (ValueMapEntry entry : map) { Bundle newBundle = Bundles.deepCopyBundle(bundle); keyField.setValue(newBundle, ValueFactory.create(entry.getKey())); valueField.setValue(newBundle, entry.getValue()); if (emitFilter == null || emitFilter.filter(newBundle)) { emitter.emit(newBundle); } } } if (emitOriginal && (emitFilter == null || emitFilter.filter(bundle))) { emitter.emit(bundle); } }  @Override void init(); @Override void process(Bundle bundle, StreamEmitter emitter); }
StreamMapSplitBuilder extends StreamBuilder { @Override public void process(Bundle bundle, StreamEmitter emitter) { ValueObject obj = field.getValue(bundle); field.removeValue(bundle); if ((obj != null) && (obj.getObjectType() == ValueObject.TYPE.MAP)) { ValueMap map = obj.asMap(); for (ValueMapEntry entry : map) { Bundle newBundle = Bundles.deepCopyBundle(bundle); keyField.setValue(newBundle, ValueFactory.create(entry.getKey())); valueField.setValue(newBundle, entry.getValue()); if (emitFilter == null || emitFilter.filter(newBundle)) { emitter.emit(newBundle); } } } if (emitOriginal && (emitFilter == null || emitFilter.filter(bundle))) { emitter.emit(bundle); } }  @Override void init(); @Override void process(Bundle bundle, StreamEmitter emitter);  }