#include<stdio.h>

#ifdef __unix__
#include <execinfo.h>
#include <unistd.h>
#endif

#include "apr.h"
#include "apr_strings.h"
#include "apr_pools.h"
#include "apr_poll.h"
#include "apr_thread_proc.h"
#include "apr_file_io.h"
#include "apr_network_io.h"
#include "apr_signal.h"
#include "apr_portable.h"
#include "apr_thread_cond.h"

#include "../nd_agent.h"
#include "../Server.h"
#include "../ndlb_mt_log.h"
#include "../ndlb_thread_handle.h"
#include "../ndlb_utiles.h"
#include "../NDParseCilFile.h"
#include "../ndlb_alloc.h"
#include "nd_proxy_server.h"
#include "../agent_thread.h"
#include "../ndlb_encode.h"
#include "../ndGlobalMacro.h"
#include "../parson.h"
#include "rudp.h"
//#include "../ndlb_log.h"

/* Proxy Multi Application support */

/////////////////////////////////////////////////////////
//         GLOBAL VARIABLES
//-------------------------------------------------------
NDProxy_t NDProxy;
//NDProxyInstance is used for TCP. NDProxyUdpInstance
NDProxyInstance_t **NDProxyInstance = NULL;
NDProxyInstance_t **NDProxyUdpInstance = NULL;
NDProxyInstance_t **NDProxyTcpUdpInstance = NULL;
apr_pool_t *pool;
controlSocketBuffer_t dataBufInfo;
controlSocketBuffer_t bufInfo;

#define WRAP_HEADER_LEN   sizeof(wrapheadervar_t)
#define UDP_PACKET_MIN_SIZE WRAP_HEADER_LEN + 5 //5 is considered for tier/server/instance
#define MSG_HEADER_LEN      sizeof(msgHdr_t)
#define MAX_TRANSACTION_RESET_LIMIT 1000
#define PARSE_UDP_HEADER(str, len, remainBytes, ptr) \
{ \
  if(remainBytes < len) \
      return 0; \
  NDLB_MALLOC_STRING(str, len + 1, "Allocating memory for str", -1);  \
  NDLB_STRNCPY(str, ptr, remainBytes,len+1);  \
  ptr += len; \
  remainBytes -= len; \
}

extern int closeCtrlThread;

char error_buf[ERROR_BUFFER_SIZE] = "";
extern int arrPropagationHeaders[HEADER_B3+1];
#define UDP_TIMEOUT 900
extern char dumpAndSendDataToNDC(char* byteArray, int len, char flushData);
extern int stop_proxy();
extern void SendHavocResponse(NDMessage_t* NDMessage);
extern void dumpLogMessage(NDPerThreadData_t* tlsData, char *logmsg);
extern int send_discovery_data(char* buffer, int buf_size, char* error_buf, int error_buf_len);
extern int send_discovery_data_to_ndc(char* buffer, int buf_size, char* error_buf, int error_buf_len);
extern void create_discovery_conn_with_ndc(struct NDProxyInstance_t* tlndApplication, char* hostname, int port, char* error_buf, int error_buf_len, struct apr_pool_t* pool);
extern void tag_record();
extern void attr_record();
extern void sendThreadDataToNDC(char *logerror, char *ndcmsg, char *clientMsg, int len);
//__thread NDProxyInstance_t *tlndApplication;
#ifdef __cplusplus
  extern "C" int parseRecdMessage(char* msg, NDMessage_t* NDMessage);
#else
  extern int parseRecdMessage(char* msg, NDMessage_t* NDMessage);
#endif

#ifdef __unix__
  __thread unsigned short tlWorkerThreadId;
#else
  __declspec(thread) unsigned short tlWorkerThreadId;
#endif

struct kwLocalStruct_t
{
  // Local variables to  store the values
  char appName[1024], ndAppServerHost[1024], tierName[1024], bciVersion[1024], ndHome[1024], hostName[1024];
  char javaHome[1024], connDestination[1024], agentName[1024];
  char agentType[64], NDCollectorIP[32], serverIp[32], jvmType[64], javaVersion[32], machineType[32], tierAttr[1024],tags[1024];
  long long int bciStartTime, pid, BCITimeStamp, lastConnectionTimeStamp, connDuration;
  int requestType, appID, ndAppServerID, tierID, NDCollectorPort, testIdx;
  char isAgentLess;
};


//Data structures used for resetting transactions

#define NUM_THREAD_OPERATIONS  10

//type of operation
enum ThreadOperationType{
  THREAD_OPERATION_FP_RESET=0
};

// required info to reset a transaction
struct threadOps {
  char operation;
  char spare[3];
  NDProxyInstance_t * appInstance;
  long long flowpathInstance;
};


struct tcp_thread_storage {
  struct perThreadOps perThreadOpsVar;
  unsigned short threadIndex;
};


struct udp_thread_storage *udpStorage = NULL;
struct tcp_thread_storage tcpStorage;
/////////////////////////////////////////////////////////


// void resetKwLocalStruct(struct kwLocalStruct_t *st)
// {
//   NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
//   memset(st,0,sizeof(struct kwLocalStruct_t));
//   NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
// }

void freeclientdata(NDProxyInstance_t* nProxyInstance) {
  if(nProxyInstance->agentName)
    { 
      FREE_AND_MAKE_NULL(nProxyInstance->agentName, "Free_AgentName", -1);
    }
    if(nProxyInstance->tierName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->tierName, "Free_TierName", -1);
    }
    if(nProxyInstance->appServerName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->appServerName, "Free_AppServer", -1);
    }
    if(nProxyInstance->appName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->appName, "Free_AgentName", -1);
    }
    // if(nProxyInstance->bciVersion)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->bciVersion, "Free_BciVersion", -1);
    // }
    // if(nProxyInstance->ndHome)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->ndHome, "Free_NdHome", -1);
    // }
     if(nProxyInstance->agentType)
     {
       FREE_AND_MAKE_NULL(nProxyInstance->agentType, "Free_AgentType", -1);
     }
    // if(nProxyInstance->serverIp)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->serverIp, "Free_ServerIp", -1);
    // }
    // if(nProxyInstance->hostName)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->hostName, "Free_HostName", -1);
    // }
    // if(nProxyInstance->jvmType)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->jvmType, "Free_JvmType", -1);
    // }
    // if(nProxyInstance->javaVersion)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->javaVersion, "Free_JavaVersion", -1);
    // }
    // if(nProxyInstance->javaHome)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->javaHome, "Free_JavaHome", -1);
    // }
    // if(nProxyInstance->machineType)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->machineType, "Free_MachineType", -1);
    // }
    // if(nProxyInstance->connDestination)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->connDestination, "Free_ConnDestination", -1);
    // }
    // if(nProxyInstance->NDCollectorIP)
    // {
    //   FREE_AND_MAKE_NULL(nProxyInstance->NDCollectorIP, "Free_NDCollectorIP", -1);
    // }
    if(ndApplication->start_instr_lock)
      apr_thread_mutex_lock(ndApplication->start_instr_lock);
    if(nProxyInstance->startInstMsg)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->startInstMsg, "Free_StartInstrMsg", -1);
    }
    if(ndApplication->start_instr_lock)
      apr_thread_mutex_unlock(ndApplication->start_instr_lock);
    if(nProxyInstance->NDBCISetting.appName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.appName, "Free_AppName", -1);
    }
    if(nProxyInstance->NDBCISetting.encodedappName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.encodedappName, "Free_encodedAppName", -1);
    }
    if(nProxyInstance->NDBCISetting.tierName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.tierName, "Free_tierName", -1);
    }
    if(nProxyInstance->NDBCISetting.encodedtierName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.encodedtierName, "Free_encodedtierName", -1);
    }
    if(nProxyInstance->NDBCISetting.appServerName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.appServerName, "Free_appServerName", -1);
    }
    if(nProxyInstance->NDBCISetting.encodedappServerName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.encodedappServerName, "Free_encodedappServerName", -1);
    }
    if(nProxyInstance->funcName)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->funcName, "Free_funcName", -1);
    }
    if(nProxyInstance->lambdaKey)
    {
      FREE_AND_MAKE_NULL(nProxyInstance->lambdaKey, "Free_funcName", -1);
    }
}
// Function  to clean proxy memory
void proxyMemClean()
{
  int i=0;
  NDTL4(trace_log_key, NULL, TL_INFO, "Method entry");
  for(i = 0; i < NDProxy.maxTCPProcesses; i++)
  {
    if(NDProxyInstance[i])
      freeclientdata(NDProxyInstance[i]);
  }
  NDTL0(trace_log_key, NULL, TL_INFO, "Free'd NDProxy Instance structure successfully.");
  NDTL4(trace_log_key, NULL, TL_INFO, "Method exit");
}

// Function to  close all the sockets at exit
void resetAndCleanProxy(int sig)
{
  int i, j;
  NDProxy.shutdownProxy = 1;
  NDTL4(trace_log_key, NULL, TL_INFO, "Method entry");
  closeCtrlThread = 1;
  NDTL0(trace_log_key, NULL, TL_INFO, "Received (kill -%d) request to stop Proxy. Closing the connections", sig);

  if(udpStorage) {
    for(int i = 0; i < NDProxy.num_udpThreads; i++) {
      if(udpStorage[i].thread_cond_var)
        apr_thread_cond_signal(udpStorage[i].thread_cond_var);
    }
  }
  apr_socket_t* tmp = NDProxy.udp_skt;
  NDProxy.udp_skt = NULL;
  if(tmp)
    apr_socket_close(tmp);
  for(i=0; i < NDProxy.maxTCPProcesses; i++) {
    if(!NDProxyInstance[i])
      continue;
    ndApplication = NDProxyInstance[i];
    toStopASThread = ND_TRUE;
    NDProxyInstance[i]->NDSys.closeCtrlConFlag = 1;
    apr_sleep(200);
  }

  for(i=0; i < NDProxy.maxTCPProcesses; i++)
  {
    if(!NDProxyInstance[i])
      continue;
    for (j = 0; j < NDProxyInstance[i]->processCount; j++)
    {
      if(NDProxyInstance[i]->processInfo[j].ctrlSocket) apr_socket_close(NDProxyInstance[i]->processInfo[j].ctrlSocket);
      NDProxyInstance[i]->processInfo[j].ctrlSocket = NULL;
      if(NDProxyInstance[i]->processInfo[j].dcSocket)  apr_socket_close(NDProxyInstance[i]->processInfo[j].dcSocket);
      NDProxyInstance[i]->processInfo[j].dcSocket = NULL;
      if(NDProxyInstance[i]->processInfo[j].asSocket) apr_socket_close(NDProxyInstance[i]->processInfo[j].asSocket);
      NDProxyInstance[i]->processInfo[j].asSocket = NULL;
    }
    if(NDProxyInstance[i]->control.skt) apr_socket_close(NDProxyInstance[i]->control.skt);
    NDProxyInstance[i]->control.skt = NULL;
    if(NDProxyInstance[i]->dataSkt.skt) apr_socket_close(NDProxyInstance[i]->dataSkt.skt);
    NDProxyInstance[i]->dataSkt.skt = NULL;
    if(NDProxyInstance[i]->autoSensor.skt) apr_socket_close(NDProxyInstance[i]->autoSensor.skt);
    NDProxyInstance[i]->autoSensor.skt = NULL;
  }
  stop_proxy();

  NDTL0(trace_log_key, NULL, TL_INFO, "All connections closed successfully.");
  proxyMemClean();

  if(i > 0) {
    apr_sleep(1000);
  }

  NDTL0(trace_log_key, NULL, TL_INFO, "Exiting Proxy. GoodBye!!!");
  exit(0);
  NDTL4(trace_log_key, NULL, TL_INFO, "Method exit");
}

void freeUDPConnections(apr_hash_t* ht) {
  apr_hash_index_t *idx = NULL;
  apr_port_t *key;
  apr_sockaddr_t *val = NULL;
  for(idx = apr_hash_first(NDProxy.proxy_pool,(apr_hash_t*) ht); idx; idx = apr_hash_next(idx)){
      apr_hash_this(idx, (const void **)&key, NULL, (void **)&val);
      if(!key)
        continue;
      apr_hash_set(ht, key, sizeof(apr_port_t), NULL);
      FREE_AND_MAKE_NULL(val, "freeing val", -1);
  }
}
void closeUDPThread(NDProxyInstance_t  *ndProxyInstanceIndex) {
    ndProxyInstanceIndex->close_udp_thread = 0;

  if(ndProxyInstanceIndex->control.skt) {
    apr_socket_close(ndProxyInstanceIndex->control.skt);
    ndProxyInstanceIndex->control.skt = NULL;
  }
  ndProxyInstanceIndex->NDSys.closeCtrlConFlag = 1;
  if(ndProxyInstanceIndex->dataSkt.skt) {
    apr_socket_close(ndProxyInstanceIndex->dataSkt.skt);
    ndProxyInstanceIndex->dataSkt.skt = NULL;
  }
  if(ndProxyInstanceIndex->autoSensor.skt) {
    apr_socket_close(ndProxyInstanceIndex->autoSensor.skt);
    ndProxyInstanceIndex->autoSensor.skt = NULL;
  }
  freeclientdata(ndProxyInstanceIndex);
  if(ndProxyInstanceIndex->fpi_tls_map)
    apr_hash_clear(ndProxyInstanceIndex->fpi_tls_map);
  if(ndProxyInstanceIndex->udp_connections) {
    freeUDPConnections(ndProxyInstanceIndex->udp_connections);
    apr_hash_clear(ndProxyInstanceIndex->udp_connections);
  }
}

/******************************************************
 * This function will set the socket buffer struct
 * only the first time and will then reset it at 
 * every call.
 * ***************************************************/

void setResetSocketBuffer(controlSocketBuffer_t *bufInfoLocal)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  bufInfoLocal->readOffset = 0;
  bufInfoLocal->writeOffset = 0;
  bufInfoLocal->readBytes = 0;
  bufInfoLocal->read_len = MAX_READ_LENGTH;
  if(!bufInfoLocal->read_socket_buf)
  {
    NDLB_MALLOC_STRING(bufInfoLocal->read_socket_buf, MAX_READ_LENGTH + 1, "control buffer read_socket_buf", -1);
  }
  bufInfoLocal->read_socket_buf[0]='\0';
    //memset(bufInfoLocal->read_socket_buf, 0, MAX_READ_LENGTH + 1);
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}


static void populateNDBCISetting(NDProxyInstance_t* nProxyInstance)
{

  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");

  if(nProxyInstance->NDBCISetting.tierName )
    FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.tierName, "NDBCISetting Tier Name", -1);

  if(nProxyInstance->NDBCISetting.appServerName)
    FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.appServerName, "NDBCISetting appServer Name", -1);

  if(nProxyInstance->NDBCISetting.appName)
    FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.appName, "NDBCISetting app Name", -1);

  if(nProxyInstance->NDBCISetting.encodedtierName )
    FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.encodedtierName, "NDBCISetting encoded Tier Name", -1);

  if(nProxyInstance->NDBCISetting.encodedappServerName)
    FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.encodedappServerName, "NDBCISetting encoded appServer Name", -1);

  if(nProxyInstance->NDBCISetting.encodedappName)
    FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.encodedappName, "NDBCISetting encoded app Name", -1);

  if (nProxyInstance->NDBCISetting.tierAttr)
    FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.tierAttr, "NDBCISetting Tier Name", -1);


  NDLB_MALLOC_AND_COPY(nProxyInstance->tierName,
      nProxyInstance->NDBCISetting.tierName,
      strlen(nProxyInstance->tierName)+1, "copy tierName", -1);

  if(nProxyInstance->NDBCISetting.tierName) {
   int size = strlen(nProxyInstance->NDBCISetting.tierName) *3;
   NDLB_MALLOC_STRING(nProxyInstance->NDBCISetting.encodedtierName, size + 1, "", -1);
   escapeHTML(nProxyInstance->NDBCISetting.encodedtierName, size, nProxyInstance->NDBCISetting.tierName, 1);
  }

  FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.appServerName,"NDProxy.ndHomePath", -1 );
  NDLB_MALLOC_AND_COPY(nProxyInstance->appServerName,
      nProxyInstance->NDBCISetting.appServerName,
      strlen(nProxyInstance->appServerName)+1, "copy appServerName ", -1);

  if(nProxyInstance->NDBCISetting.appServerName) {
   int size = strlen(nProxyInstance->NDBCISetting.appServerName) *3;
   NDLB_MALLOC_STRING(nProxyInstance->NDBCISetting.encodedappServerName, size + 1, "", -1);
   escapeHTML(nProxyInstance->NDBCISetting.encodedappServerName, size, nProxyInstance->NDBCISetting.appServerName, 1);
  }
FREE_AND_MAKE_NULL(nProxyInstance->NDBCISetting.appName,"NDProxy.ndHomePath", -1 );
  NDLB_MALLOC_AND_COPY(nProxyInstance->appName,
      nProxyInstance->NDBCISetting.appName,
      strlen(nProxyInstance->appName)+1, " copy appName", -1);

  if(nProxyInstance->NDBCISetting.appName) {
   int size = strlen(nProxyInstance->NDBCISetting.appName) *3;
   NDLB_MALLOC_STRING(nProxyInstance->NDBCISetting.encodedappName, size + 1, "", -1);
   escapeHTML(nProxyInstance->NDBCISetting.encodedappName, size, nProxyInstance->NDBCISetting.appName, 1);
  }

  if(nProxyInstance->tierAttr && strlen(nProxyInstance->tierAttr) > 0) {
    NDLB_MALLOC_AND_COPY(nProxyInstance->tierAttr,
      nProxyInstance->NDBCISetting.tierAttr,
      strlen(nProxyInstance->tierAttr) + 1, " copy appName", -1);
  }
  if(nProxyInstance->tags && strlen(nProxyInstance->tags)>0) {
  NDTL1(trace_log_key, NULL, TL_INFO, "[200~strlen(nProxyInstance->tags)=%d",strlen(nProxyInstance->tags));
    NDLB_MALLOC_AND_COPY(nProxyInstance->tags,
      nProxyInstance->NDBCISetting.json_tag,
      strlen(nProxyInstance->tags) + 1, " copy tags", -1);
  }
}


void setNDProxyInstance(NDProxyInstance_t *instance)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  //instance->isAgentLess = 0;
  instance->control_thread_id = NULL;
  NDLB_MALLOC_STRING(instance->agentName, 1024, "Allocating Memory to store agentName", -1);
  NDLB_MALLOC_STRING(instance->tierName, 1024, "Allocating Memory to store tierName", -1);
  NDLB_MALLOC_STRING(instance->appServerName, 1024, "Allocating Memory to store appServerName", -1);
  NDLB_MALLOC_STRING(instance->appName, 1024, "Allocating Memory to store appName", -1);
  //NDLB_MALLOC_STRING(instance->bciVersion, 1024, "Allocating Memory to store bciVersion", -1);
  //NDLB_MALLOC_STRING(instance->ndHome, 1024, "Allocating Memory to store ndHome", -1);
  //NDLB_MALLOC_STRING(instance->hostName, 1024, "Allocating Memory to store hostName", -1);
  //NDLB_MALLOC_STRING(instance->javaHome, 1024, "Allocating Memory to store javaHome", -1);
  //NDLB_MALLOC_STRING(instance->connDestination, 1024, "Allocating Memory to store conDestination", -1);
  NDLB_MALLOC_STRING(instance->agentType, 64, "Allocating Memory to store agentType", -1);
  //NDLB_MALLOC_STRING(instance->serverIp, 32, "Allocating Memory to store serverIp", -1);
  //NDLB_MALLOC_STRING(instance->jvmType, 64, "Allocating Memory to store jvmType", -1);
  //NDLB_MALLOC_STRING(instance->javaVersion, 32, "Allocating Memory to store javaVersion", -1);
  //NDLB_MALLOC_STRING(instance->machineType, 32, "Allocating Memory to store machineType", -1);
  //NDLB_MALLOC_STRING(instance->NDCollectorIP, 32, "Allocating Memory to store NDCollectorIP", -1);
  NDLB_MALLOC_STRING(instance->tierAttr, 1024, "Allocating Memory to store agentName", -1);
  NDLB_MALLOC_STRING(instance->tags, 1024, "Allocating Memory to store tags", -1);

  NDLB_MALLOC_STRING(instance->funcName, 1024, "Allocating Memory to store funcName", -1);
  if(!instance->fpi_tls_map)
    instance->fpi_tls_map = create_hashmap(NDProxy.proxy_pool);
  if(!instance->udp_connections)
    instance->udp_connections = create_hashmap(NDProxy.proxy_pool);
  if(!instance->udp_connections)
      NDTL0(trace_log_key, NULL, TL_ERROR, "Error in creating map for udp connections...");
  instance->tierID = 0;
  instance->serverID = 0;
  instance->appID = 0;
  //instance->NDCollectorPort = 0;
  //instance->requestType = 0;
  instance->processCount = 0;			// Total process request for a particular instance. USed asindex for and instance.
  instance->activeProcessCount = 0;		// Total number of connected process(active) for a particular instance

  //instance->lastConnectionTimeStamp = 0;
  //instance->connDuration = 0;
  //instance->bciStartTime = 0;
  //instance->BCITimeStamp = 0;


  //////////////////////////
  //instance->heartbeat_count = 0;
  //instance-> resourceId = 1;




  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}

/*****************************************************************
 * This function populates the NDProxyInstance from the values
 * received on the control connection.
 * **************************************************************/

