@Test public void testIntegerField() throws Exception { String[] directives = new String[] { "send-to-error field_calories_cnt < 0" }; List<Row> rows = Arrays.asList( new Row("field_calories_cnt", new Integer(10)), new Row("field_calories_cnt", new Integer(0)), new Row("field_calories_cnt", new Integer(-10)) ); RecipePipeline pipeline = TestingRig.execute(directives); List<Row> results = pipeline.execute(rows); List<ErrorRecord> errors = pipeline.errors(); Assert.assertEquals(1, errors.size()); Assert.assertEquals(2, results.size()); }
@Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); if (context != null) { for (String variable : context.getTransientStore().getVariables()) { ctx.set(variable, context.getTransientStore().get(variable)); } } try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } throw new ErrorRowException(NAME, message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; }
SendToError implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); if (context != null) { for (String variable : context.getTransientStore().getVariables()) { ctx.set(variable, context.getTransientStore().get(variable)); } } try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } throw new ErrorRowException(NAME, message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; } }
SendToError implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); if (context != null) { for (String variable : context.getTransientStore().getVariables()) { ctx.set(variable, context.getTransientStore().get(variable)); } } try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } throw new ErrorRowException(NAME, message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; }  }
SendToError implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); if (context != null) { for (String variable : context.getTransientStore().getVariables()) { ctx.set(variable, context.getTransientStore().get(variable)); } } try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } throw new ErrorRowException(NAME, message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); }
SendToError implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { ELContext ctx = new ELContext(context, el, row); if (context != null) { for (String variable : context.getTransientStore().getVariables()) { ctx.set(variable, context.getTransientStore().get(variable)); } } try { ELResult result = el.execute(ctx); if (result.getBoolean()) { if (metric != null && context != null) { context.getMetrics().count(metric, 1); } if (message == null) { message = condition; } throw new ErrorRowException(NAME, message, 1); } } catch (ELException e) { throw new DirectiveExecutionException(NAME, e.getMessage(), e); } results.add(row); } return results; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); static final String NAME; }