src/EPTF_MQTT_LocalTransport_Functions.ttcn - titan/titan.ApplicationLibraries.MQTT - Git at Google

 ///////////////////////////////////////////////////////////////////////////////
 //
 // 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:               EPTF_MQTT_LocalTransport_Functions.ttcn
 //  Description:
 //  Rev:                R1A
 //  Prodnr:             CNL 113 860
 //  Updated:            2017-09-01
 //  Contact:            http://ttcn.ericsson.se
 ///////////////////////////////////////////////////////////////////////////////
 module EPTF_MQTT_LocalTransport_Functions
 {
   import from IPL4asp_Types all;

   import from EPTF_CLL_Base_Functions all;
   import from EPTF_CLL_Common_Definitions all;
   import from EPTF_CLL_Logging_Definitions all;
   import from EPTF_CLL_Logging_Functions all;
   import from EPTF_CLL_HashMapStr2Int_Functions all;
   import from EPTF_CLL_HashMapInt2Int_Functions all;
   import from EPTF_CLL_FBQ_Functions all;
   import from EPTF_CLL_TransportCommPortIPL4_Functions all;
   import from EPTF_CLL_RBTScheduler_Functions all;

   import from EPTF_MQTT_LocalTransport_Definitions all;
   import from EPTF_MQTT_Transport_Definitions all;

   import from MQTT_v3_1_1_Types all;
   import from MQTT_v3_1_1_IPL4SizeFunction all;

   function f_EPTF_MQTT_LocalTransport_init()
   runs on EPTF_MQTT_LocalTransport_CT
   {
     if (v_EPTF_MQTT_LocalTransport_initialized) {return;}

     f_EPTF_MQTT_LocalTransport_initLogging();

     f_EPTF_CommPort_IPL4_init();

     v_EPTF_MQTT_Transport_stats := c_EPTF_MQTT_Transport_Statistics_empty;

     f_EPTF_MQTT_LocalTransport_socketDB_init();

     var f_IPL4_getMsgLen fcb_msglen := refers(f_GetMsgLengthMQTT);
     f_EPTF_CommPort_IPL4_setMsgLen4LGenType(fcb_msglen, {}, c_MQTT_Transport_LGenType);

     f_EPTF_CommPort_IPL4_setReceive({asp_Event := ?},refers(f_EPTF_MQTT_IPL4asp_handleEvent), c_MQTT_Transport_LGenType);
     f_EPTF_CommPort_IPL4_setReceive({asp_RecvFrom := ?},refers(f_EPTF_MQTT_IPL4asp_handleMessage), c_MQTT_Transport_LGenType);

     f_EPTF_Base_registerCleanup(refers(f_EPTF_MQTT_LocalTransport_cleanup));

     v_EPTF_MQTT_LocalTransport_initialized := true;
   }

   function f_EPTF_MQTT_LocalTransport_initLogging()
   runs on EPTF_MQTT_LocalTransport_CT
   {
 	f_EPTF_Logging_init_CT("MQTT_Transport");
     v_EPTF_MQTT_LocalTransport_loggingMaskId :=
     	f_EPTF_Logging_registerComponentMasks(
     		"MQTT_LocalTransport_Logging",
     		{"WARNING", "DEBUG", "DEBUGV", "ERROR"},
     		EPTF_Logging_CLL
     	);

 	if(tsp_EPTF_MQTT_LocalTransport_debug){
 	  f_EPTF_Logging_enableLocalMask(v_EPTF_MQTT_LocalTransport_loggingMaskId, c_MQTT_LocalTransport_Logging_DEBUG);
 	}
 	else {
 	  f_EPTF_Logging_disableLocalMask(v_EPTF_MQTT_LocalTransport_loggingMaskId, c_MQTT_LocalTransport_Logging_DEBUG);
 	}

 	if(tsp_EPTF_MQTT_LocalTransport_debugVerbose) {
 	  f_EPTF_Logging_enableLocalMask(v_EPTF_MQTT_LocalTransport_loggingMaskId, c_MQTT_LocalTransport_Logging_DEBUGV);
 	}
 	else {
 	  f_EPTF_Logging_disableLocalMask(v_EPTF_MQTT_LocalTransport_loggingMaskId, c_MQTT_LocalTransport_Logging_DEBUGV);
 	}
   }

   function f_EPTF_MQTT_LocalTransport_cleanup()
   runs on EPTF_MQTT_LocalTransport_CT
   {
     // Reset DBs, close connections

     f_EPTF_MQTT_LocalTransport_socketDB_cleanUp();

     vf_EPTF_MQTT_Transport_receiveMessage := null;
     vf_EPTF_MQTT_Transport_receiveEvent := null;

     v_EPTF_MQTT_LocalTransport_initialized := false;
   }

   function f_EPTF_MQTT_IPL4asp_handleEvent() runs on EPTF_MQTT_LocalTransport_CT
   {
     f_MQTT_Transport_Logging_DEBUG(log2str(": incoming event: ", v_EPTF_CommPort_IPL4_incomingMessage));

     //{ asp_Event := { connClosed := { connId := 1, remName := "127.0.0.1", remPort := 1883, locName := "127.0.0.1", locPort := 20000, proto := { tcp := { } }, userData := 0 } } }

     if (ischosen(v_EPTF_CommPort_IPL4_incomingMessage.asp_Event))
     {
       if (ischosen(v_EPTF_CommPort_IPL4_incomingMessage.asp_Event.connClosed))
       {
         var integer vl_socketIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp_connId(v_EPTF_CommPort_IPL4_incomingMessage.asp_Event.connClosed.connId);
         var SocketEntry vl_socket;
         if (f_EPTF_MQTT_LocalTransport_socketDB_get(vl_socketIdx, vl_socket))
         {
           var EPTF_MQTT_Transport_Response vl_apiRsp := c_EPTF_MQTT_Transport_Response_init;
           vl_apiRsp.sessionIdx := vl_socket.sessionIdx;
     	  vl_apiRsp.params.connectionClosed := vl_socket.localAddr;
     	  vl_apiRsp.succ := true;
     	  f_EPTF_MQTT_LocalTransport_sendApiResponse(vl_apiRsp);
         }
       }
     }

     if (vf_EPTF_MQTT_Transport_receiveEvent != null) {
       vf_EPTF_MQTT_Transport_receiveEvent.apply(v_EPTF_CommPort_IPL4_incomingMessage.asp_Event);
     }
   }

   function f_EPTF_MQTT_IPL4asp_handleMessage() runs on EPTF_MQTT_LocalTransport_CT
   {
     var MQTT_v3_1_1_Message vl_MQTT_MSG;
     var EPTF_MQTT_PDU  vl_EPTF_MQTT_PDU;

     f_EPTF_SchedulerComp_refreshSnapshotTime();

     f_MQTT_v3_1_1_dec(v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.msg, vl_MQTT_MSG);

     if (tsp_EPTF_MQTT_LocalTransport_debug) { action("MQTT transport receiving: \n", vl_MQTT_MSG); }

     if (ischosen(vl_MQTT_MSG.msg))
     {
       vl_EPTF_MQTT_PDU.pdu := vl_MQTT_MSG.msg;
       vl_EPTF_MQTT_PDU.transportParams.localAddress.hostName := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.locName;
       vl_EPTF_MQTT_PDU.transportParams.localAddress.portNumber := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.locPort;
       vl_EPTF_MQTT_PDU.transportParams.remoteAddress.hostName := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.remName;
       vl_EPTF_MQTT_PDU.transportParams.remoteAddress.portNumber := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.remPort;
       vl_EPTF_MQTT_PDU.transportParams.proto := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.proto;

       var SocketEntry vl_entry := c_SocketEntry_init;
       var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp({
         v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.locName,
         v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.locPort
       });
       f_EPTF_MQTT_LocalTransport_socketDB_get(vl_sockIdx, vl_entry);
       vl_EPTF_MQTT_PDU.sessionIdx := vl_entry.sessionIdx;
       if (tsp_EPTF_MQTT_LocalTransport_debug) { action("MQTT transport receiving: sessionIdx: ", vl_EPTF_MQTT_PDU.sessionIdx); }

       v_EPTF_MQTT_Transport_stats.nofReceivedMessages :=
         v_EPTF_MQTT_Transport_stats.nofReceivedMessages + 1.0;
       v_EPTF_MQTT_Transport_stats.nofReceivedBytes :=
         v_EPTF_MQTT_Transport_stats.nofReceivedBytes + int2float(lengthof(v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.msg));

       if (vf_EPTF_MQTT_Transport_receiveMessage != null) {
         vf_EPTF_MQTT_Transport_receiveMessage.apply(vl_EPTF_MQTT_PDU);
       }
     }
     else
     {
       f_MQTT_Transport_Logging_WARNING(
       	log2str(%definitionId, " Couldn't decode msg: ",v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.msg, " decoded: ", vl_MQTT_MSG)
       );
     }
   }

   function f_EPTF_MQTT_LocalTransport_send(in EPTF_MQTT_PDU pl_msg)
   runs on EPTF_MQTT_LocalTransport_CT
   {
     f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId, " ", pl_msg));

     var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(pl_msg.transportParams.localAddress);

     if (vl_sockIdx < 0)
     {
       if (not f_EPTF_MQTT_LocalTransport_connect(
       	pl_msg.transportParams.localAddress,
       	pl_msg.transportParams.remoteAddress,
       	pl_msg.sessionIdx, vl_sockIdx))
       {
         f_MQTT_Transport_Logging_WARNING(log2str(": couldn't create socket: ", pl_msg.transportParams.localAddress));
         return;
       }
     }
     f_EPTF_MQTT_LocalTransport_socketDB_get(vl_sockIdx, v_MQTT_LocalTransport_currentSocket);

     if (tsp_EPTF_MQTT_LocalTransport_debug) { action("MQTT transport sending: \n", pl_msg.pdu); }

     var octetstring v_encoded;
     f_MQTT_v3_1_1_enc(MQTT_v3_1_1_Message: {msg := pl_msg.pdu}, v_encoded);

     f_EPTF_MQTT_LocalTransport_IPL4_send
     (
       v_MQTT_LocalTransport_currentSocket.connId,
       pl_msg.transportParams.remoteAddress.hostName,
       pl_msg.transportParams.remoteAddress.portNumber,
       pl_msg.transportParams.proto,
       v_encoded
     );
   }

   function f_EPTF_MQTT_LocalTransport_transportApiRequest(EPTF_MQTT_Transport_Request pl_req)
   runs on EPTF_MQTT_LocalTransport_CT
   {
     if (tsp_EPTF_MQTT_LocalTransport_debug) { action("MQTT transport receiving: \n", pl_req); }

     if (ischosen(pl_req.params.startListening))
     {
       var EPTF_MQTT_Transport_Response vl_rsp := c_EPTF_MQTT_Transport_Response_init;
       vl_rsp.sessionIdx := pl_req.sessionIdx;

       var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(pl_req.params.startListening.localAddress);
       if (vl_sockIdx < 0)
       {
         if (not f_EPTF_MQTT_LocalTransport_startListening(pl_req.params.startListening.localAddress, pl_req.sessionIdx, vl_sockIdx))
         {
           f_MQTT_Transport_Logging_WARNING(log2str(": couldn't create socket: ", pl_req.params.startListening.localAddress));
           vl_rsp.succ := false;
         }
         else { vl_rsp.succ := true; }
       }
       else { vl_rsp.succ := false; }

       if (pl_req.expectResponse) { f_EPTF_MQTT_LocalTransport_sendApiResponse(vl_rsp);  }
     }
     else if (ischosen(pl_req.params.connect_))
     {
       var EPTF_MQTT_Transport_Response vl_rsp := c_EPTF_MQTT_Transport_Response_init;
       vl_rsp.sessionIdx := pl_req.sessionIdx;

       var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(pl_req.params.connect_.localAddress);
       if (vl_sockIdx < 0)
       {
         if (not f_EPTF_MQTT_LocalTransport_connect(
           pl_req.params.connect_.localAddress,
           pl_req.params.connect_.remoteAddress,
           pl_req.sessionIdx, vl_sockIdx))
         {
           f_MQTT_Transport_Logging_WARNING(log2str(": couldn't create socket: ", pl_req));
           vl_rsp.succ := false;
         }
         else { vl_rsp.succ := true; }
       }
       else { vl_rsp.succ := false; }
 	  if (pl_req.expectResponse) { f_EPTF_MQTT_LocalTransport_sendApiResponse(vl_rsp);  }
     }
     else if (ischosen(pl_req.params.close))
     {
       var EPTF_MQTT_Transport_Response vl_rsp := c_EPTF_MQTT_Transport_Response_init;
       vl_rsp.sessionIdx := pl_req.sessionIdx;

       var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(pl_req.params.close.localAddress);
       if (vl_sockIdx >= 0)
       {
         if (not f_EPTF_MQTT_LocalTransport_close(vl_sockIdx))
         {
           f_MQTT_Transport_Logging_WARNING(log2str(": couldn't close socket: ", pl_req));
           vl_rsp.succ := false;
         }
         else { vl_rsp.succ := true; }
 	  }
       else { vl_rsp.succ := false; }
       if (pl_req.expectResponse) { f_EPTF_MQTT_LocalTransport_sendApiResponse(vl_rsp); }
 	}
   }

   function f_EPTF_MQTT_LocalTransport_sendApiResponse(EPTF_MQTT_Transport_Response pl_rsp)
   runs on EPTF_MQTT_LocalTransport_CT
   {
     f_MQTT_Transport_Logging_WARNING(log2str(%definitionId,": ", pl_rsp));
     if (vf_EPTF_MQTT_Transport_apiResponse != null)
     {
       vf_EPTF_MQTT_Transport_apiResponse.apply(pl_rsp);
     }
   }

   function f_EPTF_MQTT_LocalTransport_startListening(in Socket p_local, in integer p_sessionIdx, inout integer p_idx)
   runs on EPTF_MQTT_LocalTransport_CT
   return boolean
   {
     var Result vl_result;
     var SocketEntry vl_socketEntry := c_SocketEntry_init;

     vl_result :=
       f_EPTF_CommPort_IPL4_send(
         {asp_Listen :=
           {
             locName := p_local.hostName,
             locPort := p_local.portNumber,
             proto := { tcp := {} }
           }
         },
         null,
         false,
         false,
         c_MQTT_Transport_LGenType);

     if (f_EPTF_MQTT_IPL4asp_handleResult(vl_result)) // Everything went OK
     {
       vl_socketEntry.localAddr := p_local;
       vl_socketEntry.remoteAddr := omit;
       vl_socketEntry.connId := vl_result.connId;
       vl_socketEntry.sessionIdx := p_sessionIdx;
       vl_socketEntry.state := OPENED;
     }
     else
     {
       f_MQTT_Transport_Logging_WARNING(log2str(%definitionId," returning false "));
       return false;
     }

     p_idx := f_EPTF_MQTT_LocalTransport_socketDB_add(vl_socketEntry);

     f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," returning true"));
     return true;
   }

   function f_EPTF_MQTT_LocalTransport_connect(in Socket p_local, in Socket p_remote, in integer p_sessionIdx, inout integer p_idx)
   runs on EPTF_MQTT_LocalTransport_CT
   return boolean
   {
     var Result vl_result;
     var SocketEntry vl_socketEntry := c_SocketEntry_init;

     vl_result :=
       f_EPTF_CommPort_IPL4_send(
         {asp_Connect :=
           {
     		remName := p_remote.hostName,
     		remPort := p_remote.portNumber,
             locName := p_local.hostName,
             locPort := p_local.portNumber,
             proto := { tcp := {} }
           }
         },
         null,
         false,
         false,
         c_MQTT_Transport_LGenType);

     if (f_EPTF_MQTT_IPL4asp_handleResult(vl_result)) // Everything went OK
     {
       vl_socketEntry.localAddr := p_local;
 	  vl_socketEntry.remoteAddr := p_remote;
       vl_socketEntry.connId := vl_result.connId;
       vl_socketEntry.sessionIdx := p_sessionIdx;
       vl_socketEntry.state := OPENED;
     }
     else
     {
       f_MQTT_Transport_Logging_WARNING(log2str(%definitionId," returning false "));
       return false;
     }

     p_idx := f_EPTF_MQTT_LocalTransport_socketDB_add(vl_socketEntry);

     f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," returning true"));
     return true;
   }

   function f_EPTF_MQTT_LocalTransport_close(in integer p_socketIdx)
   runs on EPTF_MQTT_LocalTransport_CT
   return boolean
   {
     var Result vl_result;
     var SocketEntry vl_sockEntry;

     f_EPTF_MQTT_LocalTransport_socketDB_get(p_socketIdx, vl_sockEntry);

     vl_result :=
       f_EPTF_CommPort_IPL4_send(
         {asp_Close :=
           {
 			id := vl_sockEntry.connId,
             proto := { tcp := {} }
           }
         },
         null,
         false,
         false,
         c_MQTT_Transport_LGenType
       );

     if (f_EPTF_MQTT_IPL4asp_handleResult(vl_result)) // Everything went OK
     {
       f_EPTF_MQTT_LocalTransport_socketDB_remove(p_socketIdx);
     }
     else
     {
       f_MQTT_Transport_Logging_WARNING(log2str(%definitionId," returning false "));
       return false;
     }

     f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," returning true"));
     return true;
   }

   function f_EPTF_MQTT_LocalTransport_IPL4_send(
     in integer pl_connId,
     in charstring pl_remName,
     in integer pl_remPort,
     in ProtoTuple  pl_proto,
     in octetstring pl_msg)
   runs on EPTF_MQTT_LocalTransport_CT
   {
     var Result v_res;

     v_res := f_EPTF_CommPort_IPL4_send(
       {
         asp_SendTo :=
         {
           connId := pl_connId,
           remName := pl_remName,
           remPort := pl_remPort,
           proto := pl_proto,
           msg := pl_msg
         }
       },
       null,
       false,
       false,
       c_MQTT_Transport_LGenType
     );

     if (ispresent(v_res.errorCode))
     {
       f_MQTT_Transport_Logging_WARNING(log2str(": message could not be sent."));
     }

     v_EPTF_MQTT_Transport_stats.nofSentMessages := v_EPTF_MQTT_Transport_stats.nofSentMessages + 1.0;
     v_EPTF_MQTT_Transport_stats.nofSentBytes := v_EPTF_MQTT_Transport_stats.nofSentBytes + int2float(lengthof(pl_msg));
   }

   function f_EPTF_MQTT_LocalTransport_socketDB_init()
   runs on EPTF_MQTT_LocalTransport_CT
   {
     f_EPTF_FBQ_initFreeBusyQueue(v_MQTT_LocalTransport_localSocketDB.queue);
     v_MQTT_LocalTransport_localSocketDB.data := {};
     v_MQTT_LocalTransport_localSocketDB.hashRef := f_EPTF_str2int_HashMap_New(c_EPTF_MQTT_LocalTransport_SocketDB);
     v_MQTT_LocalTransport_localSocketDB.connIdHashRef := f_EPTF_int2int_HashMap_New(c_EPTF_MQTT_LocalTransport_SocketDB_connId);
   }

   function f_EPTF_MQTT_LocalTransport_socketDB_cleanUp()
   runs on EPTF_MQTT_LocalTransport_CT
   {
     f_EPTF_FBQ_initFreeBusyQueue(v_MQTT_LocalTransport_localSocketDB.queue);
     v_MQTT_LocalTransport_localSocketDB.data := {};
     f_EPTF_str2int_HashMap_Delete(c_EPTF_MQTT_LocalTransport_SocketDB);
     f_EPTF_int2int_HashMap_Delete(c_EPTF_MQTT_LocalTransport_SocketDB_connId);
   }

   function f_EPTF_MQTT_LocalTransport_socketDB_add(in SocketEntry p_sock)
   runs on EPTF_MQTT_LocalTransport_CT
   return integer
   {
     var integer v_idx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(p_sock.localAddr);

     if (v_idx == -1)
     {
       v_idx := f_EPTF_FBQ_getOrCreateFreeSlot(v_MQTT_LocalTransport_localSocketDB.queue);
       f_EPTF_FBQ_moveFromFreeHeadToBusyTail(v_MQTT_LocalTransport_localSocketDB.queue);

       f_MQTT_Transport_Logging_DEBUG(log2str(%definitionId,": "," adding socket ", v_idx, " ", p_sock));

       f_EPTF_str2int_HashMap_Insert(v_MQTT_LocalTransport_localSocketDB.hashRef, f_EPTF_MQTT_Socket2String(p_sock.localAddr), v_idx);
       f_EPTF_int2int_HashMap_Insert(v_MQTT_LocalTransport_localSocketDB.connIdHashRef, p_sock.connId, v_idx);

       v_MQTT_LocalTransport_localSocketDB.data[v_idx] := p_sock;
     }

     return v_idx;
   }

   function f_EPTF_MQTT_LocalTransport_socketDB_lookUp(in Socket p_sock)
   runs on EPTF_MQTT_LocalTransport_CT
   return integer
   {
     var integer    vl_idx := -1;
     f_EPTF_str2int_HashMap_Find(v_MQTT_LocalTransport_localSocketDB.hashRef, f_EPTF_MQTT_Socket2String(p_sock), vl_idx);
     f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," looking up p_sock: ",p_sock," found at idx: ", vl_idx));
     return vl_idx;
   }

   function f_EPTF_MQTT_LocalTransport_socketDB_lookUp_connId(in integer p_connId)
   runs on EPTF_MQTT_LocalTransport_CT
   return integer
   {
     var integer    vl_idx := -1;
     f_EPTF_int2int_HashMap_Find(v_MQTT_LocalTransport_localSocketDB.connIdHashRef, p_connId, vl_idx);
     f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," looking up p_connId: ",p_connId," found at idx: ", vl_idx));
     return vl_idx;
   }

   function f_EPTF_MQTT_LocalTransport_socketDB_get(in integer p_idx, inout SocketEntry p_sock)
   runs on EPTF_MQTT_LocalTransport_CT
   return boolean
   {
     if (p_idx < sizeof(v_MQTT_LocalTransport_localSocketDB.data) and p_idx >= 0)
     {
       p_sock := v_MQTT_LocalTransport_localSocketDB.data[p_idx];
       return true;
     }
     return false;
   }

   function f_EPTF_MQTT_LocalTransport_socketDB_remove(in integer p_idx)
   runs on EPTF_MQTT_LocalTransport_CT
   {
     f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId, ": ",p_idx));

     f_EPTF_str2int_HashMap_Erase(
       v_MQTT_LocalTransport_localSocketDB.hashRef,
       f_EPTF_MQTT_Socket2String(v_MQTT_LocalTransport_localSocketDB.data[p_idx].localAddr)
     );
     f_EPTF_int2int_HashMap_Erase(
       v_MQTT_LocalTransport_localSocketDB.connIdHashRef,
       v_MQTT_LocalTransport_localSocketDB.data[p_idx].connId
     );
     v_MQTT_LocalTransport_localSocketDB.data[p_idx] := c_SocketEntry_init;
     f_EPTF_FBQ_moveFromBusyToFreeTail(p_idx, v_MQTT_LocalTransport_localSocketDB.queue);
   }

   function f_EPTF_MQTT_Socket2String(Socket p_sock)
   return charstring
   {
     return p_sock.hostName&":"&int2str(p_sock.portNumber);
   }

   function f_MQTT_Transport_Logging_VERBOSE(in @lazy charstring pl_message)
   runs on EPTF_MQTT_LocalTransport_CT
   {
     if (c_EPTF_Common_debugSwitch) {
     	f_EPTF_Logging_debugV2(pl_message, v_EPTF_MQTT_LocalTransport_loggingMaskId, {c_MQTT_LocalTransport_Logging_DEBUGV});
 	}
   }

   function f_MQTT_Transport_Logging_DEBUG(in @lazy charstring pl_message)
   runs on EPTF_MQTT_LocalTransport_CT
   {
     if (c_EPTF_Common_debugSwitch) {
 	    f_EPTF_Logging_debugV2(pl_message, v_EPTF_MQTT_LocalTransport_loggingMaskId, {c_MQTT_LocalTransport_Logging_DEBUG});
 	}
   }

   function f_MQTT_Transport_Logging_WARNING(in @lazy charstring pl_message)
   runs on EPTF_MQTT_LocalTransport_CT
   {
     if (c_EPTF_Common_debugSwitch) {
 	    f_EPTF_Logging_debugV2(pl_message, v_EPTF_MQTT_LocalTransport_loggingMaskId, {c_MQTT_LocalTransport_Logging_WARNING});
 	}
   }

   function f_MQTT_Transport_Logging_ERROR(in @lazy charstring pl_message)
   runs on EPTF_MQTT_LocalTransport_CT
   {
     if (c_EPTF_Common_debugSwitch) {
     	f_EPTF_Logging_debugV2(pl_message, v_EPTF_MQTT_LocalTransport_loggingMaskId, {c_MQTT_LocalTransport_Logging_ERROR});
 	}
   }

   function f_EPTF_MQTT_IPL4asp_handleResult(inout Result p_res)
   runs on EPTF_MQTT_LocalTransport_CT
   return boolean
   {
     f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId,": ", p_res));

     if (ispresent(p_res.errorCode) and (p_res.errorCode != IPL4_ERROR_TEMPORARILY_UNAVAILABLE))
     {
       f_MQTT_Transport_Logging_WARNING(log2str("Warning: f_EPTF_MQTT_IPL4asp_handleResult: IPL4 error: ", p_res));
       v_EPTF_MQTT_Transport_stats.nofTransportWarnings := v_EPTF_MQTT_Transport_stats.nofTransportWarnings + 1;
       return false;
     }
     return true;
   }

 }
	///////////////////////////////////////////////////////////////////////////////
	//
	// 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: EPTF_MQTT_LocalTransport_Functions.ttcn
	// Description:
	// Rev: R1A
	// Prodnr: CNL 113 860
	// Updated: 2017-09-01
	// Contact: http://ttcn.ericsson.se
	///////////////////////////////////////////////////////////////////////////////
	module EPTF_MQTT_LocalTransport_Functions
	{
	import from IPL4asp_Types all;

	import from EPTF_CLL_Base_Functions all;
	import from EPTF_CLL_Common_Definitions all;
	import from EPTF_CLL_Logging_Definitions all;
	import from EPTF_CLL_Logging_Functions all;
	import from EPTF_CLL_HashMapStr2Int_Functions all;
	import from EPTF_CLL_HashMapInt2Int_Functions all;
	import from EPTF_CLL_FBQ_Functions all;
	import from EPTF_CLL_TransportCommPortIPL4_Functions all;
	import from EPTF_CLL_RBTScheduler_Functions all;

	import from EPTF_MQTT_LocalTransport_Definitions all;
	import from EPTF_MQTT_Transport_Definitions all;

	import from MQTT_v3_1_1_Types all;
	import from MQTT_v3_1_1_IPL4SizeFunction all;

	function f_EPTF_MQTT_LocalTransport_init()
	runs on EPTF_MQTT_LocalTransport_CT
	{
	if (v_EPTF_MQTT_LocalTransport_initialized) {return;}

	f_EPTF_MQTT_LocalTransport_initLogging();

	f_EPTF_CommPort_IPL4_init();

	v_EPTF_MQTT_Transport_stats := c_EPTF_MQTT_Transport_Statistics_empty;

	f_EPTF_MQTT_LocalTransport_socketDB_init();

	var f_IPL4_getMsgLen fcb_msglen := refers(f_GetMsgLengthMQTT);
	f_EPTF_CommPort_IPL4_setMsgLen4LGenType(fcb_msglen, {}, c_MQTT_Transport_LGenType);

	f_EPTF_CommPort_IPL4_setReceive({asp_Event := ?},refers(f_EPTF_MQTT_IPL4asp_handleEvent), c_MQTT_Transport_LGenType);
	f_EPTF_CommPort_IPL4_setReceive({asp_RecvFrom := ?},refers(f_EPTF_MQTT_IPL4asp_handleMessage), c_MQTT_Transport_LGenType);

	f_EPTF_Base_registerCleanup(refers(f_EPTF_MQTT_LocalTransport_cleanup));

	v_EPTF_MQTT_LocalTransport_initialized := true;
	}

	function f_EPTF_MQTT_LocalTransport_initLogging()
	runs on EPTF_MQTT_LocalTransport_CT
	{
	f_EPTF_Logging_init_CT("MQTT_Transport");
	v_EPTF_MQTT_LocalTransport_loggingMaskId :=
	f_EPTF_Logging_registerComponentMasks(
	"MQTT_LocalTransport_Logging",
	{"WARNING", "DEBUG", "DEBUGV", "ERROR"},
	EPTF_Logging_CLL
	);

	if(tsp_EPTF_MQTT_LocalTransport_debug){
	f_EPTF_Logging_enableLocalMask(v_EPTF_MQTT_LocalTransport_loggingMaskId, c_MQTT_LocalTransport_Logging_DEBUG);
	}
	else {
	f_EPTF_Logging_disableLocalMask(v_EPTF_MQTT_LocalTransport_loggingMaskId, c_MQTT_LocalTransport_Logging_DEBUG);
	}

	if(tsp_EPTF_MQTT_LocalTransport_debugVerbose) {
	f_EPTF_Logging_enableLocalMask(v_EPTF_MQTT_LocalTransport_loggingMaskId, c_MQTT_LocalTransport_Logging_DEBUGV);
	}
	else {
	f_EPTF_Logging_disableLocalMask(v_EPTF_MQTT_LocalTransport_loggingMaskId, c_MQTT_LocalTransport_Logging_DEBUGV);
	}
	}

	function f_EPTF_MQTT_LocalTransport_cleanup()
	runs on EPTF_MQTT_LocalTransport_CT
	{
	// Reset DBs, close connections

	f_EPTF_MQTT_LocalTransport_socketDB_cleanUp();

	vf_EPTF_MQTT_Transport_receiveMessage := null;
	vf_EPTF_MQTT_Transport_receiveEvent := null;

	v_EPTF_MQTT_LocalTransport_initialized := false;
	}

	function f_EPTF_MQTT_IPL4asp_handleEvent() runs on EPTF_MQTT_LocalTransport_CT
	{
	f_MQTT_Transport_Logging_DEBUG(log2str(": incoming event: ", v_EPTF_CommPort_IPL4_incomingMessage));

	//{ asp_Event := { connClosed := { connId := 1, remName := "127.0.0.1", remPort := 1883, locName := "127.0.0.1", locPort := 20000, proto := { tcp := { } }, userData := 0 } } }

	if (ischosen(v_EPTF_CommPort_IPL4_incomingMessage.asp_Event))
	{
	if (ischosen(v_EPTF_CommPort_IPL4_incomingMessage.asp_Event.connClosed))
	{
	var integer vl_socketIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp_connId(v_EPTF_CommPort_IPL4_incomingMessage.asp_Event.connClosed.connId);
	var SocketEntry vl_socket;
	if (f_EPTF_MQTT_LocalTransport_socketDB_get(vl_socketIdx, vl_socket))
	{
	var EPTF_MQTT_Transport_Response vl_apiRsp := c_EPTF_MQTT_Transport_Response_init;
	vl_apiRsp.sessionIdx := vl_socket.sessionIdx;
	vl_apiRsp.params.connectionClosed := vl_socket.localAddr;
	vl_apiRsp.succ := true;
	f_EPTF_MQTT_LocalTransport_sendApiResponse(vl_apiRsp);
	}
	}
	}

	if (vf_EPTF_MQTT_Transport_receiveEvent != null) {
	vf_EPTF_MQTT_Transport_receiveEvent.apply(v_EPTF_CommPort_IPL4_incomingMessage.asp_Event);
	}
	}

	function f_EPTF_MQTT_IPL4asp_handleMessage() runs on EPTF_MQTT_LocalTransport_CT
	{
	var MQTT_v3_1_1_Message vl_MQTT_MSG;
	var EPTF_MQTT_PDU vl_EPTF_MQTT_PDU;

	f_EPTF_SchedulerComp_refreshSnapshotTime();

	f_MQTT_v3_1_1_dec(v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.msg, vl_MQTT_MSG);

	if (tsp_EPTF_MQTT_LocalTransport_debug) { action("MQTT transport receiving: \n", vl_MQTT_MSG); }

	if (ischosen(vl_MQTT_MSG.msg))
	{
	vl_EPTF_MQTT_PDU.pdu := vl_MQTT_MSG.msg;
	vl_EPTF_MQTT_PDU.transportParams.localAddress.hostName := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.locName;
	vl_EPTF_MQTT_PDU.transportParams.localAddress.portNumber := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.locPort;
	vl_EPTF_MQTT_PDU.transportParams.remoteAddress.hostName := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.remName;
	vl_EPTF_MQTT_PDU.transportParams.remoteAddress.portNumber := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.remPort;
	vl_EPTF_MQTT_PDU.transportParams.proto := v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.proto;

	var SocketEntry vl_entry := c_SocketEntry_init;
	var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp({
	v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.locName,
	v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.locPort
	});
	f_EPTF_MQTT_LocalTransport_socketDB_get(vl_sockIdx, vl_entry);
	vl_EPTF_MQTT_PDU.sessionIdx := vl_entry.sessionIdx;
	if (tsp_EPTF_MQTT_LocalTransport_debug) { action("MQTT transport receiving: sessionIdx: ", vl_EPTF_MQTT_PDU.sessionIdx); }

	v_EPTF_MQTT_Transport_stats.nofReceivedMessages :=
	v_EPTF_MQTT_Transport_stats.nofReceivedMessages + 1.0;
	v_EPTF_MQTT_Transport_stats.nofReceivedBytes :=
	v_EPTF_MQTT_Transport_stats.nofReceivedBytes + int2float(lengthof(v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.msg));

	if (vf_EPTF_MQTT_Transport_receiveMessage != null) {
	vf_EPTF_MQTT_Transport_receiveMessage.apply(vl_EPTF_MQTT_PDU);
	}
	}
	else
	{
	f_MQTT_Transport_Logging_WARNING(
	log2str(%definitionId, " Couldn't decode msg: ",v_EPTF_CommPort_IPL4_incomingMessage.asp_RecvFrom.msg, " decoded: ", vl_MQTT_MSG)
	);
	}
	}

	function f_EPTF_MQTT_LocalTransport_send(in EPTF_MQTT_PDU pl_msg)
	runs on EPTF_MQTT_LocalTransport_CT
	{
	f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId, " ", pl_msg));

	var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(pl_msg.transportParams.localAddress);

	if (vl_sockIdx < 0)
	{
	if (not f_EPTF_MQTT_LocalTransport_connect(
	pl_msg.transportParams.localAddress,
	pl_msg.transportParams.remoteAddress,
	pl_msg.sessionIdx, vl_sockIdx))
	{
	f_MQTT_Transport_Logging_WARNING(log2str(": couldn't create socket: ", pl_msg.transportParams.localAddress));
	return;
	}
	}
	f_EPTF_MQTT_LocalTransport_socketDB_get(vl_sockIdx, v_MQTT_LocalTransport_currentSocket);

	if (tsp_EPTF_MQTT_LocalTransport_debug) { action("MQTT transport sending: \n", pl_msg.pdu); }

	var octetstring v_encoded;
	f_MQTT_v3_1_1_enc(MQTT_v3_1_1_Message: {msg := pl_msg.pdu}, v_encoded);

	f_EPTF_MQTT_LocalTransport_IPL4_send
	(
	v_MQTT_LocalTransport_currentSocket.connId,
	pl_msg.transportParams.remoteAddress.hostName,
	pl_msg.transportParams.remoteAddress.portNumber,
	pl_msg.transportParams.proto,
	v_encoded
	);
	}

	function f_EPTF_MQTT_LocalTransport_transportApiRequest(EPTF_MQTT_Transport_Request pl_req)
	runs on EPTF_MQTT_LocalTransport_CT
	{
	if (tsp_EPTF_MQTT_LocalTransport_debug) { action("MQTT transport receiving: \n", pl_req); }

	if (ischosen(pl_req.params.startListening))
	{
	var EPTF_MQTT_Transport_Response vl_rsp := c_EPTF_MQTT_Transport_Response_init;
	vl_rsp.sessionIdx := pl_req.sessionIdx;

	var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(pl_req.params.startListening.localAddress);
	if (vl_sockIdx < 0)
	{
	if (not f_EPTF_MQTT_LocalTransport_startListening(pl_req.params.startListening.localAddress, pl_req.sessionIdx, vl_sockIdx))
	{
	f_MQTT_Transport_Logging_WARNING(log2str(": couldn't create socket: ", pl_req.params.startListening.localAddress));
	vl_rsp.succ := false;
	}
	else { vl_rsp.succ := true; }
	}
	else { vl_rsp.succ := false; }

	if (pl_req.expectResponse) { f_EPTF_MQTT_LocalTransport_sendApiResponse(vl_rsp); }
	}
	else if (ischosen(pl_req.params.connect_))
	{
	var EPTF_MQTT_Transport_Response vl_rsp := c_EPTF_MQTT_Transport_Response_init;
	vl_rsp.sessionIdx := pl_req.sessionIdx;

	var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(pl_req.params.connect_.localAddress);
	if (vl_sockIdx < 0)
	{
	if (not f_EPTF_MQTT_LocalTransport_connect(
	pl_req.params.connect_.localAddress,
	pl_req.params.connect_.remoteAddress,
	pl_req.sessionIdx, vl_sockIdx))
	{
	f_MQTT_Transport_Logging_WARNING(log2str(": couldn't create socket: ", pl_req));
	vl_rsp.succ := false;
	}
	else { vl_rsp.succ := true; }
	}
	else { vl_rsp.succ := false; }
	if (pl_req.expectResponse) { f_EPTF_MQTT_LocalTransport_sendApiResponse(vl_rsp); }
	}
	else if (ischosen(pl_req.params.close))
	{
	var EPTF_MQTT_Transport_Response vl_rsp := c_EPTF_MQTT_Transport_Response_init;
	vl_rsp.sessionIdx := pl_req.sessionIdx;

	var integer vl_sockIdx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(pl_req.params.close.localAddress);
	if (vl_sockIdx >= 0)
	{
	if (not f_EPTF_MQTT_LocalTransport_close(vl_sockIdx))
	{
	f_MQTT_Transport_Logging_WARNING(log2str(": couldn't close socket: ", pl_req));
	vl_rsp.succ := false;
	}
	else { vl_rsp.succ := true; }
	}
	else { vl_rsp.succ := false; }
	if (pl_req.expectResponse) { f_EPTF_MQTT_LocalTransport_sendApiResponse(vl_rsp); }
	}
	}

	function f_EPTF_MQTT_LocalTransport_sendApiResponse(EPTF_MQTT_Transport_Response pl_rsp)
	runs on EPTF_MQTT_LocalTransport_CT
	{
	f_MQTT_Transport_Logging_WARNING(log2str(%definitionId,": ", pl_rsp));
	if (vf_EPTF_MQTT_Transport_apiResponse != null)
	{
	vf_EPTF_MQTT_Transport_apiResponse.apply(pl_rsp);
	}
	}

	function f_EPTF_MQTT_LocalTransport_startListening(in Socket p_local, in integer p_sessionIdx, inout integer p_idx)
	runs on EPTF_MQTT_LocalTransport_CT
	return boolean
	{
	var Result vl_result;
	var SocketEntry vl_socketEntry := c_SocketEntry_init;

	vl_result :=
	f_EPTF_CommPort_IPL4_send(
	{asp_Listen :=
	{
	locName := p_local.hostName,
	locPort := p_local.portNumber,
	proto := { tcp := {} }
	}
	},
	null,
	false,
	false,
	c_MQTT_Transport_LGenType);

	if (f_EPTF_MQTT_IPL4asp_handleResult(vl_result)) // Everything went OK
	{
	vl_socketEntry.localAddr := p_local;
	vl_socketEntry.remoteAddr := omit;
	vl_socketEntry.connId := vl_result.connId;
	vl_socketEntry.sessionIdx := p_sessionIdx;
	vl_socketEntry.state := OPENED;
	}
	else
	{
	f_MQTT_Transport_Logging_WARNING(log2str(%definitionId," returning false "));
	return false;
	}

	p_idx := f_EPTF_MQTT_LocalTransport_socketDB_add(vl_socketEntry);

	f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," returning true"));
	return true;
	}

	function f_EPTF_MQTT_LocalTransport_connect(in Socket p_local, in Socket p_remote, in integer p_sessionIdx, inout integer p_idx)
	runs on EPTF_MQTT_LocalTransport_CT
	return boolean
	{
	var Result vl_result;
	var SocketEntry vl_socketEntry := c_SocketEntry_init;

	vl_result :=
	f_EPTF_CommPort_IPL4_send(
	{asp_Connect :=
	{
	remName := p_remote.hostName,
	remPort := p_remote.portNumber,
	locName := p_local.hostName,
	locPort := p_local.portNumber,
	proto := { tcp := {} }
	}
	},
	null,
	false,
	false,
	c_MQTT_Transport_LGenType);

	if (f_EPTF_MQTT_IPL4asp_handleResult(vl_result)) // Everything went OK
	{
	vl_socketEntry.localAddr := p_local;
	vl_socketEntry.remoteAddr := p_remote;
	vl_socketEntry.connId := vl_result.connId;
	vl_socketEntry.sessionIdx := p_sessionIdx;
	vl_socketEntry.state := OPENED;
	}
	else
	{
	f_MQTT_Transport_Logging_WARNING(log2str(%definitionId," returning false "));
	return false;
	}

	p_idx := f_EPTF_MQTT_LocalTransport_socketDB_add(vl_socketEntry);

	f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," returning true"));
	return true;
	}

	function f_EPTF_MQTT_LocalTransport_close(in integer p_socketIdx)
	runs on EPTF_MQTT_LocalTransport_CT
	return boolean
	{
	var Result vl_result;
	var SocketEntry vl_sockEntry;

	f_EPTF_MQTT_LocalTransport_socketDB_get(p_socketIdx, vl_sockEntry);

	vl_result :=
	f_EPTF_CommPort_IPL4_send(
	{asp_Close :=
	{
	id := vl_sockEntry.connId,
	proto := { tcp := {} }
	}
	},
	null,
	false,
	false,
	c_MQTT_Transport_LGenType
	);

	if (f_EPTF_MQTT_IPL4asp_handleResult(vl_result)) // Everything went OK
	{
	f_EPTF_MQTT_LocalTransport_socketDB_remove(p_socketIdx);
	}
	else
	{
	f_MQTT_Transport_Logging_WARNING(log2str(%definitionId," returning false "));
	return false;
	}

	f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," returning true"));
	return true;
	}

	function f_EPTF_MQTT_LocalTransport_IPL4_send(
	in integer pl_connId,
	in charstring pl_remName,
	in integer pl_remPort,
	in ProtoTuple pl_proto,
	in octetstring pl_msg)
	runs on EPTF_MQTT_LocalTransport_CT
	{
	var Result v_res;

	v_res := f_EPTF_CommPort_IPL4_send(
	{
	asp_SendTo :=
	{
	connId := pl_connId,
	remName := pl_remName,
	remPort := pl_remPort,
	proto := pl_proto,
	msg := pl_msg
	}
	},
	null,
	false,
	false,
	c_MQTT_Transport_LGenType
	);

	if (ispresent(v_res.errorCode))
	{
	f_MQTT_Transport_Logging_WARNING(log2str(": message could not be sent."));
	}

	v_EPTF_MQTT_Transport_stats.nofSentMessages := v_EPTF_MQTT_Transport_stats.nofSentMessages + 1.0;
	v_EPTF_MQTT_Transport_stats.nofSentBytes := v_EPTF_MQTT_Transport_stats.nofSentBytes + int2float(lengthof(pl_msg));
	}

	function f_EPTF_MQTT_LocalTransport_socketDB_init()
	runs on EPTF_MQTT_LocalTransport_CT
	{
	f_EPTF_FBQ_initFreeBusyQueue(v_MQTT_LocalTransport_localSocketDB.queue);
	v_MQTT_LocalTransport_localSocketDB.data := {};
	v_MQTT_LocalTransport_localSocketDB.hashRef := f_EPTF_str2int_HashMap_New(c_EPTF_MQTT_LocalTransport_SocketDB);
	v_MQTT_LocalTransport_localSocketDB.connIdHashRef := f_EPTF_int2int_HashMap_New(c_EPTF_MQTT_LocalTransport_SocketDB_connId);
	}

	function f_EPTF_MQTT_LocalTransport_socketDB_cleanUp()
	runs on EPTF_MQTT_LocalTransport_CT
	{
	f_EPTF_FBQ_initFreeBusyQueue(v_MQTT_LocalTransport_localSocketDB.queue);
	v_MQTT_LocalTransport_localSocketDB.data := {};
	f_EPTF_str2int_HashMap_Delete(c_EPTF_MQTT_LocalTransport_SocketDB);
	f_EPTF_int2int_HashMap_Delete(c_EPTF_MQTT_LocalTransport_SocketDB_connId);
	}

	function f_EPTF_MQTT_LocalTransport_socketDB_add(in SocketEntry p_sock)
	runs on EPTF_MQTT_LocalTransport_CT
	return integer
	{
	var integer v_idx := f_EPTF_MQTT_LocalTransport_socketDB_lookUp(p_sock.localAddr);

	if (v_idx == -1)
	{
	v_idx := f_EPTF_FBQ_getOrCreateFreeSlot(v_MQTT_LocalTransport_localSocketDB.queue);
	f_EPTF_FBQ_moveFromFreeHeadToBusyTail(v_MQTT_LocalTransport_localSocketDB.queue);

	f_MQTT_Transport_Logging_DEBUG(log2str(%definitionId,": "," adding socket ", v_idx, " ", p_sock));

	f_EPTF_str2int_HashMap_Insert(v_MQTT_LocalTransport_localSocketDB.hashRef, f_EPTF_MQTT_Socket2String(p_sock.localAddr), v_idx);
	f_EPTF_int2int_HashMap_Insert(v_MQTT_LocalTransport_localSocketDB.connIdHashRef, p_sock.connId, v_idx);

	v_MQTT_LocalTransport_localSocketDB.data[v_idx] := p_sock;
	}

	return v_idx;
	}

	function f_EPTF_MQTT_LocalTransport_socketDB_lookUp(in Socket p_sock)
	runs on EPTF_MQTT_LocalTransport_CT
	return integer
	{
	var integer vl_idx := -1;
	f_EPTF_str2int_HashMap_Find(v_MQTT_LocalTransport_localSocketDB.hashRef, f_EPTF_MQTT_Socket2String(p_sock), vl_idx);
	f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," looking up p_sock: ",p_sock," found at idx: ", vl_idx));
	return vl_idx;
	}

	function f_EPTF_MQTT_LocalTransport_socketDB_lookUp_connId(in integer p_connId)
	runs on EPTF_MQTT_LocalTransport_CT
	return integer
	{
	var integer vl_idx := -1;
	f_EPTF_int2int_HashMap_Find(v_MQTT_LocalTransport_localSocketDB.connIdHashRef, p_connId, vl_idx);
	f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId," looking up p_connId: ",p_connId," found at idx: ", vl_idx));
	return vl_idx;
	}

	function f_EPTF_MQTT_LocalTransport_socketDB_get(in integer p_idx, inout SocketEntry p_sock)
	runs on EPTF_MQTT_LocalTransport_CT
	return boolean
	{
	if (p_idx < sizeof(v_MQTT_LocalTransport_localSocketDB.data) and p_idx >= 0)
	{
	p_sock := v_MQTT_LocalTransport_localSocketDB.data[p_idx];
	return true;
	}
	return false;
	}

	function f_EPTF_MQTT_LocalTransport_socketDB_remove(in integer p_idx)
	runs on EPTF_MQTT_LocalTransport_CT
	{
	f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId, ": ",p_idx));

	f_EPTF_str2int_HashMap_Erase(
	v_MQTT_LocalTransport_localSocketDB.hashRef,
	f_EPTF_MQTT_Socket2String(v_MQTT_LocalTransport_localSocketDB.data[p_idx].localAddr)
	);
	f_EPTF_int2int_HashMap_Erase(
	v_MQTT_LocalTransport_localSocketDB.connIdHashRef,
	v_MQTT_LocalTransport_localSocketDB.data[p_idx].connId
	);
	v_MQTT_LocalTransport_localSocketDB.data[p_idx] := c_SocketEntry_init;
	f_EPTF_FBQ_moveFromBusyToFreeTail(p_idx, v_MQTT_LocalTransport_localSocketDB.queue);
	}

	function f_EPTF_MQTT_Socket2String(Socket p_sock)
	return charstring
	{
	return p_sock.hostName&":"&int2str(p_sock.portNumber);
	}

	function f_MQTT_Transport_Logging_VERBOSE(in @lazy charstring pl_message)
	runs on EPTF_MQTT_LocalTransport_CT
	{
	if (c_EPTF_Common_debugSwitch) {
	f_EPTF_Logging_debugV2(pl_message, v_EPTF_MQTT_LocalTransport_loggingMaskId, {c_MQTT_LocalTransport_Logging_DEBUGV});
	}
	}

	function f_MQTT_Transport_Logging_DEBUG(in @lazy charstring pl_message)
	runs on EPTF_MQTT_LocalTransport_CT
	{
	if (c_EPTF_Common_debugSwitch) {
	f_EPTF_Logging_debugV2(pl_message, v_EPTF_MQTT_LocalTransport_loggingMaskId, {c_MQTT_LocalTransport_Logging_DEBUG});
	}
	}

	function f_MQTT_Transport_Logging_WARNING(in @lazy charstring pl_message)
	runs on EPTF_MQTT_LocalTransport_CT
	{
	if (c_EPTF_Common_debugSwitch) {
	f_EPTF_Logging_debugV2(pl_message, v_EPTF_MQTT_LocalTransport_loggingMaskId, {c_MQTT_LocalTransport_Logging_WARNING});
	}
	}

	function f_MQTT_Transport_Logging_ERROR(in @lazy charstring pl_message)
	runs on EPTF_MQTT_LocalTransport_CT
	{
	if (c_EPTF_Common_debugSwitch) {
	f_EPTF_Logging_debugV2(pl_message, v_EPTF_MQTT_LocalTransport_loggingMaskId, {c_MQTT_LocalTransport_Logging_ERROR});
	}
	}

	function f_EPTF_MQTT_IPL4asp_handleResult(inout Result p_res)
	runs on EPTF_MQTT_LocalTransport_CT
	return boolean
	{
	f_MQTT_Transport_Logging_VERBOSE(log2str(%definitionId,": ", p_res));

	if (ispresent(p_res.errorCode) and (p_res.errorCode != IPL4_ERROR_TEMPORARILY_UNAVAILABLE))
	{
	f_MQTT_Transport_Logging_WARNING(log2str("Warning: f_EPTF_MQTT_IPL4asp_handleResult: IPL4 error: ", p_res));
	v_EPTF_MQTT_Transport_stats.nofTransportWarnings := v_EPTF_MQTT_Transport_stats.nofTransportWarnings + 1;
	return false;
	}
	return true;
	}

	}