Stop a python source once the subscriber is no longer subscribed (#493)

* When a Python generator source yields a value, and the subscriber is no longer subscribed, stop the source.
* Fix out of date docstring comment.

This is a partial fix for nv-morpheus/Morpheus#1838

Authors:
  - David Gardner (https://github.com/dagardner-nv)

Approvers:
  - Anuradha Karuppiah (https://github.com/AnuradhaKaruppiah)

URL: #493

cpp/mrc/src/internal/service.hpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -36,7 +36,7 @@ enum class ServiceState
 };
 
 /**
- * @brief Converts a `AsyncServiceState` enum to a string
+ * @brief Converts a `ServiceState` enum to a string
  *
  * @param f
  * @return std::string

python/mrc/_pymrc/src/segment.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -233,6 +233,11 @@ std::shared_ptr<mrc::segment::ObjectProperties> build_source(mrc::segment::IBuil
                 {
                     subscriber.on_next(std::move(next_val));
                 }
+                else
+                {
+                    DVLOG(10) << ctx.info() << " Source unsubscribed. Stopping";
+                    break;
+                }
             }
 
         } catch (const std::exception& e)