@Test public void basicFunctionality() throws IOException { int reps = 5; RepeatBuilder builder = decodeObject(RepeatBuilder.class, "repeatField: reps"); Bundle bundle = Bundles.decode("reps: " + reps); builder.process(bundle, emitter); verify(emitter, times(reps)).emit(bundle); }
@Override public void process(Bundle row, StreamEmitter emitter) { long emitTimes = defaultRepeats; ValueObject repeatCount = repeatField.getValue(row); if (repeatCount != null) { try { emitTimes = repeatCount.asLong().getLong(); } catch (ValueTranslationException vte) { if (failOnParseException) { throw vte; } else { log.warn("Failed to parse repeats field value={}", repeatCount.asString().asNative()); } } } for (int i = 0; i < emitTimes; i++) { Bundle toEmit = copyBundle ? Bundles.deepCopyBundle(row) : row; emitter.emit(toEmit); } }
RepeatBuilder extends StreamBuilder { @Override public void process(Bundle row, StreamEmitter emitter) { long emitTimes = defaultRepeats; ValueObject repeatCount = repeatField.getValue(row); if (repeatCount != null) { try { emitTimes = repeatCount.asLong().getLong(); } catch (ValueTranslationException vte) { if (failOnParseException) { throw vte; } else { log.warn("Failed to parse repeats field value={}", repeatCount.asString().asNative()); } } } for (int i = 0; i < emitTimes; i++) { Bundle toEmit = copyBundle ? Bundles.deepCopyBundle(row) : row; emitter.emit(toEmit); } } }
RepeatBuilder extends StreamBuilder { @Override public void process(Bundle row, StreamEmitter emitter) { long emitTimes = defaultRepeats; ValueObject repeatCount = repeatField.getValue(row); if (repeatCount != null) { try { emitTimes = repeatCount.asLong().getLong(); } catch (ValueTranslationException vte) { if (failOnParseException) { throw vte; } else { log.warn("Failed to parse repeats field value={}", repeatCount.asString().asNative()); } } } for (int i = 0; i < emitTimes; i++) { Bundle toEmit = copyBundle ? Bundles.deepCopyBundle(row) : row; emitter.emit(toEmit); } }  }
RepeatBuilder extends StreamBuilder { @Override public void process(Bundle row, StreamEmitter emitter) { long emitTimes = defaultRepeats; ValueObject repeatCount = repeatField.getValue(row); if (repeatCount != null) { try { emitTimes = repeatCount.asLong().getLong(); } catch (ValueTranslationException vte) { if (failOnParseException) { throw vte; } else { log.warn("Failed to parse repeats field value={}", repeatCount.asString().asNative()); } } } for (int i = 0; i < emitTimes; i++) { Bundle toEmit = copyBundle ? Bundles.deepCopyBundle(row) : row; emitter.emit(toEmit); } }  @Override void init(); @Override void process(Bundle row, StreamEmitter emitter); }
RepeatBuilder extends StreamBuilder { @Override public void process(Bundle row, StreamEmitter emitter) { long emitTimes = defaultRepeats; ValueObject repeatCount = repeatField.getValue(row); if (repeatCount != null) { try { emitTimes = repeatCount.asLong().getLong(); } catch (ValueTranslationException vte) { if (failOnParseException) { throw vte; } else { log.warn("Failed to parse repeats field value={}", repeatCount.asString().asNative()); } } } for (int i = 0; i < emitTimes; i++) { Bundle toEmit = copyBundle ? Bundles.deepCopyBundle(row) : row; emitter.emit(toEmit); } }  @Override void init(); @Override void process(Bundle row, StreamEmitter emitter);  }