void populateNDProxyInstance(NDProxyInstance_t* nProxyInstance,struct kwLocalStruct_t *kwLocalStruct, int udp)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  NDLB_STRNCPY(nProxyInstance->appName, kwLocalStruct->appName, strlen(kwLocalStruct->appName),1024);
  NDLB_STRNCPY(nProxyInstance->agentName, kwLocalStruct->agentName, strlen(kwLocalStruct->agentName),1024);
  NDLB_STRNCPY(nProxyInstance->tierName, kwLocalStruct->tierName, strlen(kwLocalStruct->tierName),1024);
  NDLB_STRNCPY(nProxyInstance->appServerName, kwLocalStruct->ndAppServerHost, strlen(kwLocalStruct->ndAppServerHost),1024);
  //NDLB_STRNCPY(nProxyInstance->bciVersion, kwLocalStruct->bciVersion, strlen(kwLocalStruct->bciVersion),1024);
  //NDLB_STRNCPY(nProxyInstance->ndHome, kwLocalStruct->ndHome, strlen(kwLocalStruct->ndHome),1024);
  NDLB_STRNCPY(nProxyInstance->agentType, kwLocalStruct->agentType, strlen(kwLocalStruct->agentType),64);
  //NDLB_STRNCPY(nProxyInstance->serverIp, kwLocalStruct->serverIp, strlen(kwLocalStruct->serverIp),32);
  //NDLB_STRNCPY(nProxyInstance->hostName, kwLocalStruct->hostName, strlen(kwLocalStruct->hostName),1024);
  //NDLB_STRNCPY(nProxyInstance->jvmType, kwLocalStruct->jvmType, strlen(kwLocalStruct->jvmType),64);
  //NDLB_STRNCPY(nProxyInstance->javaVersion, kwLocalStruct->javaVersion, strlen(kwLocalStruct->javaVersion),32);
  //NDLB_STRNCPY(nProxyInstance->javaHome, kwLocalStruct->javaHome, strlen(kwLocalStruct->javaHome),1024);
  //NDLB_STRNCPY(nProxyInstance->machineType, kwLocalStruct->machineType, strlen(kwLocalStruct->machineType),32);
  //NDLB_STRNCPY(nProxyInstance->connDestination, kwLocalStruct->connDestination, strlen(kwLocalStruct->connDestination),1024);
  //strncpy(nProxyInstance->NDCollectorIP, kwLocalStruct->NDCollectorIP, strlen(kwLocalStruct->NDCollectorIP));
  //nProxyInstance->lastConnectionTimeStamp = kwLocalStruct->lastConnectionTimeStamp;
  //nProxyInstance->connDuration = kwLocalStruct->connDuration;
  //nProxyInstance->bciStartTime = kwLocalStruct->bciStartTime;
  //nProxyInstance->BCITimeStamp = kwLocalStruct->BCITimeStamp;
  //nProxyInstance->requestType = kwLocalStruct->requestType;
  //nProxyInstance->isAgentLess = kwLocalStruct->isAgentLess;
  nProxyInstance->totalElapsedTime = my_gettimeofday();
  if(strlen(kwLocalStruct->tierAttr)>0)  
    NDLB_STRNCPY(nProxyInstance->tierAttr, kwLocalStruct->tierAttr, strlen(kwLocalStruct->tierAttr),1024);
  
  if(strlen(kwLocalStruct->tags)>0) {
    NDLB_STRNCPY(nProxyInstance->tags, kwLocalStruct->tags, strlen(kwLocalStruct->tags),1024);
  }
  if(udp) {
    NDLB_MALLOC_AND_MEMSET(nProxyInstance->processInfo,
  sizeof(instanceProcess_t) ,
  "Allocating memory to store the nProxyInstance->processInfo", -1);

    NDTL0(trace_log_key, NULL, TL_INFO, "Memory allocated for nProxyInstance->processInfo. Available memory = '1' block.");
  }
  else {
    // Allocating memory for process and assigning values to it
    if((nProxyInstance->processCount % 500) == 490)
    {
      instanceProcess_t *tmp_ptr = NULL, *tmp_ptr2 = nProxyInstance->processInfo;
      NDLB_REALLOC_AND_MEMSET(tmp_ptr, sizeof(instanceProcess_t) * (nProxyInstance->processCount + 10 + 500),
      sizeof(instanceProcess_t) * (nProxyInstance->processCount + 10),
      "Re-Allocating memory to store the nProxyInstance->processInfo", -1);
      memcpy(tmp_ptr, nProxyInstance->processInfo, sizeof(instanceProcess_t) * (nProxyInstance->processCount + 10));
      nProxyInstance->processInfo = tmp_ptr;

      FREE_AND_MAKE_NULL(tmp_ptr2, "tmp_ptr", -1);
      //    free(tmp_ptr2);
      //    tmp_ptr2 = NULL;
      tmp_ptr = NULL;
      NDTL0(trace_log_key, NULL, TL_INFO, "Memory reallocated for nProxyInstance->processInfo. Available memory = '%d' blocks.",
      nProxyInstance->processCount + 10 + 500);
    }
    else
    {
      // Allocate memory for 500 process in the beginning (Suggested by Sandeep sir)
      NDLB_MALLOC_AND_MEMSET(nProxyInstance->processInfo,
      sizeof(instanceProcess_t) * (nProxyInstance->processCount + 500),
      "Allocating memory to store the nProxyInstance->processInfo", -1);

      NDTL0(trace_log_key, NULL, TL_INFO, "Memory allocated for nProxyInstance->processInfo. Available memory = '500' blocks.");
    }
  }
  nProxyInstance->processInfo[0].processId = kwLocalStruct->pid;

  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}

void parse_tag_values(char *buffer, struct kwLocalStruct_t *kwLocalStruct) {
  char *fields[128];
  char *kw, *val, *ptr;
  int numFields = 0;
  numFields = ndlb_get_tokens_with_multi_delimiter(buffer, fields, ";", 128);

  for(int i = 0; i < numFields; i++)
  {
    ptr = fields[i];
    kw = apr_strtok(ptr, "=", &val);

    if(!strcmp(kw, "appName")) strcpy(kwLocalStruct->appName, val);
    else if(!strcmp(kw, "ndAppServerHost")) strcpy(kwLocalStruct->ndAppServerHost, val);
    else if(!strcmp(kw, "tierName")) strcpy(kwLocalStruct->tierName, val);
    else if(!strcmp(kw, "bciVersion")) strcpy(kwLocalStruct->bciVersion, val);
    else if(!strcmp(kw, "bciStartTime")) kwLocalStruct->bciStartTime = strtoll(val, NULL, 10);
    else if(!strcmp(kw, "ndHome")) strcpy(kwLocalStruct->ndHome, val);
    else if(!strcmp(kw, "pid")) kwLocalStruct->pid = strtoll(val, NULL, 10);
    else if(!strcmp(kw, "BCITimeStamp")) kwLocalStruct->BCITimeStamp = strtoll(val, NULL, 10);
    else if(!strcmp(kw, "agentType")) strcpy(kwLocalStruct->agentType, val);
    else if(!strcmp(kw, "serverIp")) strcpy(kwLocalStruct->serverIp, val);
    else if(!strcmp(kw, "hostName")) strcpy(kwLocalStruct->hostName, val);
    else if(!strcmp(kw, "isAgentLess")) kwLocalStruct->isAgentLess = val[0];
    else if(!strcmp(kw, "jvmType")) strcpy(kwLocalStruct->jvmType, val);
    else if(!strcmp(kw, "javaVersion")) strcpy(kwLocalStruct->javaVersion, val);
    else if(!strcmp(kw, "javaHome")) strcpy(kwLocalStruct->javaHome, val);
    else if(!strcmp(kw, "machineType")) strcpy(kwLocalStruct->machineType, val);
    else if(!strcmp(kw, "requestType")) kwLocalStruct->requestType = atoi(val);
    else if(!strcmp(kw, "lastConnectionTimeStamp")) kwLocalStruct->lastConnectionTimeStamp = strtoll(val, NULL, 10);
    else if(!strcmp(kw, "connDuration")) kwLocalStruct->connDuration = strtoll(val, NULL, 10);
    else if(!strcmp(kw, "connDestination")) strcpy(kwLocalStruct->connDestination, val);
    else if(!strcmp(kw, "agentName")) strcpy(kwLocalStruct->agentName, val);
    else if(!strcmp(kw, "appID")) kwLocalStruct->appID = (int)strtoll(val, NULL, 10);
    else if(!strcmp(kw, "ndAppServerID")) kwLocalStruct->ndAppServerID = (int)strtoll(val, NULL, 10);
    else if(!strcmp(kw, "tierID")) kwLocalStruct->tierID = (int)strtoll(val, NULL, 10);
    else if(!strcmp(kw, "testIdx")) kwLocalStruct->testIdx = (int)strtoll(val, NULL, 10);
    else if(!strcmp(kw, "NDCollectorPort")) kwLocalStruct->NDCollectorPort = (int)strtoll(val, NULL, 10);
    else if(!strcmp(kw, "NDCollectorIP")) strcpy(kwLocalStruct->NDCollectorIP, val);
    else if (!strcmp(kw, "tags")) {
      if(val && strlen(val)>0)
      {
        strcpy(kwLocalStruct->tags, val);
      }
    }
    else if (!strcmp(kw, "attributes")) {
      if(val && strlen(val)>0)
      {
        strcpy(kwLocalStruct->tierAttr,val);
      }
    }
    // else 
    // {
    //   // Invalid kw/value pair
    //   NDTL4(trace_log_key, NULL, TL_INFO, "Invalid Keyword-value pair -> %s=%s", kw, val);
    // }
  }
}

/*****************************************************************
 * Registering the socket in epoll
 * **************************************************************/

void addSocketInEpoll(apr_socket_t *skt, PxAppEpollInfo_t *ptr)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  /* Declaring all the required variables */
  char error_buf[ERROR_BUFFER_SIZE];
  apr_status_t rv;
  int port;
  char ip[128];

  // Set the values to add in epoll
  apr_pollfd_t pfd = { NDProxy.proxy_pool, APR_POLL_SOCKET, APR_POLLIN|APR_POLLHUP|APR_POLLERR|APR_POLLNVAL, 0, { NULL }, NULL };
  pfd.desc.s = skt;
  if(ptr)
  {
    pfd.client_data = (void *)ptr;
    NDTL3(trace_log_key, NULL, "INFO", "Added instance at appIndex = %d, procIndex = %d", 
	((PxAppEpollInfo_t *)pfd.client_data)->appIndex, ((PxAppEpollInfo_t *)pfd.client_data)->processIndex);
  }
  else
    pfd.client_data = NULL;

  if(skt != NDProxy.skt) 
    get_dst_addr(skt, ip, 128, &port);
  else
    get_src_addr(skt, ip, 128, &port);

  //Add proxy control fd in pollset
  if((rv = apr_pollset_add(NDProxy.pollset, &pfd)) != APR_SUCCESS)
  {
    NDTL0(trace_log_key, NULL, TL_ERROR, "Could not add '%s:%d' to poll. Error: %s",  ip, port, apr_strerror(rv, error_buf, ERROR_BUFFER_SIZE));
    return;
  }
  NDTL2(trace_log_key, NULL, TL_INFO, "Successully registered '%s:%d' in epoll.", ip, port);
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}

/*********************************************************
 * This function will remove the socket from  Epoll and 
 * set it to NULL
 * Uses CLEAN to  remove the socket from epoll,
 * else simply closes the connection.
 * ******************************************************/
void cleanAndRemoveSocket(apr_socket_t **socket, char clean_flag)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  if (!(*socket))
    return;

  int port;
  char ip[128];
  get_dst_addr(*socket, ip, 128, &port);

  if(clean_flag)
  {
    apr_status_t rv;
    apr_pollfd_t pfd = { NDProxy.proxy_pool, APR_POLL_SOCKET, APR_POLLIN|APR_POLLHUP|APR_POLLERR|APR_POLLNVAL, 0, { NULL }, NULL };
    pfd.desc.s = *socket;
    if (pfd.client_data) {
      FREE_AND_MAKE_NULL(pfd.client_data, "free epollPtr", -1);
      pfd.client_data = NULL;
    }

    rv = apr_pollset_remove(NDProxy.pollset, &pfd);

    if (rv == APR_NOTFOUND)
    {
      NDTL1(trace_log_key, NULL, TL_INFO, "Unable to remove '%s:%d' from epoll", ip, port);
    }
  }

  apr_socket_close(*socket);

  *socket = NULL;

  NDTL2(trace_log_key, NULL, TL_INFO, "'%s:%d' - Connection terminated.", ip, port);

  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}

