maxwell-1.25.1/src/main/java/com/zendesk/maxwell/scripting/Scripting.java
public class Scripting { static final Logger LOGGER = LoggerFactory.getLogger(Scripting.class); private final ScriptObjectMirror processRowFunc, processHeartbeatFunc, processDDLFunc; private ScriptObjectMirror getFunc(ScriptEngine engine, String fName, String filename) { ScriptObjectMirror f = (ScriptObjectMirror) engine.get(fName); if ( f == null ) return null; else if ( !f.isFunction() ) { throw new RuntimeException("Expected " + fName + " to be a function!"); } else { LOGGER.info("using function " + fName + " from " + filename); } return f; } public Scripting(String filename) throws IOException, ScriptException, NoSuchMethodException { ScriptEngineManager manager = new ScriptEngineManager(); ScriptEngine engine = manager.getEngineByName("nashorn"); String externJS = new String(Files.readAllBytes(Paths.get(filename))); engine.put("logger", LOGGER); engine.eval(externJS); processRowFunc = getFunc(engine, "process_row", filename); processHeartbeatFunc = getFunc(engine, "process_heartbeat", filename); processDDLFunc = getFunc(engine, "process_ddl", filename); if ( processRowFunc == null && processHeartbeatFunc == null && processDDLFunc == null ) LOGGER.warn("expected " + filename + " to define at least one of: process_row,process_heartbeat,process_ddl"); } public void invoke(RowMap row) { if ( row instanceof HeartbeatRowMap && processHeartbeatFunc != null ) processHeartbeatFunc.call(null, new WrappedHeartbeatMap((HeartbeatRowMap) row)); else if ( row instanceof DDLMap && processDDLFunc != null ) processDDLFunc.call(null, new WrappedDDLMap((DDLMap) row)); else if ( row instanceof RowMap && processRowFunc != null ) processRowFunc.call(null, new WrappedRowMap(row)); } private static ThreadLocal<ScriptEngine> stringifyEngineThreadLocal = ThreadLocal.withInitial(() -> { ScriptEngineManager manager = new ScriptEngineManager(); return manager.getEngineByName("nashorn"); }); public static String stringify(ScriptObjectMirror mirror) throws ScriptException { ScriptObjectMirror json = (ScriptObjectMirror) stringifyEngineThreadLocal.get().eval("JSON"); return (String) json.callMember("stringify", mirror); } } 复制代码
maxwell-1.25.1/src/main/java/com/zendesk/maxwell/MaxwellConfig.java
public class MaxwellConfig extends AbstractConfig { static final Logger LOGGER = LoggerFactory.getLogger(MaxwellConfig.class); //...... public String javascriptFile; public Scripting scripting; //...... public void validate() { validatePartitionBy(); validateFilter(); //...... if ( this.javascriptFile != null ) { try { this.scripting = new Scripting(this.javascriptFile); } catch ( Exception e ) { LOGGER.error("Error setting up javascript: ", e); System.exit(1); } } } //...... } 复制代码
Scripting定义了processRowFunc、processHeartbeatFunc、processDDLFunc属性;其构造器接收filename参数,它创建ScriptEngineManager,然后获取名为nashorn的ScriptEngine,之后通过filename获取externJS,并执行engine.eval(externJS),之后通过getFunc方法初始化化processRowFunc、processHeartbeatFunc、processDDLFunc;其getFunc方法通过engine.get(fName)获取ScriptObjectMirror;其invoke方法针对HeartbeatRowMap执行processHeartbeatFunc.call,针对DDLMap执行processDDLFunc.call,针对RowMap执行processRowFunc.call