Refinements

.github/workflows/ccpp.yml

@@ -10,15 +10,16 @@ jobs:
     - name: cmake configure
       run: cmake .
     - name: cmake build
-      run: cmake --build .
-
+      run: cmake --build . -j4
   linux-build:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v1
     - name: cmake configure
       run: cmake .
     - name: cmake build
-      run: cmake --build .
-#    - name: Run Tests
-#      run: cd tests && ctest --output-on-failure
+      run: cmake --build . -j4
+    - name: network
+      run: sudo apt install net-tools && ifconfig
+    - name: Run Tests
+      run: cd tests && ctest -VV

.gitignore

@@ -272,3 +272,4 @@ tests/CMakeFiles/*
 Makefile
 cmake_install.cmake
 build/
+html/

CMakeLists.txt

@@ -25,11 +25,12 @@ add_library (pubsub
   "src/Serialization.c"
   "src/System.c"
   "src/UDPTransport.c"
+  "src/Parameter.c"
 )
 if (UNIX)
 target_link_libraries(pubsub pubsub_msgs)
 else()
-target_link_libraries(pubsub winmm)
+target_link_libraries(pubsub winmm pubsub_msgs)
 endif()
 set_target_properties(pubsub PROPERTIES DEBUG_POSTFIX "d")
 target_include_directories(pubsub PUBLIC include/)

examples/CMakeLists.txt

@@ -33,9 +33,3 @@ set_property(TARGET simple_sub_cpp PROPERTY CXX_STANDARD 11)
 add_executable(read_param "read_param.c")
 target_link_libraries(read_param pubsub pubsub_msgs)
 add_dependencies(read_param pubsub pubsub_msgs)
-
-#install(TARGETS pubsubtest
-#  ARCHIVE DESTINATION "s"
-#  LIBRARY DESTINATION "s"
-#  RUNTIME DESTINATION "s"
-#)

examples/simple_logger.cpp

@@ -32,8 +32,7 @@ int main()
 
 		// okay, since we are publishing with shared pointer we actually need to allocate the string properly
 		auto shared = pubsub::msg::StringSharedPtr(new pubsub::msg::String);
-		shared->value = new char[strlen(msg.value) + 1];
-		strcpy(shared->value, msg.value);
+		shared->value = msg.value;
 		string_pub.publish(shared);
 
 		msg.value = 0;// so it doesnt get freed by the destructor since we allocated it ourself

examples/simple_pub.c

@@ -26,10 +26,10 @@ int main()
 
   // Create the publisher
   struct ps_pub_t string_pub;
-  ps_node_create_publisher(&node, "/data"/*topic name*/,
+  ps_node_create_publisher_ex(&node, "/data"/*topic name*/,
                            &pubsub__String_def/*message definition*/,
                            &string_pub,
-                           true/*true to "latch" the topic*/);
+                           true/*true to "latch" the topic*/, 1);
 
   // User is responsible for lifetime of the message they publish
   // Publish does a copy internally if necessary

examples/simple_pub.cpp

@@ -44,7 +44,7 @@ int main()
   });
 
   // Wait for the spinner to exit (on control-c)
-  spinner.wait();
+  spinner.run();
 
   return 0;
 }

examples/simple_sub.c

@@ -9,6 +9,15 @@
 
 #include <pubsub/String.msg.h>
 
+struct ps_sub_t string_sub;
+void callback(void* message, unsigned int size, void* cbdata, const struct ps_msg_info_t* info)
+{
+  // user is responsible for freeing the message and its arrays
+  struct pubsub__String* data = (struct pubsub__String*)message;
+  printf("Got message: %s\n", data->value);
+  pubsub__String_free(data, string_sub.allocator);
+}
+
 int main()
 {
   // Create the node
@@ -23,10 +32,10 @@ int main()
   ps_node_add_transport(&node, &tcp_transport);
 
   // Create the subscriber
-  struct ps_sub_t string_sub;
   struct ps_subscriber_options options;
   ps_subscriber_options_init(&options);
   options.preferred_transport = PUBSUB_TCP_TRANSPORT;// sets preferred transport to TCP
+  options.cb = callback;
   ps_node_create_subscriber_adv(&node, "/data", &pubsub__String_def, &string_sub, &options);
 
   // Loop and spin
@@ -36,18 +45,8 @@ int main()
     // Used to prevent this from using 100% CPU, but you can do that through other means
     ps_node_wait(&node, 1000/*maximum wait time in ms*/);
 
-    // Updates the node, which will queue up any received messages
+    // Updates the node, which will receive messages and call any callbacks as they come in
     ps_node_spin(&node);
-
-    // our sub has a message definition, so the queue contains real messages
-    struct pubsub__String* data;
-    while (data = (struct pubsub__String*)ps_sub_deque(&string_sub))
-    {
-      // user is responsible for freeing the message and its arrays
-      printf("Got message: %s\n", data->value);
-      free(data->value);
-      free(data);
-    }
   }
 
   // Shutdown the node to free resources

