FlinkInterpreter extends Interpreter { @Override public InterpreterResult interpret(String line, InterpreterContext context) { if (line == null || line.trim().length() == 0) { return new InterpreterResult(Code.SUCCESS); } InterpreterResult result = interpret(line.split("\n"), context); return result; } FlinkInterpreter(Properties property); @Override void open(); Object getValue(String name); @Override void close(); @Override InterpreterResult interpret(String line, InterpreterContext context); InterpreterResult interpret(String[] lines, InterpreterContext context); @Override void cancel(InterpreterContext context); @Override FormType getFormType(); @Override int getProgress(InterpreterContext context); @Override List<String> completion(String buf, int cursor);  }
@Test public void testWordCount() { flink.interpret("val text = env.fromElements(\"To be or not to be\")", context); flink.interpret("val counts = text.flatMap { _.toLowerCase.split(\" \") }.map { (_, 1) }.groupBy(0).sum(1)", context); InterpreterResult result = flink.interpret("counts.print()", context); assertEquals(Code.SUCCESS, result.code()); String[] expectedCounts = {"(to,2)", "(be,2)", "(or,1)", "(not,1)"}; Arrays.sort(expectedCounts); String[] counts = result.message().split("\n"); Arrays.sort(counts); assertArrayEquals(expectedCounts, counts); }
@Test public void testSimpleStatement() { InterpreterResult result = flink.interpret("val a=1", context); result = flink.interpret("print(a)", context); assertEquals("1", result.message()); }
@Test public void testSimpleStatementWithSystemOutput() { InterpreterResult result = flink.interpret("val a=1", context); result = flink.interpret("System.out.print(a)", context); assertEquals("1", result.message()); }
@Test public void testNextlineInvoke() { InterpreterResult result = flink.interpret("\"123\"\n .toInt", context); assertEquals("res0: Int = 123\n", result.message()); }