int populateProcessInfo(int i, int processCount, int pid, PxAppEpollInfo_t **epollPtr, apr_socket_t *instanceSkt, char flag, struct lws* wsi)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Entry");
  char error_flag = 0;
  int rv = 0, port;
  char ip[128];

  //get_dst_addr(instanceSkt, ip, 128, &port);
  get_dst_addr_wrapper(instanceSkt, wsi, ip, 128, &port,ndApplication->proxy_lws_active);

  // Fill the necessary data elements for the instance.
  NDProxyInstance[i]->processInfo[processCount].data_len = 0;
  if (NDProxyInstance[i]->processInfo[processCount].data)
    FREE_AND_MAKE_NULL(NDProxyInstance[i]->processInfo[processCount].data, "Free'ing data buffer", -1);
  NDProxyInstance[i]->processInfo[processCount].processId = pid;
  NDProxyInstance[i]->processInfo[processCount].ctrlSocket = instanceSkt;
  NDProxyInstance[i]->processInfo[processCount].ctrlLwsHandler.wsi = wsi;

  // epollPtr -> used for registering/identifying the instance element to/from the epoll list. (Instance context element)
  if(!(*epollPtr))
  {
    NDLB_MALLOC(*epollPtr, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
    (*epollPtr)->appIndex = i;
    (*epollPtr)->processIndex = processCount;
  }

  //AS : Free the already existing structure
  if(!NDProxyInstance[i]->processInfo[processCount].pxAppEpoll)
    NDLB_MALLOC(NDProxyInstance[i]->processInfo[processCount].pxAppEpoll, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
  (NDProxyInstance[i]->processInfo[processCount].pxAppEpoll)->appIndex = i;
  (NDProxyInstance[i]->processInfo[processCount].pxAppEpoll)->processIndex = processCount;

  // Getting OS specific socket description from the socket instance (for information purpose only).
  if (instanceSkt) {
    apr_os_sock_get(&NDProxyInstance[i]->processInfo[processCount].ctrlFd, 
	NDProxyInstance[i]->processInfo[processCount].ctrlSocket);
  }

  // If control connection with NDC exists, then send response for nd_ctrl_msg_req as OK.
  // This will be called only when a second request from the same application is received (second process request).
  if(NDProxyInstance[i]->control.skt || NDProxyInstance[i]->control.lws_handler.wsi)
  {
    char ctrlMsgRep[512] = "";
    int ctrlMsgLen = 0;
    ctrlMsgLen = snprintf(ctrlMsgRep,512, "nd_ctrl_msg_rep:result=Ok;procIndex=%d\n", processCount);
    if (ctrlMsgLen >= 512)
        ctrlMsgLen = 511;

    if (ndApplication->proxy_lws_active == 1) {
      rv = send_on_wsi(NDProxy.lws_handler.context, &NDProxyInstance[i]->processInfo[processCount].ctrlLwsHandler, ctrlMsgRep, ctrlMsgLen);
      if (rv != 0) {
	NDTL0(trace_log_key, NULL, TL_ERROR, "Connection Broken. Unable to send '%s' for pid = %d, '%s:%d'",
	    ctrlMsgRep, NDProxyInstance[i]->processInfo[processCount].processId, ip, port);
	NDProxyInstance[i]->processInfo[processCount].processId = -1;
	return -1;
      }
    }
    else {
      rv = ndlb_send_n_bytes(NDProxyInstance[i]->processInfo[processCount].ctrlSocket,
	  ctrlMsgRep, ctrlMsgLen, error_buf, ERROR_BUFFER_SIZE);
      if(rv == ctrlMsgLen)
      {
	NDTL0(trace_log_key, NULL, "PROXY->INSTANCE", "Sent '%s' '%s:%d'", ctrlMsgRep, ip, port);
      }
      else if( rv == NDLB_CONNECTION_CLOSE)
      {
	NDTL0(trace_log_key, NULL, TL_ERROR, "Connection Broken. Unable to send '%s' for pid = %d, '%s:%d'",
	    ctrlMsgRep, NDProxyInstance[i]->processInfo[processCount].processId, ip, port);
	NDProxyInstance[i]->processInfo[processCount].processId = -1;
	return -1;
      }
    }


    // Check if the startInstMsg exists, then send the message to the controlFd.
    apr_thread_mutex_lock(NDProxyInstance[i]->start_instr_lock);
    if(NDProxyInstance[i]->startInstMsg) {
      if (ndApplication->proxy_lws_active == 1) {
        rv = send_on_wsi(NDProxy.lws_handler.context, &NDProxyInstance[i]->processInfo[processCount].ctrlLwsHandler,
        NDProxyInstance[i]->startInstMsg, strlen(NDProxyInstance[i]->startInstMsg));

        if (rv != 0) {
          NDTL0(trace_log_key, NULL, TL_ERROR, "Connection Broken. Unable to send 'START_INSTRUMENTATION' for pid = %d, '%s:%d'",
          NDProxyInstance[i]->processInfo[processCount].processId, ip, port);
          NDProxyInstance[i]->processInfo[processCount].processId = -1;
          apr_thread_mutex_unlock(NDProxyInstance[i]->start_instr_lock);
          return -1;
        }
      }
      else {
        rv = ndlb_send_n_bytes(NDProxyInstance[i]->processInfo[processCount].ctrlSocket,
        NDProxyInstance[i]->startInstMsg, strlen(NDProxyInstance[i]->startInstMsg), error_buf, ERROR_BUFFER_SIZE);
        if( rv == NDLB_CONNECTION_CLOSE) {
          NDTL0(trace_log_key, NULL, TL_ERROR, "Connection Broken. Unable to send 'START_INSTRUMENTATION' for pid = %d, '%s:%d'",
          NDProxyInstance[i]->processInfo[processCount].processId, ip, port);
          NDProxyInstance[i]->processInfo[processCount].processId = -1;
          apr_thread_mutex_unlock(NDProxyInstance[i]->start_instr_lock);
          return -1;
        }

        rv = ndlb_send_n_bytes(NDProxyInstance[i]->processInfo[processCount].ctrlSocket, 
        "\n", strlen("\n"), error_buf, ERROR_BUFFER_SIZE);
        if( rv == NDLB_CONNECTION_CLOSE) {
          NDTL0(trace_log_key, NULL, TL_ERROR, "Connection Broken. Unable to send 'START_INSTRUMENTATION' for pid = %d, '%s:%d'",
          NDProxyInstance[i]->processInfo[processCount].processId, ip, port);
          NDProxyInstance[i]->processInfo[processCount].processId = -1;
          apr_thread_mutex_unlock(NDProxyInstance[i]->start_instr_lock);
          return -1;
        }
      }
      NDTL0(trace_log_key, NULL, TL_INFO, "Sent 'START_INSTRUMENTATION' message successfully for pid = %d, '%s:%d'",
      NDProxyInstance[i]->processInfo[processCount].processId, ip, port); 
    }
    apr_thread_mutex_unlock(NDProxyInstance[i]->start_instr_lock);
  }
  // register the fd in epoll
  if(instanceSkt && !error_flag)
    addSocketInEpoll(instanceSkt, NDProxyInstance[i]->processInfo[processCount].pxAppEpoll);

  if(flag)
    NDProxyInstance[i]->processCount += 1;
  NDProxyInstance[i]->activeProcessCount += 1;

  NDTL1(trace_log_key, NULL, TL_INFO, "Process Info populated for Application with pid = %d, '%s:%d'",  NDProxyInstance[i]->processInfo[processCount].processId, ip, port);
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
  return 0;
}

void exceptionMapInit()
{
   ndExceptionIDS.exceptionClassNameID = 1;
   ndExceptionIDS.throwingClassNameID = 1;
   ndExceptionIDS.throwingMethodNameID = 1;
   ndExceptionIDS.msgID = 1;
   ndExceptionIDS.causeID = 1;
}

int make_connection_with_ndc(NDProxyInstance_t* nProxyInstance, int apptype) {
  if (!nProxyInstance->control.skt &&
  !nProxyInstance->control.lws_handler.wsi &&
  !nProxyInstance->control_thread_id)
    {
      //thread create and make control connection
      char  error_buff[ERROR_BUFFER_SIZE];

      ndApplication = nProxyInstance;
      ndApplication->close_udp_thread = 0;
      ndApplication->apptype = apptype;
      apr_pool_create(&control_pool, NULL);
      long long currTimeforReset = my_gettimeofday();
      resetTimeInterval = currTimeforReset; //KV: This is for calling reset after a given time of interval.

      if(apptype) {
        if(!udp_havoc_timeout)
          udp_havoc_timeout = create_hashmap(control_pool);
        if(!udp_havoc_timeout)
            NDTL0(trace_log_key, NULL, TL_ERROR, "Error in creating map for udp havoc messages...");
      }

      NDSysInit("proxy");
      if(!methodEntryLock)
        apr_thread_mutex_create(&methodEntryLock, APR_THREAD_MUTEX_NESTED, control_pool);
      if(!ndApplication->start_instr_lock)
        apr_thread_mutex_create(&ndApplication->start_instr_lock, APR_THREAD_MUTEX_NESTED, control_pool);
      char traceFileName[1024] = "" ;
      snprintf(traceFileName, 1024, "%s_%s_%s_%s", gPool, ndApplication->tierName, ndApplication->appServerName, ndApplication->appName);
      //if(isndc)
        //NDTraceInit(gProcessName, traceFileName);
      NDBCISettingInit();
      //read_conf_file();
      if(read_env_var() == SUCESS ||  read_conf_file() == SUCESS)
      {
	   NDTL2(trace_log_key, NULL, TL_INFO, "configuration readiing in proxy");
      }
      setAgentType(nProxyInstance->agentType);
      initializeNDApplicationGlobal();
      populateNDBCISetting(nProxyInstance);
      initThreadBasedModelMap();
      if(!controlThreadTmpLock)
        apr_thread_mutex_create(&controlThreadTmpLock, APR_THREAD_MUTEX_NESTED, control_pool);
      gFreeTLSHead = create_tls_pool(gBusyTLSCnt);
      initNDManageTIdToTLocal();
      if(!ndTDHashMap)
        ndTDHashMap = create_hashmap(control_pool);
      exceptionMapInit();

      NDTL3(trace_log_key, NULL, TL_INFO, "\nBefore creating control thread NDApplication  'appName - %s'",ndApplication->NDBCISetting.appName);
      if (ndlb_thread_create(&nProxyInstance->control_thread_id, control_thread_entry_fun,
      (void *)nProxyInstance, error_buff, ERROR_BUFFER_SIZE, control_pool) != SUCESS)
      {
          NDTL0(trace_log_key, NULL, TL_ERROR, "Error in creating control thread... '%s'", error_buff);
      }
      else
      {
         NDTL1(trace_log_key, NULL, TL_INFO, "Control thread created successfully. ThreadId - '%lld'", nProxyInstance->control_thread_id);
      }
      return 1;
    }
    return 0;
}

/*****************************************************************
 * This function  will parse the message based on the there type
 * and will populate the data structures accordingly for that
 * NDProxyInstance.
 * This function takes 2 arguments:
 * parseProxyMsgReq (char *buffer, int con_type)
 *
 * con_type = 	1 - Control message
 * 		2 - Data Connection Message
 * 		3 - Auto Sensor Connection Message
 * **************************************************************/
void parseProxyMsgReq(apr_socket_t *instanceSkt, char *buffer, PxAppEpollInfo_t **epollPtr, int con_type, struct lws* wsi)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  int port;
  char ip[128];
  char error_flag = 0;
  int i, j, free_index = -1, app_index = -1, match_index = -1;
  struct kwLocalStruct_t kwLocalStruct;
  memset(&kwLocalStruct,0,sizeof(struct kwLocalStruct_t));
  //resetKwLocalStruct(&kwLocalStruct);
  parse_tag_values(buffer,&kwLocalStruct);

  //get_dst_addr(instanceSkt, ip, 128, &port);
  get_dst_addr_wrapper(instanceSkt, wsi, ip, 128, &port, ndApplication->proxy_lws_active);

  if (con_type == 1)	//Control connection
  {
     NDTL2(trace_log_key, NULL, TL_INFO, "con_type:kwLocalStruct.appName [%s],tags:%s,attributes:%s", 
     kwLocalStruct.appName, kwLocalStruct.tags,kwLocalStruct.tierAttr);
    // Check  for the received entry in the connected list of applications to populate structures accordingly
    // New connections may have same application name, but should have different process id.
    for (i = 1; i < NDProxy.maxTCPProcesses; i++)
    {
      if(!NDProxyInstance[i]) {
        if(app_index == -1)
          app_index = i;
        continue;
      }
      NDTL2(trace_log_key, NULL, TL_INFO, "con_type:NDProxyInstance[i]->appName [%s], kwLocalStruct.appName [%s],tags:%s,attributes:%s", 
      NDProxyInstance[i]->appName, kwLocalStruct.appName, kwLocalStruct.tags,kwLocalStruct.tierAttr);
      NDTL2(trace_log_key, NULL, TL_INFO, "con_type:NDProxyInstance[i]->tierName [%s], kwLocalStruct.tierName [%s]", 
      NDProxyInstance[i]->tierName, kwLocalStruct.tierName);
      NDTL2(trace_log_key, NULL, TL_INFO, "con_type:NDProxyInstance[i]->appServerName [%s], kwLocalStruct.ndAppServerHost [%s]", 
      NDProxyInstance[i]->appServerName, kwLocalStruct.ndAppServerHost);
      NDTL2(trace_log_key, NULL, TL_INFO, "con_type:NDProxyInstance[i]->agentName [%s], kwLocalStruct.agentName [%s]", 
      NDProxyInstance[i]->agentName, kwLocalStruct.agentName);
      if (!(strcmp(NDProxyInstance[i]->tierName, kwLocalStruct.tierName))
  	  && !(strcmp(NDProxyInstance[i]->appServerName, kwLocalStruct.ndAppServerHost))
      && !(strcmp(NDProxyInstance[i]->agentName, kwLocalStruct.agentName))
      && (!strcmp(NDProxyInstance[i]->appName, kwLocalStruct.appName) || !strcmp(NDProxyInstance[i]->NDBCISetting.appName, kwLocalStruct.appName)))
      {
        match_index = i;
        // Instance Found !
        // Check to see if the process id is unique
        for (j = 0; j < NDProxyInstance[i]->processCount; j++)
        {
          if (kwLocalStruct.pid == NDProxyInstance[i]->processInfo[j].processId)
          {
            NDTL3(trace_log_key, NULL, TL_INFO, "Control connection received again from application with pId = %d, '%s:%d'",
            kwLocalStruct.pid, ip, port);
            if (NDProxyInstance[i]->processInfo[j].ctrlSocket) {
              cleanAndRemoveSocket(&NDProxyInstance[i]->processInfo[j].ctrlSocket, CLEAN);
              NDProxyInstance[i]->activeProcessCount--;
            }
            if (NDProxyInstance[i]->processInfo[j].ctrlLwsHandler.wsi) {
              remove_wsi(NDProxy.lws_handler.context, NDProxyInstance[i]->processInfo[j].ctrlLwsHandler.wsi);
              NDProxyInstance[i]->activeProcessCount--;
            }
            if (populateProcessInfo(i, j, kwLocalStruct.pid, epollPtr, instanceSkt, 0, wsi))
            {
              NDTL0(trace_log_key, NULL, TL_ERROR, "Control connection registration failed with proxy, pId = %d, '%s:%d'", 
              kwLocalStruct.pid, ip, port);
              cleanAndRemoveSocket(&instanceSkt, NOCLEAN);
              NDProxyInstance[i]->processInfo[j].processId = -1;
              NDProxyInstance[i]->processInfo[j].ctrlSocket = NULL;
              remove_wsi(NDProxy.lws_handler.context, wsi);
              return;
            }
            break;
          }
          else if (NDProxyInstance[i]->processInfo[j].processId == -1 && (free_index == -1))
          {
            free_index = j;
          }
        }
        if (j == NDProxyInstance[i]->processCount)
        {
          if (free_index > -1)
          {
            if (populateProcessInfo(i, free_index, kwLocalStruct.pid, epollPtr, instanceSkt, 0, wsi))
            {
              NDTL0(trace_log_key, NULL, TL_ERROR, "Control connection registration failed with proxy, pId = %d, '%s:%d'", 
              kwLocalStruct.pid, ip, port);
              cleanAndRemoveSocket(&instanceSkt, NOCLEAN);
              NDProxyInstance[i]->processInfo[free_index].processId = -1;
              NDProxyInstance[i]->processInfo[free_index].ctrlSocket = NULL;
              remove_wsi(NDProxy.lws_handler.context, wsi);
              return;
            }
            break;
          }
          // Process Entry not found (New Process). Add it to the structure.
          // Re-allocate memory for the NDAppInstance structure. (keeping 10 structure elements as buffer)
          else if ((NDProxyInstance[i]->processCount % 500 == 490))
          {
            instanceProcess_t *tmp_ptr = NULL, *tmp_ptr2 = NDProxyInstance[i]->processInfo;
            NDLB_REALLOC_AND_MEMSET(tmp_ptr, sizeof(instanceProcess_t) * (NDProxyInstance[i]->processCount + 10 + 500),
            sizeof(instanceProcess_t) * (NDProxyInstance[i]->processCount + 10),
            "Reallocating Memory to store NDProxyInstance->processInfo", -1);
            memcpy(tmp_ptr, NDProxyInstance[i]->processInfo, sizeof(instanceProcess_t) * (NDProxyInstance[i]->processCount + 10));
            NDProxyInstance[i]->processInfo = tmp_ptr;

            FREE_AND_MAKE_NULL(tmp_ptr2, "Free'ing temp_ptr", -1);
            tmp_ptr = NULL;
            NDTL0(trace_log_key, NULL, TL_INFO, "Memory reallocated for NDProxyInstance->processInfo. Available memory = '%d' blocks.", 
            NDProxyInstance[i]->processCount + 10 + 500);
          }
          NDTL3(trace_log_key, NULL, TL_INFO, "Control connection received, pId = %d, '%s:%d'", kwLocalStruct.pid, ip, port);
          if (populateProcessInfo(i, NDProxyInstance[i]->processCount, kwLocalStruct.pid, epollPtr, instanceSkt, 1, wsi))
          {
            NDTL0(trace_log_key, NULL, TL_ERROR, "Control connection registration failed with proxy, pId = %d, '%s:%d'", 
            kwLocalStruct.pid, ip, port);
            cleanAndRemoveSocket(&instanceSkt, NOCLEAN);
            NDProxyInstance[i]->processInfo[NDProxyInstance[i]->processCount].processId = -1;
            NDProxyInstance[i]->processInfo[NDProxyInstance[i]->processCount].ctrlSocket = NULL;
            remove_wsi(NDProxy.lws_handler.context, wsi);
            return;
          }
          break;
        }
        break;
      }
    }
    if (match_index == -1) // New Instance Request Connection
    {
      if(app_index == -1) {
        NDLB_REALLOC_AND_MEMSET(NDProxyInstance, sizeof(NDProxyInstance_t*)*(NDProxy.maxTCPProcesses+PROXY_DELTA_CONNECTIONS),
        sizeof(NDProxyInstance_t*)*NDProxy.maxTCPProcesses,
        "Re-Allocating memory to store the nProxyInstance->processInfo", -1);

        NDLB_REALLOC_AND_MEMSET(NDProxyTcpUdpInstance, sizeof(NDProxyInstance_t*)*(NDProxy.maxTCPProcesses+PROXY_DELTA_CONNECTIONS),
        sizeof(NDProxyInstance_t*)*NDProxy.maxTCPProcesses,
        "Re-Allocating memory to store the nProxyInstance->processInfo", -1);

        app_index = NDProxy.maxTCPProcesses;
        NDProxy.maxTCPProcesses += PROXY_DELTA_CONNECTIONS;
      }
      if(!NDProxyInstance[app_index]) {
        NDLB_MALLOC_AND_MEMSET(NDProxyInstance[app_index],
        sizeof(NDProxyInstance_t) ,
        "Allocating memory to store the nProxyInstance->processInfo", -1);
      }
      NDTL2(trace_log_key, NULL, TL_INFO, "con_type:NDProxyInstance[app_index]->appName [%s], kwLocalStruct.appName [%s]", 
      NDProxyInstance[app_index]->appName, kwLocalStruct.appName);
      NDTL2(trace_log_key, NULL, TL_INFO, "con_type:NDProxyInstance[app_index]->tierName [%s], kwLocalStruct.tierName [%s]", 
      NDProxyInstance[app_index]->tierName, kwLocalStruct.tierName);
      NDTL2(trace_log_key, NULL, TL_INFO, "con_type:NDProxyInstance[app_index]->appServerName [%s], kwLocalStruct.ndAppServerHost [%s]", 
      NDProxyInstance[app_index]->appServerName, kwLocalStruct.ndAppServerHost);
      NDTL2(trace_log_key, NULL, TL_INFO, "con_type:NDProxyInstance[app_index]->agentName [%s], kwLocalStruct.agentName [%s]", 
      NDProxyInstance[app_index]->agentName, kwLocalStruct.agentName);

      // Reset the structure
      setNDProxyInstance(NDProxyInstance[app_index]);
      // Populate the structure with values received.
      populateNDProxyInstance(NDProxyInstance[app_index],&kwLocalStruct,0);

      NDProxyInstance[app_index]->processInfo[0].ctrlSocket = instanceSkt;
      NDProxyInstance[app_index]->processInfo[0].ctrlLwsHandler.wsi = wsi;
      if (!(*epollPtr))
      {
        NDLB_MALLOC(*epollPtr, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
        (*epollPtr)->appIndex = app_index;
        (*epollPtr)->processIndex = NDProxyInstance[app_index]->processCount;
      } 
      
      //AS : Free the already existing structure
      if(!NDProxyInstance[app_index]->processInfo[0].pxAppEpoll)
        NDLB_MALLOC(NDProxyInstance[app_index]->processInfo[0].pxAppEpoll, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
      (NDProxyInstance[app_index]->processInfo[0].pxAppEpoll)->appIndex = app_index;
      (NDProxyInstance[app_index]->processInfo[0].pxAppEpoll)->processIndex = NDProxyInstance[app_index]->processCount;

      NDProxyInstance[app_index]->processInfo[0].data_len = 0;
      NDProxyInstance[app_index]->processInfo[0].processId = kwLocalStruct.pid;
      if (instanceSkt) {
	     apr_os_sock_get(&NDProxyInstance[app_index]->processInfo[0].ctrlFd,
	     NDProxyInstance[app_index]->processInfo[0].ctrlSocket);
      }

      // register the fd in epoll
      if (instanceSkt && !error_flag)
	     addSocketInEpoll(instanceSkt, NDProxyInstance[app_index]->processInfo[0].pxAppEpoll);

      NDTL1(trace_log_key, NULL, TL_INFO, "Process Info populated for Application with pid = %d, '%s:%d'", 
      NDProxyInstance[app_index]->processInfo[0].processId, ip, port);

      // increment the process counter
      NDProxyInstance[app_index]->processCount += 1;
      NDProxyInstance[app_index]->activeProcessCount += 1;
      // Increment the counter for numInstances by 1
      NDProxy.numInstances += 1;
      match_index = app_index;
    }
    // Make a new control connection to NDC
    make_connection_with_ndc(NDProxyInstance[match_index],0);
  }
  else if (con_type == 2)	// Data Connection
  {
    // Search this entry in NDProxyInstance and add its socketfd in the epoll
    int i;
    for (i = 1; i < NDProxy.maxTCPProcesses; i++)
    {
      if(!NDProxyInstance[i])
        continue;
      if (!strcmp(NDProxyInstance[i]->NDBCISetting.appName, kwLocalStruct.appName) &&
	  (NDProxyInstance[i]->NDBCISetting.appID == kwLocalStruct.appID) &&
	  !(strcmp(NDProxyInstance[i]->NDBCISetting.tierName, kwLocalStruct.tierName)) &&
	  (NDProxyInstance[i]->NDBCISetting.tierID == kwLocalStruct.tierID) &&
	  !(strcmp(NDProxyInstance[i]->NDBCISetting.appServerName, kwLocalStruct.ndAppServerHost)) &&
	  (NDProxyInstance[i]->NDBCISetting.serverID == kwLocalStruct.ndAppServerID) &&
	  !(strcmp(NDProxyInstance[i]->agentName, kwLocalStruct.agentName)))
      {
	// Instance Found !
	// Check to see if the process id is unique
	NDTL1(trace_log_key, NULL, TL_INFO, "Data connection received, pId = %d, '%s:%d'", kwLocalStruct.pid, ip, port);
	for (j = 0; j < NDProxyInstance[i]->processCount; j++)
	{
	  if (kwLocalStruct.pid == NDProxyInstance[i]->processInfo[j].processId)
	  {
           if (ndApplication->proxy_lws_active == 1) {
	      if (NDProxyInstance[i]->processInfo[j].dcLwsHandler.wsi)
	      {
		remove_wsi(NDProxy.lws_handler.context, NDProxyInstance[i]->processInfo[j].dcLwsHandler.wsi);
	      }
	      // Entry Found for the process. Now add the entry of data connection fd in the NDProxyInstance
	      NDProxyInstance[i]->processInfo[j].dcLwsHandler.wsi = wsi;

	      if (!(*epollPtr))
	      {
		NDLB_MALLOC(*epollPtr, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
		(*epollPtr)->appIndex = i;
		(*epollPtr)->processIndex = j;
	      }
	      NDTL1(trace_log_key, NULL, TL_INFO, "Registered the data connection socket for pid '%d', '%s:%d'",
		  NDProxyInstance[i]->processInfo[j].processId, ip, port);
	      break;
	    }
	    else {
	      if (NDProxyInstance[i]->processInfo[j].dcSocket)
	      {
		cleanAndRemoveSocket(&NDProxyInstance[i]->processInfo[j].dcSocket, CLEAN);
		NDProxyInstance[i]->processInfo[NDProxyInstance[i]->processCount].dcFd = -1;
	      }
	      // Entry Found for the process. Now add the entry of data connection fd in the NDProxyInstance
	      NDProxyInstance[i]->processInfo[j].dcSocket = instanceSkt;

	      if (!(*epollPtr))
	      {
		NDLB_MALLOC(*epollPtr, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
		(*epollPtr)->appIndex = i;
		(*epollPtr)->processIndex = j;
	      }

	      apr_os_sock_get(&NDProxyInstance[i]->processInfo[j].dcFd, NDProxyInstance[i]->processInfo[j].dcSocket);

	      // Registering the fd in epoll
	      if (!error_flag)
		addSocketInEpoll(instanceSkt, NDProxyInstance[i]->processInfo[j].pxAppEpoll);

	      NDTL1(trace_log_key, NULL, TL_INFO, "Registered the data connection socket for pid '%d', '%s:%d'",
		  NDProxyInstance[i]->processInfo[j].processId, ip, port);
	      break;
	    }
	  }
	}
	if (j == NDProxyInstance[i]->processCount)
	{
         NDTL1(trace_log_key, NULL, TL_ERROR, "No control connection found for this data connection request with pid '%d', '%s:%d'",
	      kwLocalStruct.pid, ip, port);
	  cleanAndRemoveSocket(&instanceSkt, NOCLEAN);
	  remove_wsi(NDProxy.lws_handler.context, wsi);
	  break;
	}
      }
    }
  }

  else if (con_type == 3) 	// Auto Sensor HotSpot connection
  {
    // Search this entry in NDProxyInstance and add its socketfd in the epoll
    int i;
    for ( i=1; i< NDProxy.maxTCPProcesses; i++)
    {
      if(!NDProxyInstance[i])
        continue;
      if (!strcmp(NDProxyInstance[i]->NDBCISetting.appName, kwLocalStruct.appName) && 
	  (NDProxyInstance[i]->NDBCISetting.appID == kwLocalStruct.appID) &&
	  !(strcmp(NDProxyInstance[i]->NDBCISetting.tierName, kwLocalStruct.tierName)) && 
	  (NDProxyInstance[i]->NDBCISetting.tierID == kwLocalStruct.tierID) &&
	  !(strcmp(NDProxyInstance[i]->NDBCISetting.appServerName, kwLocalStruct.ndAppServerHost)) && 
	  (NDProxyInstance[i]->NDBCISetting.serverID == kwLocalStruct.ndAppServerID) &&
	  !(strcmp(NDProxyInstance[i]->agentName, kwLocalStruct.agentName)))
      {
	// Instance Found !
	// Check to see if the process id is unique
	NDTL1(trace_log_key, NULL, TL_INFO, "Auto Sensor connection received, pId = %d, '%s:%d'", kwLocalStruct.pid, ip, port);
	for (j = 0; j<NDProxyInstance[i]->processCount; j++)
	{
	  if (kwLocalStruct.pid == NDProxyInstance[i]->processInfo[j].processId)
	  {
           if (ndApplication->proxy_lws_active == 1) {
	      if (NDProxyInstance[i]->processInfo[j].asLwsHandler.wsi) {
		remove_wsi(NDProxy.lws_handler.context, NDProxyInstance[i]->processInfo[j].asLwsHandler.wsi);
	      }
	      NDProxyInstance[i]->processInfo[j].asLwsHandler.wsi = wsi;

	      if (!(*epollPtr))
	      {
		NDLB_MALLOC(*epollPtr, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
		(*epollPtr)->appIndex = i;
		(*epollPtr)->processIndex = j;
	      }
	      NDTL1(trace_log_key, NULL, TL_INFO, "Registered the AS connection socket for pid '%d', '%s:%d'",
		  NDProxyInstance[i]->processInfo[j].processId, ip, port);
	      break;
	    }
	    else {
	      if (NDProxyInstance[i]->processInfo[j].asSocket)
	      {
		cleanAndRemoveSocket(&NDProxyInstance[i]->processInfo[j].asSocket, CLEAN);
		NDProxyInstance[i]->processInfo[NDProxyInstance[i]->processCount].asFd = -1;
	      }
	      // Entry Found for the process. Now add the entry of data connection fd in the NDProxyInstance
	      NDProxyInstance[i]->processInfo[j].asSocket = instanceSkt;

	      if(!(*epollPtr))
	      {
		NDLB_MALLOC(*epollPtr, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
		(*epollPtr)->appIndex = i;
		(*epollPtr)->processIndex = j;
	      }

	      if (instanceSkt) {
		apr_os_sock_get(&NDProxyInstance[i]->processInfo[j].asFd, NDProxyInstance[i]->processInfo[j].asSocket);
	      }


	      // Registering the fd in epoll
	      if(!error_flag)
		addSocketInEpoll(instanceSkt, NDProxyInstance[i]->processInfo[j].pxAppEpoll);

	      NDTL1(trace_log_key, NULL, TL_INFO, "Registered the AS connection socket for pid '%d', '%s:%d'", 
		  NDProxyInstance[i]->processInfo[j].processId, ip, port);
	      break;
	    }

	  }
	}
	if (j == NDProxyInstance[i]->processCount)
	{
         NDTL1(trace_log_key, NULL, TL_ERROR, "No control connection found for this auto sensor connection request with pid '%d', '%s:%d'",
	      kwLocalStruct.pid, ip, port);
	  cleanAndRemoveSocket(&instanceSkt, NOCLEAN);
	  remove_wsi(NDProxy.lws_handler.context, wsi);
	  break;
	}
      }
    }
  }
  else if( con_type == 4)
  {
     int i;
     for (i = 1; i < NDProxy.maxTCPProcesses; i++)
     {
        if(!NDProxyInstance[i])
          continue;
        if (!strcmp(NDProxyInstance[i]->NDBCISetting.appName, kwLocalStruct.appName) &&
	   (NDProxyInstance[i]->NDBCISetting.appID == kwLocalStruct.appID) &&
	   !(strcmp(NDProxyInstance[i]->NDBCISetting.tierName, kwLocalStruct.tierName)) &&
	   (NDProxyInstance[i]->NDBCISetting.tierID == kwLocalStruct.tierID) &&
	    !(strcmp(NDProxyInstance[i]->NDBCISetting.appServerName, kwLocalStruct.ndAppServerHost)) &&
	   (NDProxyInstance[i]->NDBCISetting.serverID == kwLocalStruct.ndAppServerID) &&
	   !(strcmp(NDProxyInstance[i]->agentName, kwLocalStruct.agentName)))
        {
			  // Instance Found !
			  // Check to see if the process id is unique
	   NDTL2(trace_log_key, NULL, TL_INFO, "Discovery connection received, pId = %d, '%s:%d'", kwLocalStruct.pid, ip, port);
	   for (j = 0; j < NDProxyInstance[i]->processCount; j++)
	   {
	   	if (kwLocalStruct.pid == NDProxyInstance[i]->processInfo[j].processId)
		{
		   if (ndApplication->proxy_lws_active == 1) {
	              if (NDProxyInstance[i]->processInfo[j].disLwsHandler.wsi)
		      {
		         remove_wsi(NDProxy.lws_handler.context, NDProxyInstance[i]->processInfo[j].disLwsHandler.wsi);
		      }
		      // Entry Found for the process. Now add the entry of data connection fd in the NDProxyInstance
		      NDProxyInstance[i]->processInfo[j].disLwsHandler.wsi = wsi;
		      if (!(*epollPtr))
		      {
		         NDLB_MALLOC(*epollPtr, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
			 (*epollPtr)->appIndex = i;
			 (*epollPtr)->processIndex = j;
		      }
		      NDTL1(trace_log_key, NULL, TL_INFO, "Registered the data connection socket for pid '%d', '%s:%d'",
		      NDProxyInstance[i]->processInfo[j].processId, ip, port);
		      break;
		   }
		   else {
			if (NDProxyInstance[i]->processInfo[j].disSocket)
			{
			   cleanAndRemoveSocket(&NDProxyInstance[i]->processInfo[j].disSocket, CLEAN);
			   NDProxyInstance[i]->processInfo[NDProxyInstance[i]->processCount].disFd = -1;
			}
			// Entry Found for the process. Now add the entry of data connection fd in the NDProxyInstance
			NDProxyInstance[i]->processInfo[j].disSocket = instanceSkt;

			if (!(*epollPtr))
			{
		            NDLB_MALLOC(*epollPtr, sizeof(PxAppEpollInfo_t), "Allocating for epoll context structure", -1);
			    (*epollPtr)->appIndex = i;
			    (*epollPtr)->processIndex = j;
			}
#ifdef __unix__
			apr_os_sock_get(&NDProxyInstance[i]->processInfo[j].disFd, NDProxyInstance[i]->processInfo[j].disSocket);
#endif
			// Registering the fd in epoll
			if (!error_flag)
			   addSocketInEpoll(instanceSkt, NDProxyInstance[i]->processInfo[j].pxAppEpoll);

			   NDTL1(trace_log_key, NULL, TL_INFO, "Registered the data connection socket for pid '%d', '%s:%d'",
			   NDProxyInstance[i]->processInfo[j].processId, ip, port);
			   break;
			}
		   }
	      }
	      if (j == NDProxyInstance[i]->processCount)
	      {
	         NDTL0(trace_log_key, NULL, TL_ERROR, "No control connection found for this data connection request with pid '%d', '%s:%d'",
						  kwLocalStruct.pid, ip, port);
		 cleanAndRemoveSocket(&instanceSkt, NOCLEAN);
		 remove_wsi(NDProxy.lws_handler.context, wsi);
		 break;
	     }
         }
      }
  }
  else
  {
    // Incorrect type
    NDTL1(trace_log_key, NULL, TL_INFO, "Unknown connection type received - %d", con_type);
  }
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}


void closeAllConnectionsOfProcess(int appIndex, int procIndex) {
  cleanAndRemoveSocket(&NDProxyInstance[appIndex]->processInfo[procIndex].ctrlSocket, CLEAN);
  NDProxyInstance[appIndex]->processInfo[procIndex].ctrlFd = -1;
  //close data & auto sensor connection for any process if control connection is closed
  if(NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket) {
    cleanAndRemoveSocket(&NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket, CLEAN);
    NDProxyInstance[appIndex]->processInfo[procIndex].dcFd = -1;
  }
  if(NDProxyInstance[appIndex]->processInfo[procIndex].asSocket) {
    cleanAndRemoveSocket(&NDProxyInstance[appIndex]->processInfo[procIndex].asSocket, CLEAN);
    NDProxyInstance[appIndex]->processInfo[procIndex].asFd = -1;
  }
  NDProxyInstance[appIndex]->processInfo[procIndex].data_len = 0;
  FREE_AND_MAKE_NULL(NDProxyInstance[appIndex]->processInfo[procIndex].data, "Free'ing data buffer", -1);
  NDProxyInstance[appIndex]->processInfo[procIndex].processId = -1;
}


/*****************************************************************
 * This function takes a file name and the socket and simply 
 * transfer the contents of file on the socket.
 * **************************************************************/
void statAndSendFileContent(char* filename, PxAppEpollInfo_t** epollPtr, int instrProfileFlag)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  char error_buff[ERROR_BUFFER_SIZE];
  apr_finfo_t finfo;
  char *buf;
  char *file_buf = NULL;
  FILE* fp = NULL;
  //char file_path[MAX_READ_LENGTH] = "";
  char file_path[DOUBLE_BUFFER_LEN] = "";
  char* ptr = NULL;
  int appIndex = (int)(*epollPtr)->appIndex;
  int processIndex = (int)(*epollPtr)->processIndex;
  int fileSize = 0, len = 0;
  int rv = 0;
  char filePresent = 0;
  long fileTotalSize = 0;

  NDLB_MALLOC_STRING(buf, MAX_READ_LENGTH, "buf", -1);
  if (instrProfileFlag)
  {
#ifdef __unix__
    if ((ptr = strrchr(filename, '/')))
#else
      if ((ptr = strrchr(filename, '\\')))
#endif
      {
	ptr++;
	strcpy(filename, ptr);
      }
  }

#ifdef __unix__
  snprintf(file_path,DOUBLE_BUFFER_LEN, "%s/.cache/%s/%s", NDProxy.ndHomePath, NDProxyInstance[(*epollPtr)->appIndex]->agentName, filename);
#else
  snprintf(file_path, DOUBLE_BUFFER_LEN, "%s\\\.cache\\\%s\\\%s", NDProxy.ndHomePath, NDProxyInstance[(*epollPtr)->appIndex]->agentName, filename);
#endif

  filePresent=nd_stat(file_path, &finfo, NDProxy.proxy_pool);

  // Fix : 101561 
  if (!filePresent) 
  {
    if ((long)finfo.size > 0)
    {
      fileTotalSize = (long)finfo.size + 1024;
    }
    else
      fileTotalSize = MAX_READ_LENGTH;
    NDLB_MALLOC_STRING(file_buf, fileTotalSize, "file_buf", -1);
  }
  else
  {
    NDLB_MALLOC_STRING(file_buf, MAX_READ_LENGTH, "file_buf", -1);
    fileTotalSize = MAX_READ_LENGTH;
  }
  if (instrProfileFlag)
  {
    NDTL1(trace_log_key, NULL, "Agent->proxy", "nd_control_rep:action=instrumentation_profile;instrProfile=%s", filename);
    len = snprintf(file_buf, fileTotalSize, "nd_control_rep:action=instrumentation_profile;instrProfile=%s", filename);
    NDTL1(trace_log_key, NULL, "Agent->Proxy", "filelength =%d,file_buf is [%.200s]",strlen(file_buf)+1, file_buf);

  }
  else
    len = snprintf(file_buf, fileTotalSize, "nd_control_rep:action=file;keyword=%s", filename);


  if (!filePresent)
  {
    if (instrProfileFlag)
    {
      //   NDTL0(trace_log_key, NULL, TL_INFO, "size=%ld;lmd=%ld;\n", (long)finfo.size, (long)finfo.mtime);
      NDLB_SNPRINTF(file_buf,len,fileTotalSize,";size=%ld;lmd=%ld;\n", (long)finfo.size, (long)finfo.mtime);
      NDTL2(trace_log_key, NULL, TL_INFO, "file_buf is [%s]", file_buf + len);
    }
    else
    {
      NDLB_SNPRINTF(file_buf, len, fileTotalSize, "%%3Bsize%%3D%ld%%3Blmd%%3D%ld;\n", (long)finfo.size, (long)finfo.mtime);
    }
    fp = fopen(file_path, "rb");
    if (fp)
    {
      while (fgets(buf, MAX_READ_LENGTH, fp))
      {
	       NDLB_SNPRINTF(file_buf , len, fileTotalSize, "%s", buf);
         if(len >= fileTotalSize) {
            break;
          }
      }

      // Read done, close the file.
      fclose(fp);
      fp = NULL;

      fileSize = strlen(file_buf);
      if (ndApplication->proxy_lws_active == 1)
      {
	rv = send_on_wsi(NDProxy.lws_handler.context, &NDProxyInstance[appIndex]->processInfo[processIndex].ctrlLwsHandler, file_buf, fileSize);
	if (rv == 0) 
	{
	  NDTL3(trace_log_key, NULL, TL_INFO, "Sent file - %s and filesize is [%d]", file_buf, strlen(file_buf));
	}
	else 
	{
	  NDTL0(trace_log_key, NULL, TL_ERROR, "Send Failed, Control Connection Closed. Unable to send file - %s",
	      file_path);

	  if (!NDProxyInstance[appIndex]->processInfo[processIndex].dcSocket &&
	      !NDProxyInstance[appIndex]->processInfo[processIndex].asSocket)
	    NDProxyInstance[appIndex]->processInfo[processIndex].processId = -1;

	  NDProxyInstance[appIndex]->activeProcessCount--;
	}
      }
      else 
      {
	rv = ndlb_send_n_bytes(NDProxyInstance[appIndex]->processInfo[processIndex].ctrlSocket,
	    file_buf, fileSize, error_buff, ERROR_BUFFER_SIZE);
	if (rv == fileSize)
	{
	  NDTL3(trace_log_key, NULL, TL_INFO, "Sent file - %s and filesize is [%d]", file_buf, strlen(file_buf));
	}
	else if (rv == NDLB_CONNECTION_CLOSE)
	{
	  NDTL0(trace_log_key, NULL, TL_ERROR, "Send Failed, Control Connection Closed. Unable to send file - %s, connection fd = %d",
	      file_path, NDProxyInstance[appIndex]->processInfo[processIndex].ctrlFd);
      closeAllConnectionsOfProcess(appIndex, processIndex);
	  NDProxyInstance[appIndex]->activeProcessCount--;
	}
      }            
    }
  }
  else
  {
    int statBuffSize = 0;
    if (instrProfileFlag)
    {
      NDTL3(trace_log_key, NULL, TL_ERROR, "Send Failed and file path is %s", file_path);
      snprintf(buf, MAX_READ_LENGTH, "nd_control_rep:action=FailedInstrumentationProfile - %s\n", file_path);
      NDTL3(trace_log_key, NULL, TL_ERROR, "buf is %s and size is %d", buf, statBuffSize);
    }
    else
    {
      NDTL1(trace_log_key, NULL, TL_ERROR, "Send Failed and file path is %s", file_path);
      snprintf(buf, MAX_READ_LENGTH, "nd_control_rep:action=Failed%s - %s\n",filename, file_path);
      // NDTL0(trace_log_key, NULL, TL_ERROR, "buf is %s and size is %d", buf, statBuffSize);
    }
    statBuffSize = strlen(buf);
    if(ndApplication->proxy_lws_active ==1){
      send_on_wsi(NDProxy.lws_handler.context, &NDProxyInstance[appIndex]->processInfo[processIndex].ctrlLwsHandler, buf, statBuffSize);
    }else{
      rv = ndlb_send_n_bytes(NDProxyInstance[appIndex]->processInfo[processIndex].ctrlSocket, buf, statBuffSize, error_buff, ERROR_BUFFER_SIZE);

      if (rv == statBuffSize)
      {
	NDTL3(trace_log_key, NULL, TL_INFO, "Sent buffer - %s and buffer size is [%d]", buf, statBuffSize);
      }
      else if (rv == NDLB_CONNECTION_CLOSE)
      {
	NDTL0(trace_log_key, NULL, TL_ERROR, "Send Failed, Control Connection Closed. Unable to send buffer - %s, connection fd = %d",
	    buf, NDProxyInstance[appIndex]->processInfo[processIndex].ctrlFd);
    closeAllConnectionsOfProcess(appIndex, processIndex);
	NDProxyInstance[appIndex]->activeProcessCount--;
      }
    }
  }
  FREE_AND_MAKE_NULL(buf, "buf", -1);
  FREE_AND_MAKE_NULL(file_buf, "file_buf", -1);
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}

/******************************************************************
 * This function reads the first message on a new connection and
 * then updates the socket fd of that application with correct
 * type of connection.
 * ***************************************************************/
void processReceivedMsg(apr_socket_t* instanceSkt, char* bufInfoLocal, PxAppEpollInfo_t** epollPtr, struct lws* wsi)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called with bufInfoLocal = [%s]", bufInfoLocal);
  int processIdx, instIdx;
  int port;
  //int rv = 0;
  char ip[1024];
  //get_dst_addr(instanceSkt, ip, 128, &port);
  get_dst_addr_wrapper(instanceSkt, wsi, ip, 128, &port, ndApplication->proxy_lws_active);
  if (*epollPtr)
  {
    /*char ad_str[100];
    ad_str[0]='\0';
    int ad_len=0;
    if(ndApplication->NDBCISetting.connId>0){
      ad_len=snprintf(ad_str, 100, "ndc_rep:action=discover_loaded_classes;Id=%d;result=Success;", ndApplication->NDBCISetting.connId);
    }*/
    processIdx = (int)(*epollPtr)->processIndex;
    instIdx = (int)(*epollPtr)->appIndex;
    if (instanceSkt == NDProxyInstance[instIdx]->processInfo[processIdx].ctrlSocket ||
	wsi == NDProxyInstance[instIdx]->processInfo[processIdx].ctrlLwsHandler.wsi)
    {
      NDTL4(trace_log_key, NULL, "CLIENT->CTRL(Process)", "Ctrl Data received on '%s:%d' - '%s'", ip, port, bufInfoLocal);
      //processCtrlMsgFromClient(instanceSkt, bufInfoLocal, );
      if (!strncmp(bufInfoLocal, "nd_ctrl_msg_rep:result", strlen("nd_ctrl_msg_rep:result")))
      {
	NDTL0(trace_log_key, NULL, TL_INFO, "'%s' received for pid = %d, '%s:%d'",
	    bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].processId, ip, port);
      }
      /*  else if(!strncmp(bufInfoLocal,"nd_ctrl_tag_req:",strlen("nd_ctrl_tag_req:")))
      {

          char str[2048];
          char* tag_for_110 = NULL;
          char* attr_for_111 = NULL;
          int x=0;
          NDTL1(trace_log_key, NULL, "CLIENT->CTRL(Process)", "Ctrl Data received on '%s:%d' - '%s'", ip, port, bufInfoLocal);
          sprintf(str,"%s",bufInfoLocal);
           tag_for_110 = strstr(str, "tag_for_110:");
            if(tag_for_110 != NULL)
            {
              *tag_for_110 = '\0';
              tag_for_110 += strlen("tag_for_110:");
              attr_for_111 = strstr(tag_for_110, "attr_for_111:");
              if(attr_for_111 != NULL)
              {
                *attr_for_111 = '\0';
                attr_for_111 += strlen("attr_for_111:");
              }
                x=1;
              }
              if(x != 1){
                attr_for_111 = strstr(str, "attr_for_111:");
                if(attr_for_111 != NULL)
                {
                  *attr_for_111 = '\0';
                    attr_for_111 += strlen("attr_for_111:");
                }
            }

          //  ndApplication->tagnew= str;
            if(!ndApplication->NDBCISetting.tagnew || strcmp(ndApplication->NDBCISetting.tagnew,str))
            {
             FREE_AND_MAKE_NULL(ndApplication->NDBCISetting.tagnew, "Free'ing ndApplication->NDBCISetting.tagnew", -1);
              NDLB_MALLOC_AND_COPY(str,ndApplication->NDBCISetting.tagnew, strlen(str) + 1, "orig_message", -1);
             sprintf(str + strlen(str), "\n");
          if(ndApplication->control.skt)
          {
          rv = ndlb_send_n_bytes(ndApplication->control.skt,str,strlen(str),error_buf,ERROR_BUFFER_SIZE);
          if(rv >0)
          {
          NDTL1(trace_log_key, NULL, "PROXY->INSTANCE", "Sent '%s' '%s:%d'",str , ip, port);
          }
          else if( rv == NDLB_CONNECTION_CLOSE)
          {
            NDTL1(trace_log_key, NULL, TL_ERROR, "Connection Broken. Unable to send '%s",str);
          }
          }
        }
       if(tag_for_110 != NULL)
        {
          NDTL4(trace_log_key, NULL, TL_ERROR, "Connection Broken. Unable to send '%s %s ",tag_for_110,ndApplication->NDBCISetting.json_tag);
          if(!ndApplication->NDBCISetting.json_tag || strcmp(ndApplication->NDBCISetting.json_tag,tag_for_110))
          {
            FREE_AND_MAKE_NULL(ndApplication->NDBCISetting.json_tag, "Free'ing tag", -1);
              NDLB_MALLOC_AND_COPY(tag_for_110, ndApplication->NDBCISetting.json_tag, strlen(tag_for_110) + 1, "orig_message", -1);
          NDTL1(trace_log_key, NULL, TL_INFO, "tag_for_110!=NULL ,  %s",tag_for_110);
          tag_record();
          }

        }
        if(attr_for_111 != NULL)
        {
                    NDTL4(trace_log_key, NULL, TL_ERROR, "Connection Broken. Unable to send '%s %s ",attr_for_111,ndApplication->NDBCISetting.tierAttr);
            if(!ndApplication->NDBCISetting.tierAttr || strcmp(ndApplication->NDBCISetting.tierAttr,attr_for_111))
          {
            FREE_AND_MAKE_NULL(ndApplication->NDBCISetting.tierAttr, "Free'ing attr", -1);
            NDLB_MALLOC_AND_COPY(attr_for_111, ndApplication->NDBCISetting.tierAttr, strlen(attr_for_111) + 1, "orig_message", -1);
          NDTL1(trace_log_key, NULL, TL_INFO, "attr_for_111!=NULL ,%s",attr_for_111);
          attr_record();
        }
        }
      
      }*/
      else if (!strncmp(bufInfoLocal, "nd_control_req:action=send_file;keyword=", 
	    strlen("nd_control_req:action=send_file;keyword=")))
      {
	NDTL1(trace_log_key, NULL, TL_INFO, "'%s' received for pid = %d, '%s:%d'",
	    bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].processId, ip, port);
	//Check whether the file is present with Proxy Server.
	char *local_ptr = NULL;
	char filename[512] = "";
	if ((local_ptr = strstr(bufInfoLocal, "keyword=")))
	{
	  local_ptr += strlen("keyword=");
	  char *local_ptr2;
	  if ((local_ptr2 = strchr(local_ptr, ';')))
	  {
	    *local_ptr2 = 0;
	  }
	  strcpy(filename, local_ptr);
	  statAndSendFileContent(filename, epollPtr, 0);
	}

      }
      else if (!strncmp(bufInfoLocal, "nd_control_req:action=send_instrumentation_profile;", 
	    strlen("nd_control_req:action=send_instrumentation_profile;")))
      {
	NDTL2(trace_log_key, NULL, TL_INFO, "'%s' received for pid = %d, '%s:%d'",
	    bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].processId, ip, port);
	//Check whether the file is present with Proxy Server.
	char *local_ptr = NULL;
	char filename[1024] = "";
	char *local_ptr2 = NULL;
	if ((local_ptr = strstr(bufInfoLocal, "instrProfile=")))
	{
	  local_ptr += strlen("instrProfile=");
	  if ((local_ptr2 = strchr(local_ptr, '\n')))
	  {
	    *local_ptr2 = 0;
	  }
	  strcpy(filename, local_ptr);
	  statAndSendFileContent(filename, epollPtr, 1);
	}
      }
      /*else if (!strncmp(bufInfoLocal, "nd_meta_data_req:action=send_meta_data;", 
	    strlen("nd_meta_data_req:action=send_meta_data;")))
      {
	NDTL1(trace_log_key, NULL, TL_INFO, "'%s' received for pid = %d, '%s:%d'",
	    bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].processId, ip, port);
	processAgentMessage(bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].ctrlSocket,
	    &NDProxyInstance[instIdx]->processInfo[processIdx].ctrlLwsHandler);
      }
      else if (!strncmp(bufInfoLocal, "nd_control_rep:action=start_instrumentation;status=starting;result=Ok", 
	    strlen("nd_control_rep:action=start_instrumentation;status=starting;result=Ok")))
      {
	NDTL1(trace_log_key, NULL, TL_INFO, "'%s' received for pid = %d, '%s:%d'",
	    bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].processId, ip, port);
      }
      else if (!strncmp(bufInfoLocal, "nd_control_rep:action=modify;result=Ok;",
	    strlen("nd_control_rep:action=modify;result=Ok;")))
      {
	NDTL1(trace_log_key, NULL, TL_INFO, "'%s' received for pid = %d, '%s:%d'",
	    bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].processId, ip, port);
      }
      else if (!strncmp(bufInfoLocal, "nd_control_rep:action=stop_instrumentation;status=stopping;result=Ok;",
	    strlen("nd_control_rep:action=stop_instrumentation;status=stopping;result=Ok;")))
      {
	NDTL1(trace_log_key, NULL, TL_INFO, "'%s' received for pid = %d, '%s:%d'",
	    bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].processId, ip, port);
      }
      else if(ad_len > 0 && !strncmp(bufInfoLocal, ad_str,ad_len))
      {
            NDTL1(trace_log_key, NULL, TL_INFO, "'%s' received for pid = %d, '%s:%d'",
            bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].processId, ip, port);
            NDTL0(trace_log_key, NULL, "AGENT-LOG", "\n<agentLogs>\n%s\n</agentLogs>", bufInfoLocal);
	    if (ndApplication->lws_active == 1) {
      		send_control_data_via_lws_socket(bufInfoLocal, strlen(bufInfoLocal));
    	    }
    	    else {
    		ndlb_send_n_bytes(ndApplication->control.skt, bufInfoLocal,  strlen(bufInfoLocal), error_buf, ERROR_BUFFER_SIZE);
    	    }
      }*/
      else if(NDProxy.proxyServerFlag && (ndApplication->temp_lws_handler.wsi || disSkt))
      {
	      if((!strstr(bufInfoLocal, "totalTimeTaken=")))
	      { 
	  	    send_discovery_data_to_ndc(bufInfoLocal, strlen(bufInfoLocal), error_buf, ERROR_BUFFER_SIZE);
		    //  NDTL1(trace_log_key, NULL, TL_INFO, "send data = %s ",bufInfoLocal);
		    send_discovery_data_to_ndc("\n", 1, error_buf, ERROR_BUFFER_SIZE);
	      }
	      else
	      {
		   send_discovery_data_to_ndc(bufInfoLocal, strlen(bufInfoLocal), error_buf, ERROR_BUFFER_SIZE);
		   send_discovery_data_to_ndc("\n", 1, error_buf, ERROR_BUFFER_SIZE);
		   NDTL0(trace_log_key, NULL, TL_INFO, "Sending the method and class counters. %s",bufInfoLocal);
		   ndApplication->NDBCISetting.connId=0;
		   if(disSkt)
		   {
			 cleanAndRemoveSocket(&disSkt, CLEAN);
                         //NDProxyInstance[appIndex]->processInfo[procIndex].disFd = -1;
			 disSkt = NULL;
		  }
                  else if(ndApplication->temp_lws_handler.wsi)
                  {
                     clean_discovery_conn();
                  }
	      }
        }
        else if(!strncmp(bufInfoLocal,"auto_sensor_GCDump:",strlen("auto_sensor_GCDump:")))
        {
          NDTL4(trace_log_key, NULL, TL_INFO, "%s   %d \n", bufInfoLocal,strlen(bufInfoLocal));
           dumpAndSendDataToNDC(bufInfoLocal+strlen("auto_sensor_GCDump:"),strlen(bufInfoLocal)-strlen("auto_sensor_GCDump:"), ND_TRUE);
           dumpAndSendDataToNDC("\n",strlen("\n"),ND_TRUE);
         // parseProxyMsgReq(instanceSkt, bufInfoLocal, epollPtr, 3, wsi);
        }
        else if(!strncmp(bufInfoLocal,"thread_dump:",strlen("thread_dump:"))) {
          int len1 = strlen(bufInfoLocal);
          for(int i = 0; i < len1; i++) {
            if(bufInfoLocal[i] == '$')
              bufInfoLocal[i] = '\n';
          }
          sendThreadDataToNDC(NULL,bufInfoLocal+strlen("thread_dump:"),NULL,BUFFER_LEN*100);
        }
        // else if(!strncmp(bufInfoLocal,"auto_sensor_GoRoutineDump:",strlen("auto_sensor_GoRoutineDump::")))
        // {
        //   NDTL4(trace_log_key, NULL, TL_INFO, "%s   %d \n", bufInfoLocal,strlen(bufInfoLocal));
        //    dumpAndSendDataToNDC(bufInfoLocal+strlen("auto_sensor_GoRoutineDump::"),strlen(bufInfoLocal)-strlen("auto_sensor_GoRoutineDump::"), ND_TRUE);
        //    dumpAndSendDataToNDC("\n",strlen("\n"),ND_TRUE);
        // }
        else
        {
            NDTL1(trace_log_key, NULL, TL_INFO, "'%s' received for pid = %d, '%s:%d'",
            bufInfoLocal, NDProxyInstance[instIdx]->processInfo[processIdx].processId, ip, port);
            NDTL1(trace_log_key, NULL, "AGENT-LOG", "\n<agentLogs>\n%s\n</agentLogs>", bufInfoLocal);
        }

      } 
  }
  else
  {
    if (!strncmp(bufInfoLocal, "nd_ctrl_msg_req:", strlen("nd_ctrl_msg_req:")))
    {
      NDTL1(trace_log_key, NULL, TL_INFO, "%s ... '%s:%d,%s'", bufInfoLocal, ip, port, ndApplication->NDBCISetting.appName);
      parseProxyMsgReq(instanceSkt, bufInfoLocal + strlen("nd_ctrl_msg_req:"), epollPtr, 1, wsi);
    }
    else if (!strncmp(bufInfoLocal, "nd_data_msg_req:", strlen("nd_data_msg_req:")))
    {
      // This message would be the data connection  request from the client application/process
      NDTL2(trace_log_key, NULL, TL_INFO, "%s ... '%s:%d'", bufInfoLocal, ip, port);
      parseProxyMsgReq(instanceSkt, bufInfoLocal + strlen("nd_data_msg_req:"), epollPtr, 2, wsi);
    }
    else if (!strncmp(bufInfoLocal, "auto_sensor_thread_hotspot_data_req:", strlen("auto_sensor_thread_hotspot_data_req:")))
    {
      // This message would be the auto sensor  connection  request from the client application/process
      NDTL2(trace_log_key, NULL, TL_INFO, "%s ... '%s:%d'", bufInfoLocal, ip, port);
      parseProxyMsgReq(instanceSkt, bufInfoLocal + strlen("auto_sensor_thread_hotspot_data_req:"), epollPtr, 3, wsi);
    }
    else if(!strncmp(bufInfoLocal, "agent_post_data_req:", strlen("agent_post_data_req:"))) 
    {
	    char tmp[1024];
	    ndc_servers_t* ndc = get_active_ndc();
	    if (ActiveAgentProtocolIndex == SupportedProtocolTCP) {
		ndc->ndc_port = ndApplication->NDBCISetting.ndcPort;
	    }else if (ActiveAgentProtocolIndex == SupportedProtocolWS) {
		ndc->ndc_port = ndApplication->NDBCISetting.ndc_WS_Port;
	    }else if (ActiveAgentProtocolIndex == SupportedProtocolWSS) {
		ndc->ndc_port = ndApplication->NDBCISetting.ndc_WSS_Port;
	    }
	    else {
		 NDTL2(trace_log_key, NULL, TL_INFO, "Unknown protocol");
	   }

	    snprintf(tmp,1024,"%s",strstr(bufInfoLocal,"discover_loaded_classes"));
	    parseProxyMsgReq(instanceSkt, bufInfoLocal + strlen("agent_post_data_req:"), epollPtr, 4, wsi);
	    char msg[2*1024];
	    int len;
	    if(!disSkt && !ndApplication->temp_lws_handler.wsi)
	    {
	       create_discovery_conn_with_ndc(ndApplication, ndApplication->NDBCISetting.ndcHost, ndc->ndc_port,error_buf, ERROR_BUFFER_SIZE, control_pool);
	    }
	    len = snprintf(msg,2048, "agent_post_data_req:action=discover_loaded_classes;Id=%d;Tier=%s;Server=%s;Instance=%s;\n",ndApplication->NDBCISetting.connId,ndApplication->NDBCISetting.tierName,ndApplication->NDBCISetting.appServerName,ndApplication->NDBCISetting.appName);
	    if(disSkt || ndApplication->temp_lws_handler.wsi)
	    {
		    NDTL0(trace_log_key, NULL, TL_INFO, "Discovery socket was created successfully");
		    send_discovery_data_to_ndc(msg, len, error_buf, ERROR_BUFFER_SIZE);
	    }
	    else
	    {
		    NDTL1(trace_log_key, NULL, TL_INFO, "Discovery socket was not created connextion error");
	    }
    }
  }

  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}

/*******************************************************************
 * This function reads the data from an application/process
 * and parses it to fill data structure  for that application.
 * ****************************************************************/
int readDataFromSocket(apr_socket_t *instanceSkt, controlSocketBuffer_t *bufInfoLocal, int appIndex, int processIndex)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  apr_status_t rv;
  int prevByteCount = 0;

  // if partial data is present from the previous read on the same connection then prepending it to the current buffer.
  /*
     if (NDProxyInstance[appIndex]->processInfo[processIndex].data_len)
     {
  //NDTL0(trace_log_key, NULL, TL_ERROR, "PREVIOUS DATA LEN = %d(appIndex = %d, procIndex = %d)", NDProxyInstance[appIndex]->processInfo[processIndex].data_len, appIndex, processIndex);

  memcpy(bufInfoLocal->read_socket_buf, NDProxyInstance[appIndex]->processInfo[processIndex].data,
  NDProxyInstance[appIndex]->processInfo[processIndex].data_len);
  bufInfoLocal->writeOffset = NDProxyInstance[appIndex]->processInfo[processIndex].data_len;
  prevByteCount = NDProxyInstance[appIndex]->processInfo[processIndex].data_len;
  }
  */
  if (ndApplication->proxy_lws_active == 1) {
    rv = bufInfoLocal->readBytes;
  }
  else {
    setResetSocketBuffer(bufInfoLocal);
    rv = ndlb_read_bytes(instanceSkt, bufInfoLocal->read_socket_buf + bufInfoLocal->writeOffset,
	bufInfoLocal->read_len - bufInfoLocal->writeOffset, error_buf, ERROR_BUFFER_SIZE);
  }

  if(ndApplication->lws_active ==0 && (!ndApplication->control.skt || !ndApplication->dataSkt.skt)) // If Data/AS connection with NDC is closed (case of DR), then close the connection
  {
    if(ndApplication->dataSkt.skt==NULL)
    {
      NDTL0(trace_log_key, NULL, TL_ERROR, "Data connection with NDC is not created hence ignoring the data.");
    }
    else
    {
      NDTL0(trace_log_key, NULL, TL_ERROR, "Data connection with NDC is closed. Ignoring the data.");
    }
    return NDLB_CONNECTION_CLOSE;
  }
  else if((ndApplication->lws_active ==1 && NDProxy.proxyServerFlag) &&(!ndApplication->control.lws_handler.wsi || !ndApplication->dataSkt.lws_handler.wsi))
  {
    NDTL1(trace_log_key, NULL, TL_ERROR, "Data connection with NDC is closed. Ignoring the data.");
    return NDLB_CONNECTION_CLOSE;
  }


  if (rv == NDLB_PARTIAL_LINE)
  {
    NDTL0(trace_log_key, NULL, TL_ERROR, "Read partial message '%s', rv = %d (appIndex = %d, procIndex = %d)",
	bufInfoLocal->read_socket_buf, rv, appIndex, processIndex);
    return NDLB_PARTIAL_LINE;
  }
  else if (rv == NDLB_CONNECTION_CLOSE)
  {

    //NDTL0(trace_log_key, NULL, TL_INFO, "Data Connection buffer:%s", bufInfoLocal->read_socket_buf);
    NDTL3(trace_log_key, NULL, TL_ERROR, "Data Connection Closed, rv = %d (appIndex = %d, procIndex = %d)",
	rv, appIndex, processIndex);
    return NDLB_CONNECTION_CLOSE;
  }
  else
  {
    NDTL3(trace_log_key, NULL, TL_INFO, "Msg get at encode side:%s,and read byte:%d", bufInfoLocal->read_socket_buf, rv + prevByteCount);
    decodeMessage(bufInfoLocal->read_socket_buf, rv + prevByteCount, processIndex,0,NULL);
  }

  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
  return APR_SUCCESS;
}

/*******************************************************************
 * This function reads the first message from an application/process
 * and parses it to fill data structure  for that application.
 * NDProxyInstance is the structure
 * ****************************************************************/


int processdatamsg(apr_socket_t* instanceSkt, controlSocketBuffer_t* bufInfoLocal, PxAppEpollInfo_t** epollPtr, struct lws* wsi)
  {
    NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");

    char *ptr = NULL;
    int dataMsgFlag = 0;

    if (!(strchr(bufInfoLocal->read_socket_buf + bufInfoLocal->readOffset, '\n')))
    {
      NDTL1(trace_log_key, NULL, TL_ERROR, "Invalid Message format(Message not terminiated with newline). Closing the connection.");
      return NDLB_CONNECTION_CLOSE;
    }

    while ((ptr = strchr(bufInfoLocal->read_socket_buf + bufInfoLocal->readOffset, '\n')))
    {
      // SPECIAL CASE HANDLING: If the buffer is received on Data socket and has some data along with nd_data_msg_req
      // ---------------------  then since this data will be encoded, so we can not perform string operations on 
      //                        it and has to be processed altogether.
      if (*(ptr + 1) == '^')
      {
	dataMsgFlag = 1;
      }

      *ptr = '\0';
      int line_size = (int)(ptr - (bufInfoLocal->read_socket_buf + bufInfoLocal->readOffset)) + 1;

      if (line_size <= 1)
      {
	bufInfoLocal->readOffset += line_size;
	bufInfoLocal->readBytes -= line_size;

	continue;  // '\0' found at the start
      }
      processReceivedMsg(instanceSkt, bufInfoLocal->read_socket_buf + bufInfoLocal->readOffset, epollPtr, wsi);
      bufInfoLocal->readOffset += line_size;
      bufInfoLocal->readBytes -= line_size;

      // Process the data message part, if exists in the message.
      if (dataMsgFlag && *epollPtr)
      {
	//int appIndex = (int)(*epollPtr)->appIndex;
	int processIndex = (int)(*epollPtr)->processIndex;
	decodeMessage(bufInfoLocal->read_socket_buf + bufInfoLocal->readOffset, bufInfoLocal->readBytes, processIndex,0,NULL);
	setResetSocketBuffer(bufInfoLocal);
	break;
      }

      if (bufInfoLocal->readBytes <= 0)
      {
	bufInfoLocal->read_socket_buf[0] = '\0';
	bufInfoLocal->readOffset = 0;
	bufInfoLocal->writeOffset = 0;
	bufInfoLocal->readBytes = 0;
      }
    }
    if (bufInfoLocal->readOffset != 0)
    {
      memmove(bufInfoLocal->read_socket_buf, bufInfoLocal->read_socket_buf + bufInfoLocal->readOffset, bufInfoLocal->readBytes);
      bufInfoLocal->readOffset = 0;
      bufInfoLocal->writeOffset = bufInfoLocal->readBytes;
    }
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
    return APR_SUCCESS;
  }

int readMessageFromSocket(apr_socket_t *instanceSkt, controlSocketBuffer_t *bufInfoLocal, PxAppEpollInfo_t **epollPtr, struct lws* wsi)  {
    NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  apr_status_t rv;
  int i, j;

while(1) {
  //for AD, the proxy server should read all the data from the socket. For other sockets, it should read only one time
  if (ndApplication->proxy_lws_active == 0) {
    setResetSocketBuffer(bufInfoLocal);
    rv = ndlb_read_bytes(instanceSkt, bufInfoLocal->read_socket_buf + bufInfoLocal->writeOffset,
  bufInfoLocal->read_len - bufInfoLocal->writeOffset, error_buf, ERROR_BUFFER_SIZE);
  }
  else {
    rv = 0;
  }
  if (rv == NDLB_CONNECTION_CLOSE)
  {
    if (*epollPtr)
    {
      NDTL1(trace_log_key, NULL, TL_ERROR, "Connection closed, rv = %d (appIndex = %d, procIndex = %d).",
    rv, (int)(*epollPtr)->appIndex, (int)(*epollPtr)->processIndex);
    }
    else
      NDTL1(trace_log_key, NULL, TL_ERROR, "Connection closed, rv = %d.", rv);
    return NDLB_CONNECTION_CLOSE;
  }
  else {
    bufInfoLocal->readBytes += rv;
    bufInfoLocal->writeOffset += rv;
    rv = processdatamsg(instanceSkt, bufInfoLocal, epollPtr, wsi);
    if(rv != APR_SUCCESS)
      return rv;
  }

  if ((*epollPtr)!= NULL)
  {
      i=(*epollPtr)->appIndex;
      j=(*epollPtr)->processIndex;
      if (instanceSkt != NDProxyInstance[i]->processInfo[j].disSocket)
      {
          break;
      }
  }
  else
    break;

  NDTL3(trace_log_key, NULL, TL_INFO, "Method Called for AD ");
}
  return APR_SUCCESS;
}


/*******************************************************************
 * This function accepts a new connection on main thread from
 * any application/process only for the first time and then 
 * populates its values in the structure
 * It also add the socketfd in the epoll for that connection
 * ****************************************************************/
void acceptNewConnection(controlSocketBuffer_t *bufInfoLocal)
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  apr_socket_t *instanceSkt = NULL;
  int port;
  char ip[128];
  apr_status_t rv;
  PxAppEpollInfo_t *epollPtr = NULL;
  int appIndex = -1, procIndex = -1;

  rv = apr_socket_accept(&instanceSkt, NDProxy.skt, NDProxy.proxy_pool);

  if(rv != APR_SUCCESS)
  {
    apr_strerror(rv, error_buf, ERROR_BUFFER_SIZE);
    NDTL1(trace_log_key, NULL, TL_ERROR, "Error in accepting connection : rv = %d , Error msg - '%s'", rv, error_buf);
    return;
  }

  get_dst_addr_wrapper(instanceSkt, NULL, ip, 128, &port, ndApplication->proxy_lws_active);

  NDTL3(trace_log_key, NULL, TL_INFO, "Connection received from '%s:%d' ", ip, port);

  // Read the message on the connection
  rv = readMessageFromSocket(instanceSkt, bufInfoLocal, &epollPtr, NULL);

  if(epollPtr)
  {
    appIndex = epollPtr->appIndex;
    procIndex = epollPtr->processIndex;
  }

  if(rv != APR_SUCCESS )
  {
    if(epollPtr)
    {
      NDTL1(trace_log_key, NULL, TL_ERROR, "Read Failed due to connection closed [pid = %d] ['%s':'%d'].", 
	  NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
    }
    else
      NDTL1(trace_log_key, NULL, TL_ERROR, "Read Failed due to connection closed ['%s':'%d'].", ip, port);
    if(instanceSkt)
      cleanAndRemoveSocket(&instanceSkt, NOCLEAN);
    instanceSkt = NULL;
  }

  FREE_AND_MAKE_NULL(epollPtr, "Free'ing epollPtr", -1);

  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
  return;
}

#ifdef __unix__
void sighup_handler(int sig){
  NDTL0(trace_log_key, NULL, TL_INFO, "SIGHUP received, ignoring...");
}

void sigterm_handler(int sig){
  NDTL0(trace_log_key, NULL, TL_INFO, "SIGTERM received, ignoring...");
}

// void sigsegv_handler(int sig) {
//   void *array[64];
//   char stack[BUF_4K] = "SIGSEGV received\n";
//   size_t size;
//   char **str;
//   char *serverlist= NULL;
//   int i = 0;
//   int len = 0;

//   if(!NDProxy.proxyServerFlag) {
//     if(ndApplication->control.skt) apr_socket_close(ndApplication->control.skt);
//       ndApplication->control.skt = NULL;
//     if(ndApplication->ndcDataSocket) apr_socket_close(ndApplication->ndcDataSocket);
//       ndApplication->ndcDataSocket = NULL;
//     if(ndApplication->ndcASSocket) apr_socket_close(ndApplication->ndcASSocket);
//       ndApplication->ndcASSocket = NULL;
//   }
//   // get void*'s for all entries on the stack
//   size = backtrace(array, 64);
//   // print out all the frames to stderr
//   NDTL0(trace_log_key, NULL, TL_INFO, "Signal Received %d,%d",NDProxy.numInstances, sig);
//   if(NDProxy.proxyServerFlag)
//   {
//     NDLB_MALLOC_AND_MEMSET(serverlist, NDProxy.numInstances*1024, "serverlist", -1);
//     int i;
//     for (i = 0; i < NDProxy.numInstances; i++)
//     {
//       if(NDProxyInstance[i]->appServerName) {
//         strcat(serverlist, NDProxyInstance[i]->appServerName);
//         strcat(serverlist, "   ");
//       }
//     }

//     NDTL0(trace_log_key, NULL, TL_INFO, "Total number of applications connected with Proxy are : %s", serverlist);
//   }
//   FREE_AND_MAKE_NULL(serverlist,"serverlist",-1);
//   NDTL0(trace_log_key, NULL, TL_INFO, "Signal Received2 %d",size);
//   memset(stack,0,BUF_4K);
//   //backtrace_symbols_fd(array, size, STDERR_FILENO);
//   str = backtrace_symbols(array, size);
//   for (i = 0; i < size; i++)
//   {
//     len = strlen(stack);
//     snprintf(stack+len,BUF_4K-len,"%s\n", (char *)str[i]);
//   }
//   NDTL0(trace_log_key, NULL, TL_INFO, "\n%s\n(possible core dumped)", stack);

//   //Facilitating for core dump as per default behaviour
//   apr_signal(sig, SIG_DFL);
//   kill(getpid(),SIGSEGV);
// }
#endif


int udp_buff_list_init()
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method called");
  int i;
  int maxudpBuffers = NDProxy.num_udpThreads*4;

  rudp_packet_wrapper* node = NULL;
  rudp_packet_wrapper* curr_node = NULL;
  NDProxy.udp_writeList = NULL;
  NDProxy.udp_bufferWriteLock = NULL;

  apr_thread_mutex_create(&NDProxy.udp_bufferWriteLock, APR_THREAD_MUTEX_NESTED, NDProxy.proxy_pool);
  if (!NDProxy.udp_bufferWriteLock) {
    NDTL0(trace_log_key, NULL, TL_INFO, "Failed to create lock for UDP protocol");
    return 1;
  }

  char msg[512];
  for (i = 0; i < maxudpBuffers; i++)
  {
    snprintf(msg, 512, "udp Data buffer node : %d", i);
    NDLB_MALLOC_AND_MEMSET(node, sizeof(rudp_packet_wrapper), msg, -1);
    if (!node)
    {
        NDTL0(trace_log_key, NULL, TL_INFO, "Failed to allocate udp packet");

        return 1;

    }
    //snprintf(msg, 512, "udp Data buffer chunk : %d", i);

    //NDLB_MALLOC_AND_MEMSET(node->received_packet, sizeof(struct rudp_packet), msg, -1);
    //if (!node->received_packet) {

    //    NDTL0(trace_log_key, NULL, TL_INFO, "Failed to allocate received_packet");

    //    return 1;

    //}
   /* snprintf(msg, 512, "udp Data buffer chunk : %d", i);
    NDLB_MALLOC_STRING(node->received_packet->payload, 65000, msg, i);
    if (!node->received_packet->payload) {
        NDTL0(trace_log_key, NULL, TL_INFO, "Failed to allocate buffer for payload");
        return 1;
    }*/

    //memset(node->received_packet.payload, 0, sizeof(node->received_packet.payload)); 
 //   node->next = NULL;
  //  node->received_packet->payload_length= 65000;
    //node->received_packet.header.version=0;
    //node->received_packet.header.seqno=0;
    //node->received_packet.header.type=0;

    if (!NDProxy.udp_writeList)
    {
      NDProxy.udp_writeList = node;
      curr_node = node;
    }
    else
    {
      curr_node->next = node;
      curr_node = node;
    }
    NDProxy.udp_writeListLast = node;
  }

  NDTL4(trace_log_key, NULL, TL_INFO, "Method exit");
  return 0;
}

