Feature/brokerless p2p

CMakeLists.txt

@@ -57,6 +57,16 @@ set(UCLIENT_UDP_TRANSPORT_MTU 512 CACHE STRING "Set the UDP transport MTU.")
 set(UCLIENT_TCP_TRANSPORT_MTU 512 CACHE STRING "Set the TCP transport MTU.")
 set(UCLIENT_SERIAL_TRANSPORT_MTU 512 CACHE STRING "Set the Serial transport MTU.")
 
+option(UCLIENT_PROFILE_BROKERLESS "Enable brokerless transport (experimental)." ON)
+if(NOT UCLIENT_PROFILE_UDP)
+    set(UCLIENT_PROFILE_BROKERLESS OFF)
+endif()
+
+set(UCLIENT_BROKERLESS_PORT 9999 CACHE STRING "Brokerless UDP port (experimental)")
+set(UCLIENT_BROKERLESS_ENTITY_MAP_LEN 10 CACHE STRING "Brokerless entity map length port (experimental)")
+set(UCLIENT_BROKERLESS_MESSAGE_QUEUE_LEN 100 CACHE STRING "Brokerless message queue length (experimental)")
+set(UCLIENT_BROKERLESS_INTERNAL_BUFFER_LEN 500 CACHE STRING "Brokerless internal buffer length (experimental)")
+
 ###############################################################################
 # Dependencies
 ###############################################################################
@@ -206,6 +216,10 @@ if(UCLIENT_PROFILE_DISCOVERY)
     endif()
 endif()
 
