开始之前
输入节点可从任何类型的外部源接收数据(例如,文件系统或 FTP 连接),只要从节点的输出是正确的格式即可。对于到队列或数据库的连接,您应该使用提供的 IBM 原语节点和 API 调用,主要因为原语节点已设置了错误处理。不要使用 mqget 或 mqput 命令以直接访问 WebSphere MQ 队列。
{ static char* functionName = (char *)"_Input_run()"; void* buffer; CciTerminal* terminalObject; int buflen = 4096; int rc = CCI_SUCCESS; int rcDispatch = CCI_SUCCESS; buffer = readFromDevice(&buflen); cniSetInputBuffer(&rc, message, buffer, buflen); } /*propagate etc*/
当消息已通过消息流传播时,输入节点负责执行相应的消息处理结束。明确地,输入节点需要引起提交或回滚任何事务,并将线程返回线程池。
从每个消息流维护的线程的池分配每个消息流线程,且在 cniRun 函数中启动执行。您使用与带有合适的返回值的 cniDispatchThread 实用程序函数来确定线程的行为。
您可以从 cniRun 函数调用 cniDispatchThread 实用程序函数,以使另一个线程开始执行 cniRun 函数。 执行另一个线程最适当的时间是,您在新线程上建立可供函数处理的数据时。
通常在此处使用术语事务以描述全局协调的事务或代理控制的事务。全局协调的事务被 WebSphere MQ作为 XA 兼容资源恢复服务或 z/OS 资源恢复服务(RRS)协调。WebSphere Message Broker 通过提交(或回滚)任何数据库资源,然后提交任何 WebSphere MQ 工作单元来控制事务,但是,如果使用了用户定义的节点,代理无法自动提交资源更新。用户定义的节点使用返回值表明事务是否成功,控制事务是提交还是回滚。代理基础捕获任何未处理的意外情况,并且事务发生回滚。
下表中描述了每个支持的返回值,这影响没有有任何事务的值,和代理对当前线程做些什么。
返回值 | 对事务的影响 | 线程上的代理操作 |
---|---|---|
CCI_SUCCESS_CONTINUE | 已提交 | 在 cniRun 函数中再次调用相同的线程。 |
CCI_SUCCESS_RETURN | 已提交 | 将线程返回到线程池。 |
CCI_FAILURE_CONTINUE | 回滚 | 在 cniRun 函数中再次调用相同的线程。 |
CCI_FAILURE_RETURN | 回滚 | 将线程返回到线程池。 |
CCI_TIMEOUT | 不适用。当等待输入消息时,此函数周期性地超时。 | 在 cniRun 函数中再次调用相同的线程。 |
{ ... cniDispatchThread(&rcDispatch, ((NODE_CONTEXT_ST *)context)->nodeObject); ... if (rcDispatch == CCI_NO_THREADS_AVAILABLE) return CCI_SUCCESS_CONTINUE; else return CCI_SUCCESS_RETURN; }
在您传播消息之前,您必须确定您要传播的消息流数据,以及用于接收数据的终端。
terminalObject 从用户定义的节点自己维护的列表派生而来。
if (terminalObject) { if (cniIsTerminalAttached(&rc, terminalObject)) { if (rc == CCI_SUCCESS) { cniPropagate(&rc, terminalObject, destinationList, exceptionList, message); } }
在上面的示例中,cniIsTerminalAttached 函数用于测试消息是否可以传播到指定的终端。如果您不使用 cniIsTerminalAttached 函数,且终端没有被连接器连接到另一个节点,则此消息没有传播且不返回任何警告消息。使用 cniIsTerminalAttached 函数来阻止这种情况的发生。