audio1.pony

use "collections"
use "time"
use "lib:portaudio_sink"
use "lib:portaudio"

type Buffer is (Array[F32], Array[U8])

class StopTimer is TimerNotify
  var _a: Main
  new iso create(a: Main) =>
    _a = a

  fun ref apply(timer: Timer, count: U64): Bool =>
    _a.done()
    false

  fun ref cancel(timer: Timer) =>
    false

actor Main
  """Sets up audio stream."""
  var _env: Env
  let _buffer_count: USize
  let _frame_count: USize
  var _preroll: USize
  var _buffers: Array[Buffer]
  var _index: USize
  var _timers: Timers
  var _timer: Timer tag
  var _phasor: F64
  let _phasor_inc: F64
  let _gain: F64

  new create(env: Env) =>
    _env = env
    _buffer_count = 64 
    _frame_count = 128 
    _preroll = 0 // Will be filled in by preroll().
    _buffers = recover Array[Buffer] end
    _index = 0
    _phasor = 0.0
    _phasor_inc = 880.0 / 44100.0
    _gain = 0.2
    _timers = Timers
    let t = Timer(StopTimer(this), 10_000_000_000, 0)
    _timer = t
    _timers(consume t)

    let open_result = @init_output_stream[I32](
      _frame_count,
      _buffer_count,
      addressof this.add_buffer,
      addressof this.preroll,
      addressof this.produce,
      this)
    _env.out.print("got open_result: " + open_result.string())

    // The stream will be started by produce() when the preroll phase is done.

  be done() =>
    _env.out.print("stop!")

  be add_buffer(buf: Pointer[F32] iso, ready: Pointer[U8] iso) =>
    let frame_count = _frame_count
    let buf_array: Array[F32] iso = recover
      Array[F32].from_cstring(consume buf, frame_count)
    end
    let ready_array: Array[U8] iso = recover
      Array[U8].from_cstring(consume ready, USize(1))
    end
    _buffers.push((consume buf_array, consume ready_array))

  be preroll() =>
    _preroll = _buffers.size()
    _env.out.print("preroll out! " + _preroll.string() + " buffers.")

  be produce(timestamp: F64) =>
    try
      let buf = _buffers(_index)
      // Fill the next buffer.
      for i in Range[USize](0, _frame_count) do
        buf._1.update(i, (_phasor * _gain).f32())
        _phasor = _phasor + _phasor_inc
        if _phasor > 1.0 then
          _phasor = _phasor - 1.0
        end
      end
      // Set the "ready" flag on the buffer.
      buf._2.update(0, U8(1))
      // Advance buffer pointer.
      _index = _index + 1
      if _index == _buffers.size() then
        _index = 0
      end
      // If we're prerolling, see if it's time to start the stream.
      if _preroll > 0 then
        _preroll = _preroll - 1
        if _preroll == 0 then
          //_env.out.print("start...")
          let start_result = @start_output_stream[I32]()
          //_env.out.print("got start_result: " + start_result.string())
        end
      end
    end

portaudio_sink.c

#include <assert.h>
#include <semaphore.h>
#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <portaudio.h>
#include <pthread.h>

typedef struct _buffer {
  uint8_t ready;
  float* buf;
} _buffer;

static PaStream* g_stream = NULL;
static unsigned long g_frames_per_buffer = 0;
static _buffer* g_buffers;
static size_t g_buffer_count = 0;
static size_t g_next_buffer = 0;

static pthread_t g_produce_thread;
static sem_t g_produce_sem;
static double g_produce_timestamp;
static uint32_t g_produce_underruns;

// the ponyland callbacks
typedef void (*pony_output_add_buffer_cb)(void* pony_object, float* io_buffer, uint8_t* io_ready);
typedef void (*pony_output_preroll_cb)(void* pony_object);
typedef void (*pony_output_produce_cb)(void* pony_object, double timestamp);

static pony_output_add_buffer_cb g_add_buffer_cb = NULL;
static pony_output_preroll_cb g_preroll_cb = NULL;
static pony_output_produce_cb g_produce_cb = NULL;
static void* g_pony_object;

