InfluxDB_Functions.ttcn

///////////////////////////////////////////////////////////////////////////////
//
// Copyright (c) 2000-2018 Ericsson Telecom AB
//
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v2.0
// which accompanies this distribution, and is available at
// https://www.eclipse.org/org/documents/epl-2.0/EPL-2.0.html
///////////////////////////////////////////////////////////////////////////////
//  File:               InfluxDB_Functions.ttcn
//  Description:
//  Rev:                R1A
//  Prodnr:             LPA 108 661
//  Updated:            2017-09-01
//  Contact:            http://ttcn.ericsson.se
///////////////////////////////////////////////////////////////////////////////
module InfluxDB_Functions {

	import from InfluxDB_Definitions all;
	import from IPL4asp_Types all;
	import from IPL4asp_PortType all;

	function f_InfluxDB_init(in integer p_lgenIdx := 0) runs on InfluxDB_CT
	{
		f_InlfuxDB_log_debug(log2str(%definitionId, " started"));
		
		f_InfluxDB_connect();
		
		v_InfluxDB_ctx.reportPeriod := tsp_InfluxDB_reportPeriod;
		v_InfluxDB_ctx.localPort := tsp_InfluxDB_client_basePort + p_lgenIdx;
		
		v_InfluxDB_eventHandler := activate(as_InfluxDB_eventHandler());
		
		f_InlfuxDB_log_debug(log2str(%definitionId, " finished"));
	}
	
	function f_InfluxDB_connect() runs on InfluxDB_CT
	{
        f_InlfuxDB_log_debug(log2str(%definitionId, " started"));
        var Result vl_result;

        f_InlfuxDB_log_debug(log2str("Mapping started"));
        map(self:INFLUXDB_PCO,system:INFLUXDB_PCO);

        f_InlfuxDB_log_debug(log2str("InfluxDB context: ", v_InfluxDB_ctx));

        f_InlfuxDB_log_debug(log2str("Setting up the listening socket"));
        vl_result := f_IPL4_listen(
            INFLUXDB_PCO,
            v_InfluxDB_ctx.localHost,
            v_InfluxDB_ctx.localPort, {udp := {}},
            {{reuseAddress := {enable := true}}}
        );
        v_InfluxDB_ctx.connId_listen := vl_result.connId;

        f_InlfuxDB_log_debug(log2str("Connecting the socket to the remote"));
        vl_result := f_IPL4_connect(
            INFLUXDB_PCO,
            v_InfluxDB_ctx.remoteHost,
            v_InfluxDB_ctx.remotePort,
            v_InfluxDB_ctx.localHost,
            v_InfluxDB_ctx.localPort,
            -1, {udp := {}}, {{reuseAddress := {enable := true}}}
        );
        v_InfluxDB_ctx.connId := vl_result.connId;
        f_InlfuxDB_log_debug(log2str(%definitionId, " finished"));	  
	}

	function f_InfluxDB_send(in InfluxLineProtocol p_data) runs on InfluxDB_CT
	{
	  	f_InfluxDB_encode(p_data, v_InfluxDB_msgToSend_encoded);

		v_InfluxDB_sendAsp.connId := v_InfluxDB_ctx.connId;
		v_InfluxDB_sendAsp.remName := v_InfluxDB_ctx.remoteHost;
		v_InfluxDB_sendAsp.remPort := v_InfluxDB_ctx.remotePort;
		v_InfluxDB_sendAsp.proto := {udp := {}}

		v_InfluxDB_sendAsp.msg := char2oct(v_InfluxDB_msgToSend_encoded);

		f_InlfuxDB_log_debug(log2str("Influx: ", v_InfluxDB_msgToSend));
		INFLUXDB_PCO.send(v_InfluxDB_sendAsp);
	}

	function f_InfluxDB_encode(in InfluxLineProtocol p_prot, out charstring p_encoded)
	{
      f_InlfuxDB_log_debug(log2str("influx encoding: ", p_prot));
      p_encoded := p_prot.measurement;

      //if (sizeof(p_prot.tagSet)>0) { p_encoded := p_encoded & "," }

      for (var integer i:=0; i<sizeof(p_prot.tagSet); i:=i+1)
      {
        p_encoded := p_encoded & "," & p_prot.tagSet[i].name & "=" & p_prot.tagSet[i].val;
      }
      p_encoded := p_encoded & " "
      for (var integer i:=0; i<sizeof(p_prot.fieldSet); i:=i+1)
      {
        p_encoded := p_encoded & p_prot.fieldSet[i].name & "=" & p_prot.fieldSet[i].val;
        if (i!=sizeof(p_prot.fieldSet)-1) { p_encoded := p_encoded & "," }
      }
      f_InlfuxDB_log_debug(log2str("influx encoded: ", p_encoded));
	}
    
    function f_InfluxDB_setMeasurement(inout InfluxLineProtocol p_pdu, in charstring p_measurement)
    {
      p_pdu.measurement := p_measurement;
    }

	function f_InfluxDB_addTag(inout InfluxLineProtocol p_pdu, in InfluxAttrValue p_tag)
	{
	  p_pdu.tagSet[sizeof(p_pdu.tagSet)] := p_tag;
	}

	function f_InfluxDB_addField(inout InfluxLineProtocol p_pdu, in InfluxAttrValue p_field)
	{
	  p_pdu.fieldSet[sizeof(p_pdu.fieldSet)] := p_field;
	}
	
	function f_InfluxDB_scheduleNextReport()
	runs on InfluxDB_CT
	{
	  v_InfluxDB_reportTimer.start(v_InfluxDB_ctx.reportPeriod);
	}
	
	function f_InlfuxDB_log_debug(in charstring p_log)
	{
	  if (tsp_InfluxDB_debug){ log(p_log); }
	}
	
  altstep as_InfluxDB_eventHandler()
  runs on InfluxDB_CT
  {
    var integer vl_idx;
    
    [] v_InfluxDB_reportTimer.timeout
    {
      if (v_InfluxDB_reportHandler !=null) { v_InfluxDB_reportHandler.apply(); }

      f_InfluxDB_scheduleNextReport();
      repeat;
    }
  }
}