<commit_before>/*
 * DMAPGstInputStream class: Open a URI using dmap_gst_input_stream_new ().
 * Data is decoded using GStreamer and is then made available by the class's
 * read operations.
 *
 * Copyright (C) 2008 <NAME> <<EMAIL>>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser 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 <string.h>
#include <gst/gst.h>
#include <gst/app/gstappsink.h>

#include "dmap-gst-input-stream.h"
#include "dmap-gst-mp3-input-stream.h"
#include "dmap-gst-wav-input-stream.h"
#include "dmap-gst-qt-input-stream.h"
#include "gst-util.h"

#define GST_APP_MAX_BUFFERS 1024
#define DECODED_BUFFER_SIZE 1024 * 128
#define QUEUE_PUSH_WAIT_SECONDS 10
#define QUEUE_POP_WAIT_SECONDS 1

struct dmap_gst_format
{
	gchar *id;		/* E.g., used as command line arguments. */
	gchar *extension;	/* E.g., iTunes uses URI extension to
				 * determine stream format.
				 */
};

/* NOTE: Roku clients require lower case extension. */
static const struct dmap_gst_format dmap_gst_formats[] = {
	{"raw", "raw"},
	{"wav16", "wav"},
	{"mp3", "mp3"},
	{NULL, NULL}
};

struct DMAPGstInputStreamPrivate
{
	GQueue *buffer;
	gsize read_request;	/* Size of data asked for */
	gsize write_request;	/* Number of bytes that must be read
				 * to make room for write */
	GCond buffer_read_ready;	/* Signals when buffer >= read_req. */
	GCond buffer_write_ready;	/* Signals when buffer not full. */
	GMutex buffer_mutex;	/* Protects buffer and read_request */
	gboolean buffer_closed;	/* May close before decoding complete */
};

#define DMAP_GST_INPUT_STREAM_GET_PRIVATE(o) (G_TYPE_INSTANCE_GET_PRIVATE ((o), \
					   DMAP_TYPE_GST_INPUT_STREAM, \
					   DMAPGstInputStreamPrivate))

static goffset
dmap_gst_input_stream_tell (GSeekable * seekable)
{
	/* FIXME: implement return current position in stream. */
	g_error ("Not implemented");
	return 0;
}

static gboolean
dmap_gst_input_stream_can_seek (GSeekable * seekable)
{
	return TRUE;
}

static gboolean
dmap_gst_input_stream_seek (GSeekable * seekable,
			    goffset offset,
			    GSeekType type,
			    GCancellable * cacellable, GError ** error)
{
	// FIXME: implement: DMAPGstInputStream *stream;
	// FIXME: implement: goffset absolute;

	// FIXME: implement: stream = DMAP_GST_INPUT_STREAM (seekable);

	switch (type) {
		/* FIXME: implement:
		 * case G_SEEK_CUR:
		 *      absolute = stream->priv->pos + offset;
		 *      break;
		 *
		 * case G_SEEK_END:
		 *      absolute = stream->priv->len + offset;
		 *      break;
		 */

	case G_SEEK_SET:
		// FIXME: implement: absolute = offset;
		break;

	default:
		g_set_error (error,
			     G_IO_ERROR,
			     G_IO_ERROR_INVALID_ARGUMENT,
			     "Invalid GSeekType supplied");

		return FALSE;
	}

	/* FIXME: implement:
	 * if (absolute < 0 || absolute > stream->priv->len) {
	 *      g_set_error_literal (error,
	 *              G_IO_ERROR,
	 *              G_IO_ERROR_INVALID_ARGUMENT,
	 *              _("Invalid seek request"));
	 *      return FALSE;
	 * }
	 */

	/* FIXME:
	 * if (! gst_element_seek_simple (DMAP_GST_INPUT_STREAM (seekable)->priv->pipeline,
	 * GST_FORMAT_BYTES,
	 * GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_KEY_UNIT,
	 * absolute)) {
	 * g_set_error (error,
	 * G_IO_ERROR,
	 * G_IO_ERROR_FAILED,
	 * "Seek failed");
	 * return FALSE;
	 * }
	 */

	return TRUE;
}

static gboolean
dmap_gst_input_stream_can_truncate (GSeekable * seekable)
{
	return FALSE;
}

static gboolean
dmap_gst_input_stream_truncate (GSeekable * seekable,
				goffset offset,
				GCancellable * cancellable, GError ** error)
{
	g_set_error (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
		     "Cannot truncate DMAPGstInputStream");
	return FALSE;
}

