Use GstBus messages instead of async events

7 files changed, 126 insertions, 108 deletions

diff --git a/configure.ac b/configure.ac
index ef67921..b010c01 100644
--- a/configure.ac
+++ b/configure.ac
@@ -1,4 +1,4 @@
-AC_INIT
+AC_INIT([gst-plugin-espeak], [0.1])
 
 dnl versions of gstreamer and plugins-base
 GST_MAJORMINOR=0.10
diff --git a/src/Makefile.am b/src/Makefile.am
index fa72f07..2313a30 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -1,14 +1,6 @@
-.list.h:
-	glib-genmarshal --prefix=espeak --header $< > $*.h
-
-.list.c:
-	glib-genmarshal --prefix=espeak --body $< > $*.c
-
-marshal.c: marshal.h
-
 plugin_LTLIBRARIES = libgstespeak.la
 
-libgstespeak_la_SOURCES = marshal.c marshal.h espeak.c gstespeak.c
+libgstespeak_la_SOURCES = espeak.c gstespeak.c
 
 libgstespeak_la_CFLAGS = $(GST_CFLAGS) $(GIO_CFLAGS) $(GST_AUDIO_CFLAGS)
 libgstespeak_la_LIBADD = $(GST_LIBS) $(GIO_LIBS) $(GST_AUDIO_LIBS)
@@ -17,5 +9,3 @@ libgstespeak_la_LIBTOOLFLAGS = --tag=disable-static
 
 # headers we need but don't want installed
 noinst_HEADERS = gstespeak.h espeak.h
-
-CLEANFILES = marshal.c marshal.h
diff --git a/src/espeak.c b/src/espeak.c
index 8212af5..563ee55 100644
--- a/src/espeak.c
+++ b/src/espeak.c
@@ -15,6 +15,9 @@
  * Boston, MA 02111-1307, USA.
  */
 
+#include <unistd.h>
+
+
 #include <pthread.h>
 #include <stdlib.h>
 #include <string.h>
@@ -26,7 +29,7 @@
 #define SYNC_BUFFER_SIZE 4096
 
 #define SPIN_QUEUE_SIZE 2
-#define SPIN_FRAME_SIZE 256
+#define SPIN_FRAME_SIZE 255
 
 #include "espeak.h"
 #include "text.h"
@@ -41,8 +44,7 @@ typedef enum
 
 typedef enum
 {
-    CLOSE     = 1,
-    INPROCESS = 2
+    INPROCESS = 1
 } ContextState;
 
 typedef struct
@@ -50,6 +52,7 @@ typedef struct
     volatile SpinState state;
 
     Text text;
+    goffset last_word;
 
     GMemoryOutputStream *sound;
     goffset sound_offset;
@@ -73,6 +76,9 @@ struct _Econtext
     volatile gint rate;
     volatile gint pitch;
     volatile const gchar *voice;
+
+    GstElement *emitter;
+    GstBus *bus;
 };
 
 static inline void
@@ -82,13 +88,26 @@ spinning(Espin *base, Espin **i)
         *i = base;
 }
 
+static void
+emit_word(Econtext *self, guint offset, guint len)
+{
+    GstStructure *data = gst_structure_new("word",
+            "offset", G_TYPE_UINT, offset,
+            "len", G_TYPE_UINT, len,
+            NULL);
+    if (!self->bus)
+        self->bus = gst_element_get_bus(self->emitter);
+    GstMessage *msg = gst_message_new_element(GST_OBJECT(self->emitter), data);
+    gst_bus_post(self->bus, msg);
+}
+
 static void init();
 static void process_push(Econtext*);
 static void process_pop(Econtext*);
 
-static pthread_t process_tid;
-static pthread_mutex_t process_lock = PTHREAD_MUTEX_INITIALIZER;
-static pthread_cond_t process_cond = PTHREAD_COND_INITIALIZER;
+static GThread *process_tid = NULL;
+static GMutex *process_lock = NULL;
+static GCond  *process_cond = NULL;
 static GSList *process_queue = NULL;
 
 static gint espeak_sample_rate = 0;
@@ -99,7 +118,7 @@ static GArray *espeak_events = NULL;
 // -----------------------------------------------------------------------------
 
 Econtext*
