[Feature] Transaction counting for synapse gateway

components/apimgt/org.wso2.carbon.apimgt.gateway/src/main/java/org/wso2/carbon/apimgt/gateway/handlers/transaction/TransactionCountHandler.java

@@ -0,0 +1,171 @@
+package org.wso2.carbon.apimgt.gateway.handlers.transaction;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.synapse.AbstractExtendedSynapseHandler;
+import org.apache.synapse.MessageContext;
+import org.wso2.carbon.apimgt.gateway.handlers.transaction.consumer.TransactionRecordConsumer;
+import org.wso2.carbon.apimgt.gateway.handlers.transaction.exception.TransactionCounterConfigurationException;
+import org.wso2.carbon.apimgt.gateway.handlers.transaction.producer.TransactionRecordProducer;
+import org.wso2.carbon.apimgt.gateway.handlers.transaction.queue.TransactionRecordQueue;
+import org.wso2.carbon.apimgt.gateway.handlers.transaction.record.TransactionRecord;
+import org.wso2.carbon.apimgt.gateway.handlers.transaction.store.TransactionRecordStore;
+import org.wso2.carbon.apimgt.gateway.handlers.transaction.config.TransactionCounterConfig;
+
+import java.lang.reflect.Constructor;
+
+/**
+ * This class is the starting point of the transaction counter. This class is responsible for initializing all the
+ * components of the transaction counter and starting them. This class extends the Synapse handler interface and is
+ * responsible for intercepting the Synapse message flow and passing the message context to the transaction counting
+ * logic. This is registered in handlers.xml in case of APIM and service bus initializer in case of MI/ESB.
+ * @author - Isuru Wijesiri
+ * @version - 1.0.0
+ */
+public class TransactionCountHandler extends AbstractExtendedSynapseHandler {
+    private static final Log LOG = LogFactory.getLog(TransactionCountHandler.class);
+    private TransactionRecordQueue transactionRecordQueue;
+    private TransactionRecordProducer transactionRecordProducer;
+    private TransactionRecordConsumer transactionRecordConsumer;
+    private TransactionRecordStore transactionCountStore;
+    private static boolean enabled = false;
+
+    public TransactionCountHandler() {
+
+        // Initialize the config mechanism
+        try {
+            TransactionCounterConfig.init();
+        } catch (TransactionCounterConfigurationException e) {
+            LOG.error("Error while initializing Transaction Counter. Transaction counter will be disabled", e);
+            return;
+        }
+
+        try {
+            Class<?> clazz = Class.forName(
+                    TransactionCounterConfig.getTransactionCountStoreClass()
+            );
+            Constructor<?> constructor = clazz.getConstructor();
+            this.transactionCountStore = (TransactionRecordStore) constructor.newInstance();
+        } catch (Exception e) {
+            LOG.error("Error while initializing Transaction Counter. Transaction counter will be disabled", e);
+            return;
+        }
+
+        this.transactionRecordProducer = TransactionRecordProducer.getInstance();
+        this.transactionRecordConsumer = TransactionRecordConsumer.getInstance();
+        this.transactionRecordQueue = TransactionRecordQueue.getInstance();
+
+        TransactionRecord.init(
+                TransactionCounterConfig.getServerID(),
+                TransactionCounterConfig.getServerType().toString()
+        );
+
+        this.transactionRecordQueue.init(
+                TransactionCounterConfig.getTransactionRecordQueueSize()
+        );
+
+        this.transactionRecordProducer.init(
+                transactionRecordQueue,
+                TransactionCounterConfig.getProducerThreadPoolSize(),
+                TransactionCounterConfig.getMaxTransactionCount(),
+                TransactionCounterConfig.getMinTransactionCount(),
+                TransactionCounterConfig.getTransactionCountRecordInterval());
+
+        this.transactionCountStore.init(
+                TransactionCounterConfig.getTransactionCountService(),
+                TransactionCounterConfig.getTransactionCountServiceUsername(),
+                TransactionCounterConfig.getTransactionCountServicePassword()
+        );
+
+        this.transactionRecordConsumer.init(
+                transactionCountStore,
+                transactionRecordQueue,
+                TransactionCounterConfig.getConsumerCommitInterval(),
+                TransactionCounterConfig.getMaxRetryCount(),
+                TransactionCounterConfig.getMaxTransactionRecordsPerCommit());
+
+        enabled = true;
+    }
+
+    @Override
+    public boolean handleRequestInFlow(MessageContext messageContext) {
+        if(!enabled) {
+            return true;
+        }
+        int tCount = TransactionCountingLogic.handleRequestInFlow(messageContext);
+        if(tCount > 0) {
+            this.transactionRecordProducer.addTransaction(tCount);
+        }
+        return true;
+    }
+
+    @Override
+    public boolean handleRequestOutFlow(MessageContext messageContext) {
+        if(!enabled) {
+            return true;
+        }
+        int tCount = TransactionCountingLogic.handleRequestOutFlow(messageContext);
+        if(tCount > 0) {
+            this.transactionRecordProducer.addTransaction(tCount);
+        }
+        return true;
+    }
+
+    @Override
+    public boolean handleResponseInFlow(MessageContext messageContext) {
+        if(!enabled) {
+            return true;
+        }
+        int tCount = TransactionCountingLogic.handleResponseInFlow(messageContext);
+        if(tCount > 0) {
+            this.transactionRecordProducer.addTransaction(tCount);
+        }
+        return true;
+    }
+
+    @Override
+    public boolean handleResponseOutFlow(MessageContext messageContext) {
+        if(!enabled) {
+            return true;
+        }
+        int tCount = TransactionCountingLogic.handleResponseOutFlow(messageContext);
+        if(tCount > 0) {
+            this.transactionRecordProducer.addTransaction(tCount);
+        }
+        return true;
+    }
+
+    @Override
+    public boolean handleServerInit() {
+        // Nothing to implement
+        return true;
+    }
+
+    @Override
+    public boolean handleServerShutDown() {
+        // Clen up resources
+        transactionRecordProducer.shutdown();
+        transactionRecordConsumer.shutdown();
+        transactionRecordQueue.clenUp();
+        transactionCountStore.clenUp();
+        return true;
+    }
+
+    @Override
+    public boolean handleArtifactDeployment(String s, String s1, String s2) {
+        // Nothing to implement
+        return true;
+    }
+
+    @Override
+    public boolean handleArtifactUnDeployment(String s, String s1, String s2) {
+        // Nothing to implement
+        return true;
+    }
+
+    @Override
+    public boolean handleError(MessageContext messageContext) {
+        // Nothing to implement
+        return true;
+    }
+}

