cancel playback functionality, on write event

package.json

@@ -11,7 +11,7 @@
     "test": "node test.js"
   },
   "gypfile": true,
-  "publishConfig": { 
+  "publishConfig": {
     "registry": "https://gitlab.com/api/v4/projects/28849281/packages/npm/"
   }
 }

src/na_callbacks.c

@@ -10,6 +10,7 @@
 
 /* JS callbacks */
 napi_threadsafe_function js_on_data;
+napi_threadsafe_function js_on_playback;
 
 /* Calls js_emit_ts from the main thread and translates data to node */
 static void call_js_on_data( napi_env   env,
@@ -56,11 +57,30 @@ static void call_js_on_data( napi_env   env,
     status = napi_call_function(env, undefined, on_data, argc, &args[0], NULL);
 }
 
+static void call_js_on_playback( napi_env   env,
+                                 napi_value on_data,
+                                 void*      context,
+                                 void*      data )
+{
+    (void) context;
+    napi_status status;
+
+    napi_value channel;
+    status = napi_create_string_utf8(env, "write", NAPI_AUTO_LENGTH, &channel);
+
+    napi_value undefined;
+    status = napi_get_undefined(env, &undefined);
+
+    size_t argc = 1;
+    napi_value args[1] = { channel };
+    status = napi_call_function(env, undefined, on_data, argc, &args[0], NULL);
+}
+
 /* Takes in a Node Event Emitter and creates JS threadsafe functions to be 
  * called from the PortAudio callbacks.
  */
-void init_callbacks( napi_env env,
+void init_callbacks( napi_env          env,
-                     napi_value* js_emit, 
+                     napi_value*       js_emit,
                      PaUtilRingBuffer* inputBuffer )
 {    
     napi_status status;
@@ -70,7 +90,7 @@ void init_callbacks( napi_env env,
                                       NAPI_AUTO_LENGTH,
                                       &resource_name );
 
-    /* Create thread-safe callback js_cb_safe */
+    /* Create thread-safe js_on_data callback */
     status = napi_create_threadsafe_function( env,
                                               *js_emit,
                                               NULL,
@@ -82,7 +102,19 @@ void init_callbacks( napi_env env,
                                               inputBuffer, /* context */
                                               call_js_on_data,
                                               &js_on_data );
-    /* Optional: create another cb with js_emit here */
+
+    /* Create thread-safe js_on_write callback */
+    status = napi_create_threadsafe_function( env,
+                                              *js_emit,
+                                              NULL,
+                                              resource_name,
+                                              0,
+                                              1,
+                                              NULL,
+                                              NULL, 
+                                              inputBuffer, /* context */
+                                              call_js_on_playback,
+                                              &js_on_playback );
 }
 
 int input_callback( const void*                     input, 
@@ -136,24 +168,18 @@ int output_callback( const void*                     input,
     if (framesToRead != 0)
     {
         PaUtil_ReadRingBuffer(rb, output, framesToRead);
-        printf("Wrote %i frames\n", framesToRead); 
+
+        napi_status status;
+        status = napi_call_threadsafe_function( js_on_playback,
+                                                NULL,
+                                                napi_tsfn_nonblocking );
     }
 
     return paContinue;
 }
 
-
+// short* sample = (short*) output;
-
+// for (int i = 0; i < framesToRead * 2; ++i)
-
+// {
-
+//     printf("%hu", sample[i]);
-
+// }
-
-        // short* sample = (short*) output;
-        // for (int i = 0; i < framesToRead * 2; ++i)
-        // {
-        //     printf("%hu", sample[i]);
-        // }
-
-
-
-

src/na_callbacks.h

@@ -8,8 +8,8 @@ typedef struct
 } 
 audio_info;
 
-void init_callbacks( napi_env env, 
+void init_callbacks( napi_env          env,
-                     napi_value* js_emit, 
+                     napi_value*       js_emit,
                      PaUtilRingBuffer* inputBuffer );
 
 int input_callback( const void*                     input, 

src/na_core.c

@@ -20,11 +20,15 @@ PaUtilRingBuffer outputBuffer;
 void* inputBufferStore;
 void* outputBufferStore;
 
-/* Sample format is hard-coded paInt16 (unsigned short, 2 bytes) */
+pthread_t pb_th;
+pthread_mutex_t mutex;
 
-PaError Na_Initialize(napi_env env, napi_value* js_cb)
+PaError Na_Initialize(napi_env env, napi_value* js_emit)
 {
-    init_callbacks(env, js_cb, &inputBuffer);
+    init_callbacks(env, js_emit, &inputBuffer);
+
+    pthread_mutex_init(&mutex, NULL);
+
     PaError err;
     err = Pa_Initialize();
     return err;
@@ -132,6 +136,8 @@ PaError Na_CloseOutputStream()
     return err;
 }
 
+bool cancel_pb_th = false;
+
 static void WriteLargeDataToOutputBuffer(void* arg)
 {
     playback_data* pd = (playback_data*) arg;
@@ -140,9 +146,9 @@ static void WriteLargeDataToOutputBuffer(void* arg)
     ring_buffer_size_t framesToWrite;
     ring_buffer_size_t framesWritten;
 
-    // printing data works here...
+    pthread_mutex_lock(&mutex);
-
+    cancel_pb_th = false;
-    while (pd->len > 0)
+    while (pd->len > 0 && !cancel_pb_th)
     {
         writableSpace = PaUtil_GetRingBufferWriteAvailable(&outputBuffer);
         if (writableSpace != 0)
@@ -158,6 +164,7 @@ static void WriteLargeDataToOutputBuffer(void* arg)
             pd->data += framesWritten * 2; /* convert to bytes */
         }
     }
+    pthread_mutex_unlock(&mutex);
 
     free(arg);
 }
@@ -167,8 +174,6 @@ void Na_WriteToOutputBuffer(void* data, size_t size)
     ring_buffer_size_t writableSpace;
     writableSpace = PaUtil_GetRingBufferWriteAvailable(&outputBuffer);
 
-    // printing data works here...
-
     /* If there is not enough space to write all the data at once, we must
      * start a new thread in order to prevent blocking.
      */
@@ -178,15 +183,22 @@ void Na_WriteToOutputBuffer(void* data, size_t size)
         playback_data* pd = (playback_data*) malloc(sizeof(playback_data));
         pd->data = data;
         pd->len = len;
-        pthread_t w_th;
+        pthread_create(&pb_th, NULL, (void*) WriteLargeDataToOutputBuffer, pd);
-        pthread_create(&w_th, NULL, (void*) WriteLargeDataToOutputBuffer, pd);
     }
     else
     {
+        pthread_mutex_lock(&mutex);
         PaUtil_WriteRingBuffer(&outputBuffer, data, writableSpace);
+        pthread_mutex_unlock(&mutex);
     }
 }
 
+void Na_CancelPlayback()
+{
+    printf("NACORE:Na_CancelPlayback\n");
+    cancel_pb_th = true;
+}
+
 PaError Na_Terminate()
 {
     printf("NACORE:Na_Terminate\n");
@@ -205,25 +217,10 @@ PaError Na_Terminate()
     free_buffer(&inputBuffer);
     free_buffer(&outputBuffer);
 
+    pthread_mutex_destroy(&mutex);
+
     return err;
 }
 
 
 
-
-        // /* Data being passed into the thread must be allocated to heap so that 
-        //  * it isn't deleted when this function returns.
-        //  */
-        // playback_data* pd = (playback_data*) malloc(sizeof(playback_data));
-        // pd->data = (void*) malloc(len * 2);
-        // memcpy(pd->data, data, 2);
-        // pd->len = len;
-
-    // short* sample = (short*) pd->data;
-    // int count = (int) pd->len;
-    // for (int i = 0; i < count; i++)
-    // {
-    //     printf("%hu", sample[i]);
-    // }
-
-

src/na_core.h

@@ -8,7 +8,7 @@ typedef struct
 }
 playback_data;
 
-PaError Na_Initialize(napi_env env, napi_value* js_cb);
+PaError Na_Initialize(napi_env env, napi_value* js_emit);
 
 PaError Na_GetDefaultInputDevice(PaDeviceIndex* deviceIndex);
 
@@ -24,6 +24,8 @@ PaError Na_CloseOutputStream();
 
 void Na_WriteToOutputBuffer(void* data, size_t size);
 
+void Na_CancelPlayback();
+
 PaError Na_Terminate();
 
 #endif

src/na_front.c

@@ -161,6 +161,16 @@ napi_value WriteToOutputStream(napi_env env, napi_callback_info info)
     return NULL;
 }
 
