multistream_example.cpp

/*
 * @file multistream_example.cpp
 * @copyright This file is Confidential and Proprietary to Modulate, Inc.
 * This file may not be shared or distributed without permission from Modulate.
 * Email support@modulate.ai with requests, bug reports, or questions!
 * Last updated July 6, 2022.
 */
 
#include "tox_server.h"
 
#include <iostream>
#include <fstream>
#include <chrono>
#include <thread>
#include <vector>
#include <atomic>
#include <cstring>
#include <ogg/ogg.h>
 
// Test program to read test_clip.opus and upload the internal Opus packets
// to ToxMod's servers
 
// Game-specific data inputs - help keep track of what game or application
// is involved, who is speaking, and who they are speaking to, etc.
static const char* account_uuid = TOXMOD_ACCOUNT_ID;
static const char* api_key = TOXMOD_API_KEY;
static const char* single_tenant_prefix = "dev";
 
static const unsigned int num_players = 150;
static const unsigned int buffer_copy_sleep_ms = 250;
static const unsigned int server_begin_upload_ms = 3000;
 
// Input Opus file specific parameters
const char* input_filename = "test_clip.opus";
static const unsigned int packet_length_ms = 20;
static const unsigned int maximum_packets_in_circular_buffer = 1000 / packet_length_ms;
// Note: the included test clip has 80 bytes per packet, but other Opus files
// may have different packet sizes.  The maximum Opus packet size if no properties
// of the input file are known is 1275.
// https://datatracker.ietf.org/doc/html/rfc6716.html#section-3 3.2.1
static const unsigned int maximum_opus_packet_size = 80; // bytes
 
// Vector for holding one tox server instance per player
std::vector<tox_server_instance_t> tox_server_instances;
 
// A simple error handler for debugging
void error_handler(TOX_SERVER_ERROR err) {
  if(err)
    std::cout << "Tox server error " << tox_server_error_name(err) << std::endl;
}
 
// From
// https://stackoverflow.com/questions/440133/how-do-i-create-a-random-alpha-numeric-string-in-c/24586587
std::string random_string(int len) {
  std::string tmp_s;
  static const char alphanum[] = "0123456789"
                                 "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
                                 "abcdefghijklmnopqrstuvwxyz";
  tmp_s.reserve(len);
  for(int i = 0; i < len; ++i)
    tmp_s += alphanum[rand() % (sizeof(alphanum) - 1)];
  return tmp_s;
}
 
// Helper functions to read Opus packets from the test_clip.opus Ogg Opus file
 
void get_page(ogg_sync_state* sync_state, ogg_page* page, std::ifstream& clip_file) {
  unsigned int chunk_size = 1024;
  while((ogg_sync_pageout(sync_state, page) != 1) && !clip_file.eof()) {
    char* buffer = ogg_sync_buffer(sync_state, chunk_size);
    clip_file.read(buffer, chunk_size);
    const long bytes_read = (long)clip_file.gcount();
    ogg_sync_wrote(sync_state, bytes_read);
  }
}
 
bool check_is_opus_header(const unsigned char* packet, unsigned int packet_size) {
  // https://datatracker.ietf.org/doc/html/rfc7845.html#section-5
  if(packet_size < 8)
    return false;
  return (packet[0] == 0x4f) && (packet[1] == 0x70);
}
 
// Given an instance (which corresponds to a player session pair), finds the uuid of a random other player in the same session
void get_random_other_player_in_session(tox_server_instance_t inst,
                                        const char** found_player_uuid) {
  const char* source_session_uuid;
  const char* source_player_uuid;
  const char* found_session_uuid;
  tox_server_instance_t found_player_inst;
  TOX_SERVER_ERROR e = tox_server_get_internal_session_uuid(&inst, &source_session_uuid);
  error_handler(e);
  e = tox_server_get_internal_player_uuid(&inst, &source_player_uuid);
  error_handler(e);
  int idx = rand() % tox_server_instances.size();
  do {
    idx = (idx + 1) % tox_server_instances.size();
    found_player_inst = tox_server_instances[idx];
    e = tox_server_get_internal_session_uuid(&found_player_inst, &found_session_uuid);
    error_handler(e);
    e = tox_server_get_internal_player_uuid(&found_player_inst, found_player_uuid);
  } while(strcmp(source_session_uuid, found_session_uuid) &&
          !(strcmp(source_player_uuid, *found_player_uuid)));
  error_handler(e);
}
 
