FlinkInterpreter extends Interpreter { @Override public InterpreterResult interpret(String st, InterpreterContext context) throws InterpreterException { LOGGER.debug("Interpret code: " + st); this.z.setInterpreterContext(context); this.z.setGui(context.getGui()); this.z.setNoteGui(context.getNoteGui()); ClassLoader originClassLoader = Thread.currentThread().getContextClassLoader(); try { Thread.currentThread().setContextClassLoader(getFlinkScalaShellLoader()); createPlannerAgain(); setParallelismIfNecessary(context); setSavepointIfNecessary(context); return innerIntp.interpret(st, context); } finally { Thread.currentThread().setContextClassLoader(originClassLoader); } } FlinkInterpreter(Properties properties); @Override void open(); @Override void close(); @Override InterpreterResult interpret(String st, InterpreterContext context); @Override void cancel(InterpreterContext context); @Override FormType getFormType(); @Override int getProgress(InterpreterContext context); @Override List<InterpreterCompletion> completion(String buf,
                                                int cursor,
                                                InterpreterContext interpreterContext); void createPlannerAgain(); ClassLoader getFlinkScalaShellLoader(); FlinkScalaInterpreter getInnerIntp(); FlinkShims getFlinkShims(); void setSavepointIfNecessary(InterpreterContext context); void setParallelismIfNecessary(InterpreterContext context); FlinkVersion getFlinkVersion();  }
