Merge pull request #15328 from Atlante45/fix/injectors

Case 21925: Warn mixer about stopped injectors

assignment-client/src/audio/AudioMixer.cpp

@@ -98,7 +98,8 @@ AudioMixer::AudioMixer(ReceivedMessage& message) :
             PacketType::RequestsDomainListData,
             PacketType::PerAvatarGainSet,
             PacketType::InjectorGainSet,
-            PacketType::AudioSoloRequest },
+            PacketType::AudioSoloRequest,
+            PacketType::StopInjector },
             this, "queueAudioPacket");
 
     // packets whose consequences are global should be processed on the main thread
@@ -246,7 +247,8 @@ void AudioMixer::removeHRTFsForFinishedInjector(const QUuid& streamID) {
 
     if (injectorClientData) {
         // stage the removal of this stream, workers handle when preparing mixes for listeners
-        _workerSharedData.removedStreams.emplace_back(injectorClientData->getNodeID(), injectorClientData->getNodeLocalID(),
+        _workerSharedData.removedStreams.emplace_back(injectorClientData->getNodeID(),
+                                                      injectorClientData->getNodeLocalID(),
                                                       streamID);
     }
 }

assignment-client/src/audio/AudioMixerClientData.cpp

@@ -104,6 +104,9 @@ int AudioMixerClientData::processPackets(ConcurrentAddedStreams& addedStreams) {
             case PacketType::AudioSoloRequest:
                 parseSoloRequest(packet, node);
                 break;
+            case PacketType::StopInjector:
+                parseStopInjectorPacket(packet);
+                break;
             default:
                 Q_UNREACHABLE();
         }
@@ -574,6 +577,19 @@ int AudioMixerClientData::checkBuffersBeforeFrameSend() {
     return (int)_audioStreams.size();
 }
 
+void AudioMixerClientData::parseStopInjectorPacket(QSharedPointer<ReceivedMessage> packet) {
+    auto streamID = QUuid::fromRfc4122(packet->readWithoutCopy(NUM_BYTES_RFC4122_UUID));
+
+    auto it = std::find_if(std::begin(_audioStreams), std::end(_audioStreams), [&](auto stream) {
+        return streamID == stream->getStreamIdentifier();
+    });
+
+    if (it != std::end(_audioStreams)) {
+        _audioStreams.erase(it);
+        emit injectorStreamFinished(streamID);
+    }
+}
+
 bool AudioMixerClientData::shouldSendStats(int frameNumber) {
     return frameNumber == _frameToSendStats;
 }

assignment-client/src/audio/AudioMixerClientData.h

@@ -67,12 +67,11 @@ public:
     void parseNodeIgnoreRequest(QSharedPointer<ReceivedMessage> message, const SharedNodePointer& node);
     void parseRadiusIgnoreRequest(QSharedPointer<ReceivedMessage> message, const SharedNodePointer& node);
     void parseSoloRequest(QSharedPointer<ReceivedMessage> message, const SharedNodePointer& node);
+    void parseStopInjectorPacket(QSharedPointer<ReceivedMessage> packet);
 
     // attempt to pop a frame from each audio stream, and return the number of streams from this client
     int checkBuffersBeforeFrameSend();
 
-    void removeDeadInjectedStreams();
-
     QJsonObject getAudioStreamStats();
 
     void sendAudioStreamStatsPackets(const SharedNodePointer& destinationNode);
@@ -163,7 +162,7 @@ public:
     // end of methods called non-concurrently from single AudioMixerSlave
 
 signals:
-    void injectorStreamFinished(const QUuid& streamIdentifier);
+    void injectorStreamFinished(const QUuid& streamID);
 
 public slots:
     void handleMismatchAudioFormat(SharedNodePointer node, const QString& currentCodec, const QString& recievedCodec);

libraries/audio/src/AudioInjector.cpp

@@ -103,6 +103,8 @@ void AudioInjector::finishLocalInjection() {
 
 void AudioInjector::finish() {
     withWriteLock([&] {
+        _state |= AudioInjectorState::LocalInjectionFinished;
+        _state |= AudioInjectorState::NetworkInjectionFinished;
         _state |= AudioInjectorState::Finished;
     });
     emit finished();
@@ -252,7 +254,7 @@ int64_t AudioInjector::injectNextFrame() {
             writeStringToStream(noCodecForInjectors, audioPacketStream);
 
             // pack stream identifier (a generated UUID)
-            audioPacketStream << QUuid::createUuid();
+            audioPacketStream << _streamID;
 
             // pack the stereo/mono type of the stream
             audioPacketStream << options.stereo;
@@ -402,4 +404,17 @@ int64_t AudioInjector::injectNextFrame() {
     int64_t playNextFrameAt = ++_nextFrame * AudioConstants::NETWORK_FRAME_USECS;
 
     return std::max(INT64_C(0), playNextFrameAt - currentTime);
-}
+}
+
+
+void AudioInjector::sendStopInjectorPacket() {
+    auto nodeList = DependencyManager::get<NodeList>();
+    if (auto audioMixer = nodeList->soloNodeOfType(NodeType::AudioMixer)) {
+        // Build packet
+        auto stopInjectorPacket = NLPacket::create(PacketType::StopInjector);
+        stopInjectorPacket->write(_streamID.toRfc4122());
+
+        // Send packet
+        nodeList->sendUnreliablePacket(*stopInjectorPacket, *audioMixer);
+    }
+}

libraries/audio/src/AudioInjector.h

@@ -100,6 +100,7 @@ private:
     int64_t injectNextFrame();
     bool inject(bool(AudioInjectorManager::*injection)(const AudioInjectorPointer&));
     bool injectLocally();
+    void sendStopInjectorPacket();
 
     static AbstractAudioInterface* _localAudioInterface;
 
@@ -120,6 +121,9 @@ private:
     // when the injector is local, we need this
     AudioHRTF _localHRTF;
     AudioFOA _localFOA;
+
+    QUuid _streamID { QUuid::createUuid() };
+
     friend class AudioInjectorManager;
 };
 

libraries/audio/src/AudioInjectorManager.cpp

@@ -105,6 +105,8 @@ void AudioInjectorManager::run() {
                         if (nextCallDelta >= 0 && !injector->isFinished()) {
                             // enqueue the injector with the correct timing in our holding queue
                             heldInjectors.emplace(heldInjectors.end(), usecTimestampNow() + nextCallDelta, injector);
+                        } else {
+                            injector->sendStopInjectorPacket();
                         }
                     }
 
@@ -354,4 +356,4 @@ void AudioInjectorManager::stop(const AudioInjectorPointer& injector) {
 size_t AudioInjectorManager::getNumInjectors() {
     Lock lock(_injectorsMutex);
     return _injectors.size();
-}
+}

libraries/networking/src/udt/PacketHeaders.cpp

@@ -86,7 +86,8 @@ PacketVersion versionForPacketType(PacketType packetType) {
         case PacketType::MicrophoneAudioNoEcho:
         case PacketType::MicrophoneAudioWithEcho:
         case PacketType::AudioStreamStats:
-            return static_cast<PacketVersion>(AudioVersion::HighDynamicRangeVolume);
+        case PacketType::StopInjector:
+            return static_cast<PacketVersion>(AudioVersion::StopInjectors);
         case PacketType::DomainSettings:
             return 18;  // replace min_avatar_scale and max_avatar_scale with min_avatar_height and max_avatar_height
         case PacketType::Ping:

libraries/networking/src/udt/PacketHeaders.h

@@ -134,6 +134,7 @@ public:
         BulkAvatarTraits,
         AudioSoloRequest,
         BulkAvatarTraitsAck,
+        StopInjector,
         NUM_PACKET_TYPE
     };
 
@@ -369,6 +370,7 @@ enum class AudioVersion : PacketVersion {
     SpaceBubbleChanges,
     HasPersonalMute,
     HighDynamicRangeVolume,
+    StopInjectors
 };
 
 enum class MessageDataVersion : PacketVersion {