Merge branch 'master' of file:///Users/birarda/code/worklist/udpEcho

mixer/.gitignore

@@ -0,0 +1,4 @@
+socket
+sftp-config.json
+.DS_Store
+*.raw

mixer/socket.cpp

@@ -0,0 +1,292 @@
+#include <iostream>
+#include <math.h>
+#include <string.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <fcntl.h>
+#include <sys/time.h>
+#include <pthread.h>
+#include <errno.h>
+#include <fstream>
+#include <limits>
+
+const int MAX_AGENTS = 1000;
+const int LOGOFF_CHECK_INTERVAL = 1000;
+
+const int UDP_PORT = 55443; 
+
+const int BUFFER_LENGTH_BYTES = 1024;
+const int BUFFER_LENGTH_SAMPLES = BUFFER_LENGTH_BYTES / sizeof(int16_t);
+const float SAMPLE_RATE = 22050.0;
+const float BUFFER_SEND_INTERVAL_USECS = (BUFFER_LENGTH_SAMPLES/SAMPLE_RATE) * 1000000;
+
+const int MAX_SAMPLE_VALUE = std::numeric_limits<int16_t>::max();
+const int MIN_SAMPLE_VALUE = std::numeric_limits<int16_t>::min();
+
+const int MAX_SOURCE_BUFFERS = 10;
+
+int16_t* whiteNoiseBuffer;
+int whiteNoiseLength;
+
+#define ECHO_DEBUG_MODE 0
+
+sockaddr_in address, dest_address;
+socklen_t destLength = sizeof(dest_address);
+
+struct AgentList {
+    sockaddr_in agent_addr;
+    bool active;
+    timeval time;
+} agents[MAX_AGENTS];
+
+int num_agents = 0;
+
+struct SourceBuffer {
+    int16_t sourceAudioData[BUFFER_LENGTH_SAMPLES];
+    bool transmitted;
+} sourceBuffers[MAX_SOURCE_BUFFERS];
+
+double diffclock(timeval *clock1, timeval *clock2)
+{
+    double diffms = (clock2->tv_sec - clock1->tv_sec) * 1000.0;
+    diffms += (clock2->tv_usec - clock1->tv_usec) / 1000.0;   // us to ms
+    return diffms;
+}
+
+double usecTimestamp(timeval *time, double addedUsecs = 0) {
+    return (time->tv_sec * 1000000.0) + time->tv_usec + addedUsecs;
+}
+
+int create_socket()
+{
+    //  Create socket 
+    int handle = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+    
+    if (handle <= 0) {
+        printf("Failed to create socket: %d\n", handle);
+        return false;
+    }
+
+    return handle;
+}
+
+int network_init()
+{
+    int handle = create_socket();
+    
+    //  Bind socket to port 
+    address.sin_family = AF_INET;
+    address.sin_addr.s_addr = INADDR_ANY;
+    address.sin_port = htons( (unsigned short) UDP_PORT );
+    
+    if (bind(handle, (const sockaddr*) &address, sizeof(sockaddr_in)) < 0) {
+        printf( "failed to bind socket\n" );
+        return false;
+    }   
+        
+    return handle;
+}
+
+int addAgent(sockaddr_in dest_address, void *audioData) {
+    //  Search for agent in list and add if needed 
+    int is_new = 0; 
+    int i = 0;
+
+    for (i = 0; i < num_agents; i++) {
+        if (dest_address.sin_addr.s_addr == agents[i].agent_addr.sin_addr.s_addr
+            && dest_address.sin_port == agents[i].agent_addr.sin_port) {
+            break;
+        }        
+    }
+    
+    if ((i == num_agents) || (agents[i].active == false)) {
+        is_new = 1;
+    }
+
+    agents[i].agent_addr = dest_address; 
+    agents[i].active = true;
+    gettimeofday(&agents[i].time, NULL);
+
+    memcpy(sourceBuffers[i].sourceAudioData, audioData, BUFFER_LENGTH_BYTES);
+    sourceBuffers[i].transmitted = false;
+    
+    if (i == num_agents) {
+        num_agents++;
+    }
+
+    return is_new;
+}
+
+struct send_buffer_struct {
+    int socket_handle;
+};
+
+void *send_buffer_thread(void *args)
+{
+    struct send_buffer_struct *buffer_args = (struct send_buffer_struct *) args;
+    int handle = buffer_args->socket_handle;
+
+    int sentBytes;
+    int currentFrame = 1;
+    timeval startTime, sendTime, now;
+
+    int16_t *clientMix = new int16_t[BUFFER_LENGTH_SAMPLES];
+    int16_t *masterMix = new int16_t[BUFFER_LENGTH_SAMPLES];
+
+    gettimeofday(&startTime, NULL);
+
+    while (true) {
+        sentBytes = 0;
+
+        int sampleOffset = ((currentFrame - 1) * BUFFER_LENGTH_SAMPLES) % whiteNoiseLength;
+        memcpy(masterMix, whiteNoiseBuffer + sampleOffset, BUFFER_LENGTH_BYTES);
+
+        gettimeofday(&sendTime, NULL);
+
+        for (int a = 0; a < num_agents; a++) {
+            if (diffclock(&agents[a].time, &sendTime) <= LOGOFF_CHECK_INTERVAL) {
+                memcpy(clientMix, masterMix, BUFFER_LENGTH_BYTES);
+
+                for (int b = 0; b < MAX_SOURCE_BUFFERS; b++) {
+                    if (b != a && !sourceBuffers[b].transmitted) {
+                        for (int s =  0; s < BUFFER_LENGTH_SAMPLES; s++) {
+                            // we have source buffer data for this sample
+                            int mixSample = clientMix[s] + sourceBuffers[b].sourceAudioData[s];
+                            
+                            int sampleToAdd = std::max(mixSample, MIN_SAMPLE_VALUE);
+                            sampleToAdd = std::min(sampleToAdd, MAX_SAMPLE_VALUE);
+
+                            clientMix[s] = sampleToAdd;
+                        }
+
+                        sourceBuffers[b].transmitted = true;
+                    }
+                }
+
+                sockaddr_in dest_address = agents[a].agent_addr;
+                
+                sentBytes = sendto(handle, clientMix, BUFFER_LENGTH_BYTES,
+                                0, (sockaddr *) &dest_address, sizeof(dest_address));
+            
+                if (sentBytes < BUFFER_LENGTH_BYTES) {
+                    std::cout << "Error sending mix packet! " << sentBytes << strerror(errno) << "\n";
+                }
+            }
+        }   
+
+        gettimeofday(&now, NULL);
+        
+        double usecToSleep = usecTimestamp(&startTime, (currentFrame * BUFFER_SEND_INTERVAL_USECS)) - usecTimestamp(&now);
+
+        if (usecToSleep > 0) {
+            usleep(usecToSleep);
+        } else {
+            std::cout << "NOT SLEEPING!";
+        }
+
+        currentFrame++;
+    }  
+
+    pthread_exit(0);  
+}
+
+struct process_arg_struct {
+    int16_t *packet_data;
+    sockaddr_in dest_address;
+};
+
+void *process_client_packet(void *args)
+{
+    struct process_arg_struct *process_args = (struct process_arg_struct *) args;
+    
+    sockaddr_in dest_address = process_args->dest_address;
+
+    if (addAgent(dest_address, process_args->packet_data)) {
+        std::cout << "Added agent: " << 
+            inet_ntoa(dest_address.sin_addr) << " on " <<
+            dest_address.sin_port << "\n";
+    }    
+
+    pthread_exit(0);
+}
+
+bool different_clients(sockaddr_in addr1, sockaddr_in addr2) 
+{
+    return addr1.sin_addr.s_addr != addr2.sin_addr.s_addr ||
+            (addr1.sin_addr.s_addr == addr2.sin_addr.s_addr &&
+            addr1.sin_port != addr2.sin_port);
+}
+
+void white_noise_buffer_init() {
+    // open a pointer to the audio file
+    FILE *whiteNoiseFile = fopen("opera.raw", "r");
+    
+    // get length of file
+    std::fseek(whiteNoiseFile, 0, SEEK_END);
+    whiteNoiseLength = std::ftell(whiteNoiseFile) / sizeof(int16_t);
+    std::rewind(whiteNoiseFile);
+    
+    // read that amount of samples from the file
+    whiteNoiseBuffer = new int16_t[whiteNoiseLength];
+    std::fread(whiteNoiseBuffer, sizeof(int16_t), whiteNoiseLength, whiteNoiseFile);
+    
+    // close it
+    std::fclose(whiteNoiseFile);
+}
+
+int main(int argc, const char * argv[])
+{
+    timeval now, last_agent_update;
+    int received_bytes = 0;
+
+    // read in the workclub white noise file as a base layer of audio
+    white_noise_buffer_init();
+    
+    int handle = network_init();
+    
+    if (!handle) {
+        std::cout << "Failed to create listening socket.\n";
+        return 0;
+    } else {
+        std::cout << "Network Started.  Waiting for packets.\n";
+    }
+
+    gettimeofday(&last_agent_update, NULL);
+
+    int16_t packet_data[BUFFER_LENGTH_SAMPLES];
+
+    for (int b = 0; b < MAX_SOURCE_BUFFERS; b++) {
+        sourceBuffers[b].transmitted = true;
+    }
+
+    struct send_buffer_struct send_buffer_args;
+    send_buffer_args.socket_handle = handle;
+
+    pthread_t buffer_send_thread;
+    pthread_create(&buffer_send_thread, NULL, send_buffer_thread, (void *)&send_buffer_args);
+
+    while (true) {
+        received_bytes = recvfrom(handle, (int16_t*)packet_data, BUFFER_LENGTH_BYTES,
+                                      0, (sockaddr*)&dest_address, &destLength);  
+        if (ECHO_DEBUG_MODE) {
+            sendto(handle, packet_data, BUFFER_LENGTH_BYTES,
+                        0, (sockaddr *) &dest_address, sizeof(dest_address));
+        } else {
+            struct process_arg_struct args;
+            args.packet_data = packet_data;
+            args.dest_address = dest_address;
+
+            pthread_t client_process_thread;
+            pthread_create(&client_process_thread, NULL, process_client_packet, (void *)&args);
+            pthread_join(client_process_thread, NULL); 
+        }
+    }
+
+    pthread_join(buffer_send_thread, NULL);
+    
+    return 0;
+}
+

