/* GStreamer
 * Copyright (C) 2008 Wim Taymans <wim.taymans at gmail.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#include <gst/gst.h>
#include <gst/rtsp-server/rtsp-server.h>

static GstElement *pipeline;
static GstElement *src;
static GstElement *sink;
static GstBus *bus;

static GMutex lock;
static GCond cond;
static gboolean pad_blocked = FALSE;
static gboolean got_async_done = FALSE;

static GMutex msg_lock;
static GCond msg_cond;
static gboolean msg_element_received = FALSE;
static gboolean msg_async_done_received = FALSE;

static gboolean got_flush_stop = FALSE;

static GMainLoop *loop;

static GstPadProbeReturn
blocking_probe (GstPad * pad, GstPadProbeInfo * info, gpointer unused)
{
  GstPadProbeReturn ret = GST_PAD_PROBE_OK;
  GstEvent *event;

  if (GST_PAD_PROBE_INFO_TYPE (info) & GST_PAD_PROBE_TYPE_EVENT_FLUSH) {
    /* flush events */
    event = gst_pad_probe_info_get_event (info);
    if (GST_EVENT_TYPE (event) == GST_EVENT_FLUSH_STOP)
      got_flush_stop = TRUE;

     ret = GST_PAD_PROBE_DROP;
  } else if (GST_PAD_PROBE_INFO_TYPE (info) & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM) {
    /* downstream events */
    event = gst_pad_probe_info_get_event (info);
    GST_ERROR ("got event downstream %s, dropping it", GST_EVENT_TYPE_NAME (event));
    ret = GST_PAD_PROBE_DROP;
  } else {
    GstPad *sinkpad;
    GstBuffer *buffer;

    g_assert (GST_PAD_PROBE_INFO_TYPE (info) & GST_PAD_PROBE_TYPE_BUFFER);
    buffer = gst_pad_probe_info_get_buffer (info);
    GST_ERROR (" ***** got buffer %" GST_TIME_FORMAT, GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)));

    if (!pad_blocked) {
      /* first preroll buffer, keep blocking */
      ret = GST_PAD_PROBE_OK;

      GST_ERROR ("linking pads");
      sinkpad = gst_element_get_static_pad (sink, "sink");
      g_assert (sinkpad);
      gst_pad_link (pad, sinkpad);
      gst_element_sync_state_with_parent (sink);

      GST_ERROR ("blocking on preroll buffer, posting PadPlocked msg");
      gst_element_post_message (pipeline,
          gst_message_new_element (GST_OBJECT_CAST (pipeline),
            gst_structure_new_empty ("PadBlocked")));
    } else if (got_flush_stop) {
      /* preroll buffer after seek, remove blocking, but we never end up
       * in this case !!! */
      ret = GST_PAD_PROBE_REMOVE;
    } else {
      GST_ERROR ("we should never end up here");
      ret = GST_PAD_PROBE_REMOVE;
    }
  }

  return ret;

}

static gboolean
bus_watch (GstBus * bus, GstMessage * message, GMainLoop * loop)
{
  switch (GST_MESSAGE_TYPE (message)) {
    case GST_MESSAGE_EOS:
      g_main_loop_quit (loop);
    case GST_MESSAGE_ELEMENT:
    {
      const GstStructure *s;
      s = gst_message_get_structure (message);
      if (gst_structure_has_name (s, "PadBlocked")) {
        GST_ERROR ("pad is blocked");
        g_mutex_lock (&lock);
        pad_blocked = TRUE;
        g_cond_broadcast (&cond);
        g_mutex_unlock (&lock);
      }
      break;
    }
    case GST_MESSAGE_ASYNC_DONE:
    {
      GST_ERROR ("got async-done");
      g_mutex_lock (&lock);
      got_async_done = TRUE;
      g_cond_broadcast (&cond);
      g_mutex_unlock (&lock);
      break;
    }

    default:
      break;
  }
  return TRUE;
}