static void
dmap_gst_input_stream_seekable_iface_init (GSeekableIface * iface)
{
	iface->tell = dmap_gst_input_stream_tell;
	iface->can_seek = dmap_gst_input_stream_can_seek;
	iface->seek = dmap_gst_input_stream_seek;
	iface->can_truncate = dmap_gst_input_stream_can_truncate;
	iface->truncate_fn = dmap_gst_input_stream_truncate;
}

void
dmap_gst_input_stream_new_buffer_cb (GstElement * element,
				     DMAPGstInputStream * stream)
{
	gsize i;
	guint8 *ptr;
	gint64 end_time;
	GstSample *sample = NULL;
	GstBuffer *buffer = NULL;
	GstMemory *memory = NULL;
	GstMapInfo info;

	/* FIXME: Is this necessary? I am trying to protect against this
	 * thread manipulating data after the pipeline has been destroyed.
	 * see also dmap_gst_input_stream_close ().
	 */
	g_mutex_lock (&stream->priv->buffer_mutex);

	if (stream->priv->buffer_closed) {
		g_warning ("Unread data");
		goto _return;
	}

	end_time = end_time = g_get_monotonic_time () + QUEUE_PUSH_WAIT_SECONDS * G_TIME_SPAN_SECOND;

	sample = gst_app_sink_pull_sample (GST_APP_SINK (element));
	if (NULL == sample) {
		g_warning ("Error getting GStreamer sample");
		goto _return;
	}

	buffer = gst_sample_get_buffer (sample);
	if (NULL == buffer) {
		g_warning ("Error getting GStreamer buffer");
		goto _return;
	}

	memory = gst_buffer_get_memory (buffer, 0);
	if (NULL == memory) {
		g_warning ("Error getting GStreamer memory");
		goto _return;
	}

	if (gst_memory_map (memory, &info, GST_MAP_READ) == FALSE) {
		g_warning ("Error mapping GStreamer memory");
		goto _return;
	}

	/* FIXME: this actually allows buffer to grow larger than max. */
	if (g_queue_get_length (stream->priv->buffer) +
	    info.size > DECODED_BUFFER_SIZE) {
		stream->priv->write_request = info.size;
		if (!g_cond_wait_until (&stream->priv->buffer_write_ready,
		                        &stream->priv->buffer_mutex, end_time)) {<commit_msg>Remove double "end_time = "<commit_after>/*
 * DMAPGstInputStream class: Open a URI using dmap_gst_input_stream_new ().
 * Data is decoded using GStreamer and is then made available by the class's
 * read operations.
 *
 * Copyright (C) 2008 W. <NAME> <<EMAIL>>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser 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 <string.h>
#include <gst/gst.h>
#include <gst/app/gstappsink.h>

#include "dmap-gst-input-stream.h"
#include "dmap-gst-mp3-input-stream.h"
#include "dmap-gst-wav-input-stream.h"
#include "dmap-gst-qt-input-stream.h"
#include "gst-util.h"

#define GST_APP_MAX_BUFFERS 1024
#define DECODED_BUFFER_SIZE 1024 * 128
#define QUEUE_PUSH_WAIT_SECONDS 10
#define QUEUE_POP_WAIT_SECONDS 1

struct dmap_gst_format
{
	gchar *id;		/* E.g., used as command line arguments. */
	gchar *extension;	/* E.g., iTunes uses URI extension to
				 * determine stream format.
				 */
};

/* NOTE: Roku clients require lower case extension. */
static const struct dmap_gst_format dmap_gst_formats[] = {
	{"raw", "raw"},
	{"wav16", "wav"},
	{"mp3", "mp3"},
	{NULL, NULL}
};

struct DMAPGstInputStreamPrivate
{
	GQueue *buffer;
	gsize read_request;	/* Size of data asked for */
	gsize write_request;	/* Number of bytes that must be read
				 * to make room for write */
	GCond buffer_read_ready;	/* Signals when buffer >= read_req. */
	GCond buffer_write_ready;	/* Signals when buffer not full. */
	GMutex buffer_mutex;	/* Protects buffer and read_request */
	gboolean buffer_closed;	/* May close before decoding complete */
};

#define DMAP_GST_INPUT_STREAM_GET_PRIVATE(o) (G_TYPE_INSTANCE_GET_PRIVATE ((o), \
					   DMAP_TYPE_GST_INPUT_STREAM, \
					   DMAPGstInputStreamPrivate))

static goffset
dmap_gst_input_stream_tell (GSeekable * seekable)
{
	/* FIXME: implement return current position in stream. */
	g_error ("Not implemented");
	return 0;
}

static gboolean
dmap_gst_input_stream_can_seek (GSeekable * seekable)
{
	return TRUE;
}

