src/agent/core/push.cpp

   1 /*
   2 ** NetXMS multiplatform core agent
   3 ** Copyright (C) 2003-2017 Victor Kirhenshtein
   4 **
   5 ** This program is free software; you can redistribute it and/or modify
   6 ** it under the terms of the GNU General Public License as published by
   7 ** the Free Software Foundation; either version 2 of the License, or
   8 ** (at your option) any later version.
   9 **
  10 ** This program is distributed in the hope that it will be usefu,,
  11 ** but ITHOUT ANY WARRANTY; without even the implied warranty of
  12 ** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 ** GNU General Public License for more details.
  14 **
  15 ** You should have received a copy of the GNU General Public License
  16 ** along with this program; if not, write to the Free Software
  17 ** Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  18 **
  19 ** File: push.cpp
  20 **
  21 **/
  22
  23 #include "nxagentd.h"
  24
  25 /**
  26  * Request ID
  27  */
  28 static UINT64 s_requestIdHigh = (UINT64)time(NULL) << 32;
  29 static VolatileCounter s_requestIdLow = 0;
  30
  31 /**
  32  * Push parameter's data
  33  */
  34 bool PushData(const TCHAR *parameter, const TCHAR *value, UINT32 objectId, time_t timestamp)
  35 {
  36         NXCPMessage msg;
  37         bool success = false;
  38
  39         AgentWriteDebugLog(6, _T("PushData: \"%s\" = \"%s\""), parameter, value);
  40
  41         msg.setCode(CMD_PUSH_DCI_DATA);
  42         msg.setField(VID_NAME, parameter);
  43         msg.setField(VID_VALUE, value);
  44    msg.setField(VID_OBJECT_ID, objectId);
  45    msg.setFieldFromTime(VID_TIMESTAMP, timestamp);
  46    msg.setField(VID_REQUEST_ID, s_requestIdHigh | (UINT64)InterlockedIncrement(&s_requestIdLow));
  47
  48    if (g_dwFlags & AF_SUBAGENT_LOADER)
  49    {
  50       success = SendMessageToMasterAgent(&msg);
  51    }
  52    else
  53    {
  54       MutexLock(g_hSessionListAccess);
  55       for(DWORD i = 0; i < g_dwMaxSessions; i++)
  56          if (g_pSessionList[i] != NULL)
  57             if (g_pSessionList[i]->canAcceptTraps())
  58             {
  59                g_pSessionList[i]->sendMessage(&msg);
  60                success = true;
  61             }
  62       MutexUnlock(g_hSessionListAccess);
  63    }
  64         return success;
  65 }
  66
  67 /**
  68  * Process push request
  69  */
  70 static void ProcessPushRequest(NamedPipe *pipe, void *arg)
  71 {
  72         TCHAR buffer[256];
  73
  74         AgentWriteDebugLog(5, _T("ProcessPushRequest: connection established"));
  75    PipeMessageReceiver receiver(pipe->handle(), 8192, 1048576);  // 8K initial, 1M max
  76         while(true)
  77         {
  78       MessageReceiverResult result;
  79                 NXCPMessage *msg = receiver.readMessage(5000, &result);
  80                 if (msg == NULL)
  81                         break;
  82                 AgentWriteDebugLog(6, _T("ProcessPushRequest: received message %s"), NXCPMessageCodeName(msg->getCode(), buffer));
  83                 if (msg->getCode() == CMD_PUSH_DCI_DATA)
  84                 {
  85          UINT32 objectId = msg->getFieldAsUInt32(VID_OBJECT_ID);
  86                         UINT32 count = msg->getFieldAsUInt32(VID_NUM_ITEMS);
  87          time_t timestamp = msg->getFieldAsTime(VID_TIMESTAMP);
  88                         UINT32 varId = VID_PUSH_DCI_DATA_BASE;
  89                         for(DWORD i = 0; i < count; i++)
  90                         {
  91                                 TCHAR name[MAX_PARAM_NAME], value[MAX_RESULT_LENGTH];
  92                                 msg->getFieldAsString(varId++, name, MAX_PARAM_NAME);
  93                                 msg->getFieldAsString(varId++, value, MAX_RESULT_LENGTH);
  94                                 PushData(name, value, objectId, timestamp);
  95                         }
  96                 }
  97                 delete msg;
  98         }
  99         AgentWriteDebugLog(5, _T("ProcessPushRequest: connection closed"));
 100 }
 101
 102 /**
 103  * Pipe listener for push requests
 104  */
 105 static NamedPipeListener *s_listener;
 106
 107 /**
 108  * Start push connector
 109  */
 110 void StartPushConnector()
 111 {
 112    s_listener = NamedPipeListener::create(_T("nxagentd.push"), ProcessPushRequest, NULL);
 113    if (s_listener != NULL)
 114       s_listener->start();
 115 }