int get_ws_param_values(char *buf, int readbytes, wrapheader_t *wrapheader) {
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  char *ptr = buf;//write comments here
  if(readbytes < UDP_PACKET_MIN_SIZE || !ptr || *ptr != '^')
    return 0;
  ptr++;
  wrapheadervar_t* wrapheadervar = (wrapheadervar_t*)&ptr[0];
  if(wrapheadervar->appId != -1) {
    wrapheader->wrapheadervar.whLen = wrapheadervar->whLen;
    wrapheader->wrapheadervar.fpId = wrapheadervar->fpId;
    wrapheader->wrapheadervar.appId = wrapheadervar->appId;
    wrapheader->wrapheadervar.protocolType = wrapheadervar->protocolType;
    wrapheader->wrapheadervar.messageType = wrapheadervar->messageType;
    wrapheader->wrapheadervar.agentType = wrapheadervar->agentType;
    return 1;
  }
  int len = WRAP_HEADER_LEN;
  int remainingBytes = readbytes - 1;
  //wrapheader->wrapheadervar = *(wrapheadervar_t *)(unsigned char*)ptr;
  memcpy(&wrapheader->wrapheadervar, (wrapheadervar_t *)(unsigned char*)ptr, len);
  ptr += len;
  remainingBytes -= len;
  NDTL2(trace_log_key, NULL, TL_INFO, "wrapheader total length = %d whLen = %d, agentType = %d, apiReqLen = %d awsReqLen = %d funcNameLen = %d tagslength = %d messageType = %d", readbytes,wrapheader->wrapheadervar.whLen,
     wrapheader->wrapheadervar.agentType, wrapheader->wrapheadervar.apiReqLen,wrapheader->wrapheadervar.awsReqLen,wrapheader->wrapheadervar.funcNameLen,wrapheader->wrapheadervar.tagslength,wrapheader->wrapheadervar.messageType);

  if(wrapheader->wrapheadervar.apiReqLen > 0) {
    PARSE_UDP_HEADER(wrapheader->apiReqId, wrapheader->wrapheadervar.apiReqLen, remainingBytes, ptr);
    NDTL2(trace_log_key, NULL, TL_INFO, "wrapheader.apiReqId = %s", wrapheader->apiReqId);
  }

  if(wrapheader->wrapheadervar.awsReqLen > 0) {
    PARSE_UDP_HEADER(wrapheader->awsReqId, wrapheader->wrapheadervar.awsReqLen, remainingBytes, ptr);
    NDTL2(trace_log_key, NULL, TL_INFO, "weapHeader.awsReqId = %s", wrapheader->awsReqId);
  }

  if(wrapheader->wrapheadervar.funcNameLen > 0) {
    PARSE_UDP_HEADER(wrapheader->funcName, wrapheader->wrapheadervar.funcNameLen, remainingBytes, ptr);
    NDTL2(trace_log_key, NULL, TL_INFO, "weapHeader.funcName = %s", wrapheader->funcName);
  }

  if(wrapheader->wrapheadervar.tagslength > 0) {
    PARSE_UDP_HEADER(wrapheader->tags, wrapheader->wrapheadervar.tagslength, remainingBytes, ptr);
    NDTL2(trace_log_key, NULL, TL_INFO, "weapHeader.tags = %s", wrapheader->tags);
  }

  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
 return 1;
}