components/apimgt/org.wso2.carbon.apimgt.gateway/src/main/java/org/wso2/carbon/apimgt/gateway/handlers/transaction/TransactionCounterConstants.java

@@ -0,0 +1,54 @@
+package org.wso2.carbon.apimgt.gateway.handlers.transaction;
+
+public class TransactionCounterConstants {
+
+    public static enum ServerType {
+        GATEWAY, MI
+    }
+
+    public static final String IS_THERE_ASSOCIATED_INCOMING_REQUEST = "is_there_incoming_request";
+    public static final String TRANSPORT_WS = "ws";
+    public static final String TRANSPORT_WSS = "wss";
+
+    public static final String SERVER_ID = "serverId";
+    public static final String TRANSACTION_COUNT_STORE_CLASS = "transactionCountStoreClass";
+    public static final String TRANSACTION_RECORD_QUEUE_SIZE = "transactionRecordQueueSize";
+    public static final String PRODUCER_THREAD_POOL_SIZE = "producerThreadPoolSize";
+    public static final String TRANSACTION_COUNT_RECORD_INTERVAL = "transactionCountRecordInterval";
+    public static final String MAX_TRANSACTION_COUNT = "maxTransactionCount";
+    public static final String MIN_TRANSACTION_COUNT = "minTransactionCount";
+    public static final String CONSUMER_COMMIT_INTERVAL = "consumerCommitInterval";
+    public static final String MAX_TRANSACTION_RECORDS_PER_COMMIT = "maxTransactionRecordsPerCommit";
+    public static final String MAX_RETRY_COUNT = "maxRetryCount";
+    public static final String TRANSACTION_COUNT_SERVICE = "transactionCountService";
+    public static final String TRANSACTION_COUNT_SERVICE_USERNAME = "transactionCountServiceUsername";
+    public static final String TRANSACTION_COUNT_SERVICE_PASSWORD = "transactionCountServicePassword";
+
+    // APIM Gateway related constants
+    public static final String APIM_CONFIG_CLASS = "org.wso2.carbon.apimgt.impl.internal.ServiceReferenceHolder";
+    public static final String GATEWAY_CONFIG_ROOT = "APIGateway.TransactionCounter";
+    public static final String GATEWAY_PRODUCER_THREAD_POOL_SIZE = GATEWAY_CONFIG_ROOT +
+            ".ProducerThreadPoolSize";
+    public static final String GATEWAY_QUEUE_SIZE = GATEWAY_CONFIG_ROOT + ".QueueSize";
+    public static final String GATEWAY_STORE_CLASS = GATEWAY_CONFIG_ROOT + ".StoreClass";
+    public static final String GATEWAY_MAX_TRANSACTION_COUNT = GATEWAY_CONFIG_ROOT +
+            ".MaxTransactionCount";
+    public static final String GATEWAY_MIN_TRANSACTION_COUNT = GATEWAY_CONFIG_ROOT +
+            ".MinTransactionCount";
+    public static final String GATEWAY_RECORD_INTERVAL = GATEWAY_CONFIG_ROOT
+            + ".ProducerScheduledInterval";
+    public static final String GATEWAY_MAX_RETRY_COUNT = GATEWAY_CONFIG_ROOT + ".MaxRetryCount";
+    public static final String GATEWAY_MAX_TRANSACTION_RECORDS_PER_COMMIT = GATEWAY_CONFIG_ROOT
+            + ".MaxBatchSize";
+    public static final String GATEWAY_CONSUMER_COMMIT_INTERVAL = GATEWAY_CONFIG_ROOT
+            + ".PublisherScheduledInterval";
+    public static final String GATEWAY_SERVER_ID = GATEWAY_CONFIG_ROOT + ".ServerID";
+    public static final String GATEWAY_SERVICE = GATEWAY_CONFIG_ROOT + ".ServiceURL";
+    public static final String GATEWAY_SERVICE_USERNAME = GATEWAY_CONFIG_ROOT
+            + ".ServiceUsername";
+    public static final String GATEWAY_SERVICE_PASSWORD = GATEWAY_CONFIG_ROOT
+            + ".ServicePassword";
+
+    // MI related constants
+    public static final String MI_CONFIG_CLASS = "org.wso2.config.mapper.ConfigParser";
+}