-espeak_new()
+espeak_new(GstElement *emitter)
 {
     init();
 
@@ -125,6 +144,11 @@ espeak_new()
     self->rate = ESPEAK_DEFAULT_RATE;
     self->voice = ESPEAK_DEFAULT_VOICE;
 
+    self->emitter = emitter;
+    gst_object_ref(self->emitter);
+
+    self->bus = NULL;
+
     GST_DEBUG("[%p]", self);
 
     return self;
@@ -135,11 +159,8 @@ espeak_unref(Econtext *self)
 {
     GST_DEBUG("[%p]", self);
 
-    g_atomic_int_set(&self->state, g_atomic_int_get(&self->state) | CLOSE);
     process_pop(self);
 
-    GST_DEBUG("[%p]", self);
-
     gint i;
 
     for (i = SPIN_QUEUE_SIZE; i--;)
@@ -161,6 +182,9 @@ espeak_unref(Econtext *self)
 
     g_slist_free(self->process_chunk);
 
+    gst_object_unref(self->bus);
+    gst_object_unref(self->emitter);
+
     memset(self, 0, sizeof(Econtext));
     g_free(self);
 }
@@ -184,13 +208,8 @@ in_spinning(Econtext *self, Text *text)
         chunked = TRUE;
     }
 
-    int self_status = g_atomic_int_get(&self->state);
-
-    if (chunked && (self_status & INPROCESS) == 0)
-    {
-        g_atomic_int_set(&self->state, self_status | INPROCESS);
+    if (chunked)
         process_push(self);
-    }
 
     GST_DEBUG("[%p] text.body=%s text.offset=%ld text.frame_len=%ld",
             self, text->body, text->offset, text->frame_len);
@@ -222,7 +241,7 @@ espeak_in(Econtext *self, const gchar *str_)
 }
 
 GstBuffer*
-play(Espin *spin, gsize size_to_play, gpointer emitter)
+play(Econtext *self, Espin *spin, gsize size_to_play)
 {
     inline gsize whole(Espin *spin, gsize size_to_play)
     {
@@ -230,7 +249,7 @@ play(Espin *spin, gsize size_to_play, gpointer emitter)
         return MIN(size_to_play, spin_size);
     }
 
-    inline gsize words(Espin *spin, gsize size_to_play, gpointer emitter)
+    inline gsize words(Econtext *self, Espin *spin, gsize size_to_play)
     {
         gsize spin_size = g_memory_output_stream_get_data_size(spin->sound);
         size_to_play = MIN(size_to_play, spin_size);
@@ -266,6 +285,12 @@ play(Espin *spin, gsize size_to_play, gpointer emitter)
             }
         }
 
+        if (text_offset != -1 && text_offset > spin->last_word)
+        {
+            spin->last_word = text_offset + text_len;
+            emit_word(self, text_offset, text_len);
+        }
+
         if (sample_offset - spin->sound_offset > size_to_play)
         {
             GST_DEBUG("sample_offset=%ld spin->sound_offset=%ld",
@@ -274,20 +299,15 @@ play(Espin *spin, gsize size_to_play, gpointer emitter)
         }
 
         if (text_offset != -1)
-        {
-            GST_DEBUG("event=%ld", event);
-            g_signal_emit_by_name(emitter, "word",
-                    text_offset, text_len, G_TYPE_NONE);
             spin->events_pos = event + 1;
-        }
 
         return sample_offset - spin->sound_offset;
     }
 
     g_atomic_int_set(&spin->state, PLAY);
 
-    if (emitter)
-        size_to_play = words(spin, size_to_play, emitter);
+    if (self->emitter)
+        size_to_play = words(self, spin, size_to_play);
     else
         size_to_play = whole(spin, size_to_play);
 
@@ -305,23 +325,28 @@ play(Espin *spin, gsize size_to_play, gpointer emitter)
 }
 
 GstBuffer*
