Add rtRetain and rtRelease gladish#3

Removed rtMessage_Destroy from header
Fixed memory issue
Changed error return in rtConnection ReadUntil

Author: relangovan

dataProvider/dmProviderDatabase.cpp

@@ -112,8 +112,8 @@ class dmQueryImpl : public dmQuery
         m_results.addValue(dmNamedValue(param, dmValue(value)), status);
       }
 
-      rtMessage_Destroy(res);
-      rtMessage_Destroy(req);
+      rtMessage_Release(res);
+      rtMessage_Release(req);
     }
   }
 

dataProvider/dmProviderHost.cpp

@@ -133,11 +133,7 @@ class dmProviderHostImpl : public dmProviderHost
     rtMessage_Create(&res);
     host->encodeResult(res, results);
     rtConnection_SendResponse(m_con, hdr, res, 1000);
-    char* buffer = nullptr;
-    uint32_t num = 0;
-    rtError err = rtMessage_ToString(res, &buffer, &num);
-    printf("\n Response sent is : %s", buffer);
-    rtMessage_Destroy(res);
+    rtMessage_Release(res);
   }
 
   dmProviderOperation decodeOperation(rtMessage req)
@@ -188,7 +184,7 @@ class dmProviderHostImpl : public dmProviderHost
       delete [] param;
     }
 
-    rtMessage_Destroy(item);
+    rtMessage_Release(item);
   }
 
   void decodeSetRequest(rtMessage req, std::string& name, std::vector<dmNamedValue>& params)
@@ -212,7 +208,7 @@ class dmProviderHostImpl : public dmProviderHost
 
     dmNamedValue namedValue(param, property_value);
 
-    rtMessage_Destroy(item);
+    rtMessage_Release(item);
     delete [] param;
 
     params.push_back(namedValue);
@@ -236,7 +232,7 @@ class dmProviderHostImpl : public dmProviderHost
       }
       rtMessage_SetInt32(msg, "status", status_code);
       rtMessage_AddMessage(res, "result", msg);
-      rtMessage_Destroy(msg);
+      rtMessage_Release(msg);
     }
   }
 

rtConnection.c

@@ -32,7 +32,7 @@
 #include <unistd.h>
 
 #define RTMSG_LISTENERS_MAX 64
-#define RTMSG_SEND_BUFFER_SIZE (1024 * 4094)
+#define RTMSG_SEND_BUFFER_SIZE (1024 * 8)
 
 struct _rtListener
 {
@@ -65,7 +65,7 @@ static void onInboxMessage(rtMessageHeader const* hdr, uint8_t const* p, uint32_
     struct _rtConnection* con = (struct _rtConnection *) closure;
     if (con->response != NULL)
     {
-      rtMessage_Destroy(con->response);
+      rtMessage_Release(con->response);
       con->response = NULL;
     }
     rtMessage_FromBytes(&con->response, p, n);
@@ -154,7 +154,7 @@ rtConnection_ConnectAndRegister(rtConnection con)
       rtMessage_SetString(m, "topic", con->listeners[i].expression);
       rtMessage_SetInt32(m, "route_id", con->listeners[i].subscription_id);
       rtConnection_SendMessage(con, m, "_RTROUTED.INBOX.SUBSCRIBE");
-      rtMessage_Destroy(m);
+      rtMessage_Release(m);
     }
   }
 