components/apimgt/org.wso2.carbon.apimgt.gateway/src/main/java/org/wso2/carbon/apimgt/gateway/handlers/transaction/TransactionCountingLogic.java

@@ -0,0 +1,56 @@
+package org.wso2.carbon.apimgt.gateway.handlers.transaction;
+
+import org.apache.synapse.MessageContext;
+import org.apache.synapse.core.axis2.Axis2MessageContext;
+
+
+/**
+ * This class contains the logic for counting transactions.
+ * In case of changing the logic, this class should be modified, replaced or extended.
+ * @author - Isuru Wijesiri
+ * @version - 1.0.0
+ */
+public class TransactionCountingLogic {
+
+    public static int handleRequestInFlow(MessageContext messageContext) {
+        org.apache.axis2.context.MessageContext axis2MessageContext =
+                ((Axis2MessageContext) messageContext).getAxis2MessageContext();
+
+        // Setting this property to identify request-response pairs
+        messageContext.setProperty(TransactionCounterConstants.IS_THERE_ASSOCIATED_INCOMING_REQUEST, true);
+
+        // Counting message received via an open WebSocket
+        String transport = axis2MessageContext.getIncomingTransportName();
+        if (transport.equals(TransactionCounterConstants.TRANSPORT_WS) ||
+                transport.equals(TransactionCounterConstants.TRANSPORT_WSS)){
+            return 1;
+        }
+        return 0;
+    }
+
+    public static int handleRequestOutFlow(MessageContext messageContext) {
+        Object isThereAnAssociatedIncomingRequest = messageContext.getProperty(
+                TransactionCounterConstants.IS_THERE_ASSOCIATED_INCOMING_REQUEST);
+
+        // Counting outgoing messages that are not related to any request-response pair
+        if (isThereAnAssociatedIncomingRequest == null) {
+            return 1;
+        }
+        return 0;
+    }
+
+    public static int handleResponseInFlow(MessageContext messageContext) {
+        return 0;
+    }
+
+    public static int handleResponseOutFlow(MessageContext messageContext) {
+        Object isThereAnAssociatedIncomingRequest = messageContext.getProperty(
+                TransactionCounterConstants.IS_THERE_ASSOCIATED_INCOMING_REQUEST);
+
+        // Counting request-response pairs
+        if (isThereAnAssociatedIncomingRequest instanceof Boolean) {
+            return 1;
+        }
+        return 0;
+    }
+}