-espeak_out(Econtext *self, gsize size_to_play, gpointer emitter)
+espeak_out(Econtext *self, gsize size_to_play)
 {
     GST_DEBUG("[%p] size_to_play=%d", self, size_to_play);
 
     for (;;)
     {
-        pthread_mutex_lock(&process_lock);
-            while ((g_atomic_int_get(&self->state) & CLOSE) == 0 &&
-                    (g_atomic_int_get(&self->out->state) & (PLAY|OUT)) == 0)
-                pthread_cond_wait(&process_cond, &process_lock);
-        pthread_mutex_unlock(&process_lock);
-
-        if (g_atomic_int_get(&self->state) & CLOSE)
-        {
-            GST_DEBUG("[%p]", self);
-            return NULL;
-        }
+        g_mutex_lock(process_lock);
+            if ((g_atomic_int_get(&self->out->state) & (PLAY|OUT)) == 0)
+            {
+                if (g_atomic_int_get(&self->state) & INPROCESS) 
+                {
+                    GST_DEBUG("[%p]", self);
+                    g_cond_wait(process_cond, process_lock);
+                }
+                else
+                {
+                    GST_DEBUG("[%p]", self);
+                    g_mutex_unlock(process_lock);
+                    return NULL;
+                }
+            }
+        g_mutex_unlock(process_lock);
 
         Espin *spin = self->out;
         gsize spin_size = g_memory_output_stream_get_data_size(spin->sound);
@@ -355,7 +380,7 @@ espeak_out(Econtext *self, gsize size_to_play, gpointer emitter)
             continue;
         }
 
-        return play(spin, size_to_play, emitter);
+        return play(self, spin, size_to_play);
     }
 
     return NULL;
@@ -406,6 +431,7 @@ synth(Econtext *self, Espin *spin)
     g_array_set_size(spin->events, 0);
     spin->sound_offset = 0;
     spin->events_pos = 0;
+    spin->last_word = -1;
 
     espeak_SetParameter(espeakPITCH, g_atomic_int_get(&self->pitch), 0);
     espeak_SetParameter(espeakRATE, g_atomic_int_get(&self->rate), 0);
@@ -465,15 +491,15 @@ espeak_set_voice(Econtext *self, const gchar *value)
 
 // process ----------------------------------------------------------------------
 
-static void*
-process(void *data)
+static gpointer
+process(gpointer data)
 {
-    pthread_mutex_lock(&process_lock);
+    g_mutex_lock(process_lock);
 
     for (;;)
     {
         while (process_queue == NULL)
-            pthread_cond_wait(&process_cond, &process_lock);
+            g_cond_wait(process_cond, process_lock);
 
         while (process_queue)
         {
@@ -501,10 +527,10 @@ process(void *data)
             }
         }
 
-        pthread_cond_broadcast(&process_cond);
+        g_cond_broadcast(process_cond);
     }
 
-    pthread_mutex_unlock(&process_lock);
+    g_mutex_unlock(process_lock);
 
     return NULL;
 }
@@ -512,18 +538,33 @@ process(void *data)
 static void
 process_push(Econtext *context)
 {
-    pthread_mutex_lock(&process_lock);
-    process_queue = g_slist_concat(process_queue, context->process_chunk);
-    pthread_cond_broadcast(&process_cond);
-    pthread_mutex_unlock(&process_lock);
+    GST_DEBUG("[%p]", context);
+    g_mutex_lock(process_lock);
+
+    int self_status = g_atomic_int_get(&context->state);
+
+    if ((self_status & INPROCESS) == 0)
+    {
+        g_atomic_int_set(&context->state, self_status | INPROCESS);
+        process_queue = g_slist_concat(process_queue, context->process_chunk);
+        g_cond_broadcast(process_cond);
+    }
+
+    g_mutex_unlock(process_lock);
+    GST_DEBUG("[%p]", context);
 }
 
 static void
 process_pop(Econtext *context)
 {
-    pthread_mutex_lock(&process_lock);
+    GST_DEBUG("[%p]", context);
+    g_mutex_lock(process_lock);
+
     process_queue = g_slist_remove_link(process_queue, context->process_chunk);
-    pthread_mutex_unlock(&process_lock);
+    g_cond_broadcast(process_cond);
+
+    g_mutex_unlock(process_lock);
+    GST_DEBUG("[%p]", context);
 }
 
 // -----------------------------------------------------------------------------
@@ -541,10 +582,8 @@ init()
         espeak_SetSynthCallback(synth_cb);
         espeak_voices = espeak_ListVoices(NULL);
 
-        pthread_attr_t attr;
-        g_assert(pthread_attr_init(&attr) == 0);
-        g_assert(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) == 0);
-        g_assert(pthread_create(&process_tid, &attr, process, NULL) == 0);
-        g_assert(pthread_attr_destroy(&attr) == 0);
+        process_lock = g_mutex_new();
+        process_cond = g_cond_new();
+        process_tid = g_thread_create(process, NULL, FALSE, NULL);
     }
 }
diff --git a/src/espeak.h b/src/espeak.h
index e4f7d15..eaa7b68 100644
--- a/src/espeak.h
+++ b/src/espeak.h
@@ -25,7 +25,7 @@
 struct _Econtext;
 typedef struct _Econtext Econtext;
 