@@ -208,8 +208,11 @@ rtConnection_ReadUntil(rtConnection con, uint8_t* buff, int count, int32_t timeo
 
     ssize_t n = recv(con->fd, buff + bytes_read, (bytes_to_read - bytes_read), MSG_NOSIGNAL);
     if (n == 0)
-      return RT_ERROR;
-    //return rtErrorFromErrno(ENOTCONN);
+    {
+      rtLog_Error("Failed to read error : %s", rtStrError(rtErrorFromErrno(ENOTCONN)));
+      return rtErrorFromErrno(ENOTCONN);
+    }
+
     if (n == -1)
     {
       if (errno == EINTR)
@@ -502,7 +505,7 @@ rtConnection_AddListener(rtConnection con, char const* expression, rtMessageCall
   rtMessage_SetString(m, "topic", expression);
   rtMessage_SetInt32(m, "route_id", con->listeners[i].subscription_id); 
   rtConnection_SendMessage(con, m, "_RTROUTED.INBOX.SUBSCRIBE");
-  rtMessage_Destroy(m);
+  rtMessage_Release(m);
 
   return 0;
 }
@@ -535,6 +538,7 @@ rtConnection_TimedDispatch(rtConnection con, int32_t timeout)
   {
     con->state = rtConnectionState_ReadHeaderPreamble;
     err = rtConnection_ReadUntil(con, con->recv_buffer, 4, timeout);
+
     if (err == RT_ERROR_TIMEOUT)
       return err;
 

rtMessage.h

@@ -49,14 +49,6 @@ rtMessage_Clone(rtMessage const message, rtMessage* copy);
 rtError
 rtMessage_FromBytes(rtMessage* message, uint8_t const* buff, int n);
 
-/**
- * Destroy a message; free the storage that it occupies.
- * @param pointer to message to be destroyed
- * @return rtError
- **/
-rtError
-rtMessage_Destroy(rtMessage message);
-
 /**
  * Extract the data from a message as a byte sequence.
  * @param extract the data bytes from this message.

rtrouted.c

@@ -235,7 +235,7 @@ rtRouted_OnMessage(rtConnectedClient* sender, rtMessageHeader* hdr, uint8_t cons
     subscription->client = sender;
     rtRouted_AddRoute(rtRouted_ForwardMessage, expression, subscription);
 
-    rtMessage_Destroy(m);
+    rtMessage_Release(m);
   }
   else if (strcmp(hdr->topic, "_RTROUTED.INBOX.HELLO") == 0)
   {
@@ -250,7 +250,7 @@ rtRouted_OnMessage(rtConnectedClient* sender, rtMessageHeader* hdr, uint8_t cons
     subscription->client = sender;
     rtRouted_AddRoute(rtRouted_ForwardMessage, inbox, subscription);
 
-    rtMessage_Destroy(m);
+    rtMessage_Release(m);
   }
   else
   {

rtsub.c

@@ -38,7 +38,7 @@ static void on_message(rtMessageHeader const* hdr, uint8_t const* buff, uint32_t
   }
   printf("END MESSAGE\n");
 
-  rtMessage_Destroy(m);
+  rtMessage_Release(m);
 }
 
 int main(int argc, char* argv[])

sample_recv.c

@@ -44,7 +44,7 @@ void onMessage(rtMessageHeader const* hdr, uint8_t const* buff, uint32_t n, void
 
   free(s);
   free(itemstring);
-  rtMessage_Destroy(m);
+  rtMessage_Release(m);
 }
 
 int main()
@@ -56,7 +56,7 @@ int main()
   rtConnection_Create(&con, "APP2", "tcp://127.0.0.1:10001");
 //  rtConnection_Create(&con, "APP2", "unix:///tmp/rtrouted");
   rtConnection_AddListener(con, "A.*.C", onMessage, NULL);
-  rtConnection_AddListener(con, "A.B.C.>", onMessage, NULL);
+  rtConnection_AddListener(con, "A.B.C.*", onMessage, NULL);
 
   while (1)
   {

sample_req.c

@@ -50,8 +50,8 @@ int main()
       free(p);
     }
 
-    rtMessage_Destroy(req);
-    rtMessage_Destroy(res);
+    rtMessage_Release(req);
+    rtMessage_Release(res);
 
     sleep(1);
   }

sample_res.c

@@ -40,10 +40,10 @@ void onMessage(rtMessageHeader const* hdr, uint8_t const* buff, uint32_t n, void
     rtMessage_Create(&res);
     rtMessage_SetString(res, "reply", "reply -- Hello World");
     rtConnection_SendResponse(con, hdr, res, 1000);
-    rtMessage_Destroy(res);
+    rtMessage_Release(res);
   }
 
-  rtMessage_Destroy(req);
+  rtMessage_Release(req);
 
   rtLog_Info("flags     :%d", hdr->flags);
   rtLog_Info("is_request:%d", rtMessageHeader_IsRequest(hdr));

sample_send.c

@@ -44,7 +44,7 @@ int main()
     err = rtConnection_SendMessage(con, m, "A.B.C");
     rtLog_Info("send[%s]: %s", "A.B.C", rtStrError(err));
 
-    rtMessage_Destroy(m);
+    rtMessage_Release(m);
     sleep(1);
 
     rtMessage_Create(&m);
@@ -53,7 +53,7 @@ int main()
     err = rtConnection_SendMessage(con, m, "A.B.C.FOO.BAR");
     rtLog_Info("send[%s]: %s", "A.B.C.FOO.BAR", rtStrError(err));
 
-    rtMessage_Destroy(m);
+    rtMessage_Release(m);
     sleep(1);
   }