void init_buffers(unsigned long buffer_count) {
  int i;
  g_buffers = (_buffer*)malloc(sizeof(_buffer) * buffer_count);
  for (i = 0; i < buffer_count; ++i) {
    g_buffers[i].ready = 0;
    g_buffers[i].buf = (float*) malloc(sizeof(float) * g_frames_per_buffer);
  }
  g_buffer_count = buffer_count;
}

void preroll() {
  printf("preroll: add buffer %p preroll %p produce %p\n",
      g_add_buffer_cb,
      g_preroll_cb,
      g_produce_cb);
  // Send buffers.
  int i;
  for (i = 0; i < g_buffer_count; ++i) {
    (*g_add_buffer_cb)(g_pony_object, g_buffers[i].buf, &g_buffers[i].ready);
  }

  // Signal start of preroll phase.
  (*g_preroll_cb)(g_pony_object);

  // Fill.
  for (i = 0; i < g_buffer_count; ++i) {
    (*g_produce_cb)(g_pony_object, 0.);
  }
}

void* produce_thread(void* unused) {
  pony_register_thread();
  uint32_t underruns = g_produce_underruns;
  while (1) {
    int wait_result = sem_wait(&g_produce_sem);
    if (wait_result != 0) {
      printf("Producer thread done\n");
      return;
    }
    if (g_produce_underruns > underruns) {
      printf("%d underruns\n", g_produce_underruns - underruns);
      underruns = g_produce_underruns;
    }
    (*g_produce_cb)(g_pony_object, g_produce_timestamp);
  }
}

void init_produce_thread() {
  int result;
  g_produce_timestamp = 0;
  result = sem_init(&g_produce_sem, 0, 0);
  assert(result == 0);
  pthread_attr_t attr;
  pthread_attr_init(&attr);
  result = pthread_create(&g_produce_thread, &attr, &produce_thread, NULL);
  assert(result == 0);
}

// the portaudio callback
int output_stream_cb(const void* input,
                     void* output,
                     unsigned long frames_per_buffer,
                     const PaStreamCallbackTimeInfo* time_info,
                     PaStreamCallbackFlags status_flags,
                     void* user) {
  assert(frames_per_buffer == g_frames_per_buffer); 

  if (g_buffers[g_next_buffer].ready) {
    memcpy(output, g_buffers[g_next_buffer].buf, frames_per_buffer * sizeof(float));
    g_buffers[g_next_buffer].ready = 0;
    g_produce_timestamp = time_info->outputBufferDacTime;
    sem_post(&g_produce_sem);
    if (++g_next_buffer == g_buffer_count) {
      g_next_buffer = 0;
    }
  } else {
    // underrun!
    g_produce_underruns++;
  }
  return paContinue;
}

int init_output_stream(unsigned long frames_per_buffer,
                       unsigned long buffer_count,
                       pony_output_add_buffer_cb add_buffer_cb,
                       pony_output_preroll_cb preroll_cb,
                       pony_output_produce_cb produce_cb,
                       void* pony_object) {
  printf("init: cbs: add buffer %p preroll %p produce %p\n",
    add_buffer_cb,
    preroll_cb,
    produce_cb);
  if (NULL != g_stream) {
    return paDeviceUnavailable;
  }

  PaError result = Pa_Initialize();
  if (paNoError != result) {
    return result;
  }

  PaStream* stream;
  result = Pa_OpenDefaultStream(&stream,
                                0, // no inputs
                                1, // 1 output
                                paFloat32,
                                44100,
                                frames_per_buffer,
                                &output_stream_cb,
                                pony_object);
  if (paNoError != result) {
    return result;
  }

  g_stream = stream;
  g_frames_per_buffer = frames_per_buffer;
  g_pony_object = pony_object;
  g_add_buffer_cb = add_buffer_cb;
  g_preroll_cb = preroll_cb;
  g_produce_cb = produce_cb;
  init_buffers(buffer_count);
  preroll();

  return paNoError;
}

int start_output_stream() {
  init_produce_thread();
  PaError result = Pa_StartStream(g_stream);
  return result;
}