How can I set the message processor interval parameter using a non fixed value? (using a value stored in a property\registry\etc)

Question

I am very new to WSO2 ESB and I have the following problem with a message processor.

I have this message processor definiation that every second extract an element (an XML document previously stored) from a queue and process it calling a sequence named transferProcessorSequence. This is the configuration:

<?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">
    <parameter name="sequence">transferProcessorSequence</parameter>
    <parameter name="interval">1000</parameter>
    <parameter name="is.active">true</parameter>
    <parameter name="concurrency">1</parameter>
</messageProcessor>

It works fine. My problem is that I have not to do it every second (it should not be a fixed value) but have to use a value that I have stored into a registry (eventually I can store this value elsewhere).

Instead of having this setting:

<parameter name="interval">1000</parameter>

I have to retrieve a stored value from a registry or something like this (also a property).

How can I do something like this?

**EDIT 1: I am trying to follow the solution provided by Jean-Michel but I am finding some difficulties trying to build my project with Maven.

I created a simple Maven project, this is my pom.xml file content (at this time I have yet no implemented the Java class, I am only trying to compile an empty project):

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.mycompany.toolkit</groupId>
    <artifactId>SamplingProcessorHeaderRateLimitation</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>bundle</packaging>
    <name>Sampling Processor Header Rate Limitation</name>
    <description>Custom Sampling Mesageprocessor using response header to implement the rate limitation</description>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>2.0</version>
                <configuration>
                    <source>1.7</source>
                    <target>1.7</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.felix</groupId>
                <artifactId>maven-bundle-plugin</artifactId>
                <version>2.3.4</version>
                <extensions>true</extensions>
                <configuration>
                    <instructions>
                        <Bundle-SymbolicName>com.sample.messageprocessor</Bundle-SymbolicName>
                        <Bundle-Name>com.sample.messageprocessor</Bundle-Name>
                        <Export-Package>com.sample.*</Export-Package>
                        <DynamicImport-Package>*</DynamicImport-Package>
                        <Implementation-Build>${buildNumber}</Implementation-Build>
                        <Scm-Root>${project.scm.connection}</Scm-Root>
                    </instructions>
                </configuration>
            </plugin>
        </plugins>
    </build>
    <repositories>
        <repository>
            <releases>
                <updatePolicy>daily</updatePolicy>
                <checksumPolicy>ignore</checksumPolicy>
            </releases>
            <id>wso2-nexus</id>
            <url>http://maven.wso2.org/nexus/content/groups/wso2-public/</url>
        </repository>
    </repositories>
    <pluginRepositories>
        <pluginRepository>
            <releases>
                <updatePolicy>daily</updatePolicy>
                <checksumPolicy>ignore</checksumPolicy>
            </releases>
            <id>wso2-nexus</id>
            <url>http://maven.wso2.org/nexus/content/groups/wso2-public/</url>
        </pluginRepository>
    </pluginRepositories>
    <dependencies>
        <dependency>
            <groupId>org.apache.synapse</groupId>
            <artifactId>synapse-core</artifactId>
            <version>2.1.7-wso2v3</version>
        </dependency>
        <dependency>
            <groupId>commons-logging</groupId>
            <artifactId>commons-logging</artifactId>
            <version>1.1.1</version>
        </dependency>
    </dependencies>
    <properties>
        <CApp.type>lib/synapse/mediator</CApp.type>
    </properties>
</project>

The problem is that when I perfrom the mvn package statment for this Maven project I obtain the following error message:

