State change fixes in the threading element.

Original commit message from CVS:
State change fixes in the threading element.
Added bufferpools.

gst/Makefile.am

@@ -13,6 +13,7 @@ libgst_la_SOURCES =	\
 	$(GSTOBJECT_SRCS)	\
 	gstpad.c	\
 	gstbuffer.c	\
+	gstbufferpool.c	\
 	gstclock.c	\
 	gstelement.c	\
 	gstelementfactory.c	\
@@ -39,6 +40,7 @@ libgstinclude_HEADERS =	\
 	$(GSTOBJECT_INCLUDES)	\
 	gstpad.h	\
 	gstbuffer.h	\
+	gstbufferpool.h	\
 	gstclock.h	\
 	gstelement.h	\
 	gstbin.h	\

gst/elements/gstaudiosink.c

@@ -190,7 +190,7 @@ void gst_audiosink_chain(GstPad *pad,GstBuffer *buf) {
   audiosink = GST_AUDIOSINK(pad->parent);
 //  g_return_if_fail(GST_FLAG_IS_SET(audiosink,GST_STATE_RUNNING));
 
-  if (in_flush = GST_BUFFER_FLAG_IS_SET(buf, GST_BUFFER_FLUSH)) {
+  if ((in_flush = GST_BUFFER_FLAG_IS_SET(buf, GST_BUFFER_FLUSH))) {
     DEBUG("audiosink: flush\n");
     ioctl(audiosink->fd,SNDCTL_DSP_RESET,0);
   }
@@ -217,20 +217,13 @@ void gst_audiosink_chain(GstPad *pad,GstBuffer *buf) {
     //g_print("audiosink: writing to soundcard\n");
     if (audiosink->fd > 2) {
       if (!audiosink->mute) {
-        //if (gst_clock_current_diff(audiosink->clock, GST_BUFFER_TIMESTAMP(buf)) > 500000) {
+        gst_clock_wait(audiosink->clock, GST_BUFFER_TIMESTAMP(buf), GST_OBJECT(audiosink));
-	//}
+        ioctl(audiosink->fd,SNDCTL_DSP_GETOSPACE,&ospace);
-	//else {
+        DEBUG("audiosink: (%d bytes buffer)\n", ospace.bytes);
-          gst_clock_wait(audiosink->clock, GST_BUFFER_TIMESTAMP(buf), GST_OBJECT(audiosink));
+        write(audiosink->fd,GST_BUFFER_DATA(buf),GST_BUFFER_SIZE(buf));
-          ioctl(audiosink->fd,SNDCTL_DSP_GETOSPACE,&ospace);
-          DEBUG("audiosink: (%d bytes buffer)\n", ospace.bytes);
-          write(audiosink->fd,GST_BUFFER_DATA(buf),GST_BUFFER_SIZE(buf));
-	//}
-          //gst_clock_set(audiosink->clock, GST_BUFFER_TIMESTAMP(buf));
-	//}
       }
     }
   }
-end:
   //g_print("a unref\n");
   gst_buffer_unref(buf);
   //g_print("a done\n");
@@ -314,6 +307,7 @@ static gboolean gst_audiosink_open_audio(GstAudioSink *sink) {
     if (sink->caps & DSP_CAP_TRIGGER)  g_print("audiosink:   Trigger\n");
     if (sink->caps & DSP_CAP_MMAP)     g_print("audiosink:   Direct access\n");
     g_print("audiosink: opened audio\n");
+    GST_FLAG_SET(sink,GST_AUDIOSINK_OPEN);
     return TRUE;
   }
 
@@ -325,10 +319,10 @@ static void gst_audiosink_close_audio(GstAudioSink *sink) {
 
   close(sink->fd);
   sink->fd = -1;
+  GST_FLAG_UNSET(sink,GST_AUDIOSINK_OPEN);
   g_print("audiosink: closed sound device\n");
 }
 
-
 static GstElementStateReturn gst_audiosink_change_state(GstElement *element) {
   g_return_val_if_fail(GST_IS_AUDIOSINK(element), FALSE);
 
@@ -346,5 +340,5 @@ static GstElementStateReturn gst_audiosink_change_state(GstElement *element) {
       
   if (GST_ELEMENT_CLASS(parent_class)->change_state)
     return GST_ELEMENT_CLASS(parent_class)->change_state(element);
-  return TRUE;
+  return GST_STATE_SUCCESS;
 }

gst/elements/gstaudiosrc.c

@@ -241,7 +241,7 @@ static GstElementStateReturn gst_audiosrc_change_state(GstElement *element) {
 
   if (GST_ELEMENT_CLASS(parent_class)->change_state)
     return GST_ELEMENT_CLASS(parent_class)->change_state(element);
-  return TRUE;
+  return GST_STATE_SUCCESS;
 }
 
 static gboolean gst_audiosrc_open_audio(GstAudioSrc *src) {

gst/elements/gstdisksrc.c

@@ -286,6 +286,8 @@ static void gst_disksrc_close_file(GstDiskSrc *src) {
 static GstElementStateReturn gst_disksrc_change_state(GstElement *element) {
   g_return_val_if_fail(GST_IS_DISKSRC(element),GST_STATE_FAILURE);
 
+  DEBUG("gstdisksrc: state pending %d\n", GST_STATE_PENDING(element));
+
   /* if going down into NULL state, close the file if it's open */
   if (GST_STATE_PENDING(element) == GST_STATE_NULL) {
     if (GST_FLAG_IS_SET(element,GST_DISKSRC_OPEN))

gst/elements/gstpipefilter.c

@@ -138,9 +138,9 @@ static gboolean gst_pipefilter_read_and_push(GstPipefilter *pipefilter) {
   g_return_val_if_fail(GST_BUFFER_DATA(newbuf) != NULL, FALSE);
 
   /* read it in from the file */
-  DEBUG("attemting to read %d bytes\n", pipefilter->bytes_per_read);
+  DEBUG("attemting to read %ld bytes\n", pipefilter->bytes_per_read);
   readbytes = read(pipefilter->fdout[0],GST_BUFFER_DATA(newbuf),pipefilter->bytes_per_read);
-  DEBUG("read %d bytes\n", readbytes);
+  DEBUG("read %ld bytes\n", readbytes);
   if (readbytes < 0) {
     if (errno == EAGAIN) {
       DEBUG("no input yet\n");
@@ -185,9 +185,9 @@ void gst_pipefilter_chain(GstPad *pad,GstBuffer *buf) {
   data = GST_BUFFER_DATA(buf);
   size = GST_BUFFER_SIZE(buf);
 
-  DEBUG("attemting to write %d bytes\n", size);
+  DEBUG("attemting to write %ld bytes\n", size);
   writebytes = write(pipefilter->fdin[1],data,size);
-  DEBUG("written %d bytes\n", writebytes);
+  DEBUG("written %ld bytes\n", writebytes);
   if (writebytes < 0) {
     perror("write");
     gst_element_error(GST_ELEMENT(pipefilter),"writing");
@@ -296,12 +296,12 @@ static GstElementStateReturn gst_pipefilter_change_state(GstElement *element) {
   /* otherwise (READY or higher) we need to open the file */
   } else {
     if (!GST_FLAG_IS_SET(element,GST_PIPEFILTER_OPEN)) {
-      if (!gst_disksrc_open_file(GST_PIPEFILTER(element)))
+      if (!gst_pipefilter_open_file(GST_PIPEFILTER(element)))
         return GST_STATE_FAILURE;
     }
   }
       
   if (GST_ELEMENT_CLASS(parent_class)->change_state)
     return GST_ELEMENT_CLASS(parent_class)->change_state(element);
-  return TRUE;
+  return GST_STATE_SUCCESS;
 }

gst/gstbin.c

@@ -111,6 +111,7 @@ gst_bin_class_init(GstBinClass *klass) {
 static void gst_bin_init(GstBin *bin) {
   bin->numchildren = 0;
   bin->children = NULL;
+  bin->use_cothreads = TRUE;
 }
 
 /**
@@ -222,12 +223,12 @@ static GstElementStateReturn gst_bin_change_state(GstElement *element) {
     switch (gst_element_set_state(child,GST_STATE_PENDING(element))) {
       case GST_STATE_FAILURE:
         GST_STATE_PENDING(element) = GST_STATE_NONE_PENDING;
-        g_print("child '%s' failed to go to state %d(%s)\n",gst_element_get_name(child),
+        g_print("gstbin: child '%s' failed to go to state %d(%s)\n",gst_element_get_name(child),
                 GST_STATE_PENDING(element),_gst_print_statename(GST_STATE_PENDING(element)));
         return GST_STATE_FAILURE;
         break;
       case GST_STATE_ASYNC:
-        g_print("child '%s' is changing state asynchronously\n",gst_element_get_name(child));
+        g_print("gstbin: child '%s' is changing state asynchronously\n",gst_element_get_name(child));
         break;
     }
 //    g_print("\n");
@@ -385,6 +386,12 @@ static xmlNodePtr gst_bin_save_thyself(GstElement *element,xmlNodePtr parent) {
 	return childlist;
 }
 
+void gst_bin_use_cothreads(GstBin *bin, gboolean enabled) {
+  g_return_if_fail(GST_IS_BIN(bin));
+
+  bin->use_cothreads = enabled;
+}
+
 /**
  * gst_bin_iterate:
  * @bin: #Gstbin to iterate
@@ -396,7 +403,7 @@ void gst_bin_iterate(GstBin *bin) {
 
   oclass = GST_BIN_CLASS(GTK_OBJECT(bin)->klass);
 
-g_print("gst_bin_iterate()\n");
+  DEBUG("gst_bin_iterate()\n");
   if (oclass->iterate)
     (oclass->iterate)(bin);
 }
@@ -540,20 +547,23 @@ static void gst_bin_create_plan_func(GstBin *bin) {
   GstPad *pad, *peer;
   GstElement *outside;
 
-  g_print("creating plan for bin\n");
+  g_print("gstbin: creating plan for bin \"%s\"\n", gst_element_get_name(GST_ELEMENT(bin)));
 
   // first loop through all children to see if we need cothreads
   // we break immediately when we find we need to, why keep searching?
   elements = bin->children;
   while (elements) {
     element = GST_ELEMENT(elements->data);
+    g_print("gstbin: found element \"%s\" in bin \"%s\"\n", gst_element_get_name(element), gst_element_get_name(GST_ELEMENT(bin)));
     // if it's a loop-based element, use cothreads
     if (element->loopfunc != NULL) {
+      g_print("gstbin: loop based elenemt \"%s\" in bin \"%s\"\n", gst_element_get_name(element), gst_element_get_name(GST_ELEMENT(bin)));
       bin->need_cothreads = TRUE;
       break;
     }
     // if it's a complex element, use cothreads
     if (GST_ELEMENT_IS_MULTI_IN(element)) {
+      g_print("gstbin: complex elenemt \"%s\" in bin \"%s\"\n", gst_element_get_name(element), gst_element_get_name(GST_ELEMENT(bin)));
       bin->need_cothreads = TRUE;
       break;
     }
@@ -567,19 +577,23 @@ static void gst_bin_create_plan_func(GstBin *bin) {
       pads = g_list_next(pads);
     }
     if (sink_pads > 1) {
+      g_print("gstbin: more than 1 sinkpad for elenemt \"%s\" in bin \"%s\"\n", gst_element_get_name(element), gst_element_get_name(GST_ELEMENT(bin)));
       bin->need_cothreads = TRUE;
       break;
     }
     elements = g_list_next(elements);
   }
 
+  // FIXME
+  bin->need_cothreads &= bin->use_cothreads;
+
   if (bin->need_cothreads) {
-    g_print("BIN: need cothreads\n");
+    g_print("gstbin: need cothreads\n");
 
     // first create thread context
     if (bin->threadcontext == NULL) {
       bin->threadcontext = cothread_init();
-      g_print("initialized cothread context\n");
+      g_print("gstbin: initialized cothread context\n");
     }
 
     // walk through all the children
@@ -597,7 +611,7 @@ static void gst_bin_create_plan_func(GstBin *bin) {
       pads = gst_element_get_pad_list(element);
       while (pads) {
         pad = GST_PAD(pads->data);
-g_print("setting push&pull handlers for %s:%s\n",
+g_print("gstbin: setting push&pull handlers for %s:%s\n",
 gst_element_get_name(element),gst_pad_get_name(pad));
 //        if (pad->direction == GST_PAD_SRC)
           pad->pushfunc = gst_bin_pushfunc_wrapper;
@@ -608,7 +622,7 @@ gst_element_get_name(element),gst_pad_get_name(pad));
       elements = g_list_next(elements);
    }
   } else {
-    g_print("BIN: don't need cothreads, looking for entry points\n");
+    g_print("gstbin: don't need cothreads, looking for entry points\n");
     // clear previous plan state
     g_list_free(bin->entries);
     bin->numentries = 0;
@@ -616,10 +630,37 @@ gst_element_get_name(element),gst_pad_get_name(pad));
     elements = bin->children;
     while (elements) {
       element = GST_ELEMENT(elements->data);
+      g_print("gstbin: found element \"%s\"\n", gst_element_get_name(element));
       if (GST_IS_SRC(element)) {
-        g_print("adding '%s' as entry point\n",gst_element_get_name(element));
+        g_print("gstbin: adding '%s' as entry point\n",gst_element_get_name(element));
         bin->entries = g_list_prepend(bin->entries,element);
         bin->numentries++;
+      } else {
+        /* go through the list of pads to see if there's a Connection */
+        pads = gst_element_get_pad_list(element);
+        while (pads) {
+          pad = GST_PAD(pads->data);
+          /* we only worry about sink pads */
+          if (gst_pad_get_direction(pad) == GST_PAD_SINK) {
+            /* get the pad's peer */
+            peer = gst_pad_get_peer(pad);
+            if (!peer) break;
+            /* get the parent of the peer of the pad */
+            outside = GST_ELEMENT(gst_pad_get_parent(peer));
+            if (!outside) break;
+            /* if it's a connection and it's not ours... */
+            if (GST_IS_CONNECTION(outside) &&
+                 (gst_object_get_parent(GST_OBJECT(outside)) != GST_OBJECT(bin))) {
+              gst_info("gstbin: element \"%s\" is the external source Connection \
+				    for internal element \"%s\"\n",
+	                    gst_element_get_name(GST_ELEMENT(outside)),
+	                    gst_element_get_name(GST_ELEMENT(element)));
+	      bin->entries = g_list_prepend(bin->entries,outside);
+	      bin->numentries++;
+	    }
+	  }
+	  pads = g_list_next(pads);
+	}
       }
       elements = g_list_next(elements);
     }
@@ -630,14 +671,14 @@ void gst_bin_iterate_func(GstBin *bin) {
   GList *entries;
   GstElement *entry;
 
-  g_print("gst_bin_iterate_func()\n");
+  DEBUG("gst_bin_iterate_func() in \"%s\"\n", gst_element_get_name(GST_ELEMENT(bin)));
 
   g_return_if_fail(bin != NULL);
   g_return_if_fail(GST_IS_BIN(bin));
   g_return_if_fail(GST_STATE(bin) == GST_STATE_PLAYING);
   g_return_if_fail(bin->numentries > 0);
 
-  g_print("GstBin: iterating\n");
+  DEBUG("GstBin: iterating\n");
 
   if (bin->need_cothreads) {
     // all we really have to do is switch to the first child

gst/gstbin.h

@@ -44,7 +44,7 @@ GstElementDetails gst_bin_details;
 #define GST_IS_BIN(obj) \
   (GTK_CHECK_TYPE((obj),GST_TYPE_BIN))
 #define GST_IS_BIN_CLASS(obj) \
-  (GTK_CHECK_CLASS_TYPE((klass),GST_TYPE_BIN)))
+  (GTK_CHECK_CLASS_TYPE((klass),GST_TYPE_BIN))
 
 typedef struct _GstBin GstBin;
 typedef struct _GstBinClass GstBinClass;
@@ -62,6 +62,7 @@ struct _GstBin {
   gint numentries;
 
   cothread_context *threadcontext;
+  gboolean use_cothreads;
 };
 
 struct _GstBinClass {
@@ -100,6 +101,9 @@ gboolean gst_bin_set_state_type(GstBin *bin,
 void gst_bin_iterate(GstBin *bin);
 void gst_bin_create_plan(GstBin *bin);
 
+// hack FIXME
+void gst_bin_use_cothreads(GstBin *bin, gboolean enabled);
+
 #ifdef __cplusplus
 }
 #endif /* __cplusplus */

gst/gstbuffer.c

@@ -63,9 +63,15 @@ GstBuffer *gst_buffer_new() {
   buffer->timestamp = 0;
   buffer->metas = NULL;
   buffer->parent = NULL;
+  buffer->pool = NULL;
   return buffer;
 }
 
+GstBuffer *gst_buffer_new_from_pool(GstBufferPool *pool)
+{
+  return gst_buffer_pool_new_buffer(pool);
+}
+
 
 /**
  * gst_buffer_create_sub:
@@ -101,7 +107,7 @@ GstBuffer *gst_buffer_create_sub(GstBuffer *parent,guint32 offset,guint32 size)
   // set the data pointer, size, offset, and maxsize
   buffer->data = parent->data + offset;
   buffer->size = size;
-  buffer->offset = offset;
+  buffer->offset = parent->offset + offset;
   buffer->maxsize = parent->size - offset;
 
   // again, for lack of better, copy parent's timestamp
@@ -114,10 +120,53 @@ GstBuffer *gst_buffer_create_sub(GstBuffer *parent,guint32 offset,guint32 size)
   buffer->parent = parent;
   gst_buffer_ref(parent);
 
+  buffer->pool = NULL;
   // return the new subbuffer
   return buffer;
 }
 
+/**
+ * gst_buffer_append_:
+ * @buffer: a buffer
+ * @append: the buffer to append
+ *
+ * Creates a new buffer by appending the data of eppend to the
+ *  existing data of buffer.
+ *
+ * Returns: new buffer
+ */
+GstBuffer *gst_buffer_append(GstBuffer *buffer, GstBuffer *append) {
+  guint size;
+  GstBuffer *newbuf;
+
+  g_return_val_if_fail(buffer != NULL, NULL);
+  g_return_val_if_fail(append != NULL, NULL);
+  g_return_val_if_fail(buffer->pool == NULL, NULL);
+
+  GST_BUFFER_LOCK(buffer);
+  // the buffer is not used by anyone else
+  if (GST_BUFFER_REFCOUNT(buffer) == 1 && buffer->parent == NULL) {
+    // save the old size
+    size = buffer->size;
+    buffer->size += append->size;
+    buffer->data = g_realloc(buffer->data, buffer->size);
+    memcpy(buffer->data + size, append->data, append->size);
+    GST_BUFFER_UNLOCK(buffer);
+  }
+  // the buffer is used, create a new one 
+  else {
+    newbuf = gst_buffer_new();
+    newbuf->size = buffer->size+append->size;
+    newbuf->data = g_malloc(newbuf->size);
+    memcpy(newbuf->data, buffer->data, buffer->size);
+    memcpy(newbuf->data+buffer->size, append->data, append->size);
+    GST_BUFFER_UNLOCK(buffer);
+    gst_buffer_unref(buffer);
+    buffer = newbuf;
+  }
+  return buffer;
+}
+
 /**
  * gst_buffer_destroy:
  * @buffer: the GstBuffer to destroy
@@ -131,10 +180,10 @@ void gst_buffer_destroy(GstBuffer *buffer) {
 
   if (buffer->parent != NULL) {
     DEBUG("BUF: freeing subbuffer %p\n",buffer);
-	}
+  }
   else {
     DEBUG("BUF: freeing buffer %p\n",buffer);
-	}
+  }
 
   // free the data only if there is some, DONTFREE isn't set, and not sub
   if (GST_BUFFER_DATA(buffer) &&
@@ -233,8 +282,16 @@ void gst_buffer_unref(GstBuffer *buffer) {
 #endif
 
   /* if we ended up with the refcount at zero, destroy the buffer */
-  if (zero)
+  if (zero) {
-    gst_buffer_destroy(buffer);
+    // if it came from a pool, give it back
+    if (buffer->pool != NULL) {
+      gst_buffer_pool_destroy_buffer(buffer->pool, buffer);
+      return;
+    }
+    else {
+      gst_buffer_destroy(buffer);
+    }
+  }
 }
 
 /**

gst/gstbuffer.h

@@ -66,6 +66,8 @@ typedef enum {
 
 typedef struct _GstBuffer GstBuffer;
 
+#include "gstbufferpool.h"
+
 struct _GstBuffer {
   /* locking */
   GMutex *lock;
@@ -100,16 +102,23 @@ struct _GstBuffer {
 
   /* subbuffer support, who's my parent? */
   GstBuffer *parent;
+
+  /* this is a pointer to the buffer pool (if any) */
+  GstBufferPool *pool;
 };
 
 /* initialisation */
 void _gst_buffer_initialize();
 /* creating a new buffer from scratch */
 GstBuffer *gst_buffer_new();
+GstBuffer *gst_buffer_new_from_pool(GstBufferPool *pool);
 
 /* creating a subbuffer */
 GstBuffer *gst_buffer_create_sub(GstBuffer *parent,guint32 offset,guint32 size);
 
+/* adding data to a buffer */
+GstBuffer *gst_buffer_append(GstBuffer *buffer, GstBuffer *append);
+
 /* refcounting */
 void gst_buffer_ref(GstBuffer *buffer);
 void gst_buffer_ref_by_count(GstBuffer *buffer,int count);

gst/gstelement.h

@@ -111,7 +111,7 @@ _gst_print_statename(flag)); \
   (GTK_CHECK_CLASS_TYPE((klass),GST_TYPE_ELEMENT))
 
 typedef enum {
-  GST_ELEMENT_MULTI_IN		= (1 << 16),
+  GST_ELEMENT_MULTI_IN		= (1 << 0),
 } GstElementFlags;
 
 #define GST_ELEMENT_IS_MULTI_IN(obj)	(GST_FLAGS(obj) & GST_ELEMENT_MULTI_IN)

gst/gstpad.c

@@ -195,17 +195,18 @@ void gst_pad_push(GstPad *pad,GstBuffer *buffer) {
 
   // first check to see if there's a push handler
   if (pad->pushfunc != NULL) {
-//    g_print("-- gst_pad_push(): putting buffer in pen and calling push handler\n");
+    //g_print("-- gst_pad_push(): putting buffer in pen and calling push handler\n");
     // put the buffer in peer's holding pen
     pad->peer->bufpen = buffer;
     // now inform the handler that the peer pad has something
     (pad->pushfunc)(pad->peer);
   // otherwise we assume we're chaining directly
   } else if (pad->chainfunc != NULL) {
+    //g_print("-- gst_pad_push(): calling chain handler\n");
     (pad->chainfunc)(pad->peer,buffer);
   // else we squawk
   } else {
-//    g_print("-- gst_pad_push(): houston, we have a problem, no way of talking to peer\n");
+    //g_print("-- gst_pad_push(): houston, we have a problem, no way of talking to peer\n");
   }
 
 #ifdef OLD_STUFF

gst/gstpipeline.c

@@ -111,31 +111,24 @@ static void gst_pipeline_prepare(GstPipeline *pipeline) {
 static GstElementStateReturn gst_pipeline_change_state(GstElement *element) {
   GstPipeline *pipeline;
 
-/*
   g_return_val_if_fail(GST_IS_PIPELINE(element), FALSE);
   pipeline = GST_PIPELINE(element);
 
-  switch (state) {
+
+  switch (GST_STATE_PENDING(pipeline)) {
     case GST_STATE_READY:
       // we need to set up internal state
       g_print("preparing pipeline \"%s\" for iterations:\n",
               gst_element_get_name(GST_ELEMENT(element)));
       gst_pipeline_prepare(pipeline);
       break;
-#if OLDSTATE
-    case ~GST_STATE_RUNNING:
-      // tear down the internal state
-      g_print("tearing down pipelines's iteration state\n");
-      break;
-#endif
     default:
       break;
   }
     
   if (GST_ELEMENT_CLASS(parent_class)->change_state)
-    return GST_ELEMENT_CLASS(parent_class)->change_state(element,state);
+    return GST_ELEMENT_CLASS(parent_class)->change_state(element);
-*/
+  return GST_STATE_SUCCESS;
-  return TRUE;
 }
 
 

gst/gstthread.c

@@ -170,17 +170,19 @@ static void gst_thread_prepare(GstThread *thread) {
   GstPad *pad, *peer;
   GstElement *outside;
 
-  thread->numentries = 0;
+  GST_BIN(thread)->numentries = 0;
 
   /* first we need to find all the entry points into the thread */
   elements = GST_BIN(thread)->children;
   while (elements) {
     element = GST_ELEMENT(elements->data);
+    gst_info("gstthread: element \"%s\" found in thread\n",
+             gst_element_get_name(GST_ELEMENT(element)));
     if (GST_IS_SRC(element)) {
       gst_info("gstthread: element \"%s\" is a source entry point for the thread\n",
                gst_element_get_name(GST_ELEMENT(element)));
-      thread->entries = g_list_prepend(thread->entries,element);
+      GST_BIN(thread)->entries = g_list_prepend(GST_BIN(thread)->entries,element);
-      thread->numentries++;
+      GST_BIN(thread)->numentries++;
     } else {
       /* go through the list of pads to see if there's a Connection */
       pads = gst_element_get_pad_list(element);
@@ -201,8 +203,8 @@ static void gst_thread_prepare(GstThread *thread) {
 for internal element \"%s\"\n",
                     gst_element_get_name(GST_ELEMENT(outside)),
                     gst_element_get_name(GST_ELEMENT(element)));
-            thread->entries = g_list_prepend(thread->entries,outside);
+            GST_BIN(thread)->entries = g_list_prepend(GST_BIN(thread)->entries,outside);
-            thread->numentries++;
+            GST_BIN(thread)->numentries++;
           }
         }
         pads = g_list_next(pads);
@@ -210,7 +212,7 @@ for internal element \"%s\"\n",
     }
     elements = g_list_next(elements);
   }
-  gst_info("gstthread: have %d entries into thread\n",thread->numentries);
+  gst_info("gstthread: have %d entries into thread\n",GST_BIN(thread)->numentries);
 }
 #endif
 
@@ -218,28 +220,36 @@ for internal element \"%s\"\n",
 static GstElementStateReturn gst_thread_change_state(GstElement *element) {
   GstThread *thread;
   gboolean stateset = TRUE;
+  gint pending;
 
-/*
   g_return_val_if_fail(GST_IS_THREAD(element), FALSE);
   thread = GST_THREAD(element);
 
-  if (GST_ELEMENT_CLASS(parent_class)->change_state)
+  gst_info("gstthread: thread \"%s\" change state %d\n",
-    stateset = GST_ELEMENT_CLASS(parent_class)->change_state(element,state);
+               gst_element_get_name(GST_ELEMENT(element)), GST_STATE_PENDING(element));
 
-  switch (state) {
+  pending = GST_STATE_PENDING(element);
+
+  if (GST_ELEMENT_CLASS(parent_class)->change_state)
+    stateset = GST_ELEMENT_CLASS(parent_class)->change_state(element);
+  
+  gst_info("gstthread: stateset %d %d\n", stateset, GST_STATE_PENDING(element));
+
+  switch (pending) {
     case GST_STATE_READY:
       if (!stateset) return FALSE;
       // we want to prepare our internal state for doing the iterations
-      gst_info("preparing thread \"%s\" for iterations:\n",
+      gst_info("gstthread: preparing thread \"%s\" for iterations:\n",
                gst_element_get_name(GST_ELEMENT(element)));
-//      gst_thread_prepare(thread);
+      //gst_thread_prepare(thread);
+      
       gst_bin_create_plan(GST_BIN(thread));
 //      if (thread->numentries == 0)
 //        return FALSE;
       // set the state to idle
       GST_FLAG_UNSET(thread,GST_THREAD_STATE_SPINNING);
       // create the thread if that's what we're supposed to do
-      gst_info("flags are 0x%08x\n",GST_FLAGS(thread));
+      gst_info("gstthread: flags are 0x%08x\n",GST_FLAGS(thread));
       if (GST_FLAG_IS_SET(thread,GST_THREAD_CREATE)) {
         gst_info("gstthread: starting thread \"%s\"\n",
                  gst_element_get_name(GST_ELEMENT(element)));
@@ -249,7 +259,7 @@ static GstElementStateReturn gst_thread_change_state(GstElement *element) {
         gst_info("gstthread: NOT starting thread \"%s\"\n",
                 gst_element_get_name(GST_ELEMENT(element)));
       }
-      return TRUE;
+      return GST_STATE_SUCCESS;
       break;
 #if OLDSTATE
     case ~GST_STATE_RUNNING:
@@ -259,7 +269,7 @@ static GstElementStateReturn gst_thread_change_state(GstElement *element) {
       gst_thread_signal_thread(thread);
       pthread_join(thread->thread_id,0);
       // tear down the internal state
-      gst_info("tearing down thread's iteration state\n");
+      gst_info("gstthread: tearing down thread's iteration state\n");
       // FIXME do stuff
       break;
 #endif
@@ -269,9 +279,8 @@ static GstElementStateReturn gst_thread_change_state(GstElement *element) {
               gst_element_get_name(GST_ELEMENT(element)));
       GST_FLAG_SET(thread,GST_THREAD_STATE_SPINNING);
       gst_thread_signal_thread(thread);
-      return TRUE;
       break;  
-    case ~GST_STATE_PLAYING:
+    case GST_STATE_PAUSED:
       gst_info("gstthread: stopping thread \"%s\"\n",
               gst_element_get_name(GST_ELEMENT(element)));
       GST_FLAG_UNSET(thread,GST_THREAD_STATE_SPINNING);
@@ -280,7 +289,6 @@ static GstElementStateReturn gst_thread_change_state(GstElement *element) {
     default:
       break;
   }
-*/
 
   return stateset;
 }

gst/meta/videoraw.h

@@ -65,7 +65,7 @@ struct _MetaVideoRaw {
   GstMeta meta;
 
   /* formatting information */
-  gint format;
+  GstColorSpaceType format;
   GdkVisual *visual;
   // dimensions of the video buffer
   gint width;

plugins/elements/gstaudiosink.c

@@ -190,7 +190,7 @@ void gst_audiosink_chain(GstPad *pad,GstBuffer *buf) {
   audiosink = GST_AUDIOSINK(pad->parent);
 //  g_return_if_fail(GST_FLAG_IS_SET(audiosink,GST_STATE_RUNNING));
 
-  if (in_flush = GST_BUFFER_FLAG_IS_SET(buf, GST_BUFFER_FLUSH)) {
+  if ((in_flush = GST_BUFFER_FLAG_IS_SET(buf, GST_BUFFER_FLUSH))) {
     DEBUG("audiosink: flush\n");
     ioctl(audiosink->fd,SNDCTL_DSP_RESET,0);
   }
@@ -217,20 +217,13 @@ void gst_audiosink_chain(GstPad *pad,GstBuffer *buf) {
     //g_print("audiosink: writing to soundcard\n");
     if (audiosink->fd > 2) {
       if (!audiosink->mute) {
-        //if (gst_clock_current_diff(audiosink->clock, GST_BUFFER_TIMESTAMP(buf)) > 500000) {
+        gst_clock_wait(audiosink->clock, GST_BUFFER_TIMESTAMP(buf), GST_OBJECT(audiosink));
-	//}
+        ioctl(audiosink->fd,SNDCTL_DSP_GETOSPACE,&ospace);
-	//else {
+        DEBUG("audiosink: (%d bytes buffer)\n", ospace.bytes);
-          gst_clock_wait(audiosink->clock, GST_BUFFER_TIMESTAMP(buf), GST_OBJECT(audiosink));
+        write(audiosink->fd,GST_BUFFER_DATA(buf),GST_BUFFER_SIZE(buf));
-          ioctl(audiosink->fd,SNDCTL_DSP_GETOSPACE,&ospace);
-          DEBUG("audiosink: (%d bytes buffer)\n", ospace.bytes);
-          write(audiosink->fd,GST_BUFFER_DATA(buf),GST_BUFFER_SIZE(buf));
-	//}
-          //gst_clock_set(audiosink->clock, GST_BUFFER_TIMESTAMP(buf));
-	//}
       }
     }
   }
-end:
   //g_print("a unref\n");
   gst_buffer_unref(buf);
   //g_print("a done\n");
@@ -314,6 +307,7 @@ static gboolean gst_audiosink_open_audio(GstAudioSink *sink) {
     if (sink->caps & DSP_CAP_TRIGGER)  g_print("audiosink:   Trigger\n");
     if (sink->caps & DSP_CAP_MMAP)     g_print("audiosink:   Direct access\n");
     g_print("audiosink: opened audio\n");
+    GST_FLAG_SET(sink,GST_AUDIOSINK_OPEN);
     return TRUE;
   }
 
@@ -325,10 +319,10 @@ static void gst_audiosink_close_audio(GstAudioSink *sink) {
 
   close(sink->fd);
   sink->fd = -1;
+  GST_FLAG_UNSET(sink,GST_AUDIOSINK_OPEN);
   g_print("audiosink: closed sound device\n");
 }
 
-
 static GstElementStateReturn gst_audiosink_change_state(GstElement *element) {
   g_return_val_if_fail(GST_IS_AUDIOSINK(element), FALSE);
 
@@ -346,5 +340,5 @@ static GstElementStateReturn gst_audiosink_change_state(GstElement *element) {
       
   if (GST_ELEMENT_CLASS(parent_class)->change_state)
     return GST_ELEMENT_CLASS(parent_class)->change_state(element);
-  return TRUE;
+  return GST_STATE_SUCCESS;
 }

plugins/elements/gstaudiosrc.c

@@ -241,7 +241,7 @@ static GstElementStateReturn gst_audiosrc_change_state(GstElement *element) {
 
   if (GST_ELEMENT_CLASS(parent_class)->change_state)
     return GST_ELEMENT_CLASS(parent_class)->change_state(element);
-  return TRUE;
+  return GST_STATE_SUCCESS;
 }
 
 static gboolean gst_audiosrc_open_audio(GstAudioSrc *src) {

plugins/elements/gstdisksrc.c

@@ -286,6 +286,8 @@ static void gst_disksrc_close_file(GstDiskSrc *src) {
 static GstElementStateReturn gst_disksrc_change_state(GstElement *element) {
   g_return_val_if_fail(GST_IS_DISKSRC(element),GST_STATE_FAILURE);
 
+  DEBUG("gstdisksrc: state pending %d\n", GST_STATE_PENDING(element));
+
   /* if going down into NULL state, close the file if it's open */
   if (GST_STATE_PENDING(element) == GST_STATE_NULL) {
     if (GST_FLAG_IS_SET(element,GST_DISKSRC_OPEN))

plugins/elements/gstpipefilter.c

@@ -138,9 +138,9 @@ static gboolean gst_pipefilter_read_and_push(GstPipefilter *pipefilter) {
   g_return_val_if_fail(GST_BUFFER_DATA(newbuf) != NULL, FALSE);
 
   /* read it in from the file */
-  DEBUG("attemting to read %d bytes\n", pipefilter->bytes_per_read);
+  DEBUG("attemting to read %ld bytes\n", pipefilter->bytes_per_read);
   readbytes = read(pipefilter->fdout[0],GST_BUFFER_DATA(newbuf),pipefilter->bytes_per_read);
-  DEBUG("read %d bytes\n", readbytes);
+  DEBUG("read %ld bytes\n", readbytes);
   if (readbytes < 0) {
     if (errno == EAGAIN) {
       DEBUG("no input yet\n");
@@ -185,9 +185,9 @@ void gst_pipefilter_chain(GstPad *pad,GstBuffer *buf) {
   data = GST_BUFFER_DATA(buf);
   size = GST_BUFFER_SIZE(buf);
 
-  DEBUG("attemting to write %d bytes\n", size);
+  DEBUG("attemting to write %ld bytes\n", size);
   writebytes = write(pipefilter->fdin[1],data,size);
-  DEBUG("written %d bytes\n", writebytes);
+  DEBUG("written %ld bytes\n", writebytes);
   if (writebytes < 0) {
     perror("write");
     gst_element_error(GST_ELEMENT(pipefilter),"writing");
@@ -296,12 +296,12 @@ static GstElementStateReturn gst_pipefilter_change_state(GstElement *element) {
   /* otherwise (READY or higher) we need to open the file */
   } else {
     if (!GST_FLAG_IS_SET(element,GST_PIPEFILTER_OPEN)) {
-      if (!gst_disksrc_open_file(GST_PIPEFILTER(element)))
+      if (!gst_pipefilter_open_file(GST_PIPEFILTER(element)))
         return GST_STATE_FAILURE;
     }
   }
       
   if (GST_ELEMENT_CLASS(parent_class)->change_state)
     return GST_ELEMENT_CLASS(parent_class)->change_state(element);
-  return TRUE;
+  return GST_STATE_SUCCESS;
 }