static void
message_func (GstBus * bus, GstMessage * msg, GMainLoop * main_loop)
{
  GST_ERROR ("got signal : %s", GST_MESSAGE_TYPE_NAME (msg));

  /* make sure that the main thread is informed about the arrived messages */
  if (GST_MESSAGE_TYPE (msg) == GST_MESSAGE_ELEMENT) {
    g_mutex_lock (&msg_lock);
    msg_element_received = TRUE;
    g_cond_broadcast (&msg_cond);
    g_mutex_unlock (&msg_lock);
  } else if (GST_MESSAGE_TYPE (msg) == GST_MESSAGE_ASYNC_DONE) {
    g_mutex_lock (&msg_lock);
    msg_async_done_received = TRUE;
    g_cond_broadcast (&msg_cond);
    g_mutex_unlock (&msg_lock);
  } else {
    g_assert_not_reached ();
  }
}

/* application thread */
static gpointer
start_pipeline (gpointer unused)
{
  GstPad *pad;

  pipeline = gst_pipeline_new ("pipeline");
  src = gst_element_factory_make ("fakesrc", "fakesrc");
  sink = gst_element_factory_make ("fakesink", "fakesink");

  g_object_set (G_OBJECT (src), "is-live", FALSE, NULL);
  g_object_set (G_OBJECT (src), "num-buffers", 3, NULL);
  g_object_set (G_OBJECT (src), "do-timestamp", TRUE, NULL);
  gst_bin_add_many (GST_BIN (pipeline), src, sink, NULL);

  pad = gst_element_get_static_pad (src, "src");

  gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM |
      GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM | GST_PAD_PROBE_TYPE_EVENT_FLUSH,
      blocking_probe, NULL, NULL);

  bus = gst_element_get_bus (pipeline);
  gst_bus_add_watch (bus, (GstBusFunc) bus_watch, loop);

  gst_bus_enable_sync_message_emission (bus);
  g_signal_connect (bus, "sync-message::element",
      G_CALLBACK (message_func), NULL);
  g_signal_connect (bus, "sync-message::async-done",
      G_CALLBACK (message_func), NULL);
  gst_object_unref (bus);

  gst_element_set_state (pipeline, GST_STATE_PAUSED);

  /* wait until the source pad is blocked */
  g_mutex_lock (&lock);
  while (!pad_blocked) {
    GST_ERROR ("waiting for pad to be blocked");
    g_cond_wait (&cond, &lock);
  }
  g_mutex_unlock (&lock);

  /* we can seek now */
  gst_element_seek (pipeline, 1.0, GST_FORMAT_TIME,
        GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_KEY_UNIT,
        GST_SEEK_TYPE_NONE, 0,
        GST_SEEK_TYPE_NONE, -1);

  /* wait for seek operation to complete */
  g_mutex_lock (&lock);
  while (!got_async_done) {
    GST_ERROR ("waiting for async-done");
    g_cond_wait (&cond, &lock);
  }
  g_mutex_unlock (&lock);

  return NULL;
}

int
main (int argc, char *argv[])
{
  GThread *thread;

  pad_blocked = FALSE;

  gst_init (&argc, &argv);
  loop = g_main_loop_new (NULL, FALSE);
  GST_ERROR ("main thread started");

  g_mutex_init (&lock);
  g_cond_init (&cond);

  g_mutex_init (&msg_lock);
  g_cond_init (&msg_cond);

  thread = g_thread_new (NULL, start_pipeline, NULL);

  g_mutex_lock (&msg_lock);
  while (!msg_element_received) {
    GST_DEBUG ("waiting for pads to be blocked");
    g_cond_wait (&msg_cond, &msg_lock);
  }
  g_mutex_unlock (&msg_lock);

  /* dispatch events */
  while (g_main_context_iteration (NULL, FALSE)) {
    //g_print ("iteration\n");
  }

  g_mutex_lock (&msg_lock);
  while (!msg_async_done_received) {
    GST_DEBUG ("waiting for pads to be blocked");
    g_cond_wait (&msg_cond, &msg_lock);
  }
  g_mutex_unlock (&msg_lock);

  /* dispatch events */
  while (g_main_context_iteration (NULL, FALSE)) {
    //g_print ("iteration\n");
  }

  g_main_loop_run (loop);
  g_thread_join (thread);

  return 0;

}