int parsetagsforUDPconnection(wrapheader_t *wrapheader, struct kwLocalStruct_t *kwLocalStruct) {
  //NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  parse_tag_values(wrapheader->tags,kwLocalStruct);

  if(strlen(kwLocalStruct->ndAppServerHost) == 0)
      return 0;

  if(wrapheader->wrapheadervar.protocolType == PROTOCOL_TCP_UDP) {
    if(strlen(kwLocalStruct->tierName) == 0)
      return 0;
    if(strlen(kwLocalStruct->agentName) == 0)
      return 0;
  }

  char *agenttype = "Unknown";
  switch((enum ND_AGENT_TYPE)wrapheader->wrapheadervar.agentType) {
    case ND_AGENT_PYTHON:
      agenttype = "Python";
      break;
    case ND_AGENT_PHP:
      agenttype = "php";
      break;
    case ND_AGENT_DOTNET:
      agenttype = "Dotnet";
      break;
    case ND_AGENT_JAVA:
      agenttype = "Java";
      break;
    case ND_AGENT_GO:
      agenttype = GOAGENT_INSTANCE;
      break;
  }
  strcpy(kwLocalStruct->agentType, agenttype);

  return 1;
}

int search_connection_in_tcp_mode(struct kwLocalStruct_t *kwLocalStruct) {
  if(!kwLocalStruct->appName || !kwLocalStruct->tierName || !kwLocalStruct->ndAppServerHost || !kwLocalStruct->agentName)
    return -1;
  for(int i = 1; i < NDProxy.maxTCPProcesses; i++)
  {
    if(!NDProxyInstance[i])
        continue;
    if (!strcmp(NDProxyInstance[i]->appName, kwLocalStruct->appName)
    && !(strcmp(NDProxyInstance[i]->tierName, kwLocalStruct->tierName))
    && !(strcmp(NDProxyInstance[i]->appServerName, kwLocalStruct->ndAppServerHost))
    && !(strcmp(NDProxyInstance[i]->agentName, kwLocalStruct->agentName))) {
      NDProxyTcpUdpInstance[i] = NDProxyInstance[i];
      NDProxy.numTcpUdpInstances++;
      ndApplication = NDProxyTcpUdpInstance[i];
      return i;
    }
  }

   return -1;
}

void updateUDPClientDetails(NDProxyInstance_t *nproxyInstance, int newconn) {
  time_t start_t;
  time(&start_t);
  nproxyInstance->udpPacketReceiveTime = start_t;
}