@Test public void testScalaBasic() throws InterpreterException, IOException { InterpreterContext context = getInterpreterContext(); InterpreterResult result = interpreter.interpret("val a=\"hello world\"", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); List<InterpreterResultMessage> resultMessages = context.out.toInterpreterResultMessage(); assertEquals(InterpreterResult.Type.TEXT, resultMessages.get(0).getType()); assertEquals("a: String = hello world\n", resultMessages.get(0).getData()); context = getInterpreterContext(); result = interpreter.interpret("print(a)", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); resultMessages = context.out.toInterpreterResultMessage(); assertEquals(InterpreterResult.Type.TEXT, resultMessages.get(0).getType()); assertEquals("hello world", resultMessages.get(0).getData()); context = getInterpreterContext(); result = interpreter.interpret("System.out.print(a)", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); resultMessages = context.out.toInterpreterResultMessage(); assertEquals(InterpreterResult.Type.TEXT, resultMessages.get(0).getType()); assertEquals("hello world", resultMessages.get(0).getData()); context = getInterpreterContext(); result = interpreter.interpret("System.err.print(a)", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); resultMessages = context.out.toInterpreterResultMessage(); assertEquals(InterpreterResult.Type.TEXT, resultMessages.get(0).getType()); assertEquals("hello world", resultMessages.get(0).getData()); result = interpreter.interpret("println(a", getInterpreterContext()); assertEquals(InterpreterResult.Code.INCOMPLETE, result.code()); context = getInterpreterContext(); result = interpreter.interpret("println(b)", context); assertEquals(InterpreterResult.Code.ERROR, result.code()); resultMessages = context.out.toInterpreterResultMessage(); assertEquals(InterpreterResult.Type.TEXT, resultMessages.get(0).getType()); assertTrue(resultMessages.get(0).getData(), resultMessages.get(0).getData().contains("not found: value b")); context = getInterpreterContext(); result = interpreter.interpret("\"123\".\ntoInt", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); context = getInterpreterContext(); result = interpreter.interpret("", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); context = getInterpreterContext(); result = interpreter.interpret("\nprint(\"hello world\")", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); context = getInterpreterContext(); result = interpreter.interpret("", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); context = getInterpreterContext(); result = interpreter.interpret("def add(x:Int, y:Int)\n{ return x+y }", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("print(add(1,2))", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("print(\"hello world\")", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("class Counter {\n " + "var value: Long = 0} \n" + "object Counter {\n def apply(x: Long) = new Counter()\n}", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); context = getInterpreterContext(); result = interpreter.interpret("z.input(\"name\", \"default_name\")", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(1, context.getGui().getForms().size()); assertTrue(context.getGui().getForms().get("name") instanceof TextBox); TextBox textBox = (TextBox) context.getGui().getForms().get("name"); assertEquals("name", textBox.getName()); assertEquals("default_name", textBox.getDefaultValue()); context = getInterpreterContext(); result = interpreter.interpret("z.checkbox(\"checkbox_1\", " + "Seq(\"value_2\"), Seq((\"value_1\", \"name_1\"), (\"value_2\", \"name_2\")))", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(1, context.getGui().getForms().size()); assertTrue(context.getGui().getForms().get("checkbox_1") instanceof CheckBox); CheckBox checkBox = (CheckBox) context.getGui().getForms().get("checkbox_1"); assertEquals("checkbox_1", checkBox.getName()); assertEquals(1, checkBox.getDefaultValue().length); assertEquals("value_2", checkBox.getDefaultValue()[0]); assertEquals(2, checkBox.getOptions().length); assertEquals("value_1", checkBox.getOptions()[0].getValue()); assertEquals("name_1", checkBox.getOptions()[0].getDisplayName()); assertEquals("value_2", checkBox.getOptions()[1].getValue()); assertEquals("name_2", checkBox.getOptions()[1].getDisplayName()); context = getInterpreterContext(); result = interpreter.interpret("z.select(\"select_1\", Seq(\"value_2\"), " + "Seq((\"value_1\", \"name_1\"), (\"value_2\", \"name_2\")))", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(1, context.getGui().getForms().size()); assertTrue(context.getGui().getForms().get("select_1") instanceof Select); Select select = (Select) context.getGui().getForms().get("select_1"); assertEquals("select_1", select.getName()); assertEquals(2, select.getOptions().length); assertEquals("value_1", select.getOptions()[0].getValue()); assertEquals("name_1", select.getOptions()[0].getDisplayName()); assertEquals("value_2", select.getOptions()[1].getValue()); assertEquals("name_2", select.getOptions()[1].getDisplayName()); }
@Test public void testZShow() throws InterpreterException, IOException { InterpreterContext context = getInterpreterContext(); InterpreterResult result = interpreter.interpret( "val data = benv.fromElements((1, \"jeff\"), (2, \"andy\"), (3, \"james\"))", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); context = getInterpreterContext(); result = interpreter.interpret("z.show(data)", context); assertEquals(context.out.toString(), InterpreterResult.Code.SUCCESS, result.code()); List<InterpreterResultMessage> resultMessages = context.out.toInterpreterResultMessage(); assertEquals(InterpreterResult.Type.TABLE, resultMessages.get(0).getType()); assertEquals("_1\t_2\n1\tjeff\n2\tandy\n3\tjames\n", resultMessages.get(0).getData()); }
@Test public void testBatchWordCount() throws InterpreterException, IOException { InterpreterContext context = getInterpreterContext(); InterpreterResult result = interpreter.interpret( "val data = benv.fromElements(\"hello world\", \"hello flink\", \"hello hadoop\")", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); context = getInterpreterContext(); result = interpreter.interpret( "data.flatMap(line => line.split(\"\\\\s\"))\n" + " .map(w => (w, 1))\n" + " .groupBy(0)\n" + " .sum(1)\n" + " .print()", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); String[] expectedCounts = {"(hello,3)", "(world,1)", "(flink,1)", "(hadoop,1)"}; Arrays.sort(expectedCounts); String[] counts = context.out.toInterpreterResultMessage().get(0).getData().split("\n"); Arrays.sort(counts); assertArrayEquals(expectedCounts, counts); }
@Test public void testStreamWordCount() throws InterpreterException, IOException { InterpreterContext context = getInterpreterContext(); InterpreterResult result = interpreter.interpret( "val data = senv.fromElements(\"hello world\", \"hello flink\", \"hello hadoop\")", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); context = getInterpreterContext(); result = interpreter.interpret( "data.flatMap(line => line.split(\"\\\\s\"))\n" + " .map(w => (w, 1))\n" + " .keyBy(0)\n" + " .sum(1)\n" + " .print()\n" + "senv.execute()", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); String[] expectedCounts = {"(hello,3)", "(world,1)", "(flink,1)", "(hadoop,1)"}; String output = context.out.toInterpreterResultMessage().get(0).getData(); for (String expectedCount : expectedCounts) { assertTrue(output, output.contains(expectedCount)); } }