nobil@DESKTOP-VCON7T6 MINGW64 ~/OneDrive/Documenti/MyCompany/JAVA/workspace-sts-3.8.4.RELEASE/SamplingProcessorHeaderRateLimitation
$ mvn -e package
[INFO] Error stacktraces are turned on.
[INFO] Scanning for projects...
[INFO]
[INFO] ------------------------------------------------------------------------
[INFO] Building Sampling Processor Header Rate Limitation 0.0.1-SNAPSHOT
[INFO] ------------------------------------------------------------------------
[INFO]
[INFO] --- maven-resources-plugin:3.0.2:resources (default-resources) @ SamplingProcessorHeaderRateLimitation ---
[WARNING] Using platform encoding (Cp1252 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] Copying 0 resource
[INFO]
[INFO] --- maven-compiler-plugin:2.0:compile (default-compile) @ SamplingProcessorHeaderRateLimitation ---
[INFO] Nothing to compile - all classes are up to date
[INFO]
[INFO] --- maven-resources-plugin:3.0.2:testResources (default-testResources) @ SamplingProcessorHeaderRateLimitation ---
[WARNING] Using platform encoding (Cp1252 actually) to copy filtered resources, i.e. build is platform dependent!
[INFO] Copying 0 resource
[INFO]
[INFO] --- maven-compiler-plugin:2.0:testCompile (default-testCompile) @ SamplingProcessorHeaderRateLimitation ---
[INFO] Nothing to compile - all classes are up to date
[INFO]
[INFO] --- maven-surefire-plugin:2.20:test (default-test) @ SamplingProcessorHeaderRateLimitation ---
[INFO]
[INFO] --- maven-bundle-plugin:2.3.4:bundle (default-bundle) @ SamplingProcessorHeaderRateLimitation ---
[WARNING] Warning building bundle com.mycompany.toolkit:SamplingProcessorHeaderRateLimitation:bundle:0.0.1-SNAPSHOT : Instructions in Export-Package that are never used: com\.sample\..*|com\.sample
Classpath: Jar:.,Jar:synapse-core,Jar:org.wso2.securevault,Jar:axiom-api,Jar:geronimo-activation_1.1_spec,Jar:geronimo-javamail_1.4_spec,Jar:jaxen,Jar:geronimo-stax-api_1.0_spec,Jar:axiom-impl,Jar:wstx-asl,Jar:log4j,Jar:jline,Jar:commons-cli,Jar:commons-io,Jar:commons-codec,Jar:synapse-commons,Jar:commons-dbcp,Jar:commons-pool,Jar:activation,Jar:axis2-transport-base,Jar:snmp4j-agent,Jar:snmp4j,Jar:axis2-transport-http,Jar:spring-core,Jar:commons-vfs2,Jar:cache-api,Jar:commons-net,Jar:oro,Jar:javax.servlet-api,Jar:bcpkix-jdk15on,Jar:bcprov-jdk15on,Jar:synapse-tasks,Jar:quartz,Jar:c3p0,Jar:slf4j-api,Jar:geronimo-jta_1.1_spec,Jar:commons-collections,Jar:synapse-nhttp-transport,Jar:httpcore,Jar:httpcore-nio,Jar:httpclient,Jar:wso2caching-core,Jar:axis2-clustering,Jar:tomcat-tribes,Jar:tomcat-juli,Jar:tomcat-embed-logging-juli,Jar:xmlunit,Jar:aspectjweaver,Jar:jms,Jar:amqp-client,Jar:antlr-runtime,Jar:stringtemplate,Jar:antlr,Jar:axis2-saaj,Jar:geronimo-saaj_1.3_spec,Jar:axiom-dom,Jar:axis2-kernel,Jar:geronimo-ws-metadata_2.0_spec,Jar:servlet-api,Jar:commons-httpclient,Jar:commons-fileupload,Jar:wsdl4j,Jar:XmlSchema,Jar:woden-api,Jar:XmlSchema,Jar:woden-impl-dom,Jar:woden-impl-commons,Jar:jsr311-api,Jar:junit,Jar:hamcrest-core,Jar:handy-uri-templates,Jar:commons-lang,Jar:jettison,Jar:jettison,Jar:stax-api,Jar:json-path,Jar:json-smart,Jar:wso2eventing-api,Jar:axis2-adb,Jar:sandesha2-core,Jar:axis2-codegen,Jar:neethi,Jar:axis2-mtompolicy,Jar:saxon-dom,Jar:saxon,Jar:xalan,Jar:serializer,Jar:xml-apis,Jar:axis2-transport-local,Jar:json-schema-validator-all,Jar:json-schema-validator,Jar:jsr305,Jar:joda-time,Jar:libphonenumber,Jar:mailapi,Jar:jopt-simple,Jar:json-schema-core,Jar:rhino,Jar:uri-template,Jar:guava,Jar:msg-simple,Jar:btf,Jar:jackson-coreutils,Jar:jackson-databind,Jar:jackson-annotations,Jar:jackson-core,Jar:commons-logging