int record_process_with_ndc_conn(wrapheader_t *wrapheader) {
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");

  if(!wrapheader->tags) {
    return -1;
  }

  //for non lambda function name and api/aws id not required but fpid is required
  if(wrapheader->wrapheadervar.protocolType == PROTOCOL_UDP) {
    if(!wrapheader->apiReqId || !wrapheader->funcName) {
        return -1;
    }
  }
  struct kwLocalStruct_t kwLocalStruct;
  memset(&kwLocalStruct,0,sizeof(struct kwLocalStruct_t));
  if(!parsetagsforUDPconnection(wrapheader, &kwLocalStruct))
    return -1;
  int i;
  int appindex = -1;
  int free_index = -1;

  apr_thread_mutex_lock(NDProxy.udp_connection_lock);
  if(wrapheader->wrapheadervar.protocolType == PROTOCOL_TCP_UDP) {
    for (i = 1; i < NDProxy.maxTCPProcesses; i++) {
      if(!NDProxyTcpUdpInstance[i])
        continue;
      if (NDProxyTcpUdpInstance[i]->tierName && NDProxyTcpUdpInstance[i]->appServerName && NDProxyTcpUdpInstance[i]->agentName) {
        if(!strcmp(NDProxyTcpUdpInstance[i]->tierName, kwLocalStruct.tierName)
        && !strcmp(NDProxyTcpUdpInstance[i]->appServerName, kwLocalStruct.ndAppServerHost)
        && !strcmp(NDProxyTcpUdpInstance[i]->agentName, kwLocalStruct.agentName)) {
          NDTL2(trace_log_key, NULL, TL_INFO, "Received data for function = %s server = %s pos = %d ", wrapheader->funcName, kwLocalStruct.ndAppServerHost, i);
          appindex = i;
          break;
        }
      }
    }

    if(appindex != -1) {
      apr_thread_mutex_unlock(NDProxy.udp_connection_lock);
      return appindex;
    }

    free_index = search_connection_in_tcp_mode(&kwLocalStruct);
    //if same connection exists in tcp mode dont create new connection with ndc
    if(free_index != -1) {
      //NDLB_MALLOC_AND_COPY(wrapheader->funcName, NDProxyTcpUdpInstance[free_index]->funcName, strlen(wrapheader->funcName) + 1, "setNDCookieName", -1);
      NDLB_MALLOC_STRING(ndApplication->lambdaKey, 50, "Allocating memory for lambdakey", -1);
      appindex = free_index;
      ndApplication->iTcpUdpMode = 1;
      apr_thread_mutex_unlock(NDProxy.udp_connection_lock);
      return appindex;
    }
    apr_thread_mutex_unlock(NDProxy.udp_connection_lock);
    return -1;
  }
  for (i = 0; i < NDProxy.maxUDPProcesses; i++) {
    if(!NDProxyUdpInstance[i]) {
      if(free_index == -1)
        free_index = i;
      continue;
    }
    if (NDProxyUdpInstance[i]->funcName && NDProxyUdpInstance[i]->appServerName)  {
      if(!strcmp(NDProxyUdpInstance[i]->funcName, wrapheader->funcName) && !strcmp(NDProxyUdpInstance[i]->appServerName, kwLocalStruct.ndAppServerHost)) {
        NDTL2(trace_log_key, NULL, TL_INFO, "Received data for function = %s server = %s pos = %d ", wrapheader->funcName, kwLocalStruct.ndAppServerHost, i);
        appindex = i;
        break;
      }
    }
  }

  if (appindex == -1) {
    if(wrapheader->wrapheadervar.messageType == UDP_MESSAGE_HAVOC) {
      apr_thread_mutex_unlock(NDProxy.udp_connection_lock);
      return -1;
    }
    if(free_index == -1) {
      NDLB_REALLOC_AND_MEMSET(NDProxyUdpInstance, sizeof(NDProxyInstance_t*)*(NDProxy.maxUDPProcesses+PROXY_DELTA_CONNECTIONS),
        sizeof(NDProxyInstance_t*)*NDProxy.maxUDPProcesses,
      "Re-Allocating memory to store the nProxyInstance->processInfo", -1);
      free_index = NDProxy.maxUDPProcesses;
      NDProxy.maxUDPProcesses += PROXY_DELTA_CONNECTIONS;
    }
    if(!NDProxyUdpInstance[free_index]) {
      NDLB_MALLOC_AND_MEMSET(NDProxyUdpInstance[free_index],
    sizeof(NDProxyInstance_t) ,
    "Allocating memory to store the nProxyInstance->processInfo", -1);
    }
    setNDProxyInstance(NDProxyUdpInstance[free_index]);
    int lambdalen = 0;
    if(wrapheader->funcName) {
      lambdalen = strlen(wrapheader->funcName);
      NDLB_MALLOC_AND_COPY(wrapheader->funcName, NDProxyUdpInstance[free_index]->funcName, lambdalen + 1, "funcName", -1);
    }
    if(wrapheader->apiReqId)
      lambdalen += strlen(wrapheader->apiReqId);
    NDLB_MALLOC_STRING(NDProxyUdpInstance[free_index]->lambdaKey, lambdalen + 1, "Allocating memory for lambdakey", -1);
    sprintf(NDProxyUdpInstance[free_index]->lambdaKey, "%s%s",wrapheader->apiReqId,wrapheader->funcName);

    //if(wrapheader->funcName)
      //strncpy(NDProxyUdpInstance[free_index].funcName, wrapheader->funcName, strlen(wrapheader->funcName));
    populateNDProxyInstance(NDProxyUdpInstance[free_index],&kwLocalStruct,1);

    appindex = free_index;
    NDProxyUdpInstance[free_index]->processInfo[0].data_len = 0;
    NDProxyUdpInstance[free_index]->processInfo[0].processId = 0;
    NDTL1(trace_log_key, NULL, TL_INFO, "UDP Info populated for funcname = %s, pos %d total %d",NDProxyUdpInstance[free_index]->funcName,free_index,NDProxy.numUdpInstances);
    // increment the process counter
    NDProxyUdpInstance[free_index]->processCount += 1;
    NDProxyUdpInstance[free_index]->activeProcessCount += 1;
    // Increment the counter for numUdpInstances by 1
    NDProxy.numUdpInstances += 1;
    if(make_connection_with_ndc(NDProxyUdpInstance[free_index], 1))
      updateUDPClientDetails(NDProxyUdpInstance[free_index],1);
  }
  apr_thread_mutex_unlock(NDProxy.udp_connection_lock);
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
  return appindex;
}

int process_ws_header(char *buf, int readbytes, wrapheader_t *wrapheader) {
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  if(!get_ws_param_values(buf, readbytes, wrapheader)) {
    NDTL0(trace_log_key, NULL, TL_ERROR, "Error in parsing data from client..");
    return -1;
  }

  if(wrapheader->wrapheadervar.appId != -1)
    return wrapheader->wrapheadervar.appId;

  return record_process_with_ndc_conn(wrapheader);
}

void addUDPClientSocket(apr_sockaddr_t *addr) {
  apr_port_t *key=&addr->port;
  apr_sockaddr_t *val = NULL;

  val = search_uint16_hashmap(key,ndApplication->udp_connections);
  if(val == NULL) {
    NDTL1(trace_log_key, NULL, TL_INFO, "add client socket for function %s", ndApplication->funcName);
    key = apr_palloc(NDProxy.proxy_pool, sizeof(apr_port_t));
    *key = addr->port;
    NDLB_MALLOC_AND_COPY_STRUCT(addr, val, sizeof(apr_sockaddr_t), "copy addr", -1);
    modify_uint16_hashmap(key, val, ndApplication->udp_connections);
  }
}

void freewrapheader(wrapheader_t *wrapheader) {
  FREE_AND_MAKE_NULL(wrapheader->apiReqId, "Free'ing apiReqId", -1);
  FREE_AND_MAKE_NULL(wrapheader->awsReqId, "Free'ing awsReqId", -1);
  FREE_AND_MAKE_NULL(wrapheader->funcName, "Free'ing funcName", -1);
  FREE_AND_MAKE_NULL(wrapheader->tags, "Free'ing tags", -1);
}

void record_havoc_response_from_agent(char *buf) {
  NDMessage_t* NDMessage = NULL;
  NDLB_MALLOC_AND_MEMSET(NDMessage, sizeof(NDMessage_t), "mallocing to parse the Netdiagnostic message", -1); 
  char *orig_message = NULL;
  NDLB_MALLOC_AND_COPY(buf, orig_message, strlen(buf) + 1, "orig_message", -1);
  if(parseRecdMessage(buf, NDMessage)==SUCESS){
    if(strlen(NDMessage->NHMID) == 0) {
      NDTL1(trace_log_key, NULL, TL_WARNING, "NHMID is missing in message %s", buf);
      FREE_AND_MAKE_NULL(NDMessage, "Free'ing NDMessage", -1);
      FREE_AND_MAKE_NULL(orig_message, "Free'ing orig_message", -1);
      return;
    }
    NDMessage_t *temp = NULL;
    temp = search_hashmap(NDMessage->NHMID, udp_havoc_timeout);
    if(temp == NULL) {
      NDTL2(trace_log_key, NULL, TL_INFO, "Havoc message expired for NHMID %s", NDMessage->NHMID);
      FREE_AND_MAKE_NULL(NDMessage, "Free'ing NDMessage", -1);
      FREE_AND_MAKE_NULL(orig_message, "Free'ing orig_message", -1);
      return;
    }
    NDTL1(trace_log_key, NULL, TL_INFO, "Got response from agent %s", buf);
  }
  else {
    NDTL1(trace_log_key, NULL, TL_WARNING, "Failed to parse response from agent %s", buf);
  }
  FREE_AND_MAKE_NULL(NDMessage, "Free'ing NDMessage", -1);
  FREE_AND_MAKE_NULL(orig_message, "Free'ing orig_message", -1);
}

void SendMessageOnUDPSocket(char *msg) {
  //iterate client list and send message
  apr_port_t *key;
  apr_hash_index_t *idx = NULL;
  apr_sockaddr_t *val = NULL;
  NDTL2(trace_log_key, NULL, TL_INFO, "Send message to agent %s", msg);
    for(idx = apr_hash_first(NDProxy.proxy_pool,(apr_hash_t*) ndApplication->udp_connections); idx; idx = apr_hash_next(idx)){
        apr_hash_this(idx, (const void **)&key, NULL, (void **)&val);
        if(!key || !val)
          continue;
#ifdef __unix__
	ndlb_sendto_n_bytes(NDProxy.udp_skt, val, msg, strlen(msg));
#else
        struct in_addr* ad = (struct in_addr*)val->ipaddr_ptr;
        char tmphost[20];
        sprintf(tmphost, "%d.%d.%d.%d", ad->S_un.S_un_b.s_b1, ad->S_un.S_un_b.s_b2, ad->S_un.S_un_b.s_b3, ad->S_un.S_un_b.s_b4);
        apr_sockaddr_t* val1 = NULL;
        apr_sockaddr_info_get(&val1, tmphost, APR_INET, val->port, 0, val->pool);
        ndlb_sendto_n_bytes(NDProxy.udp_skt, val1, msg, strlen(msg));
#endif

    }
}

char *make_data_for_agent(int appindex, enum udp_data_type type, int enableNdCookie) {
  char *serialized_string = NULL;
  JSON_Value *branch = json_value_init_array();
  JSON_Array *leaves = json_value_get_array(branch);
  JSON_Value *appid_value = NULL, *tier_value = NULL, *test_value = NULL, *ndcookie_value = NULL, *nvcookie_value = NULL, *corelation_valueid = NULL, *backend_rule_value = NULL, *otel_value = NULL, *DBBACKENDNAME=NULL;
  if(type == UDP_DATA_ALL) {
    appid_value = json_value_init_object();
    JSON_Object *leaf_object = json_value_get_object(appid_value);
    json_array_append_string(leaves,"AppId");
    json_object_set_number(leaf_object,"id",appindex);
    json_array_append_value(leaves,appid_value);
  }

  if(type == UDP_DATA_ALL) {
    tier_value = json_value_init_object();
    JSON_Object *leaf_object = json_value_get_object(tier_value);
    json_array_append_string(leaves,"TierInfo");
    json_object_set_number(leaf_object,"tierID",ndApplication->tierID);
    json_object_set_number(leaf_object,"AppServerID",ndApplication->serverID);
    json_object_set_number(leaf_object,"appID",ndApplication->appID);
    json_array_append_value(leaves,tier_value);
  }

  if(type == UDP_DATA_ALL || type == UDP_DATA_TEST) {
    test_value = json_value_init_object();
    JSON_Object *leaf_object1 = json_value_get_object(test_value);
    json_array_append_string(leaves,"TestInfo");
    json_object_set_number(leaf_object1,"TestNumber",ndApplication->Server.TestRunIDValue);
    json_object_set_number(leaf_object1,"bciInstrSessionPct",ndApplication->NDSys.bciInstrSessionPct);
    json_object_set_number(leaf_object1,"captureMethodForAllFP",ndApplication->NDSys.captureMethodForAllFP);
    json_array_append_value(leaves,test_value);
  }

  if(type == UDP_DATA_ALL || type == UDP_DATA_NDCOOKIE) {
    ndcookie_value = json_value_init_object();
    JSON_Object *leaf_object = json_value_get_object(ndcookie_value);
    json_array_append_string(leaves,"NDCookie");
    json_object_set_number(leaf_object,"enableNDSession",enableNdCookie?ndApplication->NDSys.enableNDSession:0);
    json_object_set_string(leaf_object,"CookieName",enableNdCookie?ndApplication->NDSys.ndSessionCookieName:NULL);
    json_object_set_string(leaf_object,"DomainName",enableNdCookie?ndApplication->NDSys.cookieDomainName:NULL);
    json_object_set_number(leaf_object,"headerInResponse",enableNdCookie?ndApplication->NDSys.headerInResponse:0);
    json_object_set_number(leaf_object,"mpos",enableNdCookie?getCookieMethodPosition():0);
    json_object_set_number(leaf_object,"maxFPCountInSession",ndApplication->NDSys.maxFPCountInSession);
    json_object_set_number(leaf_object,"ndSessionIdleTimeOutSecs",ndApplication->NDSys.ndSessionIdleTimeOutSecs);
    json_array_append_value(leaves,ndcookie_value);
  }

  if(type == UDP_DATA_ALL || type == UDP_DATA_NVCOOKIE) {
    nvcookie_value = json_value_init_object();
    JSON_Object *leaf_object = json_value_get_object(nvcookie_value);
    json_array_append_string(leaves,"NVCookie");
    json_object_set_string(leaf_object,"NVCookieName",ndApplication->NDSys.NVCookieName);
    json_array_append_value(leaves,nvcookie_value);
  }

  if(type == UDP_DATA_ALL || type == UDP_DATA_CORELATION) {
    corelation_valueid = json_value_init_object();
    JSON_Object *leaf_object = json_value_get_object(corelation_valueid);
    json_array_append_string(leaves,"CorelationHeader");
    json_object_set_string(leaf_object,"corelationheader",ndApplication->NDSys.correlationIdHeader);
    json_array_append_value(leaves,corelation_valueid);
  }

  if(type == UDP_DATA_ALL || type == UDP_DATA_BACKEND_RULE) {
    char defaultBackend[8];
    char type[8];
    void* ptr = NULL;
    snprintf(defaultBackend, 8, "Host");
    snprintf(type, 8, "%s", "HTTP");
    if (backendNameHashMap != NULL)
        ptr = search_hashmap(type, backendNameHashMap);

    backend_rule_value = json_value_init_object();
    JSON_Object *leaf_object = json_value_get_object(backend_rule_value);
    json_array_append_string(leaves,"BackendNameRule");
    json_object_set_string(leaf_object,"BackendNameRuleValue",ptr ? ptr : defaultBackend);
    json_array_append_value(leaves,backend_rule_value);
  }

  if(type == UDP_DATA_ALL || type == UDP_DATA_OTEL) {
    otel_value = json_value_init_object();
    JSON_Object *leaf_object = json_value_get_object(otel_value);
    char otelkeyword[50];
    sprintf(otelkeyword,"%d-%d-%d-%d",ndApplication->NDSys.enableOTELStandard,arrPropagationHeaders[HEADER_OPENTELEMETRY],arrPropagationHeaders[HEADER_B3],arrPropagationHeaders[HEADER_XB3]);
    json_array_append_string(leaves,"otel");
    json_object_set_string(leaf_object,"otelkeyword",otelkeyword);
    json_array_append_value(leaves,otel_value);

  }
  if (type == UDP_DATA_ALL || type == UDP_DATA_DBBACKENDNAME) {
      DBBACKENDNAME = json_value_init_object();
      JSON_Object* leaf_object = json_value_get_object(DBBACKENDNAME);
      json_array_append_string(leaves, "enableDBBackendName");
      json_object_set_number(leaf_object, "enableDBBackendName", ndApplication->NDSys.enableDBBackendName);
      json_array_append_value(leaves, DBBACKENDNAME);
  }

  serialized_string = json_serialize_to_string_pretty(branch);

  json_array_reset(leaves);
  if(appid_value)
    json_value_free(appid_value);
  if(tier_value)
    json_value_free(tier_value);
  if(test_value)
    json_value_free(test_value);
  if(ndcookie_value)
    json_value_free(ndcookie_value);
  if(nvcookie_value)
    json_value_free(nvcookie_value);
  if(corelation_valueid)
    json_value_free(corelation_valueid);
  if(backend_rule_value)
    json_value_free(backend_rule_value);
  if(otel_value)
    json_value_free(otel_value);
  if (DBBACKENDNAME)
      json_value_free(DBBACKENDNAME);
  json_value_free(branch);

  return serialized_string;
}

void prepare_and_send_data_to_agent(int appindex, enum udp_data_type type, int enableNdCookie) {
  char *data = make_data_for_agent(appindex, type, enableNdCookie);
  SendMessageOnUDPSocket(data);
  json_free_serialized_string(data);
}

void DumpLogMessageForUDP(char *key, char *msg) {
  NDPerThreadData_t *ptr_tls = NULL;
  ptr_tls = search_hashmap(key, ndApplication->fpi_tls_map);
  if(!ptr_tls)
    return;
  dumpLogMessage(ptr_tls,msg);
}

void sendNDCookieValueToOneInstance(char *key) {
  NDPerThreadData_t *ptr_tls = NULL;

  if (ndApplication->NDSys.enableNDSession == 0)
    return;

  ptr_tls = search_hashmap(key, ndApplication->fpi_tls_map);
  if(!ptr_tls)
    return;

  char NDcookie[1024];
  snprintf(NDcookie, BUFFER_LEN, "NDcookieValue:%lld-%lld-%lld-%d-%d-%d-%d-%d-%d-%d", ptr_tls->ndSessionId, ptr_tls->tlTestRunNumber, ptr_tls->last_fptimestamp_from_cookie,
    ptr_tls->instrumented, ptr_tls->btCategory, ptr_tls->exceptionCount, ptr_tls->totalFpcount,
    ptr_tls->tierID, ptr_tls->ndAppServerID, ptr_tls->appID);

  SendMessageOnUDPSocket(NDcookie);
}

void send_havoc_to_new_connected_agent(int appindex) {
  char buf[50];
  sprintf(buf,"AppIndex:%d",appindex);
  SendMessageOnUDPSocket(buf);

  //send havoc
  int count = apr_hash_count(udp_havoc_timeout);
  if(!count)
    return;

  apr_hash_index_t *hi;
  NDMessage_t* NDMessage = NULL;
  char *key = NULL;

  for (hi = apr_hash_first(NDProxy.proxy_pool, udp_havoc_timeout); hi; hi = apr_hash_next(hi)) {
    apr_hash_this(hi, (const void **)&key, NULL, (void **)&NDMessage);
    if(!key || !NDMessage)
      continue;
      NDTL1(trace_log_key, NULL, TL_INFO, "removing record of havoc message for NHMID %s ", NDMessage->NHMID);
      if(ndApplication->iTcpUdpMode) {
        if(!strcmp(ndApplication->tierName, NDMessage->tier)
        && !strcmp(ndApplication->appServerName, NDMessage->server)
        && !strcmp(ndApplication->agentName, NDMessage->instance)) {
          SendMessageOnUDPSocket(NDMessage->orig_message);
        }
      }
      else {
        if(!strcmp(ndApplication->funcName, NDMessage->instance) && !strcmp(ndApplication->appServerName, NDMessage->server)) {
          SendMessageOnUDPSocket(NDMessage->orig_message);
        }
      }
  }
}

void send_heart_beat_to_new_connected_agent() {
  if(!udp_havoc_timeout)
    return;
  int count = apr_hash_count(udp_havoc_timeout);

  apr_port_t *keyp;
  apr_hash_index_t *idx = NULL;
  apr_sockaddr_t *val = NULL;
  int len = 0;
  char *heartbeat_msg = "Heart Beat Received";
  char msgHB[50];
  memset(msgHB,0,50);
  memcpy(msgHB,heartbeat_msg,strlen(heartbeat_msg));

  NDTL0(trace_log_key, NULL, TL_INFO, "Send HB message to agent %s", msgHB);
  for(idx = apr_hash_first(NDProxy.proxy_pool,(apr_hash_t*) ndApplication->udp_connections); idx; idx = apr_hash_next(idx)){
      apr_hash_this(idx, (const void **)&keyp, NULL, (void **)&val);
      if(!keyp || !val)
        continue;

      if(count == 0) {
#ifdef __unix__
        ndlb_sendto_n_bytes(NDProxy.udp_skt, val, msgHB, strlen(msgHB));
#else
          struct in_addr* ad = (struct in_addr*)val->ipaddr_ptr;
          char tmphost[20];
          sprintf(tmphost, "%d.%d.%d.%d", ad->S_un.S_un_b.s_b1, ad->S_un.S_un_b.s_b2, ad->S_un.S_un_b.s_b3, ad->S_un.S_un_b.s_b4);
          apr_sockaddr_t* val1 = NULL;
          apr_sockaddr_info_get(&val1, tmphost, APR_INET, val->port, 0, val->pool);
          ndlb_sendto_n_bytes(NDProxy.udp_skt, val1, msgHB, strlen(msgHB));
#endif
       // ndlb_sendto_n_bytes(NDProxy.udp_skt, val, msgHB, strlen(msgHB));
      }
      else {
          apr_hash_index_t *hi;
          NDMessage_t* NDMessage = NULL;
          char *key = NULL;

          for (hi = apr_hash_first(NDProxy.proxy_pool, udp_havoc_timeout); hi; hi = apr_hash_next(hi)) {
            apr_hash_this(hi, (const void **)&key, NULL, (void **)&NDMessage);
            if(!key || !NDMessage)
              continue;

              NDTL1(trace_log_key, NULL, TL_INFO, "removing record of havoc message for NHMID %s ", NDMessage->NHMID);
              int senddata = 0;
              if(ndApplication->iTcpUdpMode) {
                if(!strcmp(ndApplication->tierName, NDMessage->tier)
                && !strcmp(ndApplication->appServerName, NDMessage->server)
                && !strcmp(ndApplication->agentName, NDMessage->instance)) {
                  senddata = 1;
                }
              }
              else {
                if(!strcmp(ndApplication->funcName, NDMessage->instance) && !strcmp(ndApplication->appServerName, NDMessage->server))
                  senddata = 1;
              }
              if(senddata) {
               
                char *msgForAgent = NULL;
                NDLB_MALLOC_STRING(msgForAgent,strlen(NDMessage->orig_message)+ strlen(NDMessage->body)+1,"msgForAgent",-1);
                memcpy(msgForAgent,NDMessage->orig_message,strlen(NDMessage->orig_message));
                len = strlen(msgForAgent);
                msgForAgent[len]='\n';
                len += 1;
                memcpy(msgForAgent+len,NDMessage->body,strlen(NDMessage->body));
                len += strlen(NDMessage->body);
                msgForAgent[len] = '\0';
               // ndlb_sendto_n_bytes(NDProxy.udp_skt, val, msgForAgent, strlen(msgForAgent));
#ifdef __unix__
                ndlb_sendto_n_bytes(NDProxy.udp_skt, val, msgForAgent, strlen(msgForAgent));
#else
                struct in_addr* ad = (struct in_addr*)val->ipaddr_ptr;
                char tmphost[20];
                sprintf(tmphost, "%d.%d.%d.%d", ad->S_un.S_un_b.s_b1, ad->S_un.S_un_b.s_b2, ad->S_un.S_un_b.s_b3, ad->S_un.S_un_b.s_b4);
                apr_sockaddr_t* val1 = NULL;
                apr_sockaddr_info_get(&val1, tmphost, APR_INET, val->port, 0, val->pool);
                ndlb_sendto_n_bytes(NDProxy.udp_skt, val1, msgForAgent, strlen(msgForAgent));
#endif
                FREE_AND_MAKE_NULL(msgForAgent,"msgForAgent",-1);
              }
          }
      }
  }
}