+napi_value CancelPlayback(napi_env env, napi_callback_info info)
+{
+    printf("NAPI::CancelPlayback\n");
+    (void) env;
+    (void) info;
+
+    Na_CancelPlayback();
+    return NULL;
+}
+
 napi_value Terminate(napi_env env, napi_callback_info info)
 {
     printf("NAPI::Terminate\n");
@@ -190,9 +200,10 @@ napi_value Init(napi_env env, napi_value exports)
     DECLARE_NAPI_METHOD("CloseInputStream", CloseInputStream),
     DECLARE_NAPI_METHOD("CloseOutputStream", CloseOutputStream),
     DECLARE_NAPI_METHOD("WriteToOutputStream", WriteToOutputStream),
+    DECLARE_NAPI_METHOD("CancelPlayback", CancelPlayback)
   };
 
-  status = napi_define_properties(env, exports, 9, desc);
+  status = napi_define_properties(env, exports, 10, desc);
 
   return exports;
 

test/rwSwitchStream.js

@@ -21,6 +21,10 @@ emitter.on("data", (int16Arr) => {
     }
 });
 
+emitter.on("write", () => {
+    console.log("Writing frames");
+});
+
 const setStreams = () => {
 
     const defaultInput = nodeAudio.core.GetDefaultInputDevice();
@@ -48,6 +52,18 @@ const stopRecording = () => {
     /* Play recorded audio */
     console.log("Playing recorded audio...");
     nodeAudio.core.WriteToOutputStream(pcmAudio.buffer);
+
+    /* Two writes test */
+    // setTimeout(() => {
+    //     console.log("Playing again...");
+    //     nodeAudio.core.WriteToOutputStream(pcmAudio.buffer);
+    // }, 2000);
+
+    /* Cancel mid-playback test */
+    setTimeout(() => {
+        console.log("Cancelling...");
+        nodeAudio.core.CancelPlayback();
+    }, 2000);
 }
 
 nodeAudio.core.Initialize(emitter.emit.bind(emitter));
@@ -69,4 +85,4 @@ stdin.on( 'data', function( key ) {
 setTimeout(() => {
     nodeAudio.core.Terminate();
     process.exit();
-}, 30000);
+}, 20000);