// Required ToxMod non-realtime loops for buffer copy, begin uploads, and run uploads
 
std::atomic<bool> should_run_buffer_copies = {true};
void run_buffer_copies() {
  while(should_run_buffer_copies.load()) {
    for(auto&& it : tox_server_instances) {
      TOX_SERVER_ERROR err = tox_server_run_buffer_copy(&it);
      error_handler(err);
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(buffer_copy_sleep_ms));
  }
}
 
std::atomic<bool> should_run_begin_uploads = {true};
void run_begin_uploads() {
  while(should_run_begin_uploads.load()) {
    for(auto&& it : tox_server_instances) {
      TOX_SERVER_ERROR err = tox_server_begin_upload(&it);
      error_handler(err);
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(server_begin_upload_ms));
  }
}
 
std::atomic<bool> should_run_run_uploads = {true};
void run_run_uploads() {
  while(should_run_run_uploads.load()) {
    TOX_SERVER_ERROR err = tox_server_run_all_uploads_with_timeout(nullptr, 100);
    error_handler(err);
  }
}
 
// A simple malloc wrapper to demonstrate passing custom memory allocators
void* tox_alt_malloc(size_t sz) { return malloc(sz); }
 
// A wrapper around free to demonstrate passing custom memory allocators
void tox_alt_free(void* ptr) { free(ptr); }
 
// A wrapper around realloc to demonstrate passing custom memory allocators
void* tox_alt_realloc(void* ptr, size_t sz) { return realloc(ptr, sz); }
 
/*
 * By default, ToxMod sends info logging to stdout and errors to stderr. If different logging is
 * required, custom logging callbacks such as these ones can be set.
 */
void custom_logging_callback(const char* message) {
  std::cout << "[libtox] " << message << std::endl;
}
void custom_error_callback(const char* message) {
  std::cerr << "[libtox] " << message << std::endl;
}
 
