Make the trigger of the DDS Pipe callbacks configurable (#406)

* Dynamically trigger the creation of entities Signed-off-by: tempate <[email protected]> * Add none entity-creation-trigger Signed-off-by: tempate <[email protected]> * Rename entity_creation_trigger to discovery_trigger Signed-off-by: tempate <[email protected]> * Uncrustify Signed-off-by: tempate <[email protected]> * Documentation Signed-off-by: tempate <[email protected]> * Tests Signed-off-by: tempate <[email protected]> * Rename disconnects to n-unmatches Signed-off-by: tempate <[email protected]> * Add n-matches option Signed-off-by: tempate <[email protected]> * Docker tests Signed-off-by: tempate <[email protected]> * Only read n-matches args when samples == 0 Signed-off-by: tempate <[email protected]> * Fix tests Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> * Fix python linter Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> * Fix timeout_as_error python argument Signed-off-by: tempate <[email protected]> * Remove outdated warning Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> * Uncrustify Signed-off-by: tempate <[email protected]> * Fix misspell Signed-off-by: tempate <[email protected]> * Apply suggestions Signed-off-by: tempate <[email protected]> --------- Signed-off-by: tempate <[email protected]>
eProsima · Nov 22, 2023 · 5988fa4 · 5988fa4
1 parent 6140c85
commit 5988fa4
Show file tree

Hide file tree

Showing 24 changed files with 537 additions and 65 deletions.
diff --git a/ddsrouter_core/include/ddsrouter_core/configuration/SpecsConfiguration.hpp b/ddsrouter_core/include/ddsrouter_core/configuration/SpecsConfiguration.hpp
@@ -19,6 +19,7 @@
 
 #include <cpp_utils/Formatter.hpp>
 
+#include <ddspipe_core/configuration/DdsPipeConfiguration.hpp>
 #include <ddspipe_core/configuration/IConfiguration.hpp>
 #include <ddspipe_core/types/dds/TopicQoS.hpp>
 
@@ -66,6 +67,9 @@ struct SpecsConfiguration : public ddspipe::core::IConfiguration
 
     //! The globally configured Topic QoS.
     ddspipe::core::types::TopicQoS topic_qos{};
+
+    //! The type of the entities whose discovery triggers the discovery callbacks.
+    ddspipe::core::DiscoveryTrigger discovery_trigger = ddspipe::core::DiscoveryTrigger::READER;
 };
 
 } /* namespace core */

diff --git a/ddsrouter_test/compose/CMakeLists.txt b/ddsrouter_test/compose/CMakeLists.txt
@@ -62,6 +62,12 @@ set(TESTS
     manual_topics/generic
     manual_topics/participants
     manual_topics/precedence
+
+    discovery_trigger/reader
+    discovery_trigger/writer
+    discovery_trigger/none
+    discovery_trigger/any/reader
+    discovery_trigger/any/writer
 )
 
 file(COPY ${CMAKE_CURRENT_SOURCE_DIR}/test_cases DESTINATION ${CMAKE_CURRENT_BINARY_DIR})

diff --git a/ddsrouter_test/compose/docker-compose.sh b/ddsrouter_test/compose/docker-compose.sh
@@ -106,14 +106,14 @@ main ()
     echo "Docker compose file: ${COMPOSE_FILE}"
 
     # Run docker compose
-    docker-compose -f ${COMPOSE_FILE} up
+    docker compose -f ${COMPOSE_FILE} up
 
     # First this command gets the ids of every container listed on the docker compose file.
     # Then it checks the exist code of every container listed before, remove the exit codes equal
     # to 0 and count how many containers exited with an exit code different than 0.
     # As a result, the EXIT_CODE is the number of containers that exited with an exit code
     # different than 0.