void decode_udp_Message(char *buf, int readbytes, apr_sockaddr_t *saddr) {
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  
  wrapheader_t wrapheader;
  int appindex = -1;
  wrapheader.tags = wrapheader.funcName = wrapheader.awsReqId = wrapheader.apiReqId = NULL;
  wrapheader.wrapheadervar.appId = -1;
  appindex = process_ws_header(buf, readbytes, &wrapheader);
  if(appindex == -1) {
    freewrapheader(&wrapheader);
    NDTL0(trace_log_key, NULL, TL_ERROR, "Failed to process data from client..");
    return;
  }

  if(wrapheader.wrapheadervar.protocolType == PROTOCOL_UDP) {
    if(appindex >= NDProxy.maxUDPProcesses) {
      freewrapheader(&wrapheader);
      return;
    }

    ndApplication = NDProxyUdpInstance[appindex];
    if(!ndApplication)
      return;
    updateUDPClientDetails(ndApplication,0);
  }
  else {
    if(appindex >= NDProxy.maxTCPProcesses) {
      freewrapheader(&wrapheader);
      return;
    }
    if(wrapheader.wrapheadervar.messageType == UDP_MESSAGE_DEFAULT && wrapheader.wrapheadervar.fpId == 0) {
      freewrapheader(&wrapheader);
      return;
    }
    ndApplication = NDProxyTcpUdpInstance[appindex];
  }
  addUDPClientSocket(saddr);
  switch(wrapheader.wrapheadervar.messageType) {
    case UDP_MESSAGE_DEFAULT:
      if(ndApplication->iTcpUdpMode && ndApplication->lambdaKey)
        sprintf(ndApplication->lambdaKey, "%lld",wrapheader.wrapheadervar.fpId);
      if(ndApplication->lambdaKey)
        decodeMessage(buf+wrapheader.wrapheadervar.whLen, readbytes-wrapheader.wrapheadervar.whLen, 0,1,ndApplication->lambdaKey);
      break;
    case UDP_MESSAGE_HAVOC:
      record_havoc_response_from_agent(buf+wrapheader.wrapheadervar.whLen+1);
      break;
    case UDP_CLIENT_INIT:
      send_havoc_to_new_connected_agent(appindex);
      break;
    case UDP_CLIENT_HEARTBEAT:
      send_heart_beat_to_new_connected_agent();
      break;
    case UDP_CLIENT_GET_NVCOOKIE:
      prepare_and_send_data_to_agent(appindex, UDP_DATA_NVCOOKIE,1);
      break;
    case UDP_CLIENT_GET_NDCOOKIE:
      prepare_and_send_data_to_agent(appindex, UDP_DATA_NDCOOKIE,1);
      break;
    case UDP_CLIENT_GET_NDCOOKIE_VALUE:
      if(ndApplication->iTcpUdpMode && ndApplication->lambdaKey)
        sprintf(ndApplication->lambdaKey, "%lld",wrapheader.wrapheadervar.fpId);
      if(ndApplication->lambdaKey)
        sendNDCookieValueToOneInstance(ndApplication->lambdaKey);
      break;
    case UDP_CLIENT_DUMP_LOG_MESSAGE:
      if(ndApplication->iTcpUdpMode && ndApplication->lambdaKey)
        sprintf(ndApplication->lambdaKey, "%lld",wrapheader.wrapheadervar.fpId);
      if(ndApplication->lambdaKey)
        DumpLogMessageForUDP(ndApplication->lambdaKey, buf+wrapheader.wrapheadervar.whLen+1);
      break;

    case UDP_AGENT_INITIALIZE:
      prepare_and_send_data_to_agent(appindex, UDP_DATA_ALL,1);
      break;
  }

  freewrapheader(&wrapheader);
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}

//support multiple udp packets
// void decode_udp_Message_wrapper(char *buf, int readbytes, apr_sockaddr_t *saddr) {
//   int numbytes = 0;
//   int totalbytes = 0;
//   char *ptr = strchr(buf,'\n');
//   char *str = buf;

//    while (ptr != NULL) {
//     numbytes = ptr - str;
//     *ptr = '\0';
//      ptr++;

//      decode_udp_Message(str, numbytes, saddr);
//      totalbytes += numbytes;
//      str = ptr;
//      ptr = strchr(str,'\n');
//   }

//   decode_udp_Message(str, readbytes-totalbytes, saddr);
// }


 void decode_udp_Message_wrapper(char *buf, int readbytes, apr_sockaddr_t *saddr) {
   int numbytes = 0;
   char *ptr;
   char *str = buf;
   int curIndex = 0;
   int whlen = 0;
   int totallen = 0;
   int isError = 0;

   //There are following two types of messages
   //^wrapheader\n
   //^wrapheader^msg header^trans_start/trans_end/method_entry/method_exit etc

   while (curIndex<readbytes) {
    //Each record will have ^ (ascii value 94) at the first position
      if (buf[curIndex] != 94) {
        NDTL0(trace_log_key, NULL, TL_ERROR, "Failed to parse UDP packet at pos %d. First byte is not ^",curIndex);
        isError = 1;
        break;
      }

      //check wether unsigned char* is required
      whlen = *((int*)&buf[curIndex+1]);
      if (whlen < WRAP_HEADER_LEN)
      {
        isError = 1;
        NDTL0(trace_log_key, NULL, TL_ERROR, "Failed to parse UDP packet at pos %d. Wrong msg header len %d",curIndex,whlen);
        break;
      }
      curIndex += whlen;
      if(curIndex >= readbytes) {
        isError = 1;
        NDTL0(trace_log_key, NULL, TL_ERROR, "Failed to parse UDP packet at pos %d. Index is more than read bytes %d",curIndex,readbytes);
        break;
      }
      if(buf[curIndex] != 10) {
        if(curIndex > readbytes-6) { //6 is used for ^ + int size
          isError = 1;
          NDTL0(trace_log_key, NULL, TL_ERROR, "Failed to parse UDP packet at pos %d. Msg hdr pre index is more than read bytes %d",curIndex,readbytes);
          break;
        }

        totallen = *((int*)(unsigned char*)&buf[curIndex + 1+ 4]);
        if (totallen < MSG_HEADER_LEN) {
          isError = 1;
          NDTL0(trace_log_key, NULL, TL_ERROR, "Failed to parse UDP packet at pos %d. Wrong length of Msg hdr",curIndex,totallen);
          break;
        }

        curIndex += totallen-1;
        if(curIndex >= readbytes) {
          isError = 1;
          NDTL0(trace_log_key, NULL, TL_ERROR, "Failed to parse UDP packet at pos %d. Msg hdr post index is more than read bytes %d",curIndex,readbytes);
          break;
        }
      }

      //10 is ascii value of \n. newline is sent after each wrapheader
      if (buf[curIndex] == 10) {
        ptr = buf+curIndex;
        curIndex++;
        numbytes = ptr - str +1;
        ptr++;
        decode_udp_Message(str, numbytes,saddr);
        str=ptr;
      }
      else {
        isError = 1;
        NDTL0(trace_log_key, NULL, TL_ERROR, "Failed to parse UDP packet at pos %d. Msg hdr did not find newline character",curIndex);
	      break;
      }
   }
   if(isError)
    return;
   //process missing data
  //why we are calling decode_udp_Message here
   if(readbytes - curIndex > 1)
    decode_udp_Message(str, readbytes-curIndex, saddr);
}

int handleHavocMsgforUDPClient(NDMessage_t* NDMessage, int havoc_enable) {
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  NDMessage_t *temp = NULL;
  if(!udp_havoc_timeout)
    return 0;
  SendMessageOnUDPSocket(NDMessage->orig_message);
    temp = search_hashmap(NDMessage->NHMID, udp_havoc_timeout);

    if(!havoc_enable) {
      if(temp) {
        temp->havoc_timeout = 0;
        temp->stop = 1;
        modify_hashmap(temp->NHMID, temp, (apr_hash_t *)udp_havoc_timeout);
      }
      else {
        NDMessage->stop = 1;
        return 0;
      }
      FREE_AND_MAKE_NULL(NDMessage, "freeing the NDMessage structure", -1);
      return 1;
    }

    if(temp == NULL) {
      NDTL1(trace_log_key, NULL, TL_INFO, "record havoc message for NHMID %s ", NDMessage->NHMID);
      time(&NDMessage->msg_time);
      modify_hashmap(NDMessage->NHMID, NDMessage, (apr_hash_t *)udp_havoc_timeout);
    }
    else {
      apr_hash_set(udp_havoc_timeout, NDMessage->NHMID, APR_HASH_KEY_STRING, NULL);
      time(&temp->msg_time);
      NDTL1(trace_log_key, NULL, TL_INFO, "modify record of havoc message for NHMID %s ", NDMessage->NHMID);
      modify_hashmap(temp->NHMID, temp, (apr_hash_t *)udp_havoc_timeout);
      FREE_AND_MAKE_NULL(NDMessage, "freeing the NDMessage structure", -1);
    }
    NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
    return 1;
}

void* APR_THREAD_FUNC proxy_udp_havoc_handler(apr_thread_t *thd, void *data) {
  apr_hash_index_t *hi;
  char *key = NULL;
  NDProxyInstance_t* tmp_inst = NULL;
  tmp_inst = (NDProxyInstance_t*)data;
  if(!tmp_inst)
    return NULL;
  ndApplication = tmp_inst;

  NDMessage_t* NDMessage = NULL;
  time_t end_t;

  while(1) {
    if(NDProxy.shutdownProxy)
      break;
    if(NDProxy.numUdpInstances == 0) {
      for(int i = 0; i < 10; i++) {
        if(NDProxy.shutdownProxy)
          break;
        apr_sleep(1000000);
      }
      continue;
    }

    time(&end_t);
    for(int i = 0; i < NDProxy.maxUDPProcesses; i++) {
      if(!NDProxyUdpInstance[i])
	     continue;
      ndApplication = NDProxyUdpInstance[i];

      if(!udp_havoc_timeout)
        continue;
      int count = apr_hash_count(udp_havoc_timeout);
      if(!count)
        continue;

      for (hi = apr_hash_first(NDProxy.proxy_pool, udp_havoc_timeout); hi; hi = apr_hash_next(hi)) {
        apr_hash_this(hi, (const void **)&key, NULL, (void **)&NDMessage);
        if(!key || !NDMessage)
          continue;
        if(end_t - NDMessage->msg_time > NDMessage->havoc_timeout +2) {
          NDTL1(trace_log_key, NULL, TL_INFO, "removing record of havoc message for NHMID %s ", NDMessage->NHMID);
          apr_hash_set(udp_havoc_timeout, key, APR_HASH_KEY_STRING, NULL);
          //send message on NDC
          strncpy(NDMessage->status, "Success", 8);
          NDMessage->status[7]='\0';
          SendHavocResponse(NDMessage);
          FREE_AND_MAKE_NULL(NDMessage, "freeing the NDMessage structure", -1);
        }
      }
    }
    for(int i = 0; i < 10; i++) {
        if(NDProxy.shutdownProxy)
          break;
        apr_sleep(1000000);
    }
  }
  return NULL;
}

void* APR_THREAD_FUNC proxy_udp_client_handler(apr_thread_t *thd, void *data) {
  NDProxyInstance_t* tmp_inst = NULL;
  tmp_inst = (NDProxyInstance_t*)data;
  if(!tmp_inst)
    return NULL;
  ndApplication = tmp_inst;

  while(1) {
     if(NDProxy.shutdownProxy)
      break;
    if(NDProxy.numUdpInstances == 0) {
      for(int i = 0; i < 10; i++) {
        if(NDProxy.shutdownProxy)
          break;
        apr_sleep(1000000);
      }
      continue;
    }
    time_t end_t;
    time(&end_t);
    for(int i = 0; i < NDProxy.maxUDPProcesses; i++) {
      if(!NDProxyUdpInstance[i])
	     continue;
      ndApplication = NDProxyUdpInstance[i];
    if(!ndApplication->udpPacketReceiveTime)
        continue;

    if(end_t - ndApplication->udpPacketReceiveTime > NDProxy.udpTimeout) {
	      if(ndApplication->processInfo)
          ndApplication->processInfo[0].processId = -1;
        ndApplication->processCount = 0;
        NDTL1(trace_log_key, NULL, TL_INFO, "closing connection with ndc func=%s open connections %d", ndApplication->funcName,NDProxy.numUdpInstances-1);
        ndApplication->close_udp_thread = 1;
        NDProxy.numUdpInstances -= 1;
        NDProxyUdpInstance[i] = NULL;
      }
    }
    for(int i = 0; i < 10; i++) {
      if(NDProxy.shutdownProxy)
        break;
      apr_sleep(1000000);
    }
  }
  return NULL;
}

void MovetoUDPWriteList(rudp_packet_wrapper* curr_node) {
  apr_thread_mutex_lock(NDProxy.udp_bufferWriteLock);
  if(!NDProxy.udp_writeListLast) {
    NDProxy.udp_writeListLast = NDProxy.udp_writeList = curr_node;
  }
  else {
    NDProxy.udp_writeListLast->next = curr_node;
    NDProxy.udp_writeListLast = curr_node;
  }
  apr_thread_mutex_unlock(NDProxy.udp_bufferWriteLock);
}
//more than one udp tier/server/instance
void printAllThreadsOperations() {
  if(ndApplication->apptype == 0) {
    NDTL0(trace_log_key, NULL, TL_INFO, "ThreadOpsStats: worker id: %d, total queued operations: %d, total operation processed: %d, arrSize: %d",tcpStorage.threadIndex,tcpStorage.perThreadOpsVar.total_operation_queued,tcpStorage.perThreadOpsVar.total_operation_processed,tcpStorage.perThreadOpsVar.arrSize);
  }
  else {
    if(!udpStorage)
      return;
    for(int i = 0; i < NDProxy.num_udpThreads; i++) {
      NDTL0(trace_log_key, NULL, TL_INFO, "ThreadOpsStats: worker id: %d, total queued operations: %d, total operation processed: %d, arrSize: %d", udpStorage[i].threadIndex,udpStorage[i].perThreadOpsVar.total_operation_queued,udpStorage[i].perThreadOpsVar.total_operation_processed,udpStorage[i].perThreadOpsVar.arrSize);
    }
  }
}

void cleanFPResetOps() {
  if(ndApplication->apptype == 0) {
    tcpStorage.perThreadOpsVar.opCount = 0;
  }
  else {
    if(!udpStorage)
      return;
    for(int i = 0; i < NDProxy.num_udpThreads; i++) {
      udpStorage[i].perThreadOpsVar.opCount = 0;
    }
  }
}

void queueFPResetThreadOps(NDPerThreadData_t* tls) {
  struct perThreadOps *per_th_op = NULL;

  if(ndApplication->apptype == 0)
    per_th_op = &tcpStorage.perThreadOpsVar;
  else {
    per_th_op = &udpStorage[tls->threadIndex].perThreadOpsVar;
  }
  if(per_th_op->arrSize >= MAX_TRANSACTION_RESET_LIMIT)
    return;

  tls->reset_delayed_transaction = 1;
  apr_thread_mutex_lock(per_th_op->threadOpLock);

  if(per_th_op->opCount >= per_th_op->arrSize) {
    NDLB_REALLOC_AND_MEMSET(per_th_op->arrThreadOp, sizeof(struct threadOps) * (per_th_op->arrSize + NUM_THREAD_OPERATIONS),
      sizeof(struct threadOps) * per_th_op->arrSize,
      "Re-Allocating memory for arrThreadOp", -1);
    per_th_op->arrSize += NUM_THREAD_OPERATIONS;

    NDTL0(trace_log_key, NULL, TL_INFO, "ThreadOpsStats: Reallocate memory for thread index: %d array size: %d",tls->threadIndex,per_th_op->arrSize);
  }

  per_th_op->arrThreadOp[per_th_op->opCount].operation = THREAD_OPERATION_FP_RESET;
  per_th_op->arrThreadOp[per_th_op->opCount].flowpathInstance = tls->flowpathInstance;//use flowpathInstance
  per_th_op->arrThreadOp[per_th_op->opCount].appInstance = ndApplication;
  per_th_op->opCount++;
  per_th_op->total_operation_queued++;
  apr_thread_mutex_unlock(per_th_op->threadOpLock);

  NDTL0(trace_log_key, NULL, TL_INFO, "Added queued FP reset thread operation flowpathInstance %lld opCount %d worker id %d total_operation_queued %d",tls->flowpathInstance,per_th_op->opCount,tls->threadIndex,per_th_op->total_operation_queued);
}

void processThreadOps(struct perThreadOps *perThreadVar) {
  apr_thread_mutex_lock(perThreadVar->threadOpLock);
  
  int numProcessesed = 0;
  int numQueued = 0;
 for(int i = 0; i < perThreadVar->opCount; i++) {
    NDProxyInstance_t * appInstance = perThreadVar->arrThreadOp[i].appInstance;
    if(!appInstance) {
      NDTL0(trace_log_key, NULL, TL_INFO, "ThreadOpsStats: Application instance is NULL for OpCount %d",i);
      continue;
    }
    char fpinstance[64];
    ndApplication = appInstance;
    snprintf(fpinstance, 64,"%lld", perThreadVar->arrThreadOp[i].flowpathInstance);
    NDPerThreadData_t* ptr_tls = search_hashmap(fpinstance, appInstance->fpi_tls_map);
    if(!ptr_tls) {
      NDTL1(trace_log_key, NULL, TL_INFO, "ThreadOpsStats: FlowpathInstance %lld not found.",perThreadVar->arrThreadOp[i].flowpathInstance);
      perThreadVar->total_operation_ignored++;
      continue;
    }

    if(perThreadVar->arrThreadOp[i].operation == THREAD_OPERATION_FP_RESET) {
      endFP(ptr_tls, 200, NULL, 0, 0, 0, NULL);
      reset(ptr_tls);
      modify_hashmap(fpinstance, NULL, appInstance->fpi_tls_map);
    }
    else {
      NDTL0(trace_log_key, NULL, TL_INFO, "ThreadOpsStats: Invalid thread operation %d OpCount %d", perThreadVar->arrThreadOp[i].operation,i);
    }
    numProcessesed++;
    perThreadVar->total_operation_processed++;
    perThreadVar->arrThreadOp[i].flowpathInstance = 0;
    perThreadVar->arrThreadOp[i].appInstance = NULL;
  }
  numQueued = perThreadVar->total_operation_queued;
  perThreadVar->opCount = 0;
  apr_thread_mutex_unlock(perThreadVar->threadOpLock);
  NDTL0(trace_log_key, NULL, TL_INFO, "ThreadOpsStats: worker id: %d total queued operations: %d total operation processed: %d curent operation processed: %d ignore transactions: %d array size: %d", tlWorkerThreadId,numQueued, perThreadVar->total_operation_processed,numProcessesed, perThreadVar->total_operation_ignored, perThreadVar->arrSize);
}

void* APR_THREAD_FUNC proxy_udp_handler_each_thread(apr_thread_t *thd, void *data) {
  rudp_packet_wrapper* curr_node = NULL;
  NDProxyInstance_t *tmp_inst = NULL;

  struct udp_thread_storage *tstruct = (struct udp_thread_storage*)data;
  if(!tstruct)
    return NULL;
  if(!tstruct->proxyInstance)
    return NULL;

  tlWorkerThreadId = tstruct->threadIndex;
  tmp_inst = tstruct->proxyInstance;
  ndApplication = tmp_inst;

  for(;;) {
    if(NDProxy.shutdownProxy)
      break;
    apr_thread_mutex_lock(tstruct->thread_bufferReadLock);
    apr_thread_cond_wait(tstruct->thread_cond_var,tstruct->thread_bufferReadLock);
    apr_thread_mutex_unlock(tstruct->thread_bufferReadLock);
    if(!tstruct->node)
      continue;
    if(tstruct->perThreadOpsVar.opCount > 0)
      processThreadOps(&tstruct->perThreadOpsVar);
    apr_thread_mutex_lock(tstruct->thread_bufferReadLock);
    while(tstruct->node != NULL) {
      curr_node = tstruct->node;
      tstruct->node = tstruct->node->next;
      curr_node->next = NULL;
      apr_thread_mutex_unlock(tstruct->thread_bufferReadLock);
      ndApplication = tmp_inst;

      NDTL0(trace_log_key, NULL, TL_ERROR, "duplicate flag %d", curr_node->received_packet.header.duplicate_flag);

      if (curr_node->received_packet.header.duplicate_flag == 0)
      {
        //  NDTL0(trace_log_key, NULL, TL_ERROR, "duplicate flag %d", curr_node->received_packet.header.duplicate_flag);

          decode_udp_Message_wrapper(curr_node->received_packet.payload, curr_node->received_packet.payload_length, tstruct->saddr);
      }
      sendingAck(curr_node->file,curr_node->sender,&curr_node->received_packet);
      MovetoUDPWriteList(curr_node);
      apr_thread_mutex_lock(tstruct->thread_bufferReadLock);
    }
    apr_thread_mutex_unlock(tstruct->thread_bufferReadLock);
    ndApplication = tmp_inst;
  }

  return NULL;
}

int initializeThreadOperations(int i) {
  udpStorage[i].perThreadOpsVar.opCount = 0;
  udpStorage[i].perThreadOpsVar.total_operation_queued = 0;
  udpStorage[i].perThreadOpsVar.total_operation_processed = 0;
  udpStorage[i].perThreadOpsVar.total_operation_ignored = 0;
  udpStorage[i].perThreadOpsVar.total_pending_processe = 0;
  udpStorage[i].perThreadOpsVar.arrSize = NUM_THREAD_OPERATIONS;
  NDLB_MALLOC_AND_MEMSET(udpStorage[i].perThreadOpsVar.arrThreadOp, udpStorage[i].perThreadOpsVar.arrSize * sizeof(struct threadOps), "Create data structure for threadOps", -1);

  apr_thread_mutex_create(&udpStorage[i].perThreadOpsVar.threadOpLock, APR_THREAD_MUTEX_NESTED, NDProxy.proxy_pool);
  if (!udpStorage[i].perThreadOpsVar.threadOpLock)
    return 0;

  return 1;
}