[WARNING] Warning building bundle com.mycompany.toolkit:SamplingProcessorHeaderRateLimitation:bundle:0.0.1-SNAPSHOT : Superfluous export-package instructions: [com.sample.*]
[ERROR] Error building bundle com.mycompany.toolkit:SamplingProcessorHeaderRateLimitation:bundle:0.0.1-SNAPSHOT : The JAR is empty: dot
[ERROR] Error(s) found in bundle configuration
[INFO] ------------------------------------------------------------------------
[INFO] BUILD FAILURE
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 3.568 s
[INFO] Finished at: 2017-07-17T14:25:45+02:00
[INFO] Final Memory: 16M/284M
[INFO] ------------------------------------------------------------------------
[ERROR] Failed to execute goal org.apache.felix:maven-bundle-plugin:2.3.4:bundle (default-bundle) on project SamplingProcessorHeaderRateLimitation: Error(s) found in bundle configuration -> [Help 1]
org.apache.maven.lifecycle.LifecycleExecutionException: Failed to execute goal org.apache.felix:maven-bundle-plugin:2.3.4:bundle (default-bundle) on project SamplingProcessorHeaderRateLimitation: Error(s) found in bundle configuration
        at org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:212)
        at org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:153)
        at org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:145)
        at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject(LifecycleModuleBuilder.java:116)
        at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject(LifecycleModuleBuilder.java:80)
        at org.apache.maven.lifecycle.internal.builder.singlethreaded.SingleThreadedBuilder.build(SingleThreadedBuilder.java:51)
        at org.apache.maven.lifecycle.internal.LifecycleStarter.execute(LifecycleStarter.java:128)
        at org.apache.maven.DefaultMaven.doExecute(DefaultMaven.java:307)
        at org.apache.maven.DefaultMaven.doExecute(DefaultMaven.java:193)
        at org.apache.maven.DefaultMaven.execute(DefaultMaven.java:106)
        at org.apache.maven.cli.MavenCli.execute(MavenCli.java:863)
        at org.apache.maven.cli.MavenCli.doMain(MavenCli.java:288)
        at org.apache.maven.cli.MavenCli.main(MavenCli.java:199)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.codehaus.plexus.classworlds.launcher.Launcher.launchEnhanced(Launcher.java:289)
        at org.codehaus.plexus.classworlds.launcher.Launcher.launch(Launcher.java:229)
        at org.codehaus.plexus.classworlds.launcher.Launcher.mainWithExitCode(Launcher.java:415)
        at org.codehaus.plexus.classworlds.launcher.Launcher.main(Launcher.java:356)
Caused by: org.apache.maven.plugin.MojoExecutionException: Error(s) found in bundle configuration
        at org.apache.felix.bundleplugin.BundlePlugin.execute(BundlePlugin.java:341)
        at org.apache.felix.bundleplugin.BundlePlugin.execute(BundlePlugin.java:224)
        at org.apache.felix.bundleplugin.BundlePlugin.execute(BundlePlugin.java:215)
        at org.apache.maven.plugin.DefaultBuildPluginManager.executeMojo(DefaultBuildPluginManager.java:134)
        at org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:207)
        ... 20 more
Caused by: org.apache.maven.plugin.MojoFailureException: Error(s) found in bundle configuration
        at org.apache.felix.bundleplugin.BundlePlugin.execute(BundlePlugin.java:293)
        ... 24 more