-    EXIT_CODE=$(docker-compose -f ${COMPOSE_FILE} ps -q |
+    EXIT_CODE=$(docker compose -f ${COMPOSE_FILE} ps -aq |
         xargs docker inspect -f '{{ .State.ExitCode }}' |
         grep -vx "^0$" | wc -l | tr -d ' ')
 

diff --git a/ddsrouter_test/compose/scripts/execute_and_validate_publisher.py b/ddsrouter_test/compose/scripts/execute_and_validate_publisher.py
@@ -73,10 +73,15 @@ def parse_options():
     )
 
     parser.add_argument(
-        '--disconnects',
+        '--n-matches',
         type=int,
-        default=0,
-        help='Number of times the other participant is expected to disconnect.'
+        help='Number of times the other participant is expected to match.'
+    )
+
+    parser.add_argument(
+        '--n-unmatches',
+        type=int,
+        help='Number of times the other participant is expected to unmatch.'
     )
 
     return parser.parse_args()
@@ -100,45 +105,53 @@ def _publisher_command(args):
 
 def _publisher_parse_output(stdout, stderr):
     """
-    Transform the output of the program in a list of received disconnects.
+    Transform the output to a list of received matches and unmatches.
 
     :param data: Process stdout
-    :return: List of subscribers who have disconnected
+    :return: List of subscribers who have matched and unmatched
     """
-    regex = re.compile(r'^Publisher unmatched \[.+\].$')
-    lines = stdout.splitlines()
-    filtered_data = [line for line in lines if regex.match(line)]
+    match_regex = re.compile(r'^Publisher matched \[.+\].$')
+    unmatch_regex = re.compile(r'^Publisher unmatched \[.+\].$')
 
-    return filtered_data, stderr
+    filtered_data = {'matches': 0, 'unmatches': 0}
 
+    for line in stdout.splitlines():
+        if match_regex.match(line):
+            filtered_data['matches'] += 1
 
-def _publisher_get_retcode_validate(
-        disconnects):
-    if disconnects != 0:
-        return validation.validate_retcode_default
+        elif unmatch_regex.match(line):
+            filtered_data['unmatches'] += 1
+
+    return filtered_data, stderr
 
-    def accept_timeout(retcode):
-        return retcode in [
-            validation.ReturnCode.SUCCESS,
-            validation.ReturnCode.TIMEOUT]
 
-    return accept_timeout
+def _publisher_get_retcode_validate():
+    return lambda retcode: retcode == validation.ReturnCode.SUCCESS or \
+                           retcode == validation.ReturnCode.TIMEOUT
 
 
 def _publisher_validate(
         stdout_parsed,
         stderr_parsed,
-        disconnects):
+        n_matches,
+        n_unmatches):
 
     # Check default validator
     ret_code = validation.validate_default(stdout_parsed, stderr_parsed)
 
-    if len(stdout_parsed) != disconnects:
-        log.logger.error(f'Number of disconnected receivers: \
+    if n_matches is not None and stdout_parsed['matches'] != n_matches:
+        log.logger.error(f'Number of matched receivers: \
+                         {len(stdout_parsed)}. '
+                         f'Expected {n_matches}')
+
+        return validation.ReturnCode.NOT_VALID_MATCHES
+
+    if n_unmatches is not None and stdout_parsed['unmatches'] != n_unmatches:
+        log.logger.error(f'Number of unmatched receivers: \
                          {len(stdout_parsed)}. '
-                         f'Expected {disconnects}')
+                         f'Expected {n_unmatches}')
 
-        return validation.ReturnCode.NOT_VALID_DISCONNECTS
+        return validation.ReturnCode.NOT_VALID_UNMATCHES
 
     return ret_code
 
@@ -158,7 +171,8 @@ def _publisher_validate(
         _publisher_validate(
             stdout_parsed,
             stderr_parsed,
-            args.disconnects
+            args.n_matches,
+            args.n_unmatches
             )))
 
     ret_code = validation.run_and_validate(
@@ -167,8 +181,7 @@ def _publisher_validate(
         delay=args.delay,
         parse_output_function=_publisher_parse_output,
         validate_output_function=validate_func,
-        parse_retcode_function=_publisher_get_retcode_validate(
-            args.disconnects),
+        parse_retcode_function=_publisher_get_retcode_validate(),
         timeout_as_error=False)
 
     log.logger.info(f'Publisher validator exited with code {ret_code}')

diff --git a/ddsrouter_test/compose/scripts/execute_and_validate_subscriber.py b/ddsrouter_test/compose/scripts/execute_and_validate_subscriber.py
@@ -52,7 +52,6 @@ def parse_options():
         '-s',
         '--samples',
         type=int,
-        default=5,
         help='Samples to receive.'
     )
     parser.add_argument(
@@ -98,6 +97,18 @@ def parse_options():
         help='Maximum amount of seconds the command should take before finishing.'
     )
 
