int rmb_send_sys_log_for_api()

in eventmesh-sdks/eventmesh-sdk-c/src/message_log_api.c [51:193]

• 129 lines of code
• 12 McCabe index (conditional complexity)

int rmb_send_sys_log_for_api (stContextProxy * pContextProxy,
                              const char *iLogLevel, const char *cConsumerId,
                              const char *cLogName, const char *cContent,
                              const char *extFields)
{

  if (pContextProxy == NULL || NULL == extFields)
  {
    LOGRMB (RMB_LOG_ERROR, "pContextProxy or extFields is null");
    return -1;
  }

  StWemqThreadMsg stThreadMsg;
  memset (&stThreadMsg, 0x00, sizeof (StWemqThreadMsg));
  stThreadMsg.m_iCmd = THREAD_MSG_CMD_SEND_LOG;

  WEMQJSON *jsonHeader = json_object_new_object ();
  if (jsonHeader == NULL)
  {
    LOGRMB (RMB_LOG_ERROR, "json_object_new_object for jsonHeader failed");
    return -1;
  }

  json_object_object_add (jsonHeader, MSG_HEAD_COMMAND_STR,
                          json_object_new_string (SYS_LOG_TO_LOGSERVER));
  json_object_object_add (jsonHeader, MSG_HEAD_CODE_INT,
                          json_object_new_int (0));
  json_object_object_add (jsonHeader, MSG_HEAD_SEQ_INT,
                          json_object_new_int (0));

  WEMQJSON *jsonBody = json_object_new_object ();
  if (jsonBody == NULL)
  {
    LOGRMB (RMB_LOG_ERROR, "json_object_new_object for jsonBody failed");
    json_object_put (jsonHeader);
    return -1;
  }
  char cLogId[33] = { 0 };

  if (0 != get_log_id (cLogId, sizeof (cLogId)))
  {
    LOGRMB (RMB_LOG_ERROR, "get_log_id failed");
    return -1;
  }

  GetRmbNowLongTime ();
  json_object_object_add (jsonBody, LOG_MSG_COM_ID,
                          json_object_new_string (cLogId));
  json_object_object_add (jsonBody, LOG_MSG_COM_CONSUMERID,
                          json_object_new_string (cConsumerId));
  json_object_object_add (jsonBody, LOG_MSG_COM_LOGNAME,
                          json_object_new_string (cLogName));
  json_object_object_add (jsonBody, LOG_MSG_COM_TIMESTAMP,
                          json_object_new_int64 (pRmbStConfig->ulNowTtime));
  json_object_object_add (jsonBody, LOG_MSG_COM_CONTENT,
                          json_object_new_string (cContent));
  json_object_object_add (jsonBody, LOG_MSG_COM_LOGTYPE,
                          json_object_new_string ("sys"));
  json_object_object_add (jsonBody, LOG_MSG_COM_LANG,
                          json_object_new_string ("c"));
  json_object_object_add (jsonBody, LOG_MSG_COM_PROCESSID,
                          json_object_new_int ((int) getpid ()));
  json_object_object_add (jsonBody, LOG_MSG_COM_THREADID,
                          json_object_new_int64 ((long int) gettidv1 ()));
  json_object_object_add (jsonBody, LOG_MSG_COM_CONSUMERSVRID,
                          json_object_new_string (pRmbStConfig->cHostIp));
  json_object_object_add (jsonBody, LOG_MSG_COM_LEVEL,
                          json_object_new_string (iLogLevel));

  WEMQJSON *jsonExtFields = json_tokener_parse (extFields);
  WEMQJSON *tmp = NULL;
  if (jsonExtFields == NULL)
  {
    LOGRMB (RMB_LOG_ERROR, "json_tokener_parse extFields failed");
    tmp = json_tokener_parse ("{}");
    json_object_object_add (jsonBody, LOG_MSG_COM_EXTFIELDS, tmp);
  }
  else
  {
    json_object_object_add (jsonBody, LOG_MSG_COM_EXTFIELDS, jsonExtFields);
  }
  const char *header_str = json_object_get_string (jsonHeader);

  if (header_str == NULL)
  {
    LOGRMB (RMB_LOG_ERROR, "json_object_get_string for header is null");
    json_object_put (jsonBody);
    json_object_put (jsonHeader);
    return -2;
  }
  stThreadMsg.m_iHeaderLen = strlen (header_str);

  LOGRMB (RMB_LOG_INFO, "Gen thread msg header succ, len=%d, %s",
          stThreadMsg.m_iHeaderLen, header_str);
  stThreadMsg.m_pHeader =
    (char *) malloc (stThreadMsg.m_iHeaderLen * sizeof (char) + 1);
  if (stThreadMsg.m_pHeader == NULL)
  {
    LOGRMB (RMB_LOG_ERROR, "malloc for m_pHeader failed, errno=%d", errno);
    json_object_put (jsonBody);
    json_object_put (jsonHeader);
    return -1;
  }
  strncpy (stThreadMsg.m_pHeader, header_str, stThreadMsg.m_iHeaderLen);
  stThreadMsg.m_pHeader[stThreadMsg.m_iHeaderLen] = '\0';

  const char *body_str = json_object_get_string (jsonBody);
  if (body_str == NULL)
  {
    json_object_put (jsonBody);
    json_object_put (jsonHeader);
    return -1;
  }

  stThreadMsg.m_iBodyLen = strlen (body_str);
  stThreadMsg.m_pBody =
    (char *) malloc (stThreadMsg.m_iBodyLen * sizeof (char) + 1);
  if (stThreadMsg.m_pBody == NULL)
  {
    LOGRMB (RMB_LOG_ERROR, "malloc for hello body failed");
    json_object_put (jsonBody);
    json_object_put (jsonHeader);
    return -1;
  }
  memcpy (stThreadMsg.m_pBody, body_str, stThreadMsg.m_iBodyLen);
  stThreadMsg.m_pBody[stThreadMsg.m_iBodyLen] = '\0';

  json_object_put (jsonBody);
  json_object_put (jsonHeader);

  int iRet = wemq_kfifo_put (&pContextProxy->pubFifo, stThreadMsg);
  if (iRet <= 0)
  {
    LOGRMB (RMB_LOG_ERROR, "wemq_kfifo_put error!,iRet=%d", iRet);
    rmb_errno = RMB_ERROR_WORKER_PUT_FIFO_ERROR;
    return -5;
  }
  else
  {
    LOGRMB (RMB_LOG_DEBUG, "put sys log msg to fifo");
    return 0;
  }
}