java - 如何为自定义消息处理器实现自定义 SamplingService？在检索元素之后和执行序列之前记录

Question

我是 WSO2 ESB 的新手，我实现了一个具有此特定行为的自定义消息处理器:在从消息存储中检索元素之后和在执行与此消息处理器相关的序列。

我尽量详细解释。

这是我的 ESB 消息处理器定义:

<?xml version="1.0" encoding="UTF-8"?>
<!---<messageProcessor class="org.apache.synapse.message.processor.impl.sampler.SamplingProcessor" messageStore="transferFromMessageStore" name="transferFromMessageProcessor" xmlns="http://ws.apache.org/ns/synapse">-->
<messageProcessor class="com.mycompany.toolkit.messageprocessor.SamplingProcessorHeaderRateLimitation" messageStore="transferFromMessageStore" name="transferFromMessageProcessor" xmlns="http://ws.apache.org/ns/synapse">
    <parameter name="sequence">transferProcessorSequence</parameter>
    <parameter name="interval">1000</parameter>
    <parameter name="is.active">true</parameter>
    <parameter name="concurrency">1</parameter>
</messageProcessor>

它从 transferFromMessageStore(一个队列)中检索一些元素(XML 文档)并将此对象传递给使用它的 transferProcessorSequence.xml 序列。正如您此时看到的，我已经实现了一个自定义消息处理器 SamplingProcessorHeaderRateLimit，它只是扩展了 org.apache.synapse.message.processor.impl.sampler.SamplingProcessor WSO2 类.此时它仅在执行 init() 方法时显示日志。我将它部署在我的 Carbon 服务器上，它运行良好。

Here你可以找到整个项目代码。

好的，但根据我的理解，为了获得所需的行为，我不必简单地扩展 SamplingProcessor 类，因为为了在每个消息消费和分派(dispatch)到序列之间进行自定义实现，需要扩展类 SamplingService 类，这 one .

我认为我需要覆盖 execute() 或 fetch(MessageConsumer msgConsumer)。

此时应该可以插入一个日志，每次从消息存储中检索到一个元素时，以及在执行与消息处理器相关的序列之前，都会写入日志文件。

这可能吗？

所以我的主要怀疑是:

1) 我是否要创建一个类，将 SamplingService 类扩展到我正在实现自定义消息处理器的同一项目中(此行为必须仅用于此特定消息我的 WSO2 ESB 项目中的处理器，该项目中使用的所有其他消息处理器都必须使用标准的 SamplingService 实现)。

2) 另一个疑问与此自定义 SamplingService 实现如何传递到我的自定义消息处理器有关。进入 SamplingProcessor WSO2 类(如何将特定的自定义消息处理器实现与处理其生命周期的自定义 SamplingService 实现相关联)。

Answer 1

1) Have I to create a class extending the SamplingService class into the same project in which I am implementing my custom message processor (this behavior have to be used only for this specific message processor in my WSO2 ESB project, all the other message processor used in this project have to use the standard SamplingService implementation).

您的自定义 SamplingProcessorHeaderRateLimitation 将仅使用进入 transferFromMessageStore 的消息，并将其使用和处理的消息注入(inject)到序列 transferProcessorSequence 中。此消息处理器不会处理所有其他路径。

2) Another doubt is related about how this custom SamplingService implementation is passed to my custom message processor. Into the the SamplingProcessor WSO2 class (how to associate a specific custom message processor implementation with a custom SamplingService implementation handling its lifecycle).

如果您查看您实现的源代码 SamplingProcessorHeaderRateLimitation.getTask() 您已将您的自定义 SamplingService2 与您的自定义 SamplingProcessorHeaderRateLimitation 相关联/p>

@Override
protected Task getTask() {
    logger.info("getTask() START");
    System.out.println("getTask() START");
    logger.info("getTask() END");
    System.out.println("getTask() END");
    return (Task) new SamplingService2(this, synapseEnvironment, CONCURRENCY, SEQUENCE, isProcessorStartAsDeactivated());
}