@Test public void testHttpInvokeWithHeaders() throws Exception { String[] directives = new String[] { "invoke-http " + (baseURL + "/service ") + "a,b A=1,B=2,C=3" }; List<Row> rows = Arrays.asList( new Row("a", "1").add("b", 2.0), new Row("a", "3").add("b", 4.2) ); rows = TestingRig.execute(directives, rows); Assert.assertEquals(2, rows.size()); Assert.assertEquals(4, rows.get(0).width()); Assert.assertEquals(4, rows.get(1).width()); Assert.assertEquals("1:2.000000", rows.get(0).getValue("c")); Assert.assertEquals("3:4.200000", rows.get(1).getValue("c")); Assert.assertEquals("3", rows.get(0).getValue("HeaderC")); Assert.assertEquals("3", rows.get(1).getValue("HeaderC")); }
@Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { Map<String, Object> parameters = new HashMap<>(); for (String column : columns) { int idx = row.find(column); if (idx != -1) { parameters.put(column, row.getValue(idx)); } } try { Map<String, Object> result = invokeHttp(url, parameters, headers); for (Map.Entry<String, Object> entry : result.entrySet()) { row.addOrSet(entry.getKey(), entry.getValue()); } } catch (Exception e) { throw new ErrorRowException(NAME, e.getMessage(), 500); } } return rows; }
InvokeHttp implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { Map<String, Object> parameters = new HashMap<>(); for (String column : columns) { int idx = row.find(column); if (idx != -1) { parameters.put(column, row.getValue(idx)); } } try { Map<String, Object> result = invokeHttp(url, parameters, headers); for (Map.Entry<String, Object> entry : result.entrySet()) { row.addOrSet(entry.getKey(), entry.getValue()); } } catch (Exception e) { throw new ErrorRowException(NAME, e.getMessage(), 500); } } return rows; } }
InvokeHttp implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { Map<String, Object> parameters = new HashMap<>(); for (String column : columns) { int idx = row.find(column); if (idx != -1) { parameters.put(column, row.getValue(idx)); } } try { Map<String, Object> result = invokeHttp(url, parameters, headers); for (Map.Entry<String, Object> entry : result.entrySet()) { row.addOrSet(entry.getKey(), entry.getValue()); } } catch (Exception e) { throw new ErrorRowException(NAME, e.getMessage(), 500); } } return rows; }  }
InvokeHttp implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { Map<String, Object> parameters = new HashMap<>(); for (String column : columns) { int idx = row.find(column); if (idx != -1) { parameters.put(column, row.getValue(idx)); } } try { Map<String, Object> result = invokeHttp(url, parameters, headers); for (Map.Entry<String, Object> entry : result.entrySet()) { row.addOrSet(entry.getKey(), entry.getValue()); } } catch (Exception e) { throw new ErrorRowException(NAME, e.getMessage(), 500); } } return rows; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); }
InvokeHttp implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { Map<String, Object> parameters = new HashMap<>(); for (String column : columns) { int idx = row.find(column); if (idx != -1) { parameters.put(column, row.getValue(idx)); } } try { Map<String, Object> result = invokeHttp(url, parameters, headers); for (Map.Entry<String, Object> entry : result.entrySet()) { row.addOrSet(entry.getKey(), entry.getValue()); } } catch (Exception e) { throw new ErrorRowException(NAME, e.getMessage(), 500); } } return rows; }  @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; }