@Test public void emitFilterEmitOriginal() throws IOException { map.put("hello", ValueFactory.create("is it me you're looking for?")); map.put("you say", ValueFactory.create("good-bye")); map.put("I say", ValueFactory.create("hello")); bundle.setValue(bundle.getFormat().getField("foo"), map); bundle.setValue(bundle.getFormat().getField("bar"), ValueFactory.create("lyrics")); builder = Configs.decodeObject(StreamMapSplitBuilder.class, "field: foo, keyField: key, valueField: value, emitOriginal: true, emitFilter: {from.const: 42, to: baz}"); builder.process(bundle, emitter); ArgumentCaptor<Bundle> bundleCapture = ArgumentCaptor.forClass(Bundle.class); verify(emitter, times(4)).emit(bundleCapture.capture()); List<Bundle> newBundles = bundleCapture.getAllValues(); int nullPreKey = 0; for (Bundle newBundle: newBundles) { assertEquals("lyrics", newBundle.getValue(newBundle.getFormat().getField("bar")).asString().asNative()); long baz = newBundle.getValue(newBundle.getFormat().getField("baz")).asLong().asNative(); final ValueObject preKey = newBundle.getValue(newBundle.getFormat().getField("key")); if (preKey == null) { nullPreKey++; continue; } String key = preKey.asString().asNative(); String value = newBundle.getValue(newBundle.getFormat().getField("value")).asString().asNative(); assertEquals("Constant filter value set didn't work", 42L, baz); assertTrue(key.equals("hello") || key.equals("you say") || key.equals("I say")); switch (key) { case "hello": assertEquals("is it me you're looking for?", value); break; case "you say": assertEquals("good-bye", value); break; case "I say": assertEquals("hello", value); break; default: fail("An unexpected bundle was emitted"); break; } } assertEquals(1, nullPreKey); }
@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);  }