+    parser.add_argument(
+        '--n-matches',
+        type=int,
+        help='Number of times the other participant is expected to match.'
+    )
+
+    parser.add_argument(
+        '--n-unmatches',
+        type=int,
+        help='Number of times the other participant is expected to unmatch.'
+    )
+
     return parser.parse_args()
 
 
@@ -124,53 +135,77 @@ def _subscriber_parse_output(stdout, stderr):
     :param data: Process stdout
     :return: List of received messages
     """
-    regex = re.compile(r'^Message\sHelloWorld\s+\d+\sRECEIVED$')
-    lines = stdout.splitlines()
-    filtered_data = [line for line in lines if regex.match(line)]
+
+    msgs_regex = re.compile(r'^Message\sHelloWorld\s+\d+\sRECEIVED$')
+    match_regex = re.compile(r'^Subscriber matched \[.+\].$')
+    unmatch_regex = re.compile(r'^Subscriber unmatched \[.+\].$')
+
+    filtered_data = {'messages': [], 'matches': 0, 'unmatches': 0}
+
+    for line in stdout.splitlines():
+        if msgs_regex.match(line):
+            filtered_data['messages'].append(line)
+
+        elif match_regex.match(line):
+            filtered_data['matches'] += 1
+
+        elif unmatch_regex.match(line):
+            filtered_data['unmatches'] += 1
 
     return filtered_data, stderr
 
 
 def _subscriber_get_retcode_validate(
         samples):
-    if samples == 0:
-        def accept_timeout(retcode):
-            return (
-                retcode == validation.ReturnCode.SUCCESS
-                or retcode == validation.ReturnCode.TIMEOUT)
-        return accept_timeout
-    else:
-        return validation.validate_retcode_default
+    if samples is None or samples == 0:
+        return lambda retcode: retcode == validation.ReturnCode.SUCCESS or \
+                               retcode == validation.ReturnCode.TIMEOUT
+
+    return validation.validate_retcode_default
 
 
 def _subscriber_validate(
         stdout_parsed,
         stderr_parsed,
         samples,
         duplicates_allow,
-        transient):
+        transient,
+        n_matches,
+        n_unmatches):
 
     # Check default validator
-    ret_code = validation.validate_default(stdout_parsed, stderr_parsed)
+    ret_code = validation.validate_default(stdout_parsed['messages'], stderr_parsed)
 
     if duplicates_allow != -1:
-        duplicated_n = len(find_duplicates(stdout_parsed))
+        duplicated_n = len(find_duplicates(stdout_parsed['messages']))
         if duplicated_n > duplicates_allow:
             log.logger.error(
                 f'{duplicated_n} duplicated messages found. '
                 f'Maximum allowed {duplicates_allow}.')
             return validation.ReturnCode.NOT_VALID_MESSAGES
 
     if transient:
-        if not check_transient(stdout_parsed):
+        if not check_transient(stdout_parsed['messages']):
             log.logger.error('Transient messages incorrect reception.')
             return validation.ReturnCode.NOT_VALID_MESSAGES
 
-    if len(stdout_parsed) != samples:
-        log.logger.error(f'Number of messages received: {len(stdout_parsed)}. '
+    if samples is not None and len(stdout_parsed['messages']) != samples:
+        log.logger.error(f'Number of messages received: {len(stdout_parsed["messages"])}. '
                          f'Expected {samples}')
         return validation.ReturnCode.NOT_VALID_MESSAGES
 
+    if n_matches is not None and stdout_parsed['matches'] != n_matches:
+        log.logger.error(f'Number of matched receivers: {stdout_parsed["matches"]}.'
+                         f'Expected {n_matches}')
+
+        return validation.ReturnCode.NOT_VALID_MATCHES
+
+    if n_unmatches is not None and stdout_parsed['unmatches'] != n_unmatches:
+        log.logger.error(f'Number of unmatched receivers: {stdout_parsed["unmatches"]}.'
+                         f'Expected {n_unmatches}')
+
+        return validation.ReturnCode.NOT_VALID_UNMATCHES
+
     return ret_code
 
 
@@ -235,13 +270,17 @@ def check_transient(data):
 
     command = _subscriber_command(args)
 
+    timeout_as_error = args.samples is not None and args.samples > 0
+
     validate_func = (lambda stdout_parsed, stderr_parsed: (
         _subscriber_validate(
             stdout_parsed=stdout_parsed,
             stderr_parsed=stderr_parsed,
             samples=args.samples,
             duplicates_allow=args.allow_duplicates,
-            transient=args.transient
+            transient=args.transient,
+            n_matches=args.n_matches,
+            n_unmatches=args.n_unmatches
             )))
 
     ret_code = validation.run_and_validate(
@@ -251,7 +290,7 @@ def check_transient(data):
         parse_output_function=_subscriber_parse_output,
         validate_output_function=validate_func,
         parse_retcode_function=_subscriber_get_retcode_validate(args.samples),
-        timeout_as_error=args.samples > 0,
+        timeout_as_error=timeout_as_error,
         min_time=args.min_time,
         max_time=args.max_time)
 

diff --git a/ddsrouter_test/compose/scripts/validation.py b/ddsrouter_test/compose/scripts/validation.py
@@ -32,9 +32,10 @@ class ReturnCode(Enum):
     NOT_VALID_MESSAGES = 3
     COMMAND_FAIL = 4
     STDERR_OUTPUT = 5
-    NOT_VALID_DISCONNECTS = 6
-    FINISHED_TOO_QUICKLY = 7
-    FINISHED_TOO_SLOWLY = 8
+    NOT_VALID_MATCHES = 6
+    NOT_VALID_UNMATCHES = 7
+    FINISHED_TOO_QUICKLY = 8
+    FINISHED_TOO_SLOWLY = 9
 
 
 """