int main() {
  srand((unsigned int)time(NULL));
 
  // Check that account_uuid and api_key are set
  if(!strcmp(account_uuid, "your account uuid here")) {
    std::cerr << ("Account UUID has not been set.  Please set your account_uuid, api_key, "
                  "and (if applicable) single_tenant_prefix in this example program's "
                  "source code file before using it.")
              << std::endl;
    return 1;
  }
  if(!strcmp(api_key, "your api key here")) {
    std::cerr << ("API Key has not been set.  Please set your account_uuid, api_key, "
                  "and (if applicable) single_tenant_prefix in this example program's "
                  "source code file before using it.")
              << std::endl;
    return 1;
  }
 
  std::cout << "Running " << num_players << " simultaneous player chat streams from "
            << input_filename << " through tox_server" << std::endl;
 
  // Run global intialization with the account-wide parameters
  TOX_SERVER_ERROR err;
  err = tox_server_global_init_mem(account_uuid, api_key, single_tenant_prefix, tox_alt_malloc,
                                   tox_alt_free, tox_alt_realloc, nullptr, nullptr);
  error_handler(err);
 
  tox_server_set_log_info_callback(custom_logging_callback);
  tox_server_set_log_error_callback(custom_error_callback);
 
  std::string session_name = random_string(20);
  for(size_t i = 0; i < num_players; i++) {
    const std::string player_name = random_string(20);
    // 10% probability to change to a new session, gets a distribution
    // over session player counts, only one session per player
    if(rand() % 10 < 1)
      session_name = random_string(20);
    tox_server_instance_t tox_server_instance;
    tox_server_config_t tox_server_config;
    tox_server_config.player_name_unique = player_name.c_str();
    tox_server_config.session_name_unique = session_name.c_str();
    tox_server_config.circular_buffer_max_num_packets = maximum_packets_in_circular_buffer;
    tox_server_config.max_packet_size = maximum_opus_packet_size;
    tox_server_config.enable_proximity_chat = true;
 
    tox_server_create_instance(&tox_server_instance, &tox_server_config);
    tox_server_instances.push_back(tox_server_instance);
    error_handler(err);
  }
 
  std::thread buffer_copy_thread(run_buffer_copies);
  std::thread begin_uploads_thread(run_begin_uploads);
  std::thread run_uploads_thread(run_run_uploads);
 
  // Block for reading the input Opus file and runnin it through tox_server
  {
    // Prepare separate Ogg state for reading in the test clip
    // This is likely unneeded on a voice chat server with direct
    // access to Opus packets
    ogg_sync_state sync_state;
    ogg_stream_state stream_state;
    ogg_page page;
    ogg_packet packet;
    ogg_sync_init(&sync_state);
 
    // Open the test clip file
    std::ifstream clip_file(input_filename, std::ios::binary);
    if(!clip_file.good()) {
      std::cerr << "Failed to open clip file " << input_filename << std::endl;
      return 1;
    }
 
    // Loop through all of the Ogg packets in the test clip file, adding
    // them to tox_server, in approximately realtime.
    size_t counter = 0;
    size_t clip_time_ms = 0;
    while(!clip_file.eof()) {
      // Read in Ogg packets from the file
      get_page(&sync_state, &page, clip_file);
      if(ogg_page_bos(&page)) {
        const int serial_number = ogg_page_serialno(&page);
        ogg_stream_init(&stream_state, serial_number);
      }
      ogg_stream_pagein(&stream_state, &page);
      while(ogg_stream_packetout(&stream_state, &packet) != 0) {
        if(!check_is_opus_header(packet.packet, packet.bytes)) {
          // Realtime-safe, add the Opus packets to libtox_server
          for(auto&& it : tox_server_instances) {
            err = tox_server_add_packet(&it, packet.packet, packet.bytes);
            error_handler(err);
          }
          std::this_thread::sleep_for(std::chrono::milliseconds(packet_length_ms));
          counter++;
          // once per second print progress
          if(clip_time_ms % 1000 < packet_length_ms)
            std::cout << "." << std::endl; // Show progress once per second
 
          // Every 2 seconds, randomly change who a player can hear, simulating moving players with proximity chat
          if(clip_time_ms % 2000 < packet_length_ms) {
            std::cout << "Making change to can hear list" << std::endl;
            tox_server_instance_t source_player_inst =
              tox_server_instances[rand() % tox_server_instances.size()];
            std::vector<const char*> audible_players;
            int num_audible_players = rand() % 10;
            for(int i = 0; i < num_audible_players; ++i) {
              const char* player_uuid;
              get_random_other_player_in_session(source_player_inst, &player_uuid);
              audible_players.push_back(player_uuid);
            }
            // Could also use tox_server_set_proximity_chat_can_hear_player() instead, but not both
            err = tox_server_set_proximity_chat_player_can_hear(
              &source_player_inst, audible_players.data(), num_audible_players);
            error_handler(err);
          }
 
          // Every 5 seconds, randomly choose a player to mute or unmute another player
          if(clip_time_ms % 5000 < packet_length_ms) {
            tox_server_instance_t source_player_inst =
              tox_server_instances[rand() % tox_server_instances.size()];
            const char* player_uuid;
            get_random_other_player_in_session(source_player_inst, &player_uuid);
            std::cout << "Muting player: " << std::string(player_uuid) << std::endl;
            err = tox_server_set_player_muted(&source_player_inst, player_uuid, rand() % 2);
            error_handler(err);
          }
          clip_time_ms += packet_length_ms;
        }
      }
    }
 
    std::cout << "Total Opus Packets Count: " << counter << std::endl;
 
    // Cleanup the file's ogg stream
    ogg_sync_clear(&sync_state);
    ogg_stream_clear(&stream_state);
  }
 
  std::cout << "Finished Running through audio, stopping threads" << std::endl;
  should_run_buffer_copies.store(false);
  should_run_begin_uploads.store(false);
  should_run_run_uploads.store(false);
  buffer_copy_thread.join();
  begin_uploads_thread.join();
  run_uploads_thread.join();
 
  // Destroy the per-player tox_server_instance_t, and cleanup global state
  for(auto&& it : tox_server_instances) {
    err = tox_server_destroy_instance(&it);
    error_handler(err);
  }
 
  // When destroying an instance, a message is issued to the server letting it know that the
  // player has left the session. Make sure to finish uploading all such messages.
  int num_current_uploads;
  do {
    err = tox_server_run_all_uploads_with_timeout(&num_current_uploads, 100);
    error_handler(err);
  } while(num_current_uploads > 0);
 
  err = tox_server_global_cleanup();
  error_handler(err);
  return 0;
}