prevent duplicate xlink stream names

src/device/Device.cpp

@@ -265,8 +265,10 @@ bool Device::startPipelineImpl(const Pipeline& pipeline) {
         }
 
         // Create DataInputQueue's
+        auto streamName = xlinkIn->getStreamName();
+        if(inputQueueMap.count(streamName) != 0) throw std::invalid_argument(fmt::format("Streams have duplicate name '{}'", streamName));
         // set max data size, for more verbosity
-        inputQueueMap[xlinkIn->getStreamName()] = std::make_shared<DataInputQueue>(connection, xlinkIn->getStreamName(), 16, true, xlinkIn->getMaxDataSize());
+        inputQueueMap[std::move(streamName)] = std::make_shared<DataInputQueue>(connection, xlinkIn->getStreamName(), 16, true, xlinkIn->getMaxDataSize());
     }
     for(const auto& kv : pipeline.getNodeMap()) {
         const auto& node = kv.second;
@@ -275,29 +277,31 @@ bool Device::startPipelineImpl(const Pipeline& pipeline) {
             continue;
         }
 
-        auto streamName = xlinkOut->getStreamName();
         // Create DataOutputQueue's
+        auto streamName = xlinkOut->getStreamName();
+        if(outputQueueMap.count(streamName) != 0) throw std::invalid_argument(fmt::format("Streams have duplicate name '{}'", streamName));
         outputQueueMap[streamName] = std::make_shared<DataOutputQueue>(connection, streamName);
 
         // Add callback for events
-        callbackIdMap[streamName] = outputQueueMap[streamName]->addCallback([this](std::string queueName, std::shared_ptr<ADatatype>) {
-            {
-                // Lock first
-                std::unique_lock<std::mutex> lock(eventMtx);
-
-                // Check if size is equal or greater than EVENT_QUEUE_MAXIMUM_SIZE
-                if(eventQueue.size() >= EVENT_QUEUE_MAXIMUM_SIZE) {
-                    auto numToRemove = eventQueue.size() - EVENT_QUEUE_MAXIMUM_SIZE + 1;
-                    eventQueue.erase(eventQueue.begin(), eventQueue.begin() + numToRemove);
-                }
+        callbackIdMap[std::move(streamName)] =
+            outputQueueMap[xlinkOut->getStreamName()]->addCallback([this](std::string queueName, std::shared_ptr<ADatatype>) {
+                {
+                    // Lock first
+                    std::unique_lock<std::mutex> lock(eventMtx);
+
+                    // Check if size is equal or greater than EVENT_QUEUE_MAXIMUM_SIZE
+                    if(eventQueue.size() >= EVENT_QUEUE_MAXIMUM_SIZE) {
+                        auto numToRemove = eventQueue.size() - EVENT_QUEUE_MAXIMUM_SIZE + 1;
+                        eventQueue.erase(eventQueue.begin(), eventQueue.begin() + numToRemove);
+                    }
 
-                // Add to the end of event queue
-                eventQueue.push_back(std::move(queueName));
-            }
+                    // Add to the end of event queue
+                    eventQueue.push_back(std::move(queueName));
+                }
 
-            // notify the rest
-            eventCv.notify_all();
-        });
+                // notify the rest
+                eventCv.notify_all();
+            });
     }
     return DeviceBase::startPipelineImpl(pipeline);
 }

tests/src/pipeline_test.cpp

@@ -57,3 +57,27 @@ TEST_CASE("Cross pipeline link with Input and Output") {
     // Then check that actually linking throws
     REQUIRE_THROWS(xin->out.link(xout->input));
 }
+
+TEST_CASE("Duplicate xlink stream names") {
+    dai::Pipeline p;
+    auto sysInfo1 = p.create<dai::node::SystemLogger>();
+    auto sysInfo2 = p.create<dai::node::SystemLogger>();
+    auto xout1 = p.create<dai::node::XLinkOut>();
+    auto xout2 = p.create<dai::node::XLinkOut>();
+    sysInfo1->out.link(xout1->input);
+    sysInfo2->out.link(xout2->input);
+    xout1->setStreamName("test1");
+    xout2->setStreamName("test1");
+    REQUIRE_THROWS_AS(dai::Device{p}, std::invalid_argument);
+
+    p = {};
+    auto script1 = p.create<dai::node::Script>();
+    auto script2 = p.create<dai::node::Script>();
+    auto xin1 = p.create<dai::node::XLinkIn>();
+    auto xin2 = p.create<dai::node::XLinkIn>();
+    xin1->out.link(script1->inputs["in"]);
+    xin2->out.link(script2->inputs["in"]);
+    xin1->setStreamName("test2");
+    xin1->setStreamName("test2");
+    REQUIRE_THROWS_AS(dai::Device{p}, std::invalid_argument);
+}