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?

Answer 1

It doesn't look like that this is not possible out of the box with WSO2ESB. But you can try to override

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

and write custom logic to pull parameter value for interval from a registry or any other source instead of using the XML Configuration and use that overridden implementation in your XML configuration

<?xml version="1.0" encoding="UTF-8"?>
<messageProcessor class="my.custom.CustomSamplingProcessor" messageStore="transferFromMessageStore" name="transferFromMessageProcessor" xmlns="http://ws.apache.org/ns/synapse">
    <parameter name="sequence">transferProcessorSequence</parameter>
    <parameter name="is.active">true</parameter>
    <parameter name="concurrency">1</parameter>
</messageProcessor>

Answer 2

Well if you are using Spring you can use an application.properties file in your application classpath and use the PropertyPlaceholderConfigurer to load resources from this external shared properties file. This is how you should declare the PropertyPlaceholderConfigurer in your XML file:

<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
    <property name="locations">
        <value>classpath:wso2esb.properties</value>
    </property>
</bean>

<bean id="messageProcessor" 
   class="org.apache.synapse.message.processor.impl.sampler.SamplingProcessor" messageStore="transferFromMessageStore" name="transferFromMessageProcessor" xmlns="http://ws.apache.org/ns/synapse">
    <property name="sequence" value="${wso2esb.sequence}"/>
    <property name="interval" value="${wso2esb.interval}"/>
    <property name="isactive" value="${wso2esb.isactive}"/>
    <property name="concurrency" value="${wso2esb.concurrency}"/>
</bean>

And you will just need to update the properties and parametrs of your WSO2 ESB in this properties file.

For further details you can check:

• Spring PropertyPlaceholderConfigurer documentation.

• This answer on how lo load external .properties in XML configuration.

Answer 3

I believe the message processor itself does not provide such a feature. However, it is possible to do this another way. I am proposing a way without any custom Java code (which may or not be an advantage to you).

Let's assume you want to do this every X seconds (with X stored in the registry). You would then leave the message processor interval at 1 second. Then, in the sequence that your processor injects its message to, start by looking at a counter that you store as a global property. The counter keeps track of how many times you skipped the action. If the skipped counter is smaller than X - 1, you increase the counter and do nothing. If the skipped counter equals X - 1 (or is higher, which can happen if you change X in the registry), you reset the counter to zero and perform the action.

Answer 4

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

Answer 5

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

1. Built the custom message processor, and add the jar to <EI_HOME>\lib (since this jar is an none osgi jar)
2. Restart/Start the EI
3. Add the property file to registry
4. 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