static gboolean
dmap_gst_input_stream_seek (GSeekable * seekable,
			    goffset offset,
			    GSeekType type,
			    GCancellable * cacellable, GError ** error)
{
	// FIXME: implement: DMAPGstInputStream *stream;
	// FIXME: implement: goffset absolute;

	// FIXME: implement: stream = DMAP_GST_INPUT_STREAM (seekable);

	switch (type) {
		/* FIXME: implement:
		 * case G_SEEK_CUR:
		 *      absolute = stream->priv->pos + offset;
		 *      break;
		 *
		 * case G_SEEK_END:
		 *      absolute = stream->priv->len + offset;
		 *      break;
		 */

	case G_SEEK_SET:
		// FIXME: implement: absolute = offset;
		break;

	default:
		g_set_error (error,
			     G_IO_ERROR,
			     G_IO_ERROR_INVALID_ARGUMENT,
			     "Invalid GSeekType supplied");

		return FALSE;
	}

	/* FIXME: implement:
	 * if (absolute < 0 || absolute > stream->priv->len) {
	 *      g_set_error_literal (error,
	 *              G_IO_ERROR,
	 *              G_IO_ERROR_INVALID_ARGUMENT,
	 *              _("Invalid seek request"));
	 *      return FALSE;
	 * }
	 */

	/* FIXME:
	 * if (! gst_element_seek_simple (DMAP_GST_INPUT_STREAM (seekable)->priv->pipeline,
	 * GST_FORMAT_BYTES,
	 * GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_KEY_UNIT,
	 * absolute)) {
	 * g_set_error (error,
	 * G_IO_ERROR,
	 * G_IO_ERROR_FAILED,
	 * "Seek failed");
	 * return FALSE;
	 * }
	 */

	return TRUE;
}

static gboolean
dmap_gst_input_stream_can_truncate (GSeekable * seekable)
{
	return FALSE;
}

static gboolean
dmap_gst_input_stream_truncate (GSeekable * seekable,
				goffset offset,
				GCancellable * cancellable, GError ** error)
{
	g_set_error (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
		     "Cannot truncate DMAPGstInputStream");
	return FALSE;
}

static void
dmap_gst_input_stream_seekable_iface_init (GSeekableIface * iface)
{
	iface->tell = dmap_gst_input_stream_tell;
	iface->can_seek = dmap_gst_input_stream_can_seek;
	iface->seek = dmap_gst_input_stream_seek;
	iface->can_truncate = dmap_gst_input_stream_can_truncate;
	iface->truncate_fn = dmap_gst_input_stream_truncate;
}

void
dmap_gst_input_stream_new_buffer_cb (GstElement * element,
				     DMAPGstInputStream * stream)
{
	gsize i;
	guint8 *ptr;
	gint64 end_time;
	GstSample *sample = NULL;
	GstBuffer *buffer = NULL;
	GstMemory *memory = NULL;
	GstMapInfo info;

	/* FIXME: Is this necessary? I am trying to protect against this
	 * thread manipulating data after the pipeline has been destroyed.
	 * see also dmap_gst_input_stream_close ().
	 */
	g_mutex_lock (&stream->priv->buffer_mutex);

	if (stream->priv->buffer_closed) {
		g_warning ("Unread data");
		goto _return;
	}

	end_time = g_get_monotonic_time () + QUEUE_PUSH_WAIT_SECONDS * G_TIME_SPAN_SECOND;

	sample = gst_app_sink_pull_sample (GST_APP_SINK (element));
	if (NULL == sample) {
		g_warning ("Error getting GStreamer sample");
		goto _return;
	}

	buffer = gst_sample_get_buffer (sample);
	if (NULL == buffer) {
		g_warning ("Error getting GStreamer buffer");
		goto _return;
	}

	memory = gst_buffer_get_memory (buffer, 0);
	if (NULL == memory) {
		g_warning ("Error getting GStreamer memory");
		goto _return;
	}

	if (gst_memory_map (memory, &info, GST_MAP_READ) == FALSE) {
		g_warning ("Error mapping GStreamer memory");
		goto _return;
	}

	/* FIXME: this actually allows buffer to grow larger than max. */
	if (g_queue_get_length (stream->priv->buffer) +
	    info.size > DECODED_BUFFER_SIZE) {
		stream->priv->write_request = info.size;
		if (!g_cond_wait_until (&stream->priv->buffer_write_ready,
		                        &stream->priv->buffer_mutex, end_time)) {
{'fraction_non_alphanumeric': 0.10362903225806451, 'fraction_numerical': 0.0067741935483870966, 'mean_word_length': 3.8650451157316597, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5510995', 'n_tokens_mistral': 4627, 'n_tokens_neox': 4298, 'n_words': 1139}