add threaded injection of audio via audio injector class

audio-mixer/src/main.cpp

@@ -94,7 +94,7 @@ int main(int argc, const char* argv[]) {
     sockaddr* agentAddress = new sockaddr;
 
     // make sure our agent socket is non-blocking
-    agentList->getAgentSocket().setBlocking(false);
+    agentList->getAgentSocket()->setBlocking(false);
     
     int nextFrame = 0;
     timeval startTime;
@@ -256,7 +256,7 @@ int main(int argc, const char* argv[]) {
             }
             
             memcpy(clientPacket + 1, clientSamples, sizeof(clientSamples));
-            agentList->getAgentSocket().send(agent->getPublicSocket(), clientPacket, BUFFER_LENGTH_BYTES + 1);
+            agentList->getAgentSocket()->send(agent->getPublicSocket(), clientPacket, BUFFER_LENGTH_BYTES + 1);
         }
         
         // push forward the next output pointers for any audio buffers we used
@@ -274,7 +274,7 @@ int main(int argc, const char* argv[]) {
         }
         
         // pull any new audio data from agents off of the network stack
-        while (agentList->getAgentSocket().receive(agentAddress, packetData, &receivedBytes)) {
+        while (agentList->getAgentSocket()->receive(agentAddress, packetData, &receivedBytes)) {
             if (packetData[0] == PACKET_HEADER_INJECT_AUDIO || packetData[0] == PACKET_HEADER_MICROPHONE_AUDIO) {
                 char agentType = (packetData[0] == PACKET_HEADER_MICROPHONE_AUDIO)
                     ? AGENT_TYPE_AVATAR

avatar-mixer/src/main.cpp

@@ -72,7 +72,7 @@ int main(int argc, const char* argv[]) {
     uint16_t agentID = 0;
         
     while (true) {
-        if (agentList->getAgentSocket().receive(agentAddress, packetData, &receivedBytes)) {
+        if (agentList->getAgentSocket()->receive(agentAddress, packetData, &receivedBytes)) {
             switch (packetData[0]) {
                 case PACKET_HEADER_HEAD_DATA:
                     // grab the agent ID from the packet
@@ -93,7 +93,7 @@ int main(int argc, const char* argv[]) {
                         }
                     }
                     
-                    agentList->getAgentSocket().send(agentAddress, broadcastPacket, currentBufferPosition - broadcastPacket);
+                    agentList->getAgentSocket()->send(agentAddress, broadcastPacket, currentBufferPosition - broadcastPacket);
                     
                     break;
                 case PACKET_HEADER_DOMAIN:

domain-server/src/main.cpp

@@ -97,7 +97,7 @@ int main(int argc, const char * argv[])
     uint16_t packetAgentID = 0;
     
     while (true) {
-        if (agentList->getAgentSocket().receive((sockaddr *)&agentPublicAddress, packetData, &receivedBytes) &&
+        if (agentList->getAgentSocket()->receive((sockaddr *)&agentPublicAddress, packetData, &receivedBytes) &&
             (packetData[0] == PACKET_HEADER_DOMAIN_RFD || packetData[0] == PACKET_HEADER_DOMAIN_LIST_REQUEST)) {
             std::map<char, Agent *> newestSoloAgents;
             
@@ -169,7 +169,7 @@ int main(int argc, const char * argv[])
             currentBufferPos += packAgentId(currentBufferPos, packetAgentID);
             
             // send the constructed list back to this agent
-            agentList->getAgentSocket().send((sockaddr*) &agentPublicAddress,
+            agentList->getAgentSocket()->send((sockaddr*) &agentPublicAddress,
                                              broadcastPacket,
                                              (currentBufferPos - startPointer) + 1);
         }

eve/src/main.cpp

@@ -34,7 +34,6 @@ const float EVE_PELVIS_HEIGHT = 0.565925f;
 const float AUDIO_INJECT_PROXIMITY = 0.4f;
 
 bool stopReceiveAgentDataThread;
-bool injectAudioThreadRunning = false;
 
 int TEMP_AUDIO_LISTEN_PORT = 55439;
 UDPSocket audioSocket(TEMP_AUDIO_LISTEN_PORT);
@@ -47,7 +46,7 @@ void *receiveAgentData(void *args) {
     AgentList* agentList = AgentList::getInstance();
     
     while (!::stopReceiveAgentDataThread) {
-        if (agentList->getAgentSocket().receive(&senderAddress, incomingPacket, &bytesReceived)) { 
+        if (agentList->getAgentSocket()->receive(&senderAddress, incomingPacket, &bytesReceived)) { 
             switch (incomingPacket[0]) {
                 case PACKET_HEADER_BULK_AVATAR_DATA:
                     // this is the positional data for other agents
@@ -67,29 +66,6 @@ void *receiveAgentData(void *args) {
     return NULL;
 }
 
-void *injectAudio(void *args) {
-    ::injectAudioThreadRunning = true;
-    
-    AudioInjector* eveAudioInjector = (AudioInjector *)args;
-    
-    // look for an audio mixer in our agent list
-    Agent* audioMixer = AgentList::getInstance()->soloAgentOfType(AGENT_TYPE_AUDIO_MIXER);
-    
-    if (audioMixer) {
-        // until the audio mixer is setup for ping-reply, activate the public socket if it's not active
-        if (!audioMixer->getActiveSocket()) {
-            audioMixer->activatePublicSocket();
-        }
-        
-        // we have an active audio mixer we can send data to
-        eveAudioInjector->injectAudio(&::audioSocket, audioMixer->getActiveSocket());
-    }
-    
-    ::injectAudioThreadRunning = false;
-    pthread_exit(0);
-    return NULL;
-}
-
 void createAvatarDataForAgent(Agent* agent) {
     if (!agent->getLinkedData()) {
         agent->setLinkedData(new AvatarData());
@@ -138,6 +114,14 @@ int main(int argc, const char* argv[]) {
     // lower Eve's volume by setting the attentuation modifier (this is a value out of 255)
     eveAudioInjector.setAttenuationModifier(190);
     
+    // pass the agentList UDPSocket pointer to the audio injector
+    eveAudioInjector.setInjectorSocket(agentList->getAgentSocket());
+    
+    // set the position of the audio injector
+    float injectorPosition[3];
+    memcpy(injectorPosition, &eve.getPosition(), sizeof(injectorPosition));
+    eveAudioInjector.setPosition(injectorPosition);
+    
     // register the callback for agent data creation
     agentList->linkedDataCreateCallback = createAvatarDataForAgent;
     
@@ -146,8 +130,6 @@ int main(int argc, const char* argv[]) {
     
     timeval thisSend;
     double numMicrosecondsSleep = 0;
-  
-    pthread_t injectAudioThread;
     
     int handStateTimer = 0;
 
@@ -167,10 +149,10 @@ int main(int argc, const char* argv[]) {
             packetPosition += eve.getBroadcastData(packetPosition);
             
             // use the UDPSocket instance attached to our agent list to send avatar data to mixer
-            agentList->getAgentSocket().send(avatarMixer->getActiveSocket(), broadcastPacket, packetPosition - broadcastPacket);
+            agentList->getAgentSocket()->send(avatarMixer->getActiveSocket(), broadcastPacket, packetPosition - broadcastPacket);
         }        
         
-        if (!::injectAudioThreadRunning) {
+        if (!eveAudioInjector.isInjectingAudio()) {
             // enumerate the other agents to decide if one is close enough that eve should talk
             for (AgentList::iterator agent = agentList->begin(); agent != agentList->end(); agent++) {
                 AvatarData* avatarData = (AvatarData*) agent->getLinkedData();
@@ -180,7 +162,20 @@ int main(int argc, const char* argv[]) {
                     float squareDistance = glm::dot(tempVector, tempVector);
                     
                     if (squareDistance <= AUDIO_INJECT_PROXIMITY) {
-                        pthread_create(&injectAudioThread, NULL, injectAudio, (void*) &eveAudioInjector);
+                        // look for an audio mixer in our agent list
+                        Agent* audioMixer = AgentList::getInstance()->soloAgentOfType(AGENT_TYPE_AUDIO_MIXER);
+                        
+                        if (audioMixer) {
+                            // until the audio mixer is setup for ping-reply, activate the public socket if it's not active
+                            if (!audioMixer->getActiveSocket()) {
+                                audioMixer->activatePublicSocket();
+                            }
+                            
+                            eveAudioInjector.setDestinationSocket(audioMixer->getActiveSocket());
+                            
+                            // we have an active audio mixer we can send data to
+                            eveAudioInjector.threadInjectionOfAudio();
+                        }
                     }
                 }            
             }

injector/src/main.cpp

@@ -104,6 +104,7 @@ int main(int argc, char* argv[]) {
     mixerSocket.sin_family = AF_INET;
     mixerSocket.sin_addr.s_addr = inet_addr(EC2_WEST_AUDIO_SERVER);
     mixerSocket.sin_port = htons((uint16_t)AUDIO_UDP_LISTEN_PORT);
+
     
     if (processParameters(argc, argv)) {        
         if (::sourceAudioFile == NULL) {
@@ -111,6 +112,8 @@ int main(int argc, char* argv[]) {
             exit(-1);
         } else {
             AudioInjector injector(sourceAudioFile);
+            injector.setInjectorSocket(&streamSocket);
+            injector.setDestinationSocket((sockaddr*) &mixerSocket);
             
             injector.setPosition(::floatArguments);
             injector.setBearing(*(::floatArguments + 3));
@@ -120,7 +123,7 @@ int main(int argc, char* argv[]) {
             int usecDelay = 0;
             
             while (true) {
-                injector.injectAudio(&streamSocket, (sockaddr*) &mixerSocket);
+                injector.injectAudio();
                 
                 if (!::loopAudio) {
                     delay = randFloatInRange(::sleepIntervalMin, ::sleepIntervalMax);

interface/src/Application.cpp

@@ -179,7 +179,7 @@ Application::Application(int& argc, char** argv) :
     AgentList::createInstance(AGENT_TYPE_AVATAR, listenPort);
     _enableNetworkThread = !cmdOptionExists(argc, constArgv, "--nonblocking");
     if (!_enableNetworkThread) {
-        AgentList::getInstance()->getAgentSocket().setBlocking(false);
+        AgentList::getInstance()->getAgentSocket()->setBlocking(false);
     }
     
     const char* domainIP = getCmdOption(argc, constArgv, "--domain");
@@ -2022,7 +2022,7 @@ void* Application::networkReceive(void* args) {
             app->_wantToKillLocalVoxels = false;
         }
     
-        if (AgentList::getInstance()->getAgentSocket().receive(&senderAddress, app->_incomingPacket, &bytesReceived)) {
+        if (AgentList::getInstance()->getAgentSocket()->receive(&senderAddress, app->_incomingPacket, &bytesReceived)) {
             app->_packetCount++;
             app->_bytesCount += bytesReceived;
             

interface/src/Audio.cpp

@@ -146,7 +146,7 @@ int audioCallback (const void* inputBuffer,
             // copy the audio data to the last BUFFER_LENGTH_BYTES bytes of the data packet
             memcpy(currentPacketPtr, inputLeft, BUFFER_LENGTH_BYTES);
             
-            agentList->getAgentSocket().send(audioMixer->getActiveSocket(), dataPacket, BUFFER_LENGTH_BYTES + leadingBytes);
+            agentList->getAgentSocket()->send(audioMixer->getActiveSocket(), dataPacket, BUFFER_LENGTH_BYTES + leadingBytes);
         }
         
     }

interface/src/PairingHandler.cpp

@@ -19,7 +19,7 @@ const int PAIRING_SERVER_PORT = 7247;
 
 void PairingHandler::sendPairRequest() {    
     // grab the agent socket from the AgentList singleton
-    UDPSocket *agentSocket = &AgentList::getInstance()->getAgentSocket();
+    UDPSocket *agentSocket = AgentList::getInstance()->getAgentSocket();
     
     // prepare the pairing request packet
     

libraries/shared/src/AgentList.cpp

@@ -78,10 +78,6 @@ AgentList::~AgentList() {
     pthread_mutex_destroy(&mutex);
 }
 
-UDPSocket& AgentList::getAgentSocket() {
-    return agentSocket;
-}
-
 unsigned int AgentList::getSocketListenPort() {
     return socketListenPort;
 }
@@ -337,8 +333,8 @@ void *pingUnknownAgents(void *args) {
                 && (agent->getPublicSocket() != NULL && agent->getLocalSocket() != NULL)) {
                 // ping both of the sockets for the agent so we can figure out
                 // which socket we can use
-                agentList->getAgentSocket().send(agent->getPublicSocket(), &PACKET_HEADER_PING, 1);
+                agentList->getAgentSocket()->send(agent->getPublicSocket(), &PACKET_HEADER_PING, 1);
-                agentList->getAgentSocket().send(agent->getLocalSocket(), &PACKET_HEADER_PING, 1);
+                agentList->getAgentSocket()->send(agent->getLocalSocket(), &PACKET_HEADER_PING, 1);
             }
         }
         
@@ -433,7 +429,7 @@ void *checkInWithDomainServer(void *args) {
         
         packSocket(packet + 2, localAddress, htons(parentAgentList->getSocketListenPort()));
         
-        parentAgentList->getAgentSocket().send(DOMAIN_IP, DOMAINSERVER_PORT, packet, sizeof(packet));
+        parentAgentList->getAgentSocket()->send(DOMAIN_IP, DOMAINSERVER_PORT, packet, sizeof(packet));
         
         packet[0] = PACKET_HEADER_DOMAIN_LIST_REQUEST;
         

libraries/shared/src/AgentList.h

@@ -49,7 +49,7 @@ public:
     
     int size() { return _numAgents; }
     
-    UDPSocket& getAgentSocket();
+    UDPSocket* getAgentSocket() { return &agentSocket; }
     
     void lock() { pthread_mutex_lock(&mutex); }
     void unlock() { pthread_mutex_unlock(&mutex); }

libraries/shared/src/AudioInjector.cpp

@@ -24,7 +24,8 @@ AudioInjector::AudioInjector(const char* filename) :
     _position(),
     _bearing(0),
     _attenuationModifier(255),
-    _indexOfNextSlot(0)
+    _indexOfNextSlot(0),
+    _isInjectingAudio(false)
 {
     std::fstream sourceFile;
     
@@ -50,7 +51,8 @@ AudioInjector::AudioInjector(int maxNumSamples) :
     _position(),
     _bearing(0),
     _attenuationModifier(255),
-    _indexOfNextSlot(0)
+    _indexOfNextSlot(0),
+    _isInjectingAudio(false)
 {
     _audioSampleArray = new int16_t[maxNumSamples];
     memset(_audioSampleArray, 0, _numTotalSamples * sizeof(int16_t));
@@ -81,8 +83,10 @@ void AudioInjector::addSamples(int16_t* sampleBuffer, int numSamples) {
     }
 }
 
-void AudioInjector::injectAudio(UDPSocket* injectorSocket, sockaddr* destinationSocket) const {
+void AudioInjector::injectAudio() {
     if (_audioSampleArray != NULL) {
+        _isInjectingAudio = true;
+        
         timeval startTime;
         
         // one byte for header, 3 positional floats, 1 bearing float, 1 attenuation modifier byte
@@ -115,12 +119,26 @@ void AudioInjector::injectAudio(UDPSocket* injectorSocket, sockaddr* destination
             
             memcpy(currentPacketPtr, _audioSampleArray + i, numSamplesToCopy * sizeof(int16_t));
             
-            injectorSocket->send(destinationSocket, dataPacket, sizeof(dataPacket));
+            _injectorSocket->send(&_destinationSocket, dataPacket, sizeof(dataPacket));
             
             double usecToSleep = BUFFER_SEND_INTERVAL_USECS - (usecTimestampNow() - usecTimestamp(&startTime));
             if (usecToSleep > 0) {
                 usleep(usecToSleep);
             }
         }
+        
+        _isInjectingAudio = false;
     }
 }
+
+void* injectAudioViaThread(void* args) {
+    AudioInjector* parentInjector = (AudioInjector*) args;
+    parentInjector->injectAudio();
+    
+    pthread_exit(0);
+}
+
+void AudioInjector::threadInjectionOfAudio() {
+    pthread_t audioInjectThread;
+    pthread_create(&audioInjectThread, NULL, injectAudioViaThread, (void*) this);
+}

libraries/shared/src/AudioInjector.h

@@ -20,14 +20,19 @@ public:
     AudioInjector(int maxNumSamples);
     ~AudioInjector();
     
+    bool isInjectingAudio() const { return _isInjectingAudio; }
+    
     void setPosition(float* position);
     void setBearing(float bearing) { _bearing = bearing; }
     void setAttenuationModifier(unsigned char attenuationModifier) { _attenuationModifier = attenuationModifier; }
+    void setInjectorSocket(UDPSocket* injectorSocket) { _injectorSocket = injectorSocket; }
+    void setDestinationSocket(sockaddr* destinationSocket) { _destinationSocket = *destinationSocket; }
     
     void addSample(const int16_t sample);
     void addSamples(int16_t* sampleBuffer, int numSamples);
     
-    void injectAudio(UDPSocket* injectorSocket, sockaddr* destinationSocket) const;
+    void injectAudio();
+    void threadInjectionOfAudio();
 private:
     int16_t* _audioSampleArray;
     int _numTotalSamples;
@@ -35,6 +40,9 @@ private:
     float _bearing;
     unsigned char _attenuationModifier;
     int _indexOfNextSlot;
+    UDPSocket* _injectorSocket;
+    sockaddr _destinationSocket;
+    bool _isInjectingAudio;
 };
 
 #endif /* defined(__hifi__AudioInjector__) */

libraries/shared/src/UDPSocket.cpp

@@ -204,13 +204,12 @@ void UDPSocket::setBlocking(bool blocking) {
 }
 
 //  Receive data on this socket with retrieving address of sender
-bool UDPSocket::receive(void *receivedData, ssize_t *receivedBytes) {
+bool UDPSocket::receive(void *receivedData, ssize_t *receivedBytes) const {
-    
     return receive((sockaddr *)&senderAddress, receivedData, receivedBytes);
 }
 
 //  Receive data on this socket with the address of the sender 
-bool UDPSocket::receive(sockaddr *recvAddress, void *receivedData, ssize_t *receivedBytes) {
+bool UDPSocket::receive(sockaddr *recvAddress, void *receivedData, ssize_t *receivedBytes) const {
     
 #ifdef _WIN32
     int addressSize = sizeof(*recvAddress);
@@ -223,7 +222,7 @@ bool UDPSocket::receive(sockaddr *recvAddress, void *receivedData, ssize_t *rece
     return (*receivedBytes > 0);
 }
 
-int UDPSocket::send(sockaddr *destAddress, const void *data, size_t byteLength) {
+int UDPSocket::send(sockaddr *destAddress, const void *data, size_t byteLength) const {
     // send data via UDP
     int sent_bytes = sendto(handle, (const char*)data, byteLength,
                             0, (sockaddr *) destAddress, sizeof(sockaddr_in));
@@ -236,7 +235,7 @@ int UDPSocket::send(sockaddr *destAddress, const void *data, size_t byteLength)
     return sent_bytes;
 }
 
-int UDPSocket::send(char * destAddress, int destPort, const void *data, size_t byteLength) {
+int UDPSocket::send(char * destAddress, int destPort, const void *data, size_t byteLength) const {
     
     // change address and port on reusable global to passed variables
     destSockaddr.sin_addr.s_addr = inet_addr(destAddress);

libraries/shared/src/UDPSocket.h

@@ -19,19 +19,19 @@
 #define MAX_BUFFER_LENGTH_BYTES 1500
 
 class UDPSocket {    
-    public:
+public:
-        UDPSocket(int listening_port);
+    UDPSocket(int listening_port);
-        ~UDPSocket();
+    ~UDPSocket();
-        bool init();
+    bool init();
-        void setBlocking(bool blocking);
+    void setBlocking(bool blocking);
-        bool isBlocking() { return blocking; }
+    bool isBlocking() { return blocking; }
-        int send(sockaddr *destAddress, const void *data, size_t byteLength);
+    int send(sockaddr *destAddress, const void *data, size_t byteLength) const;
-        int send(char *destAddress, int destPort, const void *data, size_t byteLength);
+    int send(char *destAddress, int destPort, const void *data, size_t byteLength) const;
-        bool receive(void *receivedData, ssize_t *receivedBytes);
+    bool receive(void *receivedData, ssize_t *receivedBytes) const;
-        bool receive(sockaddr *recvAddress, void *receivedData, ssize_t *receivedBytes);
+    bool receive(sockaddr *recvAddress, void *receivedData, ssize_t *receivedBytes) const;
-    private:
+private:
-        int handle;
+    int handle;
-        bool blocking;
+    bool blocking;
 };
 
 bool socketMatch(sockaddr *first, sockaddr *second);

voxel-server/src/main.cpp

@@ -168,7 +168,7 @@ void resInVoxelDistributor(AgentList* agentList,
                 if (agentData->getAvailable() >= bytesWritten) {
                     agentData->writeToPacket(&tempOutputBuffer[0], bytesWritten);
                 } else {
-                    agentList->getAgentSocket().send(agent->getActiveSocket(), 
+                    agentList->getAgentSocket()->send(agent->getActiveSocket(),
                                                      agentData->getPacket(), agentData->getPacketLength());
                     trueBytesSent += agentData->getPacketLength();
                     truePacketsSent++;
@@ -178,8 +178,8 @@ void resInVoxelDistributor(AgentList* agentList,
                 }
             } else {
                 if (agentData->isPacketWaiting()) {
-                    agentList->getAgentSocket().send(agent->getActiveSocket(), 
+                    agentList->getAgentSocket()->send(agent->getActiveSocket(),
-                                                     agentData->getPacket(), agentData->getPacketLength());
+                                                      agentData->getPacket(), agentData->getPacketLength());
                     trueBytesSent += agentData->getPacketLength();
                     truePacketsSent++;
                     agentData->resetVoxelPacket();
@@ -190,7 +190,7 @@ void resInVoxelDistributor(AgentList* agentList,
         }
         // send the environment packet
         int envPacketLength = environmentData.getBroadcastData(tempOutputBuffer);
-        agentList->getAgentSocket().send(agent->getActiveSocket(), tempOutputBuffer, envPacketLength);
+        agentList->getAgentSocket()->send(agent->getActiveSocket(), tempOutputBuffer, envPacketLength);
         trueBytesSent += envPacketLength;
         truePacketsSent++;
         
@@ -288,7 +288,7 @@ void deepestLevelVoxelDistributor(AgentList* agentList,
                 if (agentData->getAvailable() >= bytesWritten) {
                     agentData->writeToPacket(&tempOutputBuffer[0], bytesWritten);
                 } else {
-                    agentList->getAgentSocket().send(agent->getActiveSocket(), 
+                    agentList->getAgentSocket()->send(agent->getActiveSocket(),
                                                      agentData->getPacket(), agentData->getPacketLength());
                     trueBytesSent += agentData->getPacketLength();
                     truePacketsSent++;
@@ -298,7 +298,7 @@ void deepestLevelVoxelDistributor(AgentList* agentList,
                 }
             } else {
                 if (agentData->isPacketWaiting()) {
-                    agentList->getAgentSocket().send(agent->getActiveSocket(), 
+                    agentList->getAgentSocket()->send(agent->getActiveSocket(),
                                                      agentData->getPacket(), agentData->getPacketLength());
                     trueBytesSent += agentData->getPacketLength();
                     truePacketsSent++;
@@ -310,7 +310,7 @@ void deepestLevelVoxelDistributor(AgentList* agentList,
         }
         // send the environment packet
         int envPacketLength = environmentData.getBroadcastData(tempOutputBuffer);
-        agentList->getAgentSocket().send(agent->getActiveSocket(), tempOutputBuffer, envPacketLength);
+        agentList->getAgentSocket()->send(agent->getActiveSocket(), tempOutputBuffer, envPacketLength);
         trueBytesSent += envPacketLength;
         truePacketsSent++;
         
@@ -501,7 +501,7 @@ int main(int argc, const char * argv[])
         // check to see if we need to persist our voxel state
         persistVoxelsWhenDirty();    
     
-        if (agentList->getAgentSocket().receive(&agentPublicAddress, packetData, &receivedBytes)) {
+        if (agentList->getAgentSocket()->receive(&agentPublicAddress, packetData, &receivedBytes)) {
         	// XXXBHG: Hacked in support for 'S' SET command
             if (packetData[0] == PACKET_HEADER_SET_VOXEL || packetData[0] == PACKET_HEADER_SET_VOXEL_DESTRUCTIVE) {
                 bool destructive = (packetData[0] == PACKET_HEADER_SET_VOXEL_DESTRUCTIVE);