diff --git a/CMakeLists.txt b/CMakeLists.txt index a6dca3f69..aa096ef64 100755 --- a/CMakeLists.txt +++ b/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 + $<$:src/c/brokerless/brokerless.c> + $<$:src/c/brokerless/udp_transport_broadcast_posix.c> $<$,$>: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 ############################################################################### diff --git a/examples/BrokerlessPublisher/CMakeLists.txt b/examples/BrokerlessPublisher/CMakeLists.txt new file mode 100644 index 000000000..7c171fcf8 --- /dev/null +++ b/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 $<$:-Wl,--gc-section,--no-export-dynamic>) + + if(UCLIENT_INSTALL_EXAMPLES) + install( + TARGETS + ${PROJECT_NAME} + RUNTIME DESTINATION + ${BIN_INSTALL_DIR} + ) + endif() +endif() diff --git a/examples/BrokerlessPublisher/main.c b/examples/BrokerlessPublisher/main.c new file mode 100644 index 000000000..b4009fedb --- /dev/null +++ b/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 +#include +#include + +#include //printf +#include //strcmp +#include //atoi +#include //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 = "" + "" + "" + "default_xrce_participant" + "" + "" + ""; + 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 = "" + "" + "HelloWorldTopic" + "HelloWorld" + "" + ""; + 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 = "" + "" + "" + "NO_KEY" + "HelloWorldTopic" + "HelloWorld" + "" + "" + ""; + 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; +} diff --git a/examples/BrokerlessSubscriber/CMakeLists.txt b/examples/BrokerlessSubscriber/CMakeLists.txt new file mode 100644 index 000000000..be98dd345 --- /dev/null +++ b/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 $<$:-Wl,--gc-section,--no-export-dynamic>) + + if(UCLIENT_INSTALL_EXAMPLES) + install( + TARGETS + ${PROJECT_NAME} + RUNTIME DESTINATION + ${BIN_INSTALL_DIR} + ) + endif() +endif() diff --git a/examples/BrokerlessSubscriber/main.c b/examples/BrokerlessSubscriber/main.c new file mode 100644 index 000000000..a5a901db0 --- /dev/null +++ b/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 +#include + +#include //printf +#include //strcmp +#include //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 = "" + "" + "" + "default_xrce_participant" + "" + "" + ""; + 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 = "" + "" + "HelloWorldTopic" + "HelloWorld" + "" + ""; + 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 = "" + "" + "" + "NO_KEY" + "HelloWorldTopic" + "HelloWorld" + "" + "" + ""; + 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; +} diff --git a/examples/PublishHelloWorld/main.c b/examples/PublishHelloWorld/main.c index aeed1679a..e53b03771 100644 --- a/examples/PublishHelloWorld/main.c +++ b/examples/PublishHelloWorld/main.c @@ -21,6 +21,12 @@ #include //strcmp #include //atoi +#ifndef _WIN32 +#include //sleep +#else +#include +#endif + #define STREAM_HISTORY 8 #define BUFFER_SIZE UXR_CONFIG_UDP_TRANSPORT_MTU * STREAM_HISTORY @@ -52,12 +58,12 @@ int main(int args, char** argv) if(!uxr_create_session(&session)) { printf("Error at create session.\n"); - return 1; + // return 1; } // Streams uint8_t output_reliable_stream_buffer[BUFFER_SIZE]; - uxrStreamId reliable_out = uxr_create_output_reliable_stream(&session, output_reliable_stream_buffer, BUFFER_SIZE, STREAM_HISTORY); + uxrStreamId be_out = uxr_create_output_best_effort_stream(&session, output_reliable_stream_buffer, BUFFER_SIZE); uint8_t input_reliable_stream_buffer[BUFFER_SIZE]; uxr_create_input_reliable_stream(&session, input_reliable_stream_buffer, BUFFER_SIZE, STREAM_HISTORY); @@ -71,7 +77,7 @@ int main(int args, char** argv) "" "" ""; - uint16_t participant_req = uxr_buffer_create_participant_xml(&session, reliable_out, participant_id, 0, participant_xml, UXR_REPLACE); + uint16_t participant_req = uxr_buffer_create_participant_xml(&session, be_out, participant_id, 0, participant_xml, UXR_REPLACE); uxrObjectId topic_id = uxr_object_id(0x01, UXR_TOPIC_ID); const char* topic_xml = "" @@ -80,11 +86,11 @@ int main(int args, char** argv) "HelloWorld" "" ""; - uint16_t topic_req = uxr_buffer_create_topic_xml(&session, reliable_out, topic_id, participant_id, topic_xml, UXR_REPLACE); + uint16_t topic_req = uxr_buffer_create_topic_xml(&session, be_out, 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, reliable_out, publisher_id, participant_id, publisher_xml, UXR_REPLACE); + uint16_t publisher_req = uxr_buffer_create_publisher_xml(&session, be_out, publisher_id, participant_id, publisher_xml, UXR_REPLACE); uxrObjectId datawriter_id = uxr_object_id(0x01, UXR_DATAWRITER_ID); const char* datawriter_xml = "" @@ -96,31 +102,38 @@ int main(int args, char** argv) "" "" ""; - uint16_t datawriter_req = uxr_buffer_create_datawriter_xml(&session, reliable_out, datawriter_id, publisher_id, datawriter_xml, UXR_REPLACE); - + // uint16_t datawriter_req = uxr_buffer_create_datawriter_xml(&session, be_out, datawriter_id, publisher_id, datawriter_xml, UXR_REPLACE); + uint16_t datawriter_req = uxr_buffer_create_datawriter_ref(&session, be_out, datawriter_id, publisher_id, "prueba2", UXR_REPLACE); + // Send create entities message and wait its status uint8_t status[4]; uint16_t requests[4] = {participant_req, topic_req, publisher_req, datawriter_req}; if(!uxr_run_session_until_all_status(&session, 1000, requests, status, 4)) { printf("Error at create entities: participant: %i topic: %i publisher: %i darawriter: %i\n", status[0], status[1], status[2], status[3]); - return 1; + // return 1; } // Write topics bool connected = true; uint32_t count = 0; - while(connected && count < max_topics) + while(1) { HelloWorld topic = {++count, "Hello DDS world!"}; ucdrBuffer ub; uint32_t topic_size = HelloWorld_size_of_topic(&topic, 0); - uxr_prepare_output_stream(&session, reliable_out, datawriter_id, &ub, topic_size); + uxr_prepare_output_stream(&session, be_out, datawriter_id, &ub, topic_size); HelloWorld_serialize_topic(&ub, &topic); printf("Send topic: %s, id: %i\n", topic.message, topic.index); connected = uxr_run_session_time(&session, 1000); + +#ifndef _WIN32 + sleep(1); +#else + Sleep(1000); +#endif } // Delete resources diff --git a/examples/ReplyAdder/main.c b/examples/ReplyAdder/main.c index 3c5d3938f..192407e58 100644 --- a/examples/ReplyAdder/main.c +++ b/examples/ReplyAdder/main.c @@ -72,7 +72,7 @@ int main(int args, char** argv) char* ip = argv[1]; char* port = argv[2]; - uint32_t key = (args == 4) ? (uint32_t)atoi(argv[3]) : 0xCCCCDDDD; + uint32_t key = 0xCCCCDDDD; // Transport uxrUDPTransport transport; @@ -90,7 +90,7 @@ int main(int args, char** argv) if (!uxr_create_session(&session)) { printf("Error at init session.\n"); - return 1; + // return 1; } // Streams @@ -119,7 +119,7 @@ int main(int args, char** argv) "reply_type=\"reply_type\">" "" ""; - uint16_t replier_req = uxr_buffer_create_replier_xml(&session, reliable_out, replier_id, participant_id, replier_xml, UXR_REPLACE); + uint16_t replier_req = uxr_buffer_create_replier_ref(&session, reliable_out, replier_id, participant_id, "repli1", UXR_REPLACE); // Send create entities message and wait its status uint8_t status[2]; @@ -127,7 +127,7 @@ int main(int args, char** argv) if(!uxr_run_session_until_all_status(&session, 1000, requests, status, 2)) { printf("Error at create entities: participant: %i requester: %i\n", status[0], status[1]); - return 1; + // return 1; } // Request requests @@ -137,10 +137,11 @@ int main(int args, char** argv) // Read request bool connected = true; - while (connected) + while (1) { - uint8_t read_data_status; - connected = uxr_run_session_until_all_status(&session, UXR_TIMEOUT_INF, &read_data_req, &read_data_status, 1); + // uint8_t read_data_status; + // connected = uxr_run_session_until_all_status(&session, UXR_TIMEOUT_INF, &read_data_req, &read_data_status, 1); + connected = uxr_run_session_time(&session, 100); } return 0; diff --git a/examples/RequestAdder/main.c b/examples/RequestAdder/main.c index 160f94ce8..4614dc57a 100644 --- a/examples/RequestAdder/main.c +++ b/examples/RequestAdder/main.c @@ -18,10 +18,13 @@ #include #include #include +#include #define STREAM_HISTORY 8 #define BUFFER_SIZE UXR_CONFIG_UDP_TRANSPORT_MTU * STREAM_HISTORY +bool sended = false; + void on_reply( uxrSession* session, uxrObjectId object_id, @@ -38,6 +41,8 @@ void on_reply( uint64_t result; ucdr_deserialize_uint64_t(ub, &result); + sended = false; + #ifdef WIN32 printf("Reply received: %I64u [id: %d]\n", result, reply_id); #else @@ -55,7 +60,8 @@ int main(int args, char** argv) char* ip = argv[1]; char* port = argv[2]; - uint32_t key = (args == 4) ? (uint32_t)atoi(argv[3]) : 0xAAAABBBB; + srand((unsigned int)time(NULL)); // Initialization, should only be called once. + uint32_t key = rand(); // Transport uxrUDPTransport transport; @@ -102,7 +108,7 @@ int main(int args, char** argv) "reply_type=\"reply_type\">" "" ""; - uint16_t requester_req = uxr_buffer_create_requester_xml(&session, reliable_out, requester_id, participant_id, requester_xml, UXR_REPLACE); + uint16_t requester_req = uxr_buffer_create_requester_ref(&session, reliable_out, requester_id, participant_id, "repli1", UXR_REPLACE); // Send create entities message and wait its status uint8_t status[2]; @@ -110,7 +116,7 @@ int main(int args, char** argv) if(!uxr_run_session_until_all_status(&session, 1000, requests, status, 2)) { printf("Error at create entities: participant: %i requester: %i\n", status[0], status[1]); - return 1; + // return 1; } // Request replies @@ -120,21 +126,26 @@ int main(int args, char** argv) // Write requests bool connected = true; - uint32_t count = 0; - while (connected) + uint32_t count = atoi(argv[3]); + while (1) { - uint8_t request[2 * 4] = {0}; - ucdrBuffer ub; - - ucdr_init_buffer(&ub, request, sizeof(request)); - ucdr_serialize_uint32_t(&ub, count); - ucdr_serialize_uint32_t(&ub, count); - - uint16_t request_id = uxr_buffer_request(&session, reliable_out, requester_id, request, sizeof(request)); - printf("Request sent: (%d + %d) [id: %d]\n", count, count, request_id); - connected = uxr_run_session_time(&session, 1000); - - ++count; + if (!sended) + { + uint8_t request[2 * 4] = {0}; + ucdrBuffer ub; + + ucdr_init_buffer(&ub, request, sizeof(request)); + ucdr_serialize_uint32_t(&ub, count); + ucdr_serialize_uint32_t(&ub, count); + + uint16_t request_id = uxr_buffer_request(&session, reliable_out, requester_id, request, sizeof(request)); + printf("Request sent: (%d + %d) [id: %d]\n", count, count, request_id); + + ++count; + sended = true; + } + + connected = uxr_run_session_time(&session, 10); } return 0; diff --git a/examples/SubscribeHelloWorld/main.c b/examples/SubscribeHelloWorld/main.c index cae845274..465b4f5c8 100644 --- a/examples/SubscribeHelloWorld/main.c +++ b/examples/SubscribeHelloWorld/main.c @@ -23,6 +23,7 @@ #define STREAM_HISTORY 8 #define BUFFER_SIZE UXR_CONFIG_UDP_TRANSPORT_MTU * STREAM_HISTORY +uint32_t last_index = -1; void on_topic( uxrSession* session, uxrObjectId object_id, @@ -39,6 +40,13 @@ void on_topic( printf("Received topic: %s, id: %i\n", topic.message, topic.index); + if (last_index != topic.index-1) + { + printf("LOST PACKAGE!!"); + } + + last_index = topic.index; + uint32_t* count_ptr = (uint32_t*) args; (*count_ptr)++; } @@ -75,7 +83,7 @@ int main(int args, char** argv) if(!uxr_create_session(&session)) { printf("Error at create session.\n"); - return 1; + // return 1; } // Streams @@ -119,7 +127,8 @@ int main(int args, char** argv) "" "" ""; - uint16_t datareader_req = uxr_buffer_create_datareader_xml(&session, reliable_out, datareader_id, subscriber_id, datareader_xml, UXR_REPLACE); + // uint16_t datareader_req = uxr_buffer_create_datareader_xml(&session, reliable_out, datareader_id, subscriber_id, datareader_xml, UXR_REPLACE); + uint16_t datareader_req = uxr_buffer_create_datareader_ref(&session, reliable_out, datareader_id, subscriber_id, "prueba1", UXR_REPLACE); // Send create entities message and wait its status uint8_t status[4]; @@ -127,7 +136,7 @@ int main(int args, char** argv) if(!uxr_run_session_until_all_status(&session, 1000, requests, status, 4)) { printf("Error at create entities: participant: %i topic: %i subscriber: %i datareader: %i\n", status[0], status[1], status[2], status[3]); - return 1; + // return 1; } // Request topics @@ -137,10 +146,11 @@ int main(int args, char** argv) // Read topics bool connected = true; - while(connected && count < max_topics) + while(true) { - uint8_t read_data_status; - connected = uxr_run_session_until_all_status(&session, UXR_TIMEOUT_INF, &read_data_req, &read_data_status, 1); + // uint8_t read_data_status; + // connected = uxr_run_session_until_all_status(&session, UXR_TIMEOUT_INF, &read_data_req, &read_data_status, 1); + connected = uxr_run_session_timeout(&session, 1); } // Delete resources diff --git a/include/uxr/client/brokerless/brokerless.h b/include/uxr/client/brokerless/brokerless.h new file mode 100644 index 000000000..b4a9bc595 --- /dev/null +++ b/include/uxr/client/brokerless/brokerless.h @@ -0,0 +1,55 @@ +// 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. + +#ifndef _SRC_C_BROKERLESS_BROKERLESS_H_ +#define _SRC_C_BROKERLESS_BROKERLESS_H_ + +#include + +#ifdef __cplusplus +extern "C" +{ +#endif + +bool brokerless_send_stub(void* instance, const uint8_t* buf, size_t len) +{ + (void)instance; + (void)buf; + (void)len; + return true; +} + +bool brokerless_recv_stub(void* instance, uint8_t** buf, size_t* len, int timeout) +{ + (void)instance; + (void)buf; + (void)len; + (void)timeout; + return false; +} + +uxrCommunication brokerless_comm_stub = +{ + NULL, + brokerless_send_stub, + brokerless_recv_stub, + NULL, + 0 +}; + +#ifdef __cplusplus +} +#endif + +#endif // _SRC_C_BROKERLESS_BROKERLESS_H_ diff --git a/include/uxr/client/config.h.in b/include/uxr/client/config.h.in index 220934d64..a64253651 100644 --- a/include/uxr/client/config.h.in +++ b/include/uxr/client/config.h.in @@ -25,6 +25,7 @@ #cmakedefine UCLIENT_PROFILE_UDP #cmakedefine UCLIENT_PROFILE_TCP #cmakedefine UCLIENT_PROFILE_SERIAL +#cmakedefine UCLIENT_PROFILE_BROKERLESS #cmakedefine UCLIENT_PLATFORM_POSIX #cmakedefine UCLIENT_PLATFORM_POSIX_NOPOLL @@ -54,4 +55,9 @@ #define UXR_CONFIG_SERIAL_TRANSPORT_MTU @UCLIENT_SERIAL_TRANSPORT_MTU@ #endif +#define UCLIENT_BROKERLESS_PORT @UCLIENT_BROKERLESS_PORT@ +#define UCLIENT_BROKERLESS_ENTITY_MAP_LEN @UCLIENT_BROKERLESS_ENTITY_MAP_LEN@ +#define UCLIENT_BROKERLESS_MESSAGE_QUEUE_LEN @UCLIENT_BROKERLESS_MESSAGE_QUEUE_LEN@ +#define UCLIENT_BROKERLESS_INTERNAL_BUFFER_LEN @UCLIENT_BROKERLESS_INTERNAL_BUFFER_LEN@ + #endif // _UXR_CLIENT_CONFIG_H_ diff --git a/include/uxr/client/core/session/stream/stream_id.h b/include/uxr/client/core/session/stream/stream_id.h index bd8e41575..ef4c651cf 100644 --- a/include/uxr/client/core/session/stream/stream_id.h +++ b/include/uxr/client/core/session/stream/stream_id.h @@ -44,7 +44,9 @@ typedef enum uxrStreamType /** Identifies a best-effort stream. */ UXR_BEST_EFFORT_STREAM, /** Identifies a reliable stream. */ - UXR_RELIABLE_STREAM + UXR_RELIABLE_STREAM, + /** Identifies a brokerless stream. */ + UXR_BROKERLESS_STREAM } uxrStreamType; diff --git a/src/c/brokerless/brokerless.c b/src/c/brokerless/brokerless.c new file mode 100644 index 000000000..e91d3aebc --- /dev/null +++ b/src/c/brokerless/brokerless.c @@ -0,0 +1,466 @@ +// 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. + +// TODOs: +// - Manage brokerlessMessageQueue_t and brokerlessEntityMap_t as a container that allow push and pop +// - Implement removing entities from brokerless when they are removed from user API +// - Is the hast/objectid lookup optimal? Which container structure should we use? +// - Are the UDP broadcast sending a whole packet? Can it have fragmentation in UDP packet level? +// - What happen when uCDR buffer fragments? +// - Implement hash function for "topic recognition" +// - How can we match using XML instead of references? +// - Control flow of datareaders using the API of uxr_buffer_request_data and pass the request_id to the callbacks +// - If we implement binary XML entities it would be really nice to the Brokerless architecture +// Not all QoS can be sended in Binary Format p29 s7.7.3.2 https://www.omg.org/spec/DDS-XRCE/1.0/PDF +// - Should we simplify the approach of uxr_prepare_output_stream -> serialize to something similar to the request/reply where buffer is available in uxr_buffer_request? +// - Explore cases when multiple datawriters/datareaders/requesters/repliers coexists +// - sample identity inside brokerlessMessage_t can be optimized because it is a big member + +// IDEAS: +// - A P2P system that has a node with some services for create entities in the real agent. +// - A P2P system that has a node acting as a gateware to DDS. We need a P2P standart QoS which only fill type and topic to always match. + + +#include "./brokerless_internal.h" +#include "./brokerless_transport.h" + +#include + +static uint32_t client_key; +static brokerlessMessageQueue_t brokerlessMessageQueue; +static brokerlessEntityMap_t brokerlessEntityMap; +static uint8_t brokerlessBuffer[UCLIENT_BROKERLESS_INTERNAL_BUFFER_LEN]; + +//================================================================== +// PRIVATE UTILS +//================================================================== + +// djb2 by Dan Bernstein: http://www.cse.yorku.ca/~oz/hash.html +void hash_brokerless(unsigned char *str, char* hash) +{ + hash_int_t int_hash = 5381; + int c; + + while ((c = *str++)) + { + int_hash = ((int_hash << 5) + int_hash) + (hash_int_t)c; /* hash * 33 + c */ + } + + for (size_t i = 0; i < BROKERLESS_HASH_SIZE; i++) + { + hash[i] = ((char*)&int_hash)[i]; + } +} + +// Find first occurrence of tag in XML +bool find_tag_xml(const char * xml, size_t len, char * tag, const char ** content, size_t * content_len) +{ + size_t tag_len = strlen(tag); + bool found_begin = false; + bool found_end = false; + for (size_t i = 1; i < len; i++) + { + if (!found_begin && 0 == memcmp(&xml[i], tag, tag_len) && xml[i-1] == '<') + { + size_t tag_opener_len = 0; + while(xml[i+tag_opener_len] != '>') + { + tag_opener_len++; + } + *content = &xml[i+tag_opener_len+1]; + found_begin = true; + } + else if(found_begin && 0 == memcmp(&xml[i], tag, tag_len) && xml[i-1] == '/') + { + *content_len = (size_t)(&xml[i-2] - *content); + found_end = true; + break; + } + } + + return found_begin && found_end; +} + +// Find property in first occurrence of tag in XML +bool find_tag_property(const char * xml, size_t len, const char * tag, char * property, const char ** content, size_t * content_len) +{ + size_t tag_len = strlen(tag); + size_t property_len = strlen(property); + + bool found_tag = false; + bool found_property = false; + for (size_t i = 1; i < len; i++) + { + if (!found_tag && 0 == memcmp(&xml[i], tag, tag_len) && xml[i-1] == '<') + { + found_tag = true; + } + else if(found_tag && 0 == memcmp(&xml[i], property, property_len)) + { + *content = &xml[i+property_len+2]; + i += property_len+2; + *content_len = 0; + while(xml[i + (*content_len)] != '"') + { + *content_len += 1; + } + found_property = true; + break; + } + } + + return found_tag && found_property; +} + +//================================================================== +// PRIVATE +//================================================================== + +void init_brokerless(uint32_t key) +{ + brokerlessMessageQueue.index = 0; + brokerlessEntityMap.index = 0; + brokerlessEntityMap.datareaders = 0; + brokerlessEntityMap.datawriters = 0; + brokerlessEntityMap.requesters = 0; + brokerlessEntityMap.repliers = 0; + + client_key = key; + + brokerless_init_transport(); +} + +bool add_brokerless_message(ucdrBuffer* ub, uint32_t length, uxrObjectId id) +{ + SampleIdentity sample_id = {0}; + return add_brokerless_message_with_sample_id(ub, length, id, sample_id); +} + +bool add_brokerless_message_with_sample_id(ucdrBuffer* ub, size_t length, uxrObjectId id, SampleIdentity sample_id) +{ + if (brokerlessMessageQueue.index < UCLIENT_BROKERLESS_MESSAGE_QUEUE_LEN - 1) + { + brokerlessMessageQueue.queue[brokerlessMessageQueue.index].data = ub->iterator; + brokerlessMessageQueue.queue[brokerlessMessageQueue.index].length = length; + brokerlessMessageQueue.queue[brokerlessMessageQueue.index].id = id; + brokerlessMessageQueue.queue[brokerlessMessageQueue.index].sample_id = sample_id; + + brokerlessMessageQueue.index++; + + return true; + } + + return false; +} + +bool add_brokerless_entity_hash_from_xml(const char* xml, uxrObjectId id) +{ + bool found = true; + char name_type_buffer[100]; + + switch (id.type) + { + case UXR_DATAREADER_ID: + case UXR_DATAWRITER_ID: + { + const char * data_reader_or_writer = (id.type == UXR_DATAREADER_ID) ? "data_reader\0" : "data_writer\0"; + char xml_strings[3][12] = + { + "dds", + "", + "topic" + }; + memmove(xml_strings[1], data_reader_or_writer, 12); + + const char * content_in = xml; + char * content_out; + size_t content_len_in = strlen(content_in); + size_t content_len_out; + + for (size_t i = 0; i < 3; i++) + { + if(find_tag_xml(content_in, content_len_in, xml_strings[i], (const char **)&content_out, &content_len_out)) + { + content_in = content_out; + content_len_in = content_len_out; + } + else + { + return false; + } + } + + size_t topic_name_len; + size_t type_name_len; + + found &= find_tag_xml(content_in, content_len_in, "name", (const char **)&content_out, &topic_name_len); + memcpy(name_type_buffer, content_out, topic_name_len); + + found &= find_tag_xml(content_in, content_len_in, "dataType", (const char **)&content_out, &type_name_len); + memcpy(&name_type_buffer[topic_name_len], content_out, type_name_len); + + name_type_buffer[topic_name_len+type_name_len] = '\0'; + + found &= add_brokerless_entity_hash(name_type_buffer, id); + break; + } + case UXR_REQUESTER_ID: + case UXR_REPLIER_ID: + { + char * content_out; + size_t service_name_len; + size_t request_type_name_len; + size_t reply_type_name_len; + + found &= find_tag_property(xml, + strlen(xml), + (id.type == UXR_REQUESTER_ID) ? "requester" : "replier", + "service_name", + (const char **)&content_out, + &service_name_len); + if (found) + { + memcpy(name_type_buffer, content_out, service_name_len); + } + + found &= find_tag_property(xml, + strlen(xml), + (id.type == UXR_REQUESTER_ID) ? "requester" : "replier", + "request_type", + (const char **)&content_out, + &request_type_name_len); + if (found) + { + memcpy(&name_type_buffer[service_name_len], content_out, service_name_len); + } + + found &= find_tag_property(xml, + strlen(xml), + (id.type == UXR_REQUESTER_ID) ? "requester" : "replier", + "reply_type", + (const char **)&content_out, + &reply_type_name_len); + if (found) + { + memcpy(&name_type_buffer[service_name_len+request_type_name_len], content_out, service_name_len); + name_type_buffer[service_name_len+request_type_name_len+reply_type_name_len] = '\0'; + } + + if (found) + { + found &= add_brokerless_entity_hash(name_type_buffer, id); + } + break; + } + } + + return found; +} + + +bool add_brokerless_entity_hash(const char* ref, uxrObjectId id) +{ + if (brokerlessEntityMap.index < UCLIENT_BROKERLESS_ENTITY_MAP_LEN - 1) + { + hash_brokerless((unsigned char*) ref, brokerlessEntityMap.queue[brokerlessEntityMap.index].hash); + + brokerlessEntityMap.queue[brokerlessEntityMap.index].id.id = id.id; + brokerlessEntityMap.queue[brokerlessEntityMap.index].id.type = id.type; + + switch (id.type) + { + case UXR_DATAREADER_ID: + { + brokerlessEntityMap.datareaders++; + break; + } + case UXR_DATAWRITER_ID: + { + brokerlessEntityMap.datawriters++; + break; + } + case UXR_REQUESTER_ID: + { + brokerlessEntityMap.requesters++; + break; + } + case UXR_REPLIER_ID: + { + brokerlessEntityMap.repliers++; + break; + } + } + brokerlessEntityMap.index++; + return true; + } + + return false; +} + +int32_t find_brokerless_hash_from_id(uxrObjectId id) +{ + for (size_t i = 0; i < brokerlessEntityMap.index; i++) + { + if (brokerlessEntityMap.queue[i].id.id == id.id && + brokerlessEntityMap.queue[i].id.type == id.type) + { + return (int32_t) i; + } + } + return -1; +} + +int32_t find_brokerless_hash_from_hash(char* hash) +{ + for (size_t i = 0; i < brokerlessEntityMap.index; i++) + { + if (0 == memcmp((void*) hash, (void*) brokerlessEntityMap.queue[i].hash, BROKERLESS_HASH_SIZE)) + { + return (int32_t) i; + } + } + return -1; +} + +int32_t find_brokerless_hash_from_hash_only_reader(char* hash) +{ + for (size_t i = 0; i < brokerlessEntityMap.index; i++) + { + if (0 == memcmp((void*) hash, (void*) brokerlessEntityMap.queue[i].hash, BROKERLESS_HASH_SIZE) && + brokerlessEntityMap.queue[i].id.type != UXR_DATAWRITER_ID) + { + return (int32_t) i; + } + } + return -1; +} + +bool check_brokerless_sample_id(SampleIdentity sample_id) +{ + // TODO (pablogs9): Check if requester id stored in the sample_id still exists + + return !memcmp(&sample_id.writer_guid.entityId.entityKey, (uint8_t*)(&client_key), 3) && + !memcmp(&sample_id.writer_guid.entityId.entityKind, (uint8_t*)(&client_key) + 3, 1); +} + +void fill_brokerless_sample_id(SampleIdentity* sample_id, uxrObjectId id) +{ + // TODO (pablogs9): Improve this in order to take into account the endianness + + memcpy(&sample_id->writer_guid.entityId.entityKey, (uint8_t*)(&client_key), 3); + memcpy(&sample_id->writer_guid.entityId.entityKind, (uint8_t*)(&client_key) + 3, 1); + + memcpy(&sample_id->writer_guid.guidPrefix.data, (uint8_t*)(&id.id), 2); +} + +bool flush_brokerless_queues() +{ + for (size_t i = 0; i < brokerlessMessageQueue.index; i++) + { + int32_t hash_index = find_brokerless_hash_from_id(brokerlessMessageQueue.queue[i].id); + + if (-1 != hash_index) + { + ucdrBuffer writer; + ucdr_init_buffer(&writer, brokerlessBuffer, UCLIENT_BROKERLESS_INTERNAL_BUFFER_LEN); + ucdr_serialize_array_char(&writer, brokerlessEntityMap.queue[hash_index].hash, BROKERLESS_HASH_SIZE); + + if (brokerlessMessageQueue.queue[i].id.type == UXR_REQUESTER_ID || brokerlessMessageQueue.queue[i].id.type == UXR_REPLIER_ID) + { + + ucdr_serialize_bool(&writer, brokerlessMessageQueue.queue[i].id.type == UXR_REQUESTER_ID); + + if (brokerlessMessageQueue.queue[i].id.type == UXR_REQUESTER_ID) + { + fill_brokerless_sample_id(&brokerlessMessageQueue.queue[i].sample_id, brokerlessMessageQueue.queue[i].id); + } + + uxr_serialize_SampleIdentity(&writer, &brokerlessMessageQueue.queue[i].sample_id); + } + + ucdr_serialize_sequence_char(&writer, (char*) brokerlessMessageQueue.queue[i].data, (uint32_t)brokerlessMessageQueue.queue[i].length); + + brokerless_broadcast_send(writer.init, ucdr_buffer_length(&writer)); + } + } + + brokerlessMessageQueue.index = 0; + return false; +} + +bool listen_brokerless(uxrSession* session, int timeout) +{ + size_t readed_bytes = 0; + if (brokerlessEntityMap.datareaders || brokerlessEntityMap.requesters || brokerlessEntityMap.repliers) + { + readed_bytes = brokerless_broadcast_recv(brokerlessBuffer, UCLIENT_BROKERLESS_INTERNAL_BUFFER_LEN, timeout); + } + + if(0 != readed_bytes) + { + ucdrBuffer reader; + ucdr_init_buffer(&reader, brokerlessBuffer, readed_bytes); + + char hash[BROKERLESS_HASH_SIZE]; + ucdr_deserialize_array_char(&reader, hash, BROKERLESS_HASH_SIZE); + + int32_t hash_index = find_brokerless_hash_from_hash_only_reader(hash); + + if (-1 != hash_index && brokerlessEntityMap.queue[hash_index].id.type != UXR_DATAWRITER_ID) + { + uxrObjectId * object_id = &brokerlessEntityMap.queue[hash_index].id; + + // TODO (pablogs9): request_id is related to the uxr_buffer_request_data request, so it can determine some limitations imposed into the communication -> NOT IMPLEMENTED BY NOW + if (object_id->type == UXR_DATAREADER_ID) + { + uxrStreamId stream = {0, 0, UXR_BROKERLESS_STREAM, UXR_INPUT_STREAM}; + uint32_t length; + ucdr_deserialize_uint32_t(&reader, &length); + session->on_data_flag = true; + session->on_topic(session, *object_id, 0, stream, &reader, (uint16_t)length, session->on_topic_args); + } + else + { + bool is_from_requester; + SampleIdentity sample_id; + uint32_t length; + + ucdr_deserialize_bool(&reader, &is_from_requester); + + // sample_id deserialization is done inside conditional in order to not deserialize when message should be dropped + if (is_from_requester && object_id->type == UXR_REPLIER_ID) + { + uxr_deserialize_SampleIdentity(&reader, &sample_id); + ucdr_deserialize_uint32_t(&reader, &length); + session->on_data_flag = true; + session->on_request(session, *object_id, 0, &sample_id, &reader, (uint16_t)length, session->on_request_args); + } + else if(!is_from_requester && object_id->type == UXR_REQUESTER_ID) + { + uxr_deserialize_SampleIdentity(&reader, &sample_id); + if (check_brokerless_sample_id(sample_id)) + { + ucdr_deserialize_uint32_t(&reader, &length); + session->on_data_flag = true; + session->on_reply(session, *object_id, 0, (uint16_t)sample_id.sequence_number.low, &reader, (uint16_t)length, session->on_reply_args); + } + } + } + return true; + } + } + + return false; +} + diff --git a/src/c/brokerless/brokerless_internal.h b/src/c/brokerless/brokerless_internal.h new file mode 100644 index 000000000..fd42b1af4 --- /dev/null +++ b/src/c/brokerless/brokerless_internal.h @@ -0,0 +1,87 @@ +// 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. + +#ifndef _SRC_C_BROKERLESS_BROKERLESS_INTERNAL_H_ +#define _SRC_C_BROKERLESS_BROKERLESS_INTERNAL_H_ + +#ifdef __cplusplus +extern "C" +{ +#endif + +#include +#include +#include +#include + +#include +#include + +typedef uint32_t hash_int_t; +#define BROKERLESS_HASH_SIZE sizeof(hash_int_t) + +// Outcomming message queue + +typedef struct +{ + uint8_t *data; + size_t length; + uxrObjectId id; + SampleIdentity sample_id; +} brokerlessMessage_t; + +typedef struct +{ + brokerlessMessage_t queue[UCLIENT_BROKERLESS_MESSAGE_QUEUE_LEN]; + size_t index; +} brokerlessMessageQueue_t; + +// Entities map + +typedef struct +{ + char hash[BROKERLESS_HASH_SIZE]; + uxrObjectId id; +} brokerlessEntityHash_t; + + +typedef struct +{ + brokerlessEntityHash_t queue[UCLIENT_BROKERLESS_ENTITY_MAP_LEN]; + size_t index; + uint8_t datawriters; + uint8_t datareaders; + uint8_t requesters; + uint8_t repliers; +} brokerlessEntityMap_t; + +// Internal API (TODO: add documentation) + +void init_brokerless(uint32_t key); +bool add_brokerless_message(ucdrBuffer* ub, uint32_t length, uxrObjectId id); +bool add_brokerless_message_with_sample_id(ucdrBuffer* ub, size_t length, uxrObjectId id, SampleIdentity sample_id); +bool add_brokerless_entity_hash_from_xml(const char* xml, uxrObjectId id); +bool add_brokerless_entity_hash(const char* ref, uxrObjectId id); +int32_t find_brokerless_hash_from_id(uxrObjectId id); +int32_t find_brokerless_hash_from_hash(char* hash); +bool check_brokerless_sample_id(SampleIdentity sample_id); +void fill_brokerless_sample_id(SampleIdentity* sample_id, uxrObjectId id); +bool flush_brokerless_queues(); +bool listen_brokerless(uxrSession* session, int timeout); + +#ifdef __cplusplus +} +#endif + +#endif // _SRC_C_BROKERLESS_BROKERLESS_INTERNAL_H_ diff --git a/src/c/brokerless/brokerless_transport.h b/src/c/brokerless/brokerless_transport.h new file mode 100644 index 000000000..4ae5c5b1c --- /dev/null +++ b/src/c/brokerless/brokerless_transport.h @@ -0,0 +1,44 @@ +// 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. + +#ifndef _SRC_C_BROKERLESS_BROKERLESS_TRANSPORT_H_ +#define _SRC_C_BROKERLESS_BROKERLESS_TRANSPORT_H_ + +#ifdef __cplusplus +extern "C" +{ +#endif + +#include +#include +#include + +bool brokerless_init_transport(); + +bool brokerless_close_transport(); + +size_t brokerless_broadcast_send( + const uint8_t* buf, + size_t len); + +size_t brokerless_broadcast_recv( + uint8_t* buf, + size_t len, + int timeout); + +#ifdef __cplusplus +} +#endif + +#endif // _SRC_C_BROKERLESS_BROKERLESS_TRANSPORT_H_ diff --git a/src/c/brokerless/udp_transport_broadcast_posix.c b/src/c/brokerless/udp_transport_broadcast_posix.c new file mode 100644 index 000000000..e09a2d94b --- /dev/null +++ b/src/c/brokerless/udp_transport_broadcast_posix.c @@ -0,0 +1,84 @@ + +#include "brokerless_internal.h" + +#include + +#include +#include +#include +#include +#include + +#define BROADCAST_DEFAULT_IP "255.255.255.255" + +int fd_recv; +int fd_send; +struct sockaddr_in send_addr, recv_addr; + +bool brokerless_init_transport() +{ + int trueflag = 1; + fd_recv = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + fd_send = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + + setsockopt(fd_recv, SOL_SOCKET, SO_BROADCAST, &trueflag, sizeof trueflag); + setsockopt(fd_send, SOL_SOCKET, SO_BROADCAST, &trueflag, sizeof trueflag); + + setsockopt(fd_recv, SOL_SOCKET, SO_REUSEADDR, &trueflag, sizeof trueflag); + setsockopt(fd_send, SOL_SOCKET, SO_REUSEADDR, &trueflag, sizeof trueflag); + + memset(&send_addr, 0, sizeof send_addr); + send_addr.sin_family = AF_INET; + send_addr.sin_port = (in_port_t) htons(UCLIENT_BROKERLESS_PORT); + inet_aton(BROADCAST_DEFAULT_IP, &send_addr.sin_addr); + + memset(&recv_addr, 0, sizeof recv_addr); + recv_addr.sin_family = AF_INET; + recv_addr.sin_port = (in_port_t) htons(UCLIENT_BROKERLESS_PORT); + recv_addr.sin_addr.s_addr = htonl(INADDR_ANY); + + bind(fd_recv, (struct sockaddr*) &recv_addr, sizeof recv_addr); + + return true; +} + +bool brokerless_close_transport() +{ + return ((0 == close(fd_recv)) && 0 == close(fd_send)); +} + +size_t brokerless_broadcast_send( + const uint8_t* buf, + size_t len) +{ + size_t rv = 0; + + ssize_t bytes_sent = sendto(fd_send, buf, len, 0, (struct sockaddr*) &send_addr, sizeof(send_addr)); + + rv = (0 > bytes_sent) ? 0 : (size_t)bytes_sent; + return rv; +} + +size_t brokerless_broadcast_recv( + uint8_t* buf, + size_t len, + int timeout) +{ + timeout = (timeout <= 0) ? 1 : timeout; + + struct timeval tv; + tv.tv_sec = timeout / 1000; + tv.tv_usec = (timeout % 1000) * 1000; + + setsockopt(fd_recv, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + + struct sockaddr_in from; + uint fromlen; + fromlen = sizeof(from); + + // TODO: what if received packages are not full? + ssize_t readed_bytes = recvfrom(fd_recv, (void*)buf, len, 0, (struct sockaddr *)(&from), (socklen_t *)(&fromlen)); + + return (readed_bytes > 0) ? (size_t)readed_bytes : 0; +} + diff --git a/src/c/core/session/create_entities_ref.c b/src/c/core/session/create_entities_ref.c index b4fd44e8b..4307d9d71 100644 --- a/src/c/core/session/create_entities_ref.c +++ b/src/c/core/session/create_entities_ref.c @@ -2,6 +2,7 @@ #include #include "common_create_entities_internal.h" +#include "../../brokerless/brokerless_internal.h" #include @@ -43,6 +44,10 @@ uint16_t uxr_buffer_create_datawriter_ref(uxrSession* session, uxrStreamId strea payload.object_representation.kind = DDS_XRCE_OBJK_DATAWRITER; uxr_object_id_to_raw(publisher_id, payload.object_representation._.data_writer.publisher_id.data); +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_entity_hash(ref, object_id); +#endif + return create_entity_ref(session, stream_id, object_id, ref, mode, &payload); } @@ -54,6 +59,10 @@ uint16_t uxr_buffer_create_datareader_ref(uxrSession* session, uxrStreamId strea payload.object_representation.kind = DDS_XRCE_OBJK_DATAREADER; uxr_object_id_to_raw(subscriber_id, payload.object_representation._.data_reader.subscriber_id.data); +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_entity_hash(ref, object_id); +#endif + return create_entity_ref(session, stream_id, object_id, ref, mode, &payload); } @@ -69,6 +78,10 @@ uint16_t uxr_buffer_create_requester_ref( payload.object_representation.kind = DDS_XRCE_OBJK_REQUESTER; uxr_object_id_to_raw(participant_id, payload.object_representation._.requester.participant_id.data); +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_entity_hash(ref, object_id); +#endif + return create_entity_ref(session, stream_id, object_id, ref, mode, &payload); } @@ -84,6 +97,10 @@ uint16_t uxr_buffer_create_replier_ref( payload.object_representation.kind = DDS_XRCE_OBJK_REPLIER; uxr_object_id_to_raw(participant_id, payload.object_representation._.replier.participant_id.data); +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_entity_hash(ref, object_id); +#endif + return create_entity_ref(session, stream_id, object_id, ref, mode, &payload); } diff --git a/src/c/core/session/create_entities_xml.c b/src/c/core/session/create_entities_xml.c index 0d3256f61..0252434da 100644 --- a/src/c/core/session/create_entities_xml.c +++ b/src/c/core/session/create_entities_xml.c @@ -2,6 +2,7 @@ #include #include "common_create_entities_internal.h" +#include "../../brokerless/brokerless_internal.h" #include @@ -69,6 +70,10 @@ uint16_t uxr_buffer_create_datawriter_xml(uxrSession* session, uxrStreamId strea payload.object_representation.kind = DDS_XRCE_OBJK_DATAWRITER; uxr_object_id_to_raw(publisher_id, payload.object_representation._.data_writer.publisher_id.data); +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_entity_hash_from_xml(xml, object_id); +#endif + return create_entity_xml(session, stream_id, object_id, xml, mode, &payload); } @@ -81,6 +86,10 @@ uint16_t uxr_buffer_create_datareader_xml(uxrSession* session, uxrStreamId strea payload.object_representation.kind = DDS_XRCE_OBJK_DATAREADER; uxr_object_id_to_raw(subscriber_id, payload.object_representation._.data_reader.subscriber_id.data); +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_entity_hash_from_xml(xml, object_id); +#endif + return create_entity_xml(session, stream_id, object_id, xml, mode, &payload); } @@ -96,6 +105,10 @@ uint16_t uxr_buffer_create_requester_xml( payload.object_representation.kind = DDS_XRCE_OBJK_REQUESTER; uxr_object_id_to_raw(participant_id, payload.object_representation._.requester.participant_id.data); +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_entity_hash_from_xml(xml, object_id); +#endif + return create_entity_xml(session, stream_id, object_id, xml, mode, &payload); } @@ -111,6 +124,10 @@ uint16_t uxr_buffer_create_replier_xml( payload.object_representation.kind = DDS_XRCE_OBJK_REPLIER; uxr_object_id_to_raw(participant_id, payload.object_representation._.replier.participant_id.data); +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_entity_hash_from_xml(xml, object_id); +#endif + return create_entity_xml(session, stream_id, object_id, xml, mode, &payload); } diff --git a/src/c/core/session/session.c b/src/c/core/session/session.c index e394fad55..e382c5ef4 100644 --- a/src/c/core/session/session.c +++ b/src/c/core/session/session.c @@ -16,6 +16,7 @@ #include "stream/seq_num_internal.h" #include "../log/log_internal.h" #include "../../util/time_internal.h" +#include "../../brokerless/brokerless_internal.h" #define CREATE_SESSION_MAX_MSG_SIZE (MAX_HEADER_SIZE + SUBHEADER_SIZE + CREATE_CLIENT_PAYLOAD_SIZE) #define DELETE_SESSION_MAX_MSG_SIZE (MAX_HEADER_SIZE + SUBHEADER_SIZE + DELETE_CLIENT_PAYLOAD_SIZE) @@ -81,6 +82,10 @@ void uxr_init_session(uxrSession* session, uxrCommunication* comm, uint32_t key) uxr_init_session_info(&session->info, 0x81, key); uxr_init_stream_storage(&session->streams); + +#ifdef UCLIENT_PROFILE_BROKERLESS + init_brokerless(key); // TODO: should it be desirable to give the user an uxr_init_brokerless_transport() method? +#endif } void uxr_set_status_callback(uxrSession* session, uxrOnStatusFunc on_status_func, void* args) @@ -370,6 +375,11 @@ bool uxr_buffer_performance(uxrSession *session, void uxr_flash_output_streams(uxrSession* session) { + +#ifdef UCLIENT_PROFILE_BROKERLESS + flush_brokerless_queues(); +#endif + for(uint8_t i = 0; i < session->streams.output_best_effort_size; ++i) { uxrOutputBestEffortStream* stream = &session->streams.output_best_effort[i]; @@ -411,6 +421,10 @@ bool listen_message(uxrSession* session, int poll_ms) read_message(session, &ub); } +#ifdef UCLIENT_PROFILE_BROKERLESS + listen_brokerless(session, poll_ms); +#endif + return must_be_read; } @@ -480,6 +494,7 @@ inline bool send_message(const uxrSession* session, uint8_t* buffer, size_t leng inline bool recv_message(const uxrSession* session, uint8_t**buffer, size_t* length, int poll_ms) { bool received = session->comm->recv_msg(session->comm->instance, buffer, length, poll_ms); + //LISTEN P2P PROPOSAL if(received) { UXR_DEBUG_PRINT_MESSAGE(UXR_RECV, *buffer, *length, session->info.key); diff --git a/src/c/core/session/stream/stream_id.c b/src/c/core/session/stream/stream_id.c index 27a15867b..cfcb7bd9c 100644 --- a/src/c/core/session/stream/stream_id.c +++ b/src/c/core/session/stream/stream_id.c @@ -20,6 +20,7 @@ uxrStreamId uxr_stream_id(uint8_t index, uxrStreamType type, uxrStreamDirection case UXR_NONE_STREAM: stream_id.raw = 0; break; + case UXR_BROKERLESS_STREAM: case UXR_BEST_EFFORT_STREAM: stream_id.raw = (uint8_t)(index + BEST_EFFORT_STREAM_THRESHOLD); break; diff --git a/src/c/core/session/write_access.c b/src/c/core/session/write_access.c index 7a60d155d..c8b61a550 100644 --- a/src/c/core/session/write_access.c +++ b/src/c/core/session/write_access.c @@ -4,6 +4,7 @@ #include "session_internal.h" #include "session_info_internal.h" #include "submessage_internal.h" +#include "../../brokerless/brokerless_internal.h" #define WRITE_DATA_PAYLOAD_SIZE 4 #define SAMPLE_IDENTITY_SIZE 24 @@ -28,6 +29,12 @@ uint16_t uxr_buffer_request( WRITE_DATA_Payload_Data payload; rv = uxr_init_base_object_request(&session->info, requester_id, &payload.base); uxr_serialize_WRITE_DATA_Payload_Data(&ub, &payload); + +#ifdef UCLIENT_PROFILE_BROKERLESS + SampleIdentity sample_id; + sample_id.sequence_number.low = rv; + add_brokerless_message_with_sample_id(&ub, len, requester_id, sample_id); +#endif ucdr_serialize_array_uint8_t(&ub, buffer, len); } @@ -53,6 +60,11 @@ uint16_t uxr_buffer_reply( rv = uxr_init_base_object_request(&session->info, replier_id, &payload.base); uxr_serialize_WRITE_DATA_Payload_Data(&ub, &payload); uxr_serialize_SampleIdentity(&ub, sample_id); + +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_message_with_sample_id(&ub, len, replier_id, *sample_id); +#endif + ucdr_serialize_array_uint8_t(&ub, buffer, len); } @@ -74,6 +86,10 @@ bool uxr_prepare_output_stream(uxrSession* session, uxrStreamId stream_id, uxrOb void* args = ub->args; ucdr_init_buffer(ub, ub->iterator, (size_t)(ub->final - ub->iterator)); ucdr_set_on_full_buffer_callback(ub, on_full_buffer, args); + +#ifdef UCLIENT_PROFILE_BROKERLESS + add_brokerless_message(ub, topic_size, datawriter_id); +#endif } return !ub->error; diff --git a/test/unitary/session/Session.cpp b/test/unitary/session/Session.cpp index 0706a8a8b..f2dc77bb1 100644 --- a/test/unitary/session/Session.cpp +++ b/test/unitary/session/Session.cpp @@ -23,6 +23,11 @@ extern "C" #include +#ifdef UCLIENT_PROFILE_BROKERLESS +#include +#include +#endif // UCLIENT_PROFILE_BROKERLESS + #undef UXR_MESSAGE_LOG #undef UXR_SERIALIZATION_LOG #include diff --git a/test/unitary/session/WriteReadAccess.cpp b/test/unitary/session/WriteReadAccess.cpp index 8f543c668..d2f08c48a 100644 --- a/test/unitary/session/WriteReadAccess.cpp +++ b/test/unitary/session/WriteReadAccess.cpp @@ -20,6 +20,11 @@ extern "C" #include #include + +#ifdef UCLIENT_PROFILE_BROKERLESS +#include +#include +#endif // UCLIENT_PROFILE_BROKERLESS } #include