diff --git a/ddsrouter_test/compose/test_cases/discovery_trigger/any/ddsrouter.yaml b/ddsrouter_test/compose/test_cases/discovery_trigger/any/ddsrouter.yaml
@@ -0,0 +1,13 @@
+version: v4.0
+
+participants:
+  - name: Local_80
+    kind: local
+    domain: 80
+
+  - name: Local_81
+    kind: local
+    domain: 81
+
+specs:
+  discovery-trigger: any
diff --git a/ddsrouter_test/compose/test_cases/discovery_trigger/any/reader/compose.yml b/ddsrouter_test/compose/test_cases/discovery_trigger/any/reader/compose.yml
@@ -0,0 +1,38 @@
+# Test description:
+#   This test checks that if we set up a router with any as the discovery trigger,
+#   a subscriber does match with the router (since it triggers the creation of entities).
+#
+# Test architecture:
+#
+# ┌──────────┐    ┌─────────┐
+# │subscriber│    │ddsrouter│
+# │          ├────►         │
+# │(local)   │    │(local)  │
+# └──────────┘    └─────────┘
+
+services:
+
+  # ROUTER
+  ddsrouter:
+    image: ${DDSROUTER_COMPOSE_TEST_DOCKER_IMAGE}
+    container_name: ddsrouter
+    networks:
+      - std_net
+    volumes:
+      - ../ddsrouter.yaml:/config.yaml
+    command: ddsrouter -c /config.yaml --timeout 4
+
+  # DOMAIN 0
+  subscriber_0_t0:
+    image: ${DDSROUTER_COMPOSE_TEST_DOCKER_IMAGE}
+    container_name: subscriber_0_t0
+    networks:
+      - std_net
+    volumes:
+      - ../../../../scripts:/scripts
+    command: python3 /scripts/execute_and_validate_subscriber.py --exe install/AdvancedConfigurationExample/examples/cpp/dds/AdvancedConfigurationExample/AdvancedConfigurationExample --n-matches 1 --timeout 4 --args "--domain 80"
+
+networks:
+  std_net:
+  default:
+    driver: none