-Econtext*  espeak_new();
+Econtext*  espeak_new(GstElement*);
 void       espeak_unref(Econtext*);
 
 gint       espeak_get_sample_rate();
@@ -35,6 +35,6 @@ void       espeak_set_rate(Econtext*, guint);
 void       espeak_set_voice(Econtext*, const gchar*);
 
 void       espeak_in(Econtext*, const gchar *str);
-GstBuffer* espeak_out(Econtext*, gsize size_to_play, gpointer emitter);
+GstBuffer* espeak_out(Econtext*, gsize size_to_play);
 
 #endif
diff --git a/src/gstespeak.c b/src/gstespeak.c
index 221abf4..cfbf86d 100644
--- a/src/gstespeak.c
+++ b/src/gstespeak.c
@@ -23,7 +23,7 @@
  * <refsect2>
  * <title>Example launch line</title>
  * |[
- * gst-launch-0.10 espeak text="Hello world" pitch=99 rate=300 voice=default ! alsasink
+ * gst-launch-0.10 espeak text="Hello world" ! autoaudiosink
  * ]|
  * </refsect2>
  */
@@ -37,19 +37,12 @@
 
 #include "gstespeak.h"
 #include "espeak.h"
-#include "marshal.h"
 
 GST_DEBUG_CATEGORY_STATIC (gst_espeak_debug);
 #define GST_CAT_DEFAULT gst_espeak_debug
 
 enum
 {
-    SIGNAL_WORD,
-    LAST_SIGNAL
-};
-
-enum
-{
     PROP_0,
     PROP_TEXT,
     PROP_PITCH,
@@ -81,8 +74,6 @@ static GstCaps *gst_espeak_getcaps(GstBaseSrc*);
 GST_BOILERPLATE_FULL(GstEspeak, gst_espeak, GstBaseSrc, GST_TYPE_BASE_SRC,
         gst_espeak_init_uri);
 
-static gint signals[LAST_SIGNAL];
-
 /******************************************************************************/
 
 static void
@@ -102,7 +93,7 @@ gst_espeak_base_init (gpointer gclass)
 
 /* initialize the espeak's class */
 static void
-gst_espeak_class_init (GstEspeakClass * klass)
+gst_espeak_class_init(GstEspeakClass * klass)
 {
     GObjectClass *gobject_class = (GObjectClass *) klass;
     GstBaseSrcClass *basesrc_class = (GstBaseSrcClass *) klass;
@@ -110,7 +101,6 @@ gst_espeak_class_init (GstEspeakClass * klass)
     basesrc_class->create = gst_espeak_create;
     basesrc_class->start = gst_espeak_start;
     basesrc_class->stop = gst_espeak_stop;
-    basesrc_class->stop = gst_espeak_stop;
     basesrc_class->is_seekable = gst_espeak_is_seekable;
     basesrc_class->get_caps = gst_espeak_getcaps;
 
@@ -124,30 +114,24 @@ gst_espeak_class_init (GstEspeakClass * klass)
                 G_PARAM_WRITABLE | G_PARAM_STATIC_STRINGS));
     g_object_class_install_property(gobject_class, PROP_PITCH,
             g_param_spec_uint("pitch", "Pitch adjustment",
-                "Pitch adjustment", 0, 99, 50,
+                "Pitch adjustment", 0, 99, ESPEAK_DEFAULT_PITCH,
                 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
     g_object_class_install_property(gobject_class, PROP_RATE,
             g_param_spec_uint("rate", "Speed in words per minute",
-                "Speed in words per minute", 80, 390, 170,
+                "Speed in words per minute", 80, 390, ESPEAK_DEFAULT_RATE,
                 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
     g_object_class_install_property(gobject_class, PROP_VOICE,
             g_param_spec_string("voice", "Current voice",
-                "Current voice", "default",
+                "Current voice", ESPEAK_DEFAULT_VOICE,
                 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
     g_object_class_install_property(gobject_class, PROP_VOICES,
             g_param_spec_boxed("voices", "List of voices",
                 "List of voices", G_TYPE_STRV,
                 G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
-    g_object_class_install_property (gobject_class, PROP_CAPS,
-            g_param_spec_boxed ("caps", "Caps",
-                "Caps describing the format of the data.", GST_TYPE_CAPS,
+    g_object_class_install_property(gobject_class, PROP_CAPS,
+            g_param_spec_boxed("caps", "Caps",
+                "Caps describing the format of the data", GST_TYPE_CAPS,
                 G_PARAM_READABLE));
-
-    signals[SIGNAL_WORD] = g_signal_new("word",
-            G_TYPE_FROM_CLASS(klass), G_SIGNAL_RUN_LAST,
-            G_STRUCT_OFFSET(GstEspeakClass, word),
-            NULL, NULL, espeak_VOID__UINT_UINT, G_TYPE_NONE,
-            2, G_TYPE_UINT, G_TYPE_UINT);
 }
 
 /* initialize the new element
@@ -163,7 +147,7 @@ gst_espeak_init (GstEspeak * self,
     self->rate = ESPEAK_DEFAULT_RATE;
     self->voice = g_strdup(ESPEAK_DEFAULT_VOICE);
     self->voices = espeak_get_voices();
-    self->speak = espeak_new();
+    self->speak = espeak_new(GST_ELEMENT(self));
 
     self->caps = gst_caps_new_simple("audio/x-raw-int",
             "rate", G_TYPE_INT, espeak_get_sample_rate(),
@@ -178,6 +162,7 @@ gst_espeak_init (GstEspeak * self,
 static void
 gst_espeak_finalize(GObject * self_)
 {
+fprintf(stderr, "0!!!!!!\n");
     GstEspeak *self = GST_ESPEAK(self_);
 
     gst_caps_unref(self->caps); self->caps = NULL;
@@ -253,8 +238,16 @@ gst_espeak_create(GstBaseSrc * self_, guint64 offset, guint size,
         GstBuffer **buf)
 {
     GstEspeak *self = GST_ESPEAK(self_);
-    *buf = espeak_out(self->speak, size, self);
-    return *buf ? GST_FLOW_OK : GST_FLOW_UNEXPECTED;
+
+    *buf = espeak_out(self->speak, size);
+
+    if (*buf)
+        return GST_FLOW_OK;
+    else
+    {
+        //gst_element_set_state(GST_ELEMENT(self), GST_STATE_NULL);
+        return GST_FLOW_UNEXPECTED;
+    }
 }
 
 static gboolean
@@ -266,6 +259,7 @@ gst_espeak_start(GstBaseSrc * self_)
 static gboolean
 gst_espeak_stop(GstBaseSrc * self)
 {
+    GST_DEBUG("!!!!!!!!!!");
     return TRUE;
 }
 
@@ -348,7 +342,7 @@ gst_espeak_init_uri(GType filesrc_type)
  * register the element factories and other features
  */
 static gboolean
-espeak_init (GstPlugin * espeak)
+espeak_init(GstPlugin *espeak)
 {
   /* debug category for fltering log messages
    *
@@ -361,18 +355,14 @@ espeak_init (GstPlugin * espeak)
       GST_TYPE_ESPEAK);
 }
 
-/* gstreamer looks for this structure to register espeaks
- *
- * exchange the string 'Template espeak' with your espeak description
- */
-GST_PLUGIN_DEFINE (
+GST_PLUGIN_DEFINE(
     GST_VERSION_MAJOR,
     GST_VERSION_MINOR,
     "espeak",
-    "Template espeak",
+    "Uses eSpeak library as a sound source for GStreamer",
     espeak_init,
-    VERSION,
+    PACKAGE_VERSION,
     "LGPL",
-    "GStreamer",
-    "http://gstreamer.net/"
+    PACKAGE_NAME,
+    "http://sugarlabs.org/go/DevelopmentTeam/gst-plugins-espeak"
 )
diff --git a/src/gstespeak.h b/src/gstespeak.h
index 31986f4..df17118 100644
--- a/src/gstespeak.h
+++ b/src/gstespeak.h
@@ -48,12 +48,12 @@ struct _GstEspeak
     gchar *voice;
     gchar **voices;
     GstCaps *caps;
+    gboolean poll;
 };
 
 struct _GstEspeakClass 
 {
     GstAudioSrcClass parent_class;
-    void (*word) (GstEspeak*, goffset, gsize);
 };
 
 GType gst_espeak_get_type (void);
diff --git a/src/marshal.list b/src/marshal.list
deleted file mode 100644
index 6cb7431..0000000
--- a/src/marshal.list
+++ /dev/null
@@ -1 +0,0 @@
-VOID:UINT,UINT