Как реализовать пользовательский метод SamplingService для журнала обработки сообщений? После этого элемент извлекается и перед выполнением последовательности

Я новичок в 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>

Он извлекает некоторые элементы (документы XML) из transferFromMessageStore (очередь) и передает этот объект в последовательность transferProcessorSequence.xml, которая его использует. Как вы можете видеть, в это время я внедрил обработчик пользовательских сообщений SamplingProcessorHeaderRateLimit, который просто расширяет класс org.apache.synapse.message.processor.impl.sampler.SamplingProcessor WSO2, В это время он показывает только журнал, когда выполняется init(). Я развернул его на своем сервере Carbon, и он работает.

Здесь вы можете найти весь код проекта.

Хорошо, но из того, что я понял, чтобы получить желаемое поведение, мне не нужно просто расширять класс SamplingProcessor, потому что для выполнения пользовательской реализации между каждым потреблением сообщений и отправкой в последовательность необходимо расширьте класс SamplingService, этот один.

Я думаю, что мне нужно переопределить execute() или выборку (MessageConsumer msgConsumer).

В это время должно быть хорошо также вставить журнал, который записывается в файл журнала каждый раз, когда элемент извлекается из хранилища сообщений, и до этого выполняется последовательность, относящаяся к процессору сообщений.

Возможно ли это?

Итак, мои основные главные сомнения:

1) Мне нужно создать класс, расширяющий класс SamplingService в тот же проект, в котором я реализую свой обработчик сообщений (. Это поведение должно использоваться только для этого конкретного сообщения процессор в моем проекте ESO WSO2, все другие обработчики сообщений, используемые в этом проекте, должны использовать стандартную реализацию SamplingService).

2) Еще одно сомнение связано с тем, как эта пользовательская реализация SamplingService передается моему настраиваемому процессору сообщений. В класс SamplingProcessor WSO2 (, как связать конкретную реализацию пользовательского процессора сообщений с пользовательской реализацией SamplingService, обрабатывающей жизненный цикл).

Ответ 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, и будет вводить сообщения, которые он потребляет и обрабатывает, только для последовательности 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

@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());
}

Ответ 2

Вы настроили Task в XML?

<task name="CheckPrice" class="org.wso2.esb.tutorial.tasks.PlaceStockOrderTask"> 
  <trigger interval="5000"/>
</task>

или

<task name="CheckPrice" class="org.wso2.esb.tutorial.tasks.PlaceStockOrderTask">
  <trigger interval="5000" count="10"/>
</task>

Ответ 3

Q1:

Мне нужно создать класс, расширяющий класс SamplingService в тот же проект, в котором я реализую свой обработчик сообщений (это поведение должно использоваться только для этого конкретного сообщения процессор в моем проекте WSO2 ESB, используется весь другой обработчик сообщений в этом проекте должны использоваться стандартные SamplingService реализация).

<messageProcessor class="com.mycompany.toolkit.messageprocessor.SamplingProcessorHeaderRateLimitation" messageStore="transferFromMessageStore" name="transferFromMessageProcessor" xmlns="http://ws.apache.org/ns/synapse">
<parameter name="sequence">transferProcessorSequence</parameter>

Пользовательский процессор и служба будут использоваться, только если он указан как класс процессора, как указано выше в вашем примере.

Q2:

Еще одно сомнение связано с тем, как этот пользовательский SamplingService реализация передается в мой обработчик сообщений. В Класс SamplingProcessor WSO2 (как связать определенный пользовательский реализация процессора сообщений с помощью настраиваемого SamplingService реализация с его жизненным циклом).

Вы можете напрямую расширить/реализовать ScheduledMessageProcessor, а не SamplingProcessor, поскольку он реализует только getTask() и инициализирует объект просмотра в методе init(). Также ваш SamplingService2 класс должен расширять Task, интерфейсы ManagedLifecycle.