examples/simple_sub.cpp

@@ -15,7 +15,7 @@ int main()
   // Create the subscriber, the provided callback will be called each time a message comes in
   pubsub::Subscriber<pubsub::msg::String> subscriber(node, "data"/*topic name*/,
     [](const pubsub::msg::StringSharedPtr& msg) {
-      printf("Got message %s\n", msg->value);
+      printf("Got message %s\n", msg->value.c_str());
     }, 10/*maximum queue size, after this many messages build up the oldest will get dropped*/);
 
   // Create the "spinner" which executes callbacks and timers in a background thread

include/pubsub/Node.h

@@ -29,7 +29,8 @@ struct ps_endpoint_t;
 struct ps_client_t;
 struct ps_subscribe_req_t;
 struct ps_allocator_t;
-typedef void(*ps_transport_fn_pub_t)(struct ps_transport_t* transport, struct ps_pub_t* publisher, struct ps_client_t* client, const void* message, uint32_t length);
+struct ps_msg_ref_t;
+typedef void(*ps_transport_fn_pub_t)(struct ps_transport_t* transport, struct ps_pub_t* publisher, struct ps_client_t* client, struct ps_msg_ref_t* message);
 typedef int(*ps_transport_fn_spin_t)(struct ps_transport_t* transport, struct ps_node_t* node);
 typedef void(*ps_transport_fn_add_publisher_t)(struct ps_transport_t* transport, struct ps_pub_t* publisher);
 typedef void(*ps_transport_fn_remove_publisher_t)(struct ps_transport_t* transport, struct ps_pub_t* publisher);
@@ -66,6 +67,7 @@ typedef void(*ps_param_confirm_cb_t)(const char* name, double value, void* data)
 struct ps_node_t
 {
 	const char* name;
+	const char* description;
 	unsigned int num_pubs;
 	struct ps_pub_t** pubs;
 	unsigned int num_subs;
@@ -129,10 +131,9 @@ struct ps_msg_info_t
 struct ps_msg_header
 {
 	uint8_t pid;//packet type id
+	uint32_t length;//message length
 	uint32_t id;//stream id
 	uint16_t seq;//sequence number
-	uint8_t index;
-	uint8_t count;
 };
 #pragma pack(pop)
 
@@ -161,20 +162,22 @@ struct ps_advertise_req_t
 	uint32_t type_hash;// to see if the type is correct
 	uint32_t group_id;// unique (hopefully) id that indicates which process this node is a part of
 };
-#pragma pack(pop)
 
-#pragma pack(push)
-#pragma pack(1)
 struct ps_subscribe_req_t
 {
   uint8_t id;
   int32_t addr;
   uint16_t port;
 };
-#pragma pack(pop)
 
