Merge pull request #5 from anypointcloud/3-missing-parameter-error-in-json-logger-when-flowing-logs-to-amqp

3 missing parameter error in json logger when flowing logs to amqp

Author: mtransier

README.md

@@ -86,6 +86,7 @@ PART 2: https://blogs.mulesoft.com/dev/api-dev/json-logging-in-mule-4/
 * Replaced joda-time by JDK Date and Time API
 * Upgraded depdendencies
 * Removed unused dependencies
+* Fixed AMQP destination
 
 ### 2.2.1 version - Release notes
 

pom.xml

@@ -138,21 +138,7 @@
         <dependency>
             <groupId>com.mulesoft.muleesb.modules</groupId>
             <artifactId>anypoint-mq-rest-client</artifactId>
-            <version>4.0.13-1.fix-race-condition</version>
-            <exclusions>
-                <exclusion>
-                    <artifactId>async-http-client</artifactId>
-                    <groupId>com.ning</groupId>
-                </exclusion>
-                <exclusion>
-                    <artifactId>spring-core</artifactId>
-                    <groupId>org.springframework</groupId>
-                </exclusion>
-                <exclusion>
-                    <groupId>com.google.code.gson</groupId>
-                    <artifactId>gson</artifactId>
-                </exclusion>
-            </exclusions>
+            <version>4.0.12</version>
             <scope>compile</scope>
         </dependency>
         <dependency>

src/main/java/org/mule/extension/jsonlogger/internal/destinations/AMQPDestination.java

@@ -1,6 +1,5 @@
 package org.mule.extension.jsonlogger.internal.destinations;
 
-import com.mule.extensions.amqp.api.message.AmqpMessageBuilder;
 import com.mule.extensions.amqp.api.message.AmqpProperties;
 import org.mule.runtime.api.metadata.TypedValue;
 import org.mule.runtime.extension.api.annotation.param.NullSafe;
@@ -9,16 +8,14 @@
 import org.mule.runtime.extension.api.annotation.param.display.DisplayName;
 import org.mule.runtime.extension.api.annotation.param.display.Summary;
 import org.mule.runtime.extension.api.annotation.param.reference.ConfigReference;
-import org.mule.runtime.extension.api.client.DefaultOperationParameters;
 import org.mule.runtime.extension.api.client.ExtensionsClient;
-import org.mule.runtime.extension.api.client.OperationParameters;
+import org.mule.runtime.extension.api.client.OperationParameterizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.inject.Inject;
 import java.util.ArrayList;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
 
 import static org.mule.runtime.api.metadata.DataType.JSON_STRING;
 
@@ -72,13 +69,12 @@ public ArrayList<String> getSupportedCategories() {
     @Override
     public void sendToExternalDestination(String finalLog) {
         try {
-            OperationParameters parameters = DefaultOperationParameters.builder().configName(this.amqpConfigurationRef)
-                    .addParameter("exchangeName", this.exchangeDestination)
-                    .addParameter("messageBuilder", AmqpMessageBuilder.class, DefaultOperationParameters.builder()
-                            .addParameter("body", new TypedValue<>(finalLog, JSON_STRING))
-                            .addParameter("properties", new AmqpProperties()))
-                    .build();
-            extensionsClient.executeAsync("AMQP", "publish", parameters);
+        	Consumer<OperationParameterizer> parameters = operationParameterizer ->
+        		operationParameterizer.withConfigRef(this.amqpConfigurationRef)
+        			.withParameter("exchangeName", this.exchangeDestination)
+        			.withParameter("body", new TypedValue<>(finalLog, JSON_STRING))
+        			.withParameter("properties", new AmqpProperties());
+            extensionsClient.execute("AMQP", "publish", parameters);
         } catch (Exception e) {
             LOGGER.error("Error: " + e.getMessage());
             e.printStackTrace();