void* APR_THREAD_FUNC proxy_udp_handler(apr_thread_t *thd, void *data) {
  char error_buf[ERROR_BUFFER_SIZE];
  apr_sockaddr_t *saddr = NULL;
  apr_status_t rv;
  NDProxyInstance_t* tmp_inst = NULL;
  tmp_inst = (NDProxyInstance_t*)data;
  ndApplication = tmp_inst;
  if(NDProxy.udpTimeout <= 0)
    NDProxy.udpTimeout = UDP_TIMEOUT;

  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");
  if(udp_buff_list_init()) {
    NDTL0(trace_log_key, NULL, TL_ERROR, "Error in creating udp buffer...");
    return NULL;
  }
  NDLB_MALLOC_AND_MEMSET(NDProxyUdpInstance,
  sizeof(NDProxyInstance_t*)*NDProxy.maxUDPProcesses ,
  "Allocating memory to store the nProxyInstance->processInfo", -1);
  NDProxy.numUdpInstances = 0;

  NDLB_MALLOC_AND_MEMSET(NDProxyTcpUdpInstance,
  sizeof(NDProxyInstance_t*)*NDProxy.maxTCPProcesses ,
  "Allocating memory to store the nProxyInstance->processInfo", -1);

  apr_thread_mutex_create(&NDProxy.udp_connection_lock, APR_THREAD_MUTEX_NESTED, NDProxy.proxy_pool);

  apr_thread_t *udp_client_thread_id = NULL;
  rv = apr_thread_create(&udp_client_thread_id, NULL, proxy_udp_client_handler, tmp_inst, NDProxy.proxy_pool);
  if(rv != APR_SUCCESS) {
    NDTL0(trace_log_key, NULL, TL_ERROR, "Error in creating udp client thread... '%s'", apr_strerror(rv, error_buf, ERROR_BUFFER_SIZE));
  }

   apr_thread_t *udp_havoc_thread_id = NULL;
   rv = apr_thread_create(&udp_havoc_thread_id, NULL, proxy_udp_havoc_handler, tmp_inst, NDProxy.proxy_pool);
   if(rv != APR_SUCCESS) {
      NDTL0(trace_log_key, NULL, TL_ERROR, "Error in creating udp Havoc thread... '%s'", apr_strerror(rv, error_buf, ERROR_BUFFER_SIZE));
   }

  while(1)
  {
    NDProxy.udp_skt = rudp_socket(saddr,NDProxy.proxyPort,NDProxy.proxy_pool);
    //fprintf(stderr,"NDProxy.udp_skt %p\n",NDProxy.udp_skt);
    if(!NDProxy.udp_skt)
    {
      NDTL0(trace_log_key, NULL, TL_ERROR, "Proxy Server Failed to create UDP connection '%s'. Retry in 10 seconds.", error_buf);
      apr_sleep(10000000);
      //continue;
    }
    else {
      NDTL1(trace_log_key, NULL, TL_INFO, "Successully created socket for UDP.");
      break;
    }
  }

  apr_thread_t **udp_thread_id;
  int *rvstatus=NULL;//use malloc_and_memset
  NDLB_MALLOC(udp_thread_id, sizeof(apr_thread_t *)*NDProxy.num_udpThreads, "Allocating for udp_thread_id", -1);
  NDLB_MALLOC(rvstatus, sizeof(int)*NDProxy.num_udpThreads, "mallocing rvstatus", -1);
  for(int j = 0; j < NDProxy.num_udpThreads; j++) {
    udp_thread_id[j]=NULL;//dont use this loop
    rvstatus[j]=0;
  }
//move tinstance in ndproxy
  NDLB_MALLOC(udpStorage, sizeof(struct udp_thread_storage)*NDProxy.num_udpThreads, "Allocating for epoll context structure", -1);
  for(int i = 0; i < NDProxy.num_udpThreads; i++) {
    udpStorage[i].saddr = saddr;//pass NULL
    udpStorage[i].node = NULL;
    udpStorage[i].threadIndex = i;
    udpStorage[i].proxyInstance = tmp_inst;
    apr_thread_mutex_create(&udpStorage[i].thread_bufferReadLock, APR_THREAD_MUTEX_NESTED, NDProxy.proxy_pool);
    if (!udpStorage[i].thread_bufferReadLock)
      return NULL;

    if(!initializeThreadOperations(i))
      return NULL;

    apr_thread_cond_create(&udpStorage[i].thread_cond_var,NDProxy.proxy_pool);
    if (!udpStorage[i].thread_cond_var)
      return NULL;

    rvstatus[i] = apr_thread_create(&udp_thread_id[i], NULL, proxy_udp_handler_each_thread, (void*)&udpStorage[i], NDProxy.proxy_pool);
    if(rvstatus[i] != APR_SUCCESS)
      break;
  }


   eventloop(NDProxy.proxy_pool);
   for (int k = 0; k < NDProxy.num_udpThreads; k++) {
       apr_thread_join(&rvstatus[k], udp_thread_id[k]);
       break;
   }

   FREE_AND_MAKE_NULL(udp_thread_id, "Free'ing udp_thread_id", -1);
   FREE_AND_MAKE_NULL(udpStorage, "Free'ing tInstance", -1);
   FREE_AND_MAKE_NULL(rvstatus, "Free'ing rvstatus", -1);
   if(NDProxy.udp_skt)
   apr_socket_close(NDProxy.udp_skt);
   NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
  return NULL;
}
/**************************************
 * Initializing the ND Proxy Server 
 * and populating its data structure
 * ***********************************/
void ndProxyInit()
{
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");

  json_set_escape_slashes(0);
  /* Declaring all the required variables */
  char error_buf[ERROR_BUFFER_SIZE];
  apr_int32_t num;
  apr_status_t rv;
  int i;
  controlSocketBuffer_t disbufInfo;
  memset(&disbufInfo,0,sizeof(disbufInfo));
  // First mandatory call to apr library.
  apr_initialize();
  atexit(apr_terminate);

  apr_pool_create(&NDProxy.proxy_pool, NULL);

  apr_signal_init(NDProxy.proxy_pool);
  apr_signal(13, SIG_IGN);    //SIGPIPE
  apr_signal(10, resetAndCleanProxy);    // for graceful exit SIGUSR1
  apr_signal(2, resetAndCleanProxy);    // for graceful exit SIGINT

#ifdef __unix__
  apr_signal(SIGHUP, sighup_handler);   
  apr_signal(SIGTERM, sigterm_handler);   
#endif

  apr_thread_mutex_create(&tcpStorage.perThreadOpsVar.threadOpLock, APR_THREAD_MUTEX_NESTED, NDProxy.proxy_pool);
  if (!tcpStorage.perThreadOpsVar.threadOpLock)
    return;
  tcpStorage.perThreadOpsVar.opCount = 0;
  tcpStorage.threadIndex = 0;

  tcpStorage.perThreadOpsVar.arrSize = NUM_THREAD_OPERATIONS;
  NDLB_MALLOC_AND_MEMSET(tcpStorage.perThreadOpsVar.arrThreadOp, tcpStorage.perThreadOpsVar.arrSize * sizeof(struct threadOps), "Create data structure for threadOps", -1);

  tlWorkerThreadId = 0;

  //initialize NDProxy structure
  //initializeProxyStruct();
  setResetSocketBuffer(&bufInfo);
  setResetSocketBuffer(&dataBufInfo);
  setResetSocketBuffer(&disbufInfo);
   apr_thread_t *udp_thread_id = NULL;
   if(NDProxy.proxyPort > 0) {
     rv = apr_thread_create(&udp_thread_id, NULL, proxy_udp_handler, ndApplication, NDProxy.proxy_pool);
     if(rv != APR_SUCCESS)
     {
        NDTL0(trace_log_key, NULL, TL_ERROR, "Error in creating udp thread... '%s'", apr_strerror(rv, error_buf, ERROR_BUFFER_SIZE));
     }
     else {
        NDTL1(trace_log_key, NULL, TL_INFO, "Created udp thread with port '%d' and timeout %ld (secs)", NDProxy.proxyPort, NDProxy.udpTimeout>0?NDProxy.udpTimeout:UDP_TIMEOUT);
     }
   }
  // create socket based on the connection mode i.e. Unix domain socket/INET socket
  // The socket created will be INET in case of distributed Client-Server and UNIX if both
  // lie on  the same docker.
  while(1)
  {
    if(!NDProxy.proxyConType)
    {
      // create INET socket
      NDProxy.skt = create_socket_and_bind(NDProxy.proxyPort, error_buf, ERROR_BUFFER_SIZE, NDProxy.proxy_pool);
      if (NDProxy.skt)
	get_src_addr(NDProxy.skt, NDProxy.proxyHost, BUFFER_LEN, (int *)&NDProxy.proxyPort);
    }
    else
    {
      // create Unix domain socket
      NDProxy.skt = create_unix_domain_socket_and_bind(NDProxy.ndProxyUnixDomainPath, error_buf, ERROR_BUFFER_SIZE, NDProxy.proxy_pool);
    }
    if(!NDProxy.skt)
    {
      NDTL0(trace_log_key, NULL, TL_ERROR, "Proxy Server Failed to Start '%s'", error_buf);
      NDTL0(trace_log_key, NULL, TL_ERROR, "Retry in 10 seconds.");
      apr_sleep(10000000);
      continue;
    }
    else 
      break;
  }

  //apr_os_sock_get(&NDProxy.sktFd, NDProxy.skt);
  NDTL0(trace_log_key, NULL, TL_INFO, "Proxy Started at '%s:%d'", NDProxy.proxyHost, NDProxy.proxyPort);
  // Creating a polling set for epoll
  if((rv = apr_pollset_create(&NDProxy.pollset, 2048, NDProxy.proxy_pool, 0)) != APR_SUCCESS)
  {
    NDTL0(trace_log_key, NULL, TL_ERROR, "Error in creating poll. Error: %s\n", apr_strerror(rv, error_buf, ERROR_BUFFER_SIZE));
    return;
  }

  // Register the proxy main socket in Epoll
  addSocketInEpoll(NDProxy.skt, NULL);

  int appIndex, procIndex, port; 
  PxAppEpollInfo_t *ptr = NULL;
  char ip[128];

  //Start listening to any message on control(this main) thread.

  for(;;)				// Loop 1
  {
    num = 0;
    rv = apr_pollset_poll(NDProxy.pollset, 10000000, &num, &NDProxy.ret_pfd);
    NDTL3(trace_log_key, NULL, TL_INFO, "Epoll set with 10 sec timeout.");
    NDTL3(trace_log_key, NULL, TL_INFO, "Returned epoll events = %d", num);

    /* If error occurs, continue */
    if(rv != APR_SUCCESS || num == 0)
    {
      NDTL3(trace_log_key, NULL, TL_ERROR, "Agent select/epoll_wait. Error = %s\n", apr_strerror(rv, error_buf, ERROR_BUFFER_SIZE));
      continue;
    }
    //Loop through epoll list 
    for(i = 0; i < num; i++)		// Loop 2
    {
      if (NDProxy.ret_pfd[i].rtnevents & APR_POLLNVAL)
      {
	NDTL0(trace_log_key, NULL, TL_ERROR, "Invalid descriptor. Closing socket.");
	cleanAndRemoveSocket((apr_socket_t **)&NDProxy.ret_pfd[i].desc.s, CLEAN);
	continue;
      }
      // If the event is received on the Proxy listener (new connection)
      if(NDProxy.ret_pfd[i].desc.s == NDProxy.skt)
      {
	NDTL2(trace_log_key, NULL, "CLIENT->PROXY", "============\nNEW CONNECTION -%d\n=============',buff info:%s ", NDProxy.sktFd,bufInfo);
	acceptNewConnection(&bufInfo);
      }
      else
      {
	// If the event is received on an already registered connection (Ctrl/Data/AS)
	ptr = NULL;
	appIndex = -1;
	procIndex = -1;
	ptr = (PxAppEpollInfo_t *)NDProxy.ret_pfd[i].client_data;
	if(ptr)
	{
	  appIndex = ptr->appIndex;
	  procIndex = ptr->processIndex;

	  // Error Handling : Sometimes we experienced some error values being returned in NDProxy.ret_pfd[i].client_data.
	  // ---------------- To handle that, we ensure that the returned appindex and process index are valid.
	  if((appIndex >= NDProxy.maxTCPProcesses) || (appIndex == -1))
	  {
	    NDTL0(trace_log_key, NULL, TL_ERROR, "Invalid epoll context data, appIndex = %d, "
		"procIndex = %d (numInstance = %d, numProcess = %d) i = %d, num = %d", 
		appIndex, procIndex, NDProxy.numInstances, NDProxyInstance[0]->processCount, i, num);
	    cleanAndRemoveSocket((apr_socket_t **)&NDProxy.ret_pfd[i].desc.s, CLEAN);
	    continue;
	  }
	  if((procIndex >= NDProxyInstance[appIndex]->processCount) || (procIndex == -1))
	  {
	    NDTL0(trace_log_key, NULL, TL_ERROR, "Invalid epoll context data, appIndex = %d, "
		"procIndex = %d (numInstance = %d, numProcess = %d) i = %d, num = %d", appIndex, procIndex,
		NDProxy.numInstances, NDProxyInstance[appIndex]->processCount, i , num);
	    cleanAndRemoveSocket((apr_socket_t **)&NDProxy.ret_pfd[i].desc.s, CLEAN);
	    continue;
	  }

	  /* Extract the IP : Port of the receiving connection*/
	  get_dst_addr(NDProxy.ret_pfd[i].desc.s, ip, 128, &port);

	  ndApplication = NDProxyInstance[appIndex];
	  //NDTL0(trace_log_key, NULL, TL_INFO, "NDApplication is switched 'appName - %s'", ndApplication->NDBCISetting.appName);
    if(tcpStorage.perThreadOpsVar.opCount > 0) {
      processThreadOps(&tcpStorage.perThreadOpsVar);
      ndApplication = NDProxyInstance[appIndex];
    }

	  if (NDProxy.ret_pfd[i].rtnevents & APR_POLLHUP || NDProxy.ret_pfd[i].rtnevents & APR_POLLERR)
	  {
	    if(NDProxy.ret_pfd[i].rtnevents & APR_POLLHUP)
	    {
	      NDTL0(trace_log_key, NULL, TL_INFO, "Received POLLHUP event");
	    }
	    else
	    { 
	      NDTL0(trace_log_key, NULL, TL_INFO, "Received POLLERR event");
	    }

	    // TODO : Make functions to handle these specfic event cases.
	    if (NDProxy.ret_pfd[i].desc.s == NDProxyInstance[appIndex]->processInfo[procIndex].ctrlSocket)
	    {
	      NDTL0(trace_log_key, NULL, TL_INFO, "HANGUP received. (Control) [pid = %d] ['%s':'%d']",
		  NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
        closeAllConnectionsOfProcess(appIndex, procIndex);
        NDProxyInstance[appIndex]->activeProcessCount--;
	    }
	    else if (NDProxy.ret_pfd[i].desc.s == NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket)
	    {
	      NDTL0(trace_log_key, NULL, TL_INFO, "HANGUP received. (Data) [pid = %d] ['%s':'%d']",
		  NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
	      cleanAndRemoveSocket(&NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket, CLEAN);
	      NDProxyInstance[appIndex]->processInfo[procIndex].dcFd = -1;

	      NDProxyInstance[appIndex]->processInfo[procIndex].data_len = 0;
	      if(NDProxyInstance[appIndex]->processInfo[procIndex].data)
		FREE_AND_MAKE_NULL(NDProxyInstance[appIndex]->processInfo[procIndex].data, "Free'ing data buffer", -1);

	      if(!NDProxyInstance[appIndex]->processInfo[procIndex].ctrlSocket && !NDProxyInstance[appIndex]->processInfo[procIndex].asSocket)
		NDProxyInstance[appIndex]->processInfo[procIndex].processId = -1;
	    }
	    else if (NDProxy.ret_pfd[i].desc.s == NDProxyInstance[appIndex]->processInfo[procIndex].asSocket)
	    { 
	      NDTL0(trace_log_key, NULL, TL_INFO, "HANGUP received. (AS) [pid = %d] ['%s':'%d']",
		  NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
	      cleanAndRemoveSocket(&NDProxyInstance[appIndex]->processInfo[procIndex].asSocket, CLEAN);
	      NDProxyInstance[appIndex]->processInfo[procIndex].asFd = -1;

	      if(!NDProxyInstance[appIndex]->processInfo[procIndex].ctrlSocket && !NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket)
		NDProxyInstance[appIndex]->processInfo[procIndex].processId = -1;
	    }
	    else
	    {
	      NDTL0(trace_log_key, NULL, TL_INFO, "Received HANGUP event on unknown fd. Ignoring...");
	      cleanAndRemoveSocket((apr_socket_t **)&NDProxy.ret_pfd[i].desc.s, CLEAN);
	    }
	  }
	  // If any POLLIN event is received, then we read on that particular socket.
	  else if(NDProxy.ret_pfd[i].rtnevents & APR_POLLIN)
	  {
	    // If the event is received on  Control connection for the specified appIndex-processIndex
	    if(NDProxy.ret_pfd[i].desc.s == NDProxyInstance[appIndex]->processInfo[procIndex].ctrlSocket)
	    {
	      NDTL4(trace_log_key, NULL, "CLIENT->CTRL_PROCESS", "Received for app %s, pid %d '(%s:%d)' ",
		  NDProxyInstance[appIndex]->appName, NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);

	      rv = readMessageFromSocket(NDProxyInstance[appIndex]->processInfo[procIndex].ctrlSocket, &bufInfo, &ptr, NULL);
	      if (rv != APR_SUCCESS)
	      {
		NDTL0(trace_log_key, NULL, TL_INFO, "Read Failed (Control) [pid = %d] ['%s':'%d']. Closing connection.", 
		    NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
    closeAllConnectionsOfProcess(appIndex, procIndex);
		NDProxyInstance[appIndex]->activeProcessCount--;
	      }
	    }
	    // If the event is received on  Data connection for the specified appIndex-processIndex
	    else if (NDProxy.ret_pfd[i].desc.s == NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket)
	    {
	      NDTL3(trace_log_key, NULL, "CLIENT->DATA_PROCESS", "Data Received for app %s, pid %d '(%s:%d)' ",
		  NDProxyInstance[appIndex]->appName, NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);

	      rv = readDataFromSocket(NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket, &dataBufInfo, appIndex, procIndex);
	      if (rv != APR_SUCCESS)
	      {
		NDTL1(trace_log_key, NULL, TL_INFO, "Read Failed (Data) [pid = %d] ['%s':'%d']. Closing connection.", 
		    NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
		cleanAndRemoveSocket(&NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket, CLEAN);
		NDProxyInstance[appIndex]->processInfo[procIndex].dcFd = -1;
		NDProxyInstance[appIndex]->processInfo[procIndex].data_len = 0;
		if(NDProxyInstance[appIndex]->processInfo[procIndex].data)
		  FREE_AND_MAKE_NULL(NDProxyInstance[appIndex]->processInfo[procIndex].data, "Free'ing data buffer", -1);

		if(!NDProxyInstance[appIndex]->processInfo[procIndex].ctrlSocket && !NDProxyInstance[appIndex]->processInfo[procIndex].asSocket)
		  NDProxyInstance[appIndex]->processInfo[procIndex].processId = -1;
	      }
	    }
	    // If the event is received on AS connection for the specified appIndex-processIndex
	    else if (NDProxy.ret_pfd[i].desc.s == NDProxyInstance[appIndex]->processInfo[procIndex].asSocket)
	    {  
	      NDTL3(trace_log_key, NULL, "CLIENT->AS_PROCESS", "AS Received for app %s, pid %d on sockfd '(%s:%d)' ",
		  NDProxyInstance[appIndex]->appName, NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
	      rv = readMessageFromSocket(NDProxyInstance[appIndex]->processInfo[procIndex].asSocket, &bufInfo, &ptr, NULL);
	      if (rv != APR_SUCCESS)
	      {
		NDTL0(trace_log_key, NULL, TL_INFO, "Read Failed (AS) [pid = %d] ['%s':'%d']. Closing connection.", 
		    NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
		cleanAndRemoveSocket(&NDProxyInstance[appIndex]->processInfo[procIndex].asSocket, CLEAN);
		NDProxyInstance[appIndex]->processInfo[procIndex].asFd = -1;

		if(!NDProxyInstance[appIndex]->processInfo[procIndex].ctrlSocket && !NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket)
		  NDProxyInstance[appIndex]->processInfo[procIndex].processId = -1;
	      }
	    }
	    else if (NDProxy.ret_pfd[i].desc.s == NDProxyInstance[appIndex]->processInfo[procIndex].disSocket)
            {
              NDTL3(trace_log_key, NULL, "CLIENT->Temp_Connection", "AS Received for app %s, pid %d on sockfd '(%s:%d)' ",
                  NDProxyInstance[appIndex]->appName, NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
              rv = readMessageFromSocket(NDProxyInstance[appIndex]->processInfo[procIndex].disSocket, &disbufInfo, &ptr, NULL);
              if (rv != APR_SUCCESS && !disSkt)
              {
                NDTL3(trace_log_key, NULL, TL_INFO, "Read Failed (DISCOVERY) [pid = %d] ['%s':'%d']. Closing connection.",
                   NDProxyInstance[appIndex]->processInfo[procIndex].processId, ip, port);
                cleanAndRemoveSocket(&NDProxyInstance[appIndex]->processInfo[procIndex].disSocket, CLEAN);
               NDProxyInstance[appIndex]->processInfo[procIndex].disFd = -1;

               if(!NDProxyInstance[appIndex]->processInfo[procIndex].ctrlSocket && !NDProxyInstance[appIndex]->processInfo[procIndex].dcSocket && !NDProxyInstance[appIndex]->processInfo[procIndex].asSocket)
                  NDProxyInstance[appIndex]->processInfo[procIndex].processId = -1;
              }
	      
            }
	    else    // This is the case when some event is received on unknown socket. 
	    {
	      NDTL0(trace_log_key, NULL, TL_INFO, "Received EPOLLIN event on unknown fd. Ignoring...");
	      cleanAndRemoveSocket((apr_socket_t **)&NDProxy.ret_pfd[i].desc.s, CLEAN);
	    }
	  }
	  else
	  {
	    NDTL0(trace_log_key, NULL, TL_INFO, "Received unknown event on socket, '%s:%d'", ip, port);
	  }
	}
	else
	{
	  NDTL0(trace_log_key, NULL, TL_INFO, "Received event on, '%s:%d'. Closed the connection.", ip, port);
	  cleanAndRemoveSocket((apr_socket_t **)&NDProxy.ret_pfd[i].desc.s, CLEAN); 
	}
      }
    }
  }
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}

void ndProxyInitLws() {
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Called");

  apr_initialize();
  atexit(apr_terminate);

  apr_pool_create(&NDProxy.proxy_pool, NULL);

  apr_signal_init(NDProxy.proxy_pool);
  apr_signal(13, SIG_IGN);    //SIGPIPE
  apr_signal(10, resetAndCleanProxy);    // for graceful exit
  apr_signal(2, resetAndCleanProxy);    // for graceful exit

  //initialize NDProxy structure
  //initializeProxyStruct();
  setResetSocketBuffer(&bufInfo);
  setResetSocketBuffer(&dataBufInfo);

  // create socket based on the connection mode i.e. Unix domain socket/INET socket
  // The socket created will be INET in case of distributed Client-Server and UNIX if both
  // lie on  the same docker.
  while (1)
  {
    if (!NDProxy.proxyConType)
    {
      create_proxy_server(NDProxy.proxyPort, &bufInfo);
    }
    else
    {
      // create Unix domain socket
      //NDProxy.skt = create_unix_domain_socket_and_bind(NDProxy.ndProxyUnixDomainPath, error_buf, ERROR_BUFFER_SIZE, NDProxy.proxy_pool);
    }

    if (!NDProxy.lws_handler.context)
    {
      NDTL0(trace_log_key, NULL, TL_ERROR, "Proxy Server Failed to Start");
      NDTL0(trace_log_key, NULL, TL_ERROR, "Retry in 10 seconds.");
      apr_sleep(10000000);
      continue;
    }
    else {
      //if (NDProxy.skt)
      //get_src_addr_wrapper(NDProxy.skt, NDProxy.proxyHost, BUFFER_LEN, (int*)&NDProxy.proxyPort,ndApplication->proxy_lws_active);
      break;
    }
  }

  NDTL0(trace_log_key, NULL, TL_INFO, "Proxy Started at '%% s:%d'", /*NDProxy.proxyHost,*/ NDProxy.proxyPort);

  //Start listening to any message on control(this main) thread.

  while (1)				// Loop 1
  {
    int rv = 0;
    while (rv >= 0) {
      rv = lws_service(NDProxy.lws_handler.context, 10000000);
    }
  }
  NDTL4(trace_log_key, NULL, TL_INFO, "Method Exit");
}

/********************************** 
 * Entry point for ND Proxy Server
 **********************************/
/*
   int main()
   {
//initialize the ND Proxy Server
ndProxyInit();
return 0;
}
*/