-#pragma pack(push)
-#pragma pack(1)
+struct ps_unsubscribe_req_t
+{
+  uint8_t id;
+  uint32_t addr;
+  uint16_t port;
+  uint32_t stream_id;
+};
+
 struct ps_subscribe_accept_t
 {
 	uint8_t pid;// packet type identifier
@@ -193,24 +196,20 @@ void ps_node_init_ex(struct ps_node_t* node, const char* name, const char* ip, b
 
 void ps_node_create_publisher(struct ps_node_t* node, const char* topic, const struct ps_message_definition_t* type, struct ps_pub_t* pub, bool latched);
 
-void ps_node_create_subscriber(struct ps_node_t* node, const char* topic, const struct ps_message_definition_t* type,
-	struct ps_sub_t* sub,
-	unsigned int queue_size,//make >= 1
-	struct ps_allocator_t* allocator,//give null to use default
-	bool ignore_local);// if ignore local is set, this node ignores publications from itself
-							 // this facilitiates passing messages through shared memory
+void ps_node_create_publisher_ex(struct ps_node_t* node, const char* topic, const struct ps_message_definition_t* type, struct ps_pub_t* pub, bool latched, unsigned int recommended_transport, struct ps_allocator_t* allocator);
 
 
 typedef void(*ps_subscriber_fn_cb_t)(void* message, unsigned int size, void* data, const struct ps_msg_info_t* info);
 struct ps_subscriber_options
 {
-	unsigned int queue_size;
 	bool ignore_local;
 	struct ps_allocator_t* allocator;
 	unsigned int skip;// skips to every nth message for throttling
 	ps_subscriber_fn_cb_t cb;
+	ps_subscriber_fn_cb_t cb_raw;
 	void* cb_data;
-    uint32_t preferred_transport;// falls back to udp otherwise
+  int32_t preferred_transport;// falls back to udp otherwise
+  const char* description;
 };
 
 void ps_subscriber_options_init(struct ps_subscriber_options* options);
@@ -219,7 +218,7 @@ void ps_node_create_subscriber_adv(struct ps_node_t* node, const char* topic, co
 	struct ps_sub_t* sub,
 	const struct ps_subscriber_options* options);
 
-void ps_node_create_subscriber_cb(struct ps_node_t* node, const char* topic, const struct ps_message_definition_t* type,
+void ps_node_create_subscriber(struct ps_node_t* node, const char* topic, const struct ps_message_definition_t* type,
 	struct ps_sub_t* sub,
 	ps_subscriber_fn_cb_t cb,
 	void* cb_data,

include/pubsub/Parameter.h

@@ -0,0 +1,40 @@
+#pragma once
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+//#include <cstdlib>
+#include <stdio.h>
+
+#include <pubsub/Node.h>
+#include <pubsub/Publisher.h>
+#include <pubsub/Subscriber.h>
+#include <pubsub/System.h>
+
+#include <pubsub/Parameters.msg.h>
+
+#include <math.h>
+
+typedef void(*ps_param_fancy_cb_t)(const char* name, double value, void* data);
+
+struct ps_parameters
+{
+	struct pubsub__Parameters msg;
+	struct ps_pub_t param_pub;
+	ps_param_fancy_cb_t callback;
+	void* cb_data;
+};
+
+void ps_create_parameters(struct ps_node_t* node, struct ps_parameters* params_out, ps_param_fancy_cb_t callback, void* data);
+
+void ps_destroy_parameters(struct ps_parameters* params);
+
+void ps_add_parameter_double(struct ps_parameters* params,
+ const char* name, const char* description,
+ double value, double min, double max);
+
+#ifdef __cplusplus
+}
+#endif

include/pubsub/Publisher.h

@@ -14,20 +14,20 @@ struct ps_message_definition_t;
 
 struct ps_endpoint_t
 {
-	unsigned short port;
-	int address;
+  unsigned short port;
+  int address;
 	//bool multicast;// this is probably unnecessary
 };
 
 // publisher client to network to
 struct ps_client_t
 {
-	struct ps_endpoint_t endpoint;
-	unsigned short sequence_number;// sequence of the networked packets, incremented with each one
-	unsigned long long last_keepalive;// timestamp of the last keepalive message, used to know when to deactiveate this connection
-	unsigned int stream_id;// user-unique identifier of what topic this came from
-	unsigned int modulo;
-    struct ps_transport_t* transport;
+  struct ps_endpoint_t endpoint;
+  unsigned short sequence_number;// sequence of the networked packets, incremented with each one
+  unsigned long long last_keepalive;// timestamp of the last keepalive message, used to know when to deactiveate this connection
+  unsigned int stream_id;// user-unique identifier of what topic this came from
+  unsigned int modulo;
+  struct ps_transport_t* transport;
 };
 
 struct ps_pub_t
@@ -39,10 +39,13 @@ struct ps_pub_t
 	struct ps_node_t* node;
 	unsigned int num_clients;
 	struct ps_client_t* clients;
+
+	struct ps_allocator_t* allocator;
 
 	bool latched;// todo make this an enum of options if we add more
+	uint8_t recommended_transport;
 
-	struct ps_msg_t last_message;//only used if latched
+	struct ps_msg_ref_t* last_message;//only used if latched
 	unsigned int sequence_number;
 };