mixer/socket.rb

@@ -0,0 +1,66 @@
+require 'socket'
+BUFFER_SLEEP_DURATION_MSECS = 20
+CLIENT_LIST_MEMORY_DURATION_SECS = 1
+
+begin 
+  # create the UDPSocket object
+  socket = UDPSocket.new
+
+  $dataBuffer = nil
+  $receivers = Hash.new
+
+  listenThread = Thread.new(socket) {
+    # bind it to 0.0.0.0 at the given port
+    port = 55443
+    socket.bind "0.0.0.0", port
+    puts "UDP Socket bound to port #{port} and listening."
+
+    # while true loop to keep listening for new packets
+    while true do
+      data, sender = socket.recvfrom 1024
+      puts "Recieved #{data.size} bytes from #{sender[3]} on #{sender[1]}"
+
+      # there are 2 bytes per sample
+      newSamples = data.unpack("s<*")
+      
+      if $dataBuffer.nil?
+        $dataBuffer = newSamples
+      else
+        $dataBuffer.each_with_index do |sample, index|
+          sample = (sample + newSamples[index]) / 2
+        end
+      end
+
+      $receivers["#{sender[3]}:#{sender[1]}"] = Time.now.to_f
+    end  
+  }
+
+  sendThread = Thread.new(socket) {
+    while true do
+      sleep(BUFFER_SLEEP_DURATION_MSECS/1000)
+      
+      unless $dataBuffer.nil?
+        $receivers.each do |ip_port, time|
+          if (time > Time.now.to_f - CLIENT_LIST_MEMORY_DURATION_SECS)
+            ip, port = ip_port.split(':')
+            socket.send $dataBuffer.pack("s<*"), 0, ip, port
+            puts "Sent mixed frame to #{ip} on #{port}"
+          else
+            puts "Nobody to send mixed frame to"
+            $receivers.delete(ip_port)
+          end
+        end
+
+        $dataBuffer = nil
+      end      
+    end
+  }
+
+  listenThread.join
+  sendThread.join
+
+  rescue SystemExit, Interrupt => e
+    puts
+    puts "Okay, fine. I'll stop listening."
+    socket.close
+end