[ERROR]
[ERROR] Re-run Maven using the -X switch to enable full debug logging.
[ERROR]
[ERROR] For more information about the errors and possible solutions, please read the following articles:
[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoExecutionException

So, why? What is the problem? I think that maybe I have to configure the manven-bundle-plugin plugin. But what exactly have I to do to correctly configure it? And what this plugin does?

Jenananthan · Accepted Answer

As others have mentioned you will need to create your own message processor by extending the SampleProcessor and overriding the setParameter methods.

org.apache.synapse.message.processor.impl.sampler.SamplingProcessor

Where in setParameters method read the parameters from registry and over write the parameter map.

Custom message processor which reads parameters from a property file which is defined in governance space of registry

package org.apache.synapse.message.custom.processor.impl;

import java.io.IOException;
import java.io.StringReader;
import java.util.Map;
import java.util.Properties;

import org.wso2.carbon.context.CarbonContext;
import org.wso2.carbon.context.RegistryType;
import org.wso2.carbon.registry.api.Registry;
import org.wso2.carbon.registry.api.RegistryException;
import org.wso2.carbon.registry.api.Resource;

import org.apache.synapse.message.processor.impl.sampler.SamplingProcessor;;

public class CustomSamplingProcessor extends SamplingProcessor {
    private final static String REGISTRY_PATH = "reg";

    @Override
    public void setParameters(Map<String, Object> parameters) {
        if (parameters.get(REGISTRY_PATH) != null) {
            String resourcePath = (String) parameters.get(REGISTRY_PATH);
            parameters = populateParamsFromReg(resourcePath, parameters);
        }
        System.out.println("############################## set params################");
        super.setParameters(parameters);
    }

    private Map<String, Object> populateParamsFromReg(String resourcePath, Map<String, Object> parameters) {
        Properties prop = readPropertyFile(resourcePath);
        if (prop != null) {
            for (String key : prop.stringPropertyNames()) {
                String value = prop.getProperty(key);
                parameters.put(key, value);
            }
        }
        return parameters;
    }

    private Properties readPropertyFile(String resourcePath) {
        Properties prop = null;
        try {
            CarbonContext cCtx = CarbonContext.getThreadLocalCarbonContext();
            Registry registry = cCtx.getRegistry(RegistryType.USER_GOVERNANCE);
            Resource resource = registry.get(resourcePath);
            Object content = resource.getContent();
            String output = new String((byte[]) content);
            System.out.println(output);
            prop = parseProperties(output);
        } catch (RegistryException e) {
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return prop;
    }

    public Properties parseProperties(String fileContent) throws IOException {
        final Properties properties = new Properties();
        properties.load(new StringReader(fileContent));
        return properties;
    }
}

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.apache.synapse.message.custom.processor.impl</groupId>
    <artifactId>custom-message-processor</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>

    <name>message.custom.processor</name>
    <url>http://maven.apache.org</url>
    <repositories>
        <repository>
            <id>wso2-nexus</id>
            <name>WSO2 internal Repository</name>
            <url>http://maven.wso2.org/nexus/content/groups/wso2-public/</url>
            <releases>
                <enabled>true</enabled>
                <updatePolicy>daily</updatePolicy>
                <checksumPolicy>fail</checksumPolicy>
            </releases>
        </repository>
    </repositories>


    <dependencies>
        <dependency>
            <groupId>org.apache.synapse</groupId>
            <artifactId>synapse-core</artifactId>
            <version>2.1.2-wso2v4</version>
        </dependency>
        <dependency>
            <groupId>org.wso2.carbon</groupId>
            <artifactId>org.wso2.carbon.registry.api</artifactId>
            <version>4.3.0</version>
        </dependency>

        <dependency>
            <groupId>org.wso2.carbon</groupId>
            <artifactId>org.wso2.carbon.registry.core</artifactId>
            <version>4.3.0</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.1</version>
                <configuration>
                    <source>1.7</source>
                    <target>1.7</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

Create a property file and add the parameters which should be read from registry e.g

File location in registry

/_system/governance/custom-message-processor/customprocessor.properties

content

interval:1000
concurrency:1
sequence:replySeq
is.active:true

How to use the custom processor in EI

Built the custom message processor, and add the jar to <EI_HOME>\lib (since this jar is an none osgi jar)
Restart/Start the EI
Add the property file to registry
Create custom message processor and add the registry file path of the property file as parameter to custom message processor configuration with parameter name "reg" (Give the relative registry path since in code level we read from governance space. i.e /custom-message-processor/customprocessor.properties).

  <messageProcessor xmlns="http://ws.apache.org/ns/synapse" class="org.apache.synapse.message.custom.processor.impl.CustomSamplingProcessor" name="CustomSamplingProcessor" messageStore="JMSMS">   
       <parameter name="reg">/custom-message-processor/customprocessor.properties</parameter>   
  </messageProcessor>

Full source code can be found here[1] https://github.com/jenananthan/wso2-esb-customization/tree/master/sampling_custom_message_processor

Jean-Michel · Answer

Option 1 : Develop a custom MessageProcessor that extends the SamplingProcessor class from WSO2. (Have a look to https://github.com/wso2/wso2-synapse modules\core\src\main\java\org\apache\synapse\message\processor\impl)

Extends org.apache.synapse.message.processor.impl.sampler.SamplingProcessor (or directly org.apache.synapse.message.processor.impl.ScheduledMessageProcessor) and override method setParameters

Call super.setParameters and then update interval attribute with a value that you can read where you want.

In your MessageProcessor definition, use your own class behind class XML attribute

Option 2 : Use admin services

edit carbon.xml and set HideAdminServiceWSDLs to false
admin services listen on https://host:9443/carbon
you will find an admin service named MessageProcessorAdminService

Sample pom.xml to compile / package your custom message processor :

<?xml version="1.0" encoding="UTF-8"?>
<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.sample</groupId>
    <artifactId>MyMessageProcessors</artifactId>
    <version>1.0.0</version>
    <packaging>bundle</packaging>
    <name>custom MessageProcessors</name>
    <description>sample</description>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>2.0</version>
                <configuration>
                    <source>1.7</source>
                    <target>1.7</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.felix</groupId>
                <artifactId>maven-bundle-plugin</artifactId>
                <version>2.3.4</version>
                <extensions>true</extensions>
                <configuration>
                    <instructions>
                        <Bundle-SymbolicName>com.sample.messageprocessor</Bundle-SymbolicName>
                        <Bundle-Name>com.sample.messageprocessor</Bundle-Name>
                        <Export-Package>com.sample.*</Export-Package>
                        <DynamicImport-Package>*</DynamicImport-Package>
                        <Implementation-Build>${buildNumber}</Implementation-Build>
                        <Scm-Root>${project.scm.connection}</Scm-Root>
                    </instructions>
                </configuration>
            </plugin>
        </plugins>
    </build>
    <repositories>
        <repository>
            <releases>
                <updatePolicy>daily</updatePolicy>
                <checksumPolicy>ignore</checksumPolicy>
            </releases>
            <id>wso2-nexus</id>
            <url>http://maven.wso2.org/nexus/content/groups/wso2-public/</url>
        </repository>
    </repositories>
    <pluginRepositories>
        <pluginRepository>
            <releases>
                <updatePolicy>daily</updatePolicy>
                <checksumPolicy>ignore</checksumPolicy>
            </releases>
            <id>wso2-nexus</id>
            <url>http://maven.wso2.org/nexus/content/groups/wso2-public/</url>
        </pluginRepository>
    </pluginRepositories>
    <dependencies>
        <dependency>
            <groupId>org.apache.synapse</groupId>
            <artifactId>synapse-core</artifactId>
            <version>2.1.7-wso2v3</version>
        </dependency>
        <dependency>
            <groupId>commons-logging</groupId>
            <artifactId>commons-logging</artifactId>
            <version>1.1.1</version>
        </dependency>
    </dependencies>
    <properties>
        <CApp.type>lib/synapse/mediator</CApp.type>
    </properties>
</project>

adapt groupId, artifactId, and version
adapt packages in maven-bundle-plugin config
execute mvn package from the command line

How can I set the message processor interval parameter using a non fixed value? (using a value stored in a property\registry\etc)

Tags:

java

esb

wso2

wso2-esb

wso2-message-broker

AndreaNobili

2 Answers

Jenananthan

Jean-Michel

Recent Activity

Donate For Us

How can I set the message processor interval parameter using a non fixed value? (using a value stored in a property\registry\etc)

Tags:

java

esb

wso2

wso2-esb

wso2-message-broker

AndreaNobili

2 Answers

Jenananthan

Jean-Michel

Related questions

Recent Activity

Donate For Us