@Test public void noErrorsShouldBeThrown() { builder.process(bundle, emitter); bundle.setValue(bundle.getFormat().getField("foo"), ValueFactory.create("notamap")); builder.process(bundle, emitter); verify(emitter, never()).emit(any()); }
@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);  }