+if(UCLIENT_PROFILE_BROKERLESS AND UCLIENT_PLATFORM_WINDOWS)
+    set(UCLIENT_PROFILE_BROKERLESS OFF)
+endif()
+
 # Other sources
 set(SRCS
     src/c/core/session/stream/input_best_effort_stream.c
@@ -228,6 +242,8 @@ set(SRCS
     src/c/core/session/create_entities_xml.c
     src/c/core/session/read_access.c
     src/c/core/session/write_access.c
+    $<$<BOOL:${UCLIENT_PROFILE_BROKERLESS}>:src/c/brokerless/brokerless.c>
+    $<$<BOOL:${UCLIENT_PROFILE_BROKERLESS}>:src/c/brokerless/udp_transport_broadcast_posix.c>
     $<$<OR:$<BOOL:${UCLIENT_VERBOSE_MESSAGE}>,$<BOOL:${UCLIENT_VERBOSE_SERIALIZATION}>>:src/c/core/log/log.c>
     ${_transport_src}
     )
@@ -348,6 +364,11 @@ if(UCLIENT_BUILD_EXAMPLES)
     add_subdirectory(examples/SubscribeHelloWorldP2P)
 endif()
 
+if(UCLIENT_BUILD_EXAMPLES AND UCLIENT_PROFILE_BROKERLESS)
+    add_subdirectory(examples/BrokerlessPublisher)
+    add_subdirectory(examples/BrokerlessSubscriber)
+endif()
+
 ###############################################################################
 # Tests
 ###############################################################################

examples/BrokerlessPublisher/CMakeLists.txt

@@ -0,0 +1,49 @@
+# Copyright 2017 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+cmake_minimum_required(VERSION 2.8.12)
+if (${CMAKE_VERSION} VERSION_GREATER 3.0)
+    cmake_policy(SET CMP0048 NEW)
+endif()
+
+project(BrokerlessPublisher)
+
+if(NOT UCLIENT_BUILD_EXAMPLES)
+    find_package(microxrcedds_client REQUIRED)
+endif()
+
+if(NOT UCLIENT_PROFILE_UDP OR NOT UCLIENT_PROFILE_BROKERLESS)
+    message(WARNING "Can not compile example: The UCLIENT_PROFILE_UDP and UCLIENT_PROFILE_BROKERLESS must be enabled.")
+else()
+    add_executable(${PROJECT_NAME} main.c)
+    if(MSVC OR MSVC_IDE)
+        target_compile_options(${PROJECT_NAME} PRIVATE /wd4996)
+    endif()
+
+    set_target_properties(${PROJECT_NAME} PROPERTIES
+        C_STANDARD 99
+        C_STANDARD_REQUIRED YES
+        )
+
+    target_link_libraries(${PROJECT_NAME} microxrcedds_client $<$<C_COMPILER_ID:GNU>:-Wl,--gc-section,--no-export-dynamic>)
+
+    if(UCLIENT_INSTALL_EXAMPLES)
+        install(
+            TARGETS
+                ${PROJECT_NAME}
+            RUNTIME DESTINATION
+                ${BIN_INSTALL_DIR}
+            )
+    endif()
+endif()

examples/BrokerlessPublisher/main.c

@@ -0,0 +1,96 @@
+// Copyright 2017 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <uxr/client/client.h>
+#include <uxr/client/brokerless/brokerless.h>
+#include <ucdr/microcdr.h>
+
+#include <stdio.h> //printf
+#include <string.h> //strcmp
+#include <stdlib.h> //atoi
+#include <unistd.h> //sleep
+
+#define BUFFER_SIZE     UXR_CONFIG_UDP_TRANSPORT_MTU
+
+int main(int args, char** argv)
+{
+    // Session
+    uxrSession session;
+    uxr_init_session(&session, &brokerless_comm_stub, 0xAAAABBBB);
+
+    // Streams
+    uint8_t output_reliable_stream_buffer[BUFFER_SIZE];
+    uxrStreamId best_effort_output = uxr_create_output_best_effort_stream(&session, output_reliable_stream_buffer, BUFFER_SIZE);
+
+    // Create entities
+    uxrObjectId participant_id = uxr_object_id(0x01, UXR_PARTICIPANT_ID);
+    const char* participant_xml = "<dds>"
+                                      "<participant>"
+                                          "<rtps>"
+                                              "<name>default_xrce_participant</name>"
+                                          "</rtps>"
+                                      "</participant>"
+                                  "</dds>";
+    uint16_t participant_req = uxr_buffer_create_participant_xml(&session, best_effort_output, participant_id, 0, participant_xml, UXR_REPLACE);
+
+    uxrObjectId topic_id = uxr_object_id(0x01, UXR_TOPIC_ID);
+    const char* topic_xml = "<dds>"
+                                "<topic>"
+                                    "<name>HelloWorldTopic</name>"
+                                    "<dataType>HelloWorld</dataType>"
+                                "</topic>"
+                            "</dds>";
+    uint16_t topic_req = uxr_buffer_create_topic_xml(&session, best_effort_output, topic_id, participant_id, topic_xml, UXR_REPLACE);
+
+    uxrObjectId publisher_id = uxr_object_id(0x01, UXR_PUBLISHER_ID);
+    const char* publisher_xml = "";
+    uint16_t publisher_req = uxr_buffer_create_publisher_xml(&session, best_effort_output, publisher_id, participant_id, publisher_xml, UXR_REPLACE);
+
+    uxrObjectId datawriter_id = uxr_object_id(0x01, UXR_DATAWRITER_ID);
+    const char* datawriter_xml = "<dds>"
+                                     "<data_writer>"
+                                         "<topic>"
+                                             "<kind>NO_KEY</kind>"
+                                             "<name>HelloWorldTopic</name>"
+                                             "<dataType>HelloWorld</dataType>"
+                                         "</topic>"
+                                     "</data_writer>"
+                                 "</dds>";
+    uint16_t datawriter_req = uxr_buffer_create_datawriter_xml(&session, best_effort_output, datawriter_id, publisher_id, datawriter_xml, UXR_REPLACE);
+
+    // Write topics
+    char data[50];  
+    data[49] = '\0';
+    memset(data, 96+1, 49);
+
+    while(1)
+    {
+        ucdrBuffer ub;
+
+        memset(data, 96+((data[0]-96+1)%25), 49);
+
+        uxr_prepare_output_stream(&session, best_effort_output, datawriter_id, &ub, 50);
+        ucdr_serialize_array_char(&ub, data, 50);
+
+        printf("Send topic: %s\n", data);
+
+        uxr_run_session_timeout(&session, 10);
+        sleep(1);
+    }
+
+    // Delete resources
+    uxr_delete_session(&session);
+
+    return 0;
+}

examples/BrokerlessSubscriber/CMakeLists.txt

@@ -0,0 +1,50 @@
+# Copyright 2017 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+cmake_minimum_required(VERSION 2.8.12)
+if (${CMAKE_VERSION} VERSION_GREATER 3.0)
+    cmake_policy(SET CMP0048 NEW)
+endif()
+
+project(BrokerlessSubscriber)
+
+if(NOT UCLIENT_BUILD_EXAMPLES)
+    find_package(microxrcedds_client REQUIRED)
+endif()
+
+if(NOT UCLIENT_PROFILE_UDP OR NOT UCLIENT_PROFILE_BROKERLESS)
+    message(WARNING "Can not compile example: The UCLIENT_PROFILE_UDP and UCLIENT_PROFILE_BROKERLESS must be enabled.")
+else()
+    add_executable(${PROJECT_NAME} main.c)
+    if(MSVC OR MSVC_IDE)
+        target_compile_options(${PROJECT_NAME} PRIVATE /wd4996)
+    endif()
+
+    set_target_properties(${PROJECT_NAME} PROPERTIES
+        C_STANDARD 99
+        C_STANDARD_REQUIRED YES
+        )
+
+
+    target_link_libraries(${PROJECT_NAME} microxrcedds_client $<$<C_COMPILER_ID:GNU>:-Wl,--gc-section,--no-export-dynamic>)
+
+    if(UCLIENT_INSTALL_EXAMPLES)
+        install(
+            TARGETS
+                ${PROJECT_NAME}
+            RUNTIME DESTINATION
+                ${BIN_INSTALL_DIR}
+            )
+    endif()
+endif()

examples/BrokerlessSubscriber/main.c

@@ -0,0 +1,108 @@
+// Copyright 2017 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <uxr/client/client.h>
+#include <uxr/client/brokerless/brokerless.h>
+
+#include <stdio.h> //printf
+#include <string.h> //strcmp
+#include <stdlib.h> //atoi
+
+#define BUFFER_SIZE     UXR_CONFIG_UDP_TRANSPORT_MTU
+
+uint32_t last_index = -1;
+void on_topic(
+        uxrSession* session,
+        uxrObjectId object_id,
+        uint16_t request_id,
+        uxrStreamId stream_id,
+        struct ucdrBuffer* ub,
+        uint16_t length,
+        void* args)
+{
+    (void) session; (void) object_id; (void) request_id; (void) stream_id; (void) length;
+
+    char data[50];  
+    ucdr_deserialize_array_char(ub, data, 50);
+
+    printf("Received topic: %s\n", data);
+}
+
+int main(int args, char** argv)
+{
+
+    // Session
+    uxrSession session;
+    uxr_init_session(&session, &brokerless_comm_stub, 0xCCCCDDDD);
+    uxr_set_topic_callback(&session, on_topic, NULL);
+
+    // Streams
+    uint8_t output_best_effort_stream_buffer[BUFFER_SIZE];
+    uxrStreamId best_effort_output = uxr_create_output_best_effort_stream(&session, output_best_effort_stream_buffer, BUFFER_SIZE);
+
+    uxrStreamId best_effort_input = uxr_create_input_best_effort_stream(&session);
+
+    // Create entities
+    uxrObjectId participant_id = uxr_object_id(0x01, UXR_PARTICIPANT_ID);
+    const char* participant_xml = "<dds>"
+                                      "<participant>"
+                                          "<rtps>"
+                                              "<name>default_xrce_participant</name>"
+                                          "</rtps>"
+                                      "</participant>"
+                                  "</dds>";
+    uint16_t participant_req = uxr_buffer_create_participant_xml(&session, best_effort_output, participant_id, 0, participant_xml, UXR_REPLACE);
+
+    uxrObjectId topic_id = uxr_object_id(0x01, UXR_TOPIC_ID);
+    const char* topic_xml = "<dds>"
+                                "<topic>"
+                                    "<name>HelloWorldTopic</name>"
+                                    "<dataType>HelloWorld</dataType>"
+                                "</topic>"
+                            "</dds>";
+    uint16_t topic_req = uxr_buffer_create_topic_xml(&session, best_effort_output, topic_id, participant_id, topic_xml, UXR_REPLACE);
+
+    uxrObjectId subscriber_id = uxr_object_id(0x01, UXR_SUBSCRIBER_ID);
+    const char* subscriber_xml = "";
+    uint16_t subscriber_req = uxr_buffer_create_subscriber_xml(&session, best_effort_output, subscriber_id, participant_id, subscriber_xml, UXR_REPLACE);
+
+    uxrObjectId datareader_id = uxr_object_id(0x01, UXR_DATAREADER_ID);
+    const char* datareader_xml = "<dds>"
+                                     "<data_reader>"
+                                         "<topic>"
+                                             "<kind>NO_KEY</kind>"
+                                             "<name>HelloWorldTopic</name>"
+                                             "<dataType>HelloWorld</dataType>"
+                                         "</topic>"
+                                     "</data_reader>"
+                                 "</dds>";
+    uint16_t datareader_req = uxr_buffer_create_datareader_xml(&session, best_effort_output, datareader_id, subscriber_id, datareader_xml, UXR_REPLACE);
+
+    // Request topics
+    uxrDeliveryControl delivery_control = {0};
+    delivery_control.max_samples = UXR_MAX_SAMPLES_UNLIMITED;
+    uint16_t read_data_req = uxr_buffer_request_data(&session, best_effort_output, datareader_id, best_effort_input, &delivery_control);
+
+    // Read topics
+    bool connected = true;
+    while(true)
+    {
+        connected = uxr_run_session_timeout(&session, 10);
+    }
+
+    // Delete resources
+    uxr_delete_session(&session);
+
+    return 0;
+}