Groovy

概述

这是最"Groovy"的 Groovy 脚本 :)

脚本绑定：

SQL Map 详情

示例： 如果你定义了属性 `SQL.mydb` 并把它链接到任意 DBCPService，就可以在代码中这样访问： SQL.mydb.rows('select * from mytable')

该 Processor 会在执行脚本前自动从 dbcp 服务获取连接，并尝试处理事务： 脚本发生异常时数据库事务自动回滚，成功时提交。 你也可以手动管理事务。 注意：脚本不能断开连接。

SessionFile - FlowFile 扩展

(cn.org.acal.lynxus.processors.groovyx.flow.SessionFile) 是 Extended Groovy Processor 中 session 返回的实际对象。 该 FlowFile 是一个引用 session 和真实 FlowFile 的容器。 这允许使用简化的语法处理文件 Attribute 和内容：

设置新的 Attribute 值

flowFile.ATTRIBUTE_NAME = ATTRIBUTE_VALUE
flowFile.'mime.type' = 'text/xml'
flowFile.putAttribute("ATTRIBUTE_NAME", ATTRIBUTE_VALUE)
// 等同于
flowFile = session.putAttribute(flowFile, "ATTRIBUTE_NAME", ATTRIBUTE_VALUE)

移除 Attribute

flowFile.ATTRIBUTE_NAME = null
// 等同于
flowFile = session.removeAttribute(flowFile, "ATTRIBUTE_NAME")

获取 Attribute 值

String a = flowFile.ATTRIBUTE_NAME

写入内容

flowFile.write("UTF-8", "THE CharSequence to write into flow file replacing current content")
flowFile.write("UTF-8") { writer ->
    // 对
    // java.io.Writer 做些处理...
}
flowFile.write { outStream ->
    // 对输出流做些处理...
}
flowFile.write { inStream, outStream ->
    // 对输入流和输出流做些处理...
}

获取内容

InputStream i = flowFile.read()
def json = new groovy.json.JsonSlurper().parse(flowFile.read())
String text = flowFile.read().getText("UTF-8")

把 FlowFile 转移到 success Relationship

REL_SUCCESS << flowFile
flowFile.transfer(REL_SUCCESS)
// 等同于：
session.transfer(flowFile, REL_SUCCESS)

使用 dbcp

import groovy.sql.Sql

// 定义名为
// `SQL.db`、连接到
// DBCPConnectionPool Controller Service 的属性
// 本例是一个
// H2 数据库示例

// 使用预编译语句从数据库读取值
// 并赋给
// FlowFile Attribute `db.yesterday`
def daysAdd = -1
def row = SQL.db.firstRow("select dateadd('DAY', ${daysAdd}, sysdate) as DB_DATE from dual")
flowFile.'db.yesterday' = row.DB_DATE

// 处理数据库中的
// BLOB 和
// CLOB 时，
// 使用
// groovy.sql.Sql.BLOB(Stream) 和
// groovy.sql.Sql.CLOB(Reader) 进行参数类型转换

// 把
// FlowFile 的内容写入数据库
// blob
flowFile.read { rawIn ->
    def parms = [
            p_id  : flowFile.ID as Long, // 获取名为
                                         // \`ID\` 的
                                         // FlowFile Attribute
            p_data: Sql.BLOB(rawIn),   // 把输入流用作
                                       // BLOB SQL 参数
    ]
    SQL.db.executeUpdate(parms, "update mytable set data = :p_data where id = :p_id")
}

处理

Processor 启动与停止

在 extended groovy processor 中，你可以通过提供对应的静态方法来捕获 `start`、`stop` 和 `unscheduled` 事件：

import cn.org.acal.lynxus.processor.ProcessContext
import java.util.concurrent.atomic.AtomicLong

class Const {
    static Date startTime = null;
    static AtomicLong triggerCount = null;
}

static onStart(ProcessContext context) {
    Const.startTime = new Date()
    Const.triggerCount = new AtomicLong(0)
    println "onStart $context ${Const.startTime}"
}

static onStop(ProcessContext context) {
    def alive = (System.currentTimeMillis() - Const.startTime.getTime()) / 1000
    println "onStop $context executed ${Const.triggerCount} times during ${alive} seconds"
}

static onUnscheduled(ProcessContext context) {
    def alive = (System.currentTimeMillis() - Const.startTime.getTime()) / 1000
    println "onUnscheduled $context executed ${Const.triggerCount} times during ${alive} seconds"
}

flowFile.'trigger.count' = Const.triggerCount.incrementAndGet()
REL_SUCCESS << flowFile