mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-01 22:58:50 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

More work on implementing remote protocol

Browse source 
Rework things so that we negotiate buffer pools beforehand and only pass
buffer ids around We can then remove the refcount of buffers, events and
commands.
More work on buffer reuse
Use the node state changes to trigger the next step in the configuration
sequence.
Move most of the client-node to a plugin
Do buffer allocation in the port link.
This commit is contained in:
Wim Taymans 2016-08-02 16:34:44 +02:00
parent 05829f33e6
commit 3ace7e9648
36 changed files with 1780 additions and 1450 deletions
Download patch file Download diff file Expand all files Collapse all files

91
spa/plugins/alsa/alsa-sink.c
View file

@ -89,7 +89,9 @@ struct _SpaALSASink {
SpaPortInfo info;
SpaPortStatus status;
SpaBuffer *input_buffer;
SpaBuffer *buffers;
unsigned int n_buffers;
uint32_t input_buffer;
ALSABuffer buffer;
};
@ -233,13 +235,13 @@ spa_alsa_sink_node_send_command (SpaNode *node,
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STARTED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = ≻
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_STREAMING;
this->event_cb (node, &event, this->user_data);
}
@ -249,13 +251,13 @@ spa_alsa_sink_node_send_command (SpaNode *node,
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STOPPED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = ≻
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_STREAMING;
this->event_cb (node, &event, this->user_data);
}
@ -465,6 +467,36 @@ spa_alsa_sink_node_port_set_props (SpaNode *node,
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_alsa_sink_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_alsa_sink_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_alsa_sink_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id,
off_t offset,
size_t size)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_alsa_sink_node_port_get_status (SpaNode *node,
uint32_t port_id,
@ -485,32 +517,6 @@ spa_alsa_sink_node_port_get_status (SpaNode *node,
return SPA_RESULT_OK;
}
static SpaResult
spa_alsa_sink_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_alsa_sink_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaBuffer *
find_buffer (SpaALSASink *this, uint32_t id)
{
return NULL;
}
static SpaResult
spa_alsa_sink_node_port_push_input (SpaNode *node,
unsigned int n_info,
@ -526,29 +532,25 @@ spa_alsa_sink_node_port_push_input (SpaNode *node,
this = (SpaALSASink *) node->handle;
for (i = 0; i < n_info; i++) {
SpaBuffer *buffer;
if (info[i].port_id != 0) {
info[i].status = SPA_RESULT_INVALID_PORT;
have_error = true;
continue;
}
buffer = find_buffer (this, info[i].id);
if (buffer != NULL) {
if (info[i].buffer_id != SPA_ID_INVALID) {
if (!this->have_format) {
info[i].status = SPA_RESULT_NO_FORMAT;
have_error = true;
continue;
}
if (this->input_buffer != NULL) {
if (this->input_buffer != -1) {
info[i].status = SPA_RESULT_HAVE_ENOUGH_INPUT;
have_enough = true;
continue;
}
this->input_buffer = spa_buffer_ref (buffer);
this->input_buffer = info[i].buffer_id;
}
info[i].status = SPA_RESULT_OK;
}
@ -587,6 +589,7 @@ static const SpaNode alsasink_node = {
spa_alsa_sink_node_port_set_props,
spa_alsa_sink_node_port_use_buffers,
spa_alsa_sink_node_port_alloc_buffers,
spa_alsa_sink_node_port_reuse_buffer,
spa_alsa_sink_node_port_get_status,
spa_alsa_sink_node_port_push_input,
spa_alsa_sink_node_port_pull_output,

43
spa/plugins/alsa/alsa-utils.c
View file

@ -205,40 +205,18 @@ static void
pull_input (SpaALSASink *this, void *data, snd_pcm_uframes_t frames)
{
SpaEvent event;
ALSABuffer *buffer = &this->buffer;
SpaEventPullInput pi;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_PULL_INPUT;
event.port_id = 0;
event.size = frames * sizeof (uint16_t) * 2;
event.data = buffer;
event.data = &pi;
buffer->buffer.refcount = 1;
buffer->buffer.notify = NULL;
buffer->buffer.size = frames * sizeof (uint16_t) * 2;
buffer->buffer.n_metas = 1;
buffer->buffer.metas = buffer->meta;
buffer->buffer.n_datas = 1;
buffer->buffer.datas = buffer->data;
pi.buffer_id = this->buffer.buffer.id;
pi.offset = 0;
pi.size = frames * sizeof (uint16_t) * 2;
buffer->header.flags = 0;
buffer->header.seq = 0;
buffer->header.pts = 0;
buffer->header.dts_offset = 0;
buffer->meta[0].type = SPA_META_TYPE_HEADER;
buffer->meta[0].data = &buffer->header;
buffer->meta[0].size = sizeof (buffer->header);
buffer->data[0].type = SPA_DATA_TYPE_MEMPTR;
buffer->data[0].ptr = data;
buffer->data[0].ptr_type = "sysmem";
buffer->data[0].size = frames * sizeof (uint16_t) * 2;
this->event_cb (&this->node, &event,this->user_data);
spa_buffer_unref ((SpaBuffer *)event.data);
this->event_cb (&this->node, &event, this->user_data);
}
static int
@ -273,11 +251,10 @@ mmap_write (SpaALSASink *this)
frames);
if (this->input_buffer) {
if (this->input_buffer != &this->buffer.buffer) {
if (this->input_buffer != this->buffer.buffer.id) {
/* FIXME, copy input */
}
spa_buffer_unref (this->input_buffer);
this->input_buffer = NULL;
this->input_buffer = -1;
}
commitres = snd_pcm_mmap_commit (handle, offset, frames);
@ -366,8 +343,6 @@ spa_alsa_start (SpaALSASink *this)
return err;
}
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_ADD_POLL;
event.port_id = 0;
event.data = &state->poll;
@ -394,8 +369,6 @@ spa_alsa_stop (SpaALSASink *this)
snd_pcm_drop (state->handle);
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_REMOVE_POLL;
event.port_id = 0;
event.data = &state->poll;

110
spa/plugins/audiomixer/audiomixer.c
View file

@ -148,13 +148,13 @@ spa_audiomixer_node_send_command (SpaNode *node,
case SPA_COMMAND_START:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STARTED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_STREAMING;
this->event_cb (node, &event, this->user_data);
}
@ -163,13 +163,13 @@ spa_audiomixer_node_send_command (SpaNode *node,
case SPA_COMMAND_STOP:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STOPPED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_PAUSED;
this->event_cb (node, &event, this->user_data);
}
@ -302,7 +302,6 @@ spa_audiomixer_node_remove_port (SpaNode *node,
this->ports[port_id].valid = false;
this->port_count--;
if (this->ports[port_id].buffer) {
spa_buffer_unref (this->ports[port_id].buffer);
this->ports[port_id].buffer = NULL;
this->port_queued--;
}
@ -446,6 +445,36 @@ spa_audiomixer_node_port_set_props (SpaNode *node,
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiomixer_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiomixer_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiomixer_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id,
off_t offset,
size_t size)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiomixer_node_port_get_status (SpaNode *node,
uint32_t port_id,
@ -471,26 +500,6 @@ spa_audiomixer_node_port_get_status (SpaNode *node,
return SPA_RESULT_OK;
}
static SpaResult
spa_audiomixer_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiomixer_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiomixer_node_port_push_input (SpaNode *node,
unsigned int n_info,
@ -519,7 +528,7 @@ spa_audiomixer_node_port_push_input (SpaNode *node,
continue;
}
port = &this->ports[idx];
buffer = port->buffers[info[i].id];
buffer = port->buffers[info[i].buffer_id];
if (buffer == NULL) {
info[i].status = SPA_RESULT_INVALID_ARGUMENTS;
@ -539,7 +548,7 @@ spa_audiomixer_node_port_push_input (SpaNode *node,
have_error = true;
continue;
}
this->ports[idx].buffer = spa_buffer_ref (buffer);
this->ports[idx].buffer = buffer;
this->ports[idx].buffer_queued = buffer->size;
this->ports[idx].buffer_index = 0;
this->ports[idx].buffer_offset = 0;
@ -562,35 +571,14 @@ pull_port (SpaAudioMixer *this, uint32_t port_id, SpaOutputInfo *info, size_t pu
{
SpaEvent event;
MixerBuffer *buffer = &this->ports[port_id].mix;
SpaEventPullInput pi;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_PULL_INPUT;
event.port_id = port_id;
event.data = buffer;
buffer->buffer.refcount = 1;
buffer->buffer.notify = NULL;
buffer->buffer.size = pull_size;
buffer->buffer.n_metas = 1;
buffer->buffer.metas = buffer->meta;
buffer->buffer.n_datas = 1;
buffer->buffer.datas = buffer->data;
buffer->header.flags = 0;
buffer->header.seq = 0;
buffer->header.pts = 0;
buffer->header.dts_offset = 0;
buffer->meta[0].type = SPA_META_TYPE_HEADER;
buffer->meta[0].data = &buffer->header;
buffer->meta[0].size = sizeof (buffer->header);
buffer->data[0].type = SPA_DATA_TYPE_MEMPTR;
buffer->data[0].ptr = buffer->samples;
buffer->data[0].ptr_type = "sysmem";
buffer->data[0].size = pull_size;
event.data = &pi;
pi.buffer_id = buffer->buffer.id;
pi.offset = 0;
pi.size = pull_size;
this->event_cb (&this->node, &event, this->user_data);
}
@ -622,7 +610,6 @@ add_port_data (SpaAudioMixer *this, SpaBuffer *out, SpaAudioMixerPort *port)
if ((is -= chunk) == 0) {
if (++port->buffer_index == port->buffer->n_datas) {
spa_buffer_unref (port->buffer);
port->buffer = NULL;
port->status.flags = SPA_PORT_STATUS_FLAG_NEED_INPUT;
this->ports[0].status.flags &= ~SPA_PORT_STATUS_FLAG_HAVE_OUTPUT;
@ -677,7 +664,7 @@ mix_data (SpaAudioMixer *this, SpaOutputInfo *info)
return SPA_RESULT_NEED_MORE_INPUT;
buf = this->ports[min_port].buffer;
info->id = buf->id;
info->buffer_id = buf->id;
this->ports[min_port].buffer = NULL;
this->ports[min_port].status.flags = SPA_PORT_STATUS_FLAG_NEED_INPUT;
this->ports[0].status.flags &= ~SPA_PORT_STATUS_FLAG_HAVE_OUTPUT;
@ -757,6 +744,7 @@ static const SpaNode audiomixer_node = {
spa_audiomixer_node_port_set_props,
spa_audiomixer_node_port_use_buffers,
spa_audiomixer_node_port_alloc_buffers,
spa_audiomixer_node_port_reuse_buffer,
spa_audiomixer_node_port_get_status,
spa_audiomixer_node_port_push_input,
spa_audiomixer_node_port_pull_output,

71
spa/plugins/audiotestsrc/audiotestsrc.c
View file

@ -179,13 +179,13 @@ spa_audiotestsrc_node_send_command (SpaNode *node,
case SPA_COMMAND_START:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STARTED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_STREAMING;
this->event_cb (node, &event, this->user_data);
}
@ -194,13 +194,13 @@ spa_audiotestsrc_node_send_command (SpaNode *node,
case SPA_COMMAND_STOP:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STOPPED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_PAUSED;
this->event_cb (node, &event, this->user_data);
}
@ -407,6 +407,36 @@ spa_audiotestsrc_node_port_set_props (SpaNode *node,
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiotestsrc_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiotestsrc_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiotestsrc_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id,
off_t offset,
size_t size)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiotestsrc_node_port_get_status (SpaNode *node,
uint32_t port_id,
@ -430,26 +460,6 @@ spa_audiotestsrc_node_port_get_status (SpaNode *node,
return SPA_RESULT_OK;
}
static SpaResult
spa_audiotestsrc_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiotestsrc_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_audiotestsrc_node_port_push_input (SpaNode *node,
@ -528,6 +538,7 @@ static const SpaNode audiotestsrc_node = {
spa_audiotestsrc_node_port_set_props,
spa_audiotestsrc_node_port_use_buffers,
spa_audiotestsrc_node_port_alloc_buffers,
spa_audiotestsrc_node_port_reuse_buffer,
spa_audiotestsrc_node_port_get_status,
spa_audiotestsrc_node_port_push_input,
spa_audiotestsrc_node_port_pull_output,

90
spa/plugins/ffmpeg/ffmpeg-dec.c
View file

@ -145,13 +145,13 @@ spa_ffmpeg_dec_node_send_command (SpaNode *node,
case SPA_COMMAND_START:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STARTED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_STREAMING;
this->event_cb (node, &event, this->user_data);
}
@ -159,13 +159,13 @@ spa_ffmpeg_dec_node_send_command (SpaNode *node,
case SPA_COMMAND_STOP:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STOPPED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_PAUSED;
this->event_cb (node, &event, this->user_data);
}
@ -399,6 +399,48 @@ spa_ffmpeg_dec_node_port_set_props (SpaNode *node,
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_dec_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
if (!IS_VALID_PORT (port_id))
return SPA_RESULT_INVALID_PORT;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_dec_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_dec_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id,
off_t offse,
size_t size)
{
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
if (!IS_VALID_PORT (port_id))
return SPA_RESULT_INVALID_PORT;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_dec_node_port_get_status (SpaNode *node,
uint32_t port_id,
@ -419,33 +461,6 @@ spa_ffmpeg_dec_node_port_get_status (SpaNode *node,
return SPA_RESULT_OK;
}
static SpaResult
spa_ffmpeg_dec_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
if (!IS_VALID_PORT (port_id))
return SPA_RESULT_INVALID_PORT;
return SPA_RESULT_OK;
}
static SpaResult
spa_ffmpeg_dec_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_dec_node_port_push_input (SpaNode *node,
unsigned int n_info,
@ -518,6 +533,7 @@ static const SpaNode ffmpeg_dec_node = {
spa_ffmpeg_dec_node_port_set_props,
spa_ffmpeg_dec_node_port_use_buffers,
spa_ffmpeg_dec_node_port_alloc_buffers,
spa_ffmpeg_dec_node_port_reuse_buffer,
spa_ffmpeg_dec_node_port_get_status,
spa_ffmpeg_dec_node_port_push_input,
spa_ffmpeg_dec_node_port_pull_output,

90
spa/plugins/ffmpeg/ffmpeg-enc.c
View file

@ -145,13 +145,13 @@ spa_ffmpeg_enc_node_send_command (SpaNode *node,
case SPA_COMMAND_START:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STARTED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_STREAMING;
this->event_cb (node, &event, this->user_data);
}
@ -159,13 +159,13 @@ spa_ffmpeg_enc_node_send_command (SpaNode *node,
case SPA_COMMAND_STOP:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STOPPED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_PAUSED;
this->event_cb (node, &event, this->user_data);
}
@ -399,6 +399,48 @@ spa_ffmpeg_enc_node_port_set_props (SpaNode *node,
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_enc_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
if (!IS_VALID_PORT (port_id))
return SPA_RESULT_INVALID_PORT;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_enc_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_enc_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id,
off_t offset,
size_t size)
{
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
if (!IS_VALID_PORT (port_id))
return SPA_RESULT_INVALID_PORT;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_enc_node_port_get_status (SpaNode *node,
uint32_t port_id,
@ -419,33 +461,6 @@ spa_ffmpeg_enc_node_port_get_status (SpaNode *node,
return SPA_RESULT_OK;
}
static SpaResult
spa_ffmpeg_enc_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
if (!IS_VALID_PORT (port_id))
return SPA_RESULT_INVALID_PORT;
return SPA_RESULT_OK;
}
static SpaResult
spa_ffmpeg_enc_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_ffmpeg_enc_node_port_push_input (SpaNode *node,
unsigned int n_info,
@ -518,6 +533,7 @@ static const SpaNode ffmpeg_enc_node = {
spa_ffmpeg_enc_node_port_set_props,
spa_ffmpeg_enc_node_port_use_buffers,
spa_ffmpeg_enc_node_port_alloc_buffers,
spa_ffmpeg_enc_node_port_reuse_buffer,
spa_ffmpeg_enc_node_port_get_status,
spa_ffmpeg_enc_node_port_push_input,
spa_ffmpeg_enc_node_port_pull_output,

467
spa/plugins/remote/proxy.c
View file

@ -37,7 +37,8 @@
#define MAX_OUTPUTS 64
#define MAX_PORTS (MAX_INPUTS + MAX_OUTPUTS)
#define CHECK_PORT_ID(this,id) ((id) < MAX_PORTS && (this)->ports[id].valid)
#define CHECK_FREE_PORT_ID(this,id) ((id) < MAX_PORTS && !(this)->ports[id].valid)
#define CHECK_PORT_ID(this,id) ((id) < MAX_PORTS && (this)->ports[id].valid)
#define CHECK_PORT_ID_DIR(this,id,dir) (CHECK_PORT_ID(this,id) && (this)->ports[i].direction == (dir))
typedef struct _SpaProxy SpaProxy;
@ -61,6 +62,7 @@ typedef struct {
struct _SpaProxy {
SpaHandle handle;
SpaNode node;
SpaNodeState state;
SpaProxyProps props[2];
@ -108,8 +110,6 @@ update_poll (SpaProxy *this, int socketfd)
p = &this->props[1];
if (p->socketfd != -1) {
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_REMOVE_POLL;
event.port_id = 0;
event.data = &this->poll;
@ -120,8 +120,6 @@ update_poll (SpaProxy *this, int socketfd)
if (p->socketfd != -1) {
this->fds[0].fd = p->socketfd;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_ADD_POLL;
event.port_id = 0;
event.data = &this->poll;
@ -130,6 +128,25 @@ update_poll (SpaProxy *this, int socketfd)
}
}
static SpaResult
update_state (SpaProxy *this, SpaNodeState state)
{
if (this->state != state) {
SpaEvent event;
SpaEventStateChange sc;
this->state = state;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = &sc;
event.size = sizeof (sc);
sc.state = state;
this->event_cb (&this->node, &event, this->user_data);
}
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_get_props (SpaNode *node,
SpaProps **props)
@ -186,6 +203,7 @@ spa_proxy_node_send_command (SpaNode *node,
SpaCommand *command)
{
SpaProxy *this;
SpaResult res;
if (node == NULL || node->handle == NULL || command == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
@ -197,34 +215,40 @@ spa_proxy_node_send_command (SpaNode *node,
return SPA_RESULT_INVALID_COMMAND;
case SPA_COMMAND_START:
if (this->event_cb) {
SpaEvent event;
{
SpaControlBuilder builder;
SpaControl control;
uint8_t buf[128];
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STARTED;
event.port_id = -1;
event.data = NULL;
event.size = 0;
/* send start */
spa_control_builder_init_into (&builder, buf, sizeof (buf), NULL, 0);
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_START, NULL);
spa_control_builder_end (&builder, &control);
this->event_cb (&this->node, &event, this->user_data);
}
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
fprintf (stderr, "proxy %p: error writing control %d\n", this, res);
spa_control_clear (&control);
break;
}
case SPA_COMMAND_STOP:
if (this->event_cb) {
SpaEvent event;
{
SpaControlBuilder builder;
SpaControl control;
uint8_t buf[128];
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STOPPED;
event.port_id = -1;
event.data = NULL;
event.size = 0;
/* send start */
spa_control_builder_init_into (&builder, buf, sizeof (buf), NULL, 0);
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_STOP, NULL);
spa_control_builder_end (&builder, &control);
this->event_cb (&this->node, &event, this->user_data);
}
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
fprintf (stderr, "proxy %p: error writing control %d\n", this, res);
spa_control_clear (&control);
break;
}
case SPA_COMMAND_FLUSH:
case SPA_COMMAND_DRAIN:
@ -315,14 +339,46 @@ spa_proxy_node_add_port (SpaNode *node,
SpaDirection direction,
uint32_t port_id)
{
return SPA_RESULT_NOT_IMPLEMENTED;
SpaProxy *this;
SpaProxyPort *port;
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = (SpaProxy *) node->handle;
if (!CHECK_FREE_PORT_ID (this, port_id))
return SPA_RESULT_INVALID_PORT;
fprintf (stderr, "%p: adding port %d, %d\n", node, port_id, direction);
port = &this->ports[port_id];
port->direction = direction;
port->valid = true;
port->have_format = false;
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_remove_port (SpaNode *node,
uint32_t port_id)
{
return SPA_RESULT_NOT_IMPLEMENTED;
SpaProxy *this;
SpaProxyPort *port;
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = (SpaProxy *) node->handle;
if (!CHECK_PORT_ID (this, port_id))
return SPA_RESULT_INVALID_PORT;
port = &this->ports[port_id];
port->valid = false;
return SPA_RESULT_OK;
}
static SpaResult
@ -385,13 +441,15 @@ spa_proxy_node_port_set_format (SpaNode *node,
port = &this->ports[port_id];
spa_control_builder_init_into (&builder, buf, sizeof (buf), NULL, 0);
sf.port = port_id;
sf.port_id = port_id;
sf.format = format;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_SET_FORMAT, &sf);
spa_control_builder_end (&builder, &control);
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
fprintf (stderr, "proxy %p: error writing control", this);
fprintf (stderr, "proxy %p: error writing control\n", this);
spa_control_clear (&control);
port->have_format = format != NULL;
@ -489,6 +547,127 @@ spa_proxy_node_port_get_status (SpaNode *node,
return SPA_RESULT_OK;
}
static int
tmpfile_create (void *data, size_t size)
{
char filename[] = "/dev/shm/tmpfilepay.XXXXXX";
int fd;
fd = mkostemp (filename, O_CLOEXEC);
if (fd == -1) {
fprintf (stderr, "Failed to create temporary file: %s\n", strerror (errno));
return -1;
}
unlink (filename);
if (write (fd, data, size) != (ssize_t) size)
fprintf (stderr, "Failed to write data: %s\n", strerror (errno));
return fd;
}
typedef struct {
SpaBuffer buffer;
SpaData datas[16];
int idx[16];
SpaBuffer *orig;
} MyBuffer;
static SpaResult
add_buffer (SpaProxy *this, uint32_t port_id, SpaBuffer *buffer)
{
SpaControl control;
SpaControlBuilder builder;
uint8_t buf[1024];
int fds[16];
SpaControlCmdAddBuffer ab;
bool tmpfile = false;
unsigned int i;
MyBuffer b;
SpaResult res;
spa_control_builder_init_into (&builder, buf, sizeof (buf), fds, sizeof (fds));
b.buffer.id = buffer->id;
b.buffer.size = buffer->size;
b.buffer.n_metas = buffer->n_metas;
b.buffer.metas = buffer->metas;
b.buffer.n_datas = buffer->n_datas;
b.buffer.datas = b.datas;
for (i = 0; i < buffer->n_datas; i++) {
SpaData *d = &buffer->datas[i];
int fd;
SpaControlCmdAddMem am;
if (d->type == SPA_DATA_TYPE_FD) {
fd = *((int *)d->ptr);
} else {
fd = tmpfile_create (d->ptr, d->size + d->offset);
tmpfile = true;
}
am.port_id = port_id;
am.mem_id = buffer->id * 64 + i;
am.mem_type = 0;
am.fd_index = spa_control_builder_add_fd (&builder, fd, tmpfile ? true : false);
am.offset = 0;
am.size = d->offset + d->size;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_ADD_MEM, &am);
b.idx[i] = am.mem_id;
b.datas[i].type = SPA_DATA_TYPE_MEMID;
b.datas[i].ptr_type = NULL;
b.datas[i].ptr = &b.idx[i];
b.datas[i].offset = d->offset;
b.datas[i].size = d->size;
b.datas[i].stride = d->stride;
}
ab.port_id = port_id;
ab.buffer = &b.buffer;
fprintf (stderr, "proxy %p: add buffer %d\n", this, b.buffer.id);
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_ADD_BUFFER, &ab);
spa_control_builder_end (&builder, &control);
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
fprintf (stderr, "proxy %p: error writing control\n", this);
spa_control_clear (&control);
return SPA_RESULT_OK;
}
static SpaResult
remove_buffer (SpaProxy *this, uint32_t port_id, SpaBuffer *buffer)
{
SpaControl control;
SpaControlBuilder builder;
uint8_t buf[1024];
SpaControlCmdRemoveBuffer rb;
unsigned int i;
SpaResult res;
spa_control_builder_init_into (&builder, buf, sizeof (buf), NULL, 0);
rb.port_id = port_id;
rb.buffer_id = buffer->id;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_REMOVE_BUFFER, &rb);
for (i = 0; i < buffer->n_datas; i++) {
SpaControlCmdRemoveMem rm;
rm.port_id = port_id;
rm.mem_id = i;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_REMOVE_MEM, &rm);
}
spa_control_builder_end (&builder, &control);
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
fprintf (stderr, "proxy %p: error writing control\n", this);
spa_control_clear (&control);
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
@ -497,11 +676,13 @@ spa_proxy_node_port_use_buffers (SpaNode *node,
{
SpaProxy *this;
SpaProxyPort *port;
unsigned int i;
if (node == NULL || node->handle == NULL || buffers == NULL)
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = (SpaProxy *) node->handle;
fprintf (stderr, "proxy %p: use buffers %p %u\n", this, buffers, n_buffers);
if (!CHECK_PORT_ID (this, port_id))
return SPA_RESULT_INVALID_PORT;
@ -511,8 +692,18 @@ spa_proxy_node_port_use_buffers (SpaNode *node,
if (!port->have_format)
return SPA_RESULT_NO_FORMAT;
port->buffers = buffers;
port->n_buffers = n_buffers;
for (i = 0; i < port->n_buffers; i++)
remove_buffer (this, port_id, port->buffers[i]);
if (buffers == NULL || n_buffers == 0) {
port->buffers = NULL;
port->n_buffers = 0;
} else {
port->buffers = buffers;
port->n_buffers = n_buffers;
}
for (i = 0; i < port->n_buffers; i++)
add_buffer (this, port_id, port->buffers[i]);
return SPA_RESULT_OK;
}
@ -541,112 +732,17 @@ spa_proxy_node_port_alloc_buffers (SpaNode *node,
if (!port->have_format)
return SPA_RESULT_NO_FORMAT;
return SPA_RESULT_OK;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static int
tmpfile_create (void *data, size_t size)
{
char filename[] = "/dev/shm/tmpfilepay.XXXXXX";
int fd;
fd = mkostemp (filename, O_CLOEXEC);
if (fd == -1) {
fprintf (stderr, "Failed to create temporary file: %s", strerror (errno));
return -1;
}
unlink (filename);
if (write (fd, data, size) != (ssize_t) size)
fprintf (stderr, "Failed to write data: %s", strerror (errno));
return fd;
}
typedef struct {
SpaBuffer buffer;
SpaData datas[16];
int idx[16];
SpaBuffer *orig;
} MyBuffer;
static SpaResult
send_buffer (SpaProxy *this, SpaBuffer *buffer)
spa_proxy_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id,
off_t offset,
size_t size)
{
SpaControl control;
SpaControlBuilder builder;
uint8_t buf[1024];
int fds[16];
SpaControlCmdAddBuffer ab;
SpaControlCmdProcessBuffer pb;
SpaControlCmdRemoveBuffer rb;
bool tmpfile = false;
unsigned int i;
MyBuffer b;
SpaResult res;
spa_control_builder_init_into (&builder, buf, 1024, fds, 16);
b.buffer.refcount = 1;
b.buffer.notify = NULL;
b.buffer.id = buffer->id;
b.buffer.size = buffer->size;
b.buffer.n_metas = buffer->n_metas;
b.buffer.metas = buffer->metas;
b.buffer.n_datas = buffer->n_datas;
b.buffer.datas = b.datas;
for (i = 0; i < buffer->n_datas; i++) {
SpaData *d = &buffer->datas[i];
int fd;
SpaControlCmdAddMem am;
if (d->type == SPA_DATA_TYPE_FD) {
fd = *((int *)d->ptr);
} else {
fd = tmpfile_create (d->ptr, d->size + d->offset);
tmpfile = true;
}
am.port = 0;
am.id = i;
am.type = 0;
am.fd_index = spa_control_builder_add_fd (&builder, fd, tmpfile ? true : false);
am.offset = 0;
am.size = d->offset + d->size;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_ADD_MEM, &am);
b.idx[i] = i;
b.datas[i].type = SPA_DATA_TYPE_MEMID;
b.datas[i].ptr_type = NULL;
b.datas[i].ptr = &b.idx[i];
b.datas[i].offset = d->offset;
b.datas[i].size = d->size;
b.datas[i].stride = d->stride;
}
ab.port = 0;
ab.buffer = &b.buffer;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_ADD_BUFFER, &ab);
pb.port = 0;
pb.id = b.buffer.id;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_PROCESS_BUFFER, &pb);
rb.port = 0;
rb.id = b.buffer.id;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_REMOVE_BUFFER, &rb);
for (i = 0; i < buffer->n_datas; i++) {
SpaControlCmdRemoveMem rm;
rm.port = 0;
rm.id = i;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_REMOVE_MEM, &rm);
}
spa_control_builder_end (&builder, &control);
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
fprintf (stderr, "proxy %p: error writing control", this);
spa_control_clear (&control);
return SPA_RESULT_OK;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
@ -659,12 +755,19 @@ spa_proxy_node_port_push_input (SpaNode *node,
unsigned int i;
bool have_error = false;
bool have_enough = false;
SpaControl control;
SpaControlBuilder builder;
SpaControlCmdProcessBuffer pb;
uint8_t buf[64];
SpaResult res;
if (node == NULL || node->handle == NULL || n_info == 0 || info == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = (SpaProxy *) node->handle;
spa_control_builder_init_into (&builder, buf, sizeof(buf), NULL, 0);
for (i = 0; i < n_info; i++) {
if (!CHECK_PORT_ID_DIR (this, info[i].port_id, SPA_DIRECTION_INPUT)) {
info[i].status = SPA_RESULT_INVALID_PORT;
@ -678,14 +781,33 @@ spa_proxy_node_port_push_input (SpaNode *node,
have_error = true;
continue;
}
if (info[i].buffer_id >= port->n_buffers) {
if (port->n_buffers == 0)
info[i].status = SPA_RESULT_NO_BUFFERS;
else
info[i].status = SPA_RESULT_INVALID_BUFFER_ID;
have_error = true;
continue;
}
pb.port_id = info[i].port_id;
pb.buffer_id = info[i].buffer_id;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_PROCESS_BUFFER, &pb);
info[i].status = SPA_RESULT_OK;
}
spa_control_builder_end (&builder, &control);
if (have_error)
return SPA_RESULT_ERROR;
if (have_enough)
return SPA_RESULT_HAVE_ENOUGH_INPUT;
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
fprintf (stderr, "proxy %p: error writing control\n", this);
spa_control_clear (&control);
return SPA_RESULT_OK;
}
@ -728,73 +850,56 @@ spa_proxy_node_port_pull_output (SpaNode *node,
return SPA_RESULT_OK;
}
static SpaResult
spa_proxy_node_port_push_event (SpaNode *node,
uint32_t port_id,
SpaEvent *event)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
parse_control (SpaProxy *this,
SpaControl *ctrl)
{
SpaControlIter it;
SpaResult res;
spa_control_iter_init (&it, ctrl);
while (spa_control_iter_next (&it) == SPA_RESULT_OK) {
SpaControlCmd cmd = spa_control_iter_get_cmd (&it);
fprintf (stderr, "proxy %p: got control %d\n", this, cmd);
switch (cmd) {
case SPA_CONTROL_CMD_ADD_PORT:
case SPA_CONTROL_CMD_REMOVE_PORT:
case SPA_CONTROL_CMD_SET_FORMAT:
case SPA_CONTROL_CMD_SET_PROPERTY:
case SPA_CONTROL_CMD_END_CONFIGURE:
case SPA_CONTROL_CMD_PAUSE:
case SPA_CONTROL_CMD_START:
case SPA_CONTROL_CMD_STOP:
fprintf (stderr, "proxy %p: got unexpected control %d", this, cmd);
fprintf (stderr, "proxy %p: got unexpected control %d\n", this, cmd);
break;
case SPA_CONTROL_CMD_NODE_UPDATE:
case SPA_CONTROL_CMD_PORT_UPDATE:
case SPA_CONTROL_CMD_PORT_REMOVED:
fprintf (stderr, "proxy %p: command not implemented %d", this, cmd);
fprintf (stderr, "proxy %p: command not implemented %d\n", this, cmd);
break;
case SPA_CONTROL_CMD_START_CONFIGURE:
case SPA_CONTROL_CMD_STATE_CHANGE:
{
SpaControlBuilder builder;
SpaControl control;
uint8_t buf[128];
SpaControlCmdStateChange sc;
/* set port format */
if (spa_control_iter_parse_cmd (&it, &sc) < 0)
break;
/* send end-configure */
spa_control_builder_init_into (&builder, buf, sizeof (buf), NULL, 0);
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_END_CONFIGURE, NULL);
spa_control_builder_end (&builder, &control);
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
fprintf (stderr, "proxy %p: error writing control: %d", this, res);
fprintf (stderr, "proxy %p: got state-change to %d\n", this, sc.state);
update_state (this, sc.state);
break;
}
case SPA_CONTROL_CMD_PORT_STATUS_CHANGE:
{
fprintf (stderr, "proxy %p: command not implemented %d", this, cmd);
break;
}
case SPA_CONTROL_CMD_START_ALLOC:
{
SpaControlBuilder builder;
SpaControl control;
uint8_t buf[128];
/* FIXME read port memory requirements */
/* FIXME add_mem */
/* send start */
spa_control_builder_init_into (&builder, buf, sizeof (buf), NULL, 0);
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_START, NULL);
spa_control_builder_end (&builder, &control);
if ((res = spa_control_write (&control, this->fds[0].fd)) < 0)
fprintf (stderr, "proxy %p: error writing control %d", this, res);
fprintf (stderr, "proxy %p: command not implemented %d\n", this, cmd);
break;
}
case SPA_CONTROL_CMD_NEED_INPUT:
@ -821,10 +926,26 @@ parse_control (SpaProxy *this,
}
case SPA_CONTROL_CMD_REUSE_BUFFER:
{
SpaEvent event;
SpaEventReuseBuffer rb;
SpaControlCmdReuseBuffer crb;
if (spa_control_iter_parse_cmd (&it, &crb) < 0)
break;
event.type = SPA_EVENT_TYPE_REUSE_BUFFER;
event.port_id = crb.port_id;
event.data = &rb;
event.size = sizeof (rb);
rb.buffer_id = crb.buffer_id;
rb.offset = crb.offset;
rb.size = crb.size;
this->event_cb (&this->node, &event, this->user_data);
break;
}
default:
fprintf (stderr, "proxy %p: command unhandled %d", this, cmd);
fprintf (stderr, "proxy %p: command unhandled %d\n", this, cmd);
break;
}
}
@ -845,7 +966,7 @@ proxy_on_fd_events (SpaPollNotifyData *data)
int fds[16];
if ((res = spa_control_read (&control, data->fds[0].fd, buf, 1024, fds, 16)) < 0) {
fprintf (stderr, "proxy %p: failed to read control: %d", this, res);
fprintf (stderr, "proxy %p: failed to read control: %d\n", this, res);
return 0;
}
parse_control (this, &control);
@ -873,9 +994,11 @@ static const SpaNode proxy_node = {
spa_proxy_node_port_set_props,
spa_proxy_node_port_use_buffers,
spa_proxy_node_port_alloc_buffers,
spa_proxy_node_port_reuse_buffer,
spa_proxy_node_port_get_status,
spa_proxy_node_port_push_input,
spa_proxy_node_port_pull_output,
spa_proxy_node_port_push_event,
};
static SpaResult

85
spa/plugins/v4l2/v4l2-source.c
View file

@ -228,13 +228,13 @@ spa_v4l2_source_node_send_command (SpaNode *node,
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STARTED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_STREAMING;
this->event_cb (node, &event, this->user_data);
}
@ -244,13 +244,13 @@ spa_v4l2_source_node_send_command (SpaNode *node,
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STOPPED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_PAUSED;
this->event_cb (node, &event, this->user_data);
}
@ -496,26 +496,6 @@ spa_v4l2_source_node_port_set_props (SpaNode *node,
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_v4l2_source_node_port_get_status (SpaNode *node,
uint32_t port_id,
const SpaPortStatus **status)
{
SpaV4l2Source *this;
if (node == NULL || node->handle == NULL || status == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = (SpaV4l2Source *) node->handle;
if (port_id != 0)
return SPA_RESULT_INVALID_PORT;
*status = &this->state[port_id].status;
return SPA_RESULT_OK;
}
static SpaResult
spa_v4l2_source_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
@ -560,6 +540,47 @@ spa_v4l2_source_node_port_alloc_buffers (SpaNode *node,
return SPA_RESULT_OK;
}
static SpaResult
spa_v4l2_source_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id,
off_t offset,
size_t size)
{
SpaV4l2Source *this;
if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = (SpaV4l2Source *) node->handle;
if (port_id != 0)
return SPA_RESULT_INVALID_PORT;
spa_v4l2_buffer_recycle (this, buffer_id);
return SPA_RESULT_OK;
}
static SpaResult
spa_v4l2_source_node_port_get_status (SpaNode *node,
uint32_t port_id,
const SpaPortStatus **status)
{
SpaV4l2Source *this;
if (node == NULL || node->handle == NULL || status == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = (SpaV4l2Source *) node->handle;
if (port_id != 0)
return SPA_RESULT_INVALID_PORT;
*status = &this->state[port_id].status;
return SPA_RESULT_OK;
}
static SpaResult
spa_v4l2_source_node_port_push_input (SpaNode *node,
@ -584,7 +605,6 @@ spa_v4l2_source_node_port_pull_output (SpaNode *node,
this = (SpaV4l2Source *) node->handle;
for (i = 0; i < n_info; i++) {
V4l2Buffer *b;
@ -612,7 +632,7 @@ spa_v4l2_source_node_port_pull_output (SpaNode *node,
b->outstanding = true;
info[i].id = b->buffer.id;
info[i].buffer_id = b->buffer.id;
info[i].status = SPA_RESULT_OK;
}
if (have_error)
@ -649,6 +669,7 @@ static const SpaNode v4l2source_node = {
spa_v4l2_source_node_port_set_props,
spa_v4l2_source_node_port_use_buffers,
spa_v4l2_source_node_port_alloc_buffers,
spa_v4l2_source_node_port_reuse_buffer,
spa_v4l2_source_node_port_get_status,
spa_v4l2_source_node_port_push_input,
spa_v4l2_source_node_port_pull_output,

100
spa/plugins/v4l2/v4l2-utils.c
View file

@ -437,7 +437,7 @@ spa_v4l2_close (SpaV4l2Source *this)
return 0;
}
static int
static SpaResult
mmap_read (SpaV4l2Source *this)
{
SpaV4l2State *state = &this->state[0];
@ -455,7 +455,7 @@ mmap_read (SpaV4l2Source *this)
case EIO:
default:
perror ("VIDIOC_DQBUF");
return -1;
return SPA_RESULT_ERROR;
}
}
@ -464,7 +464,7 @@ mmap_read (SpaV4l2Source *this)
state->ready = b;
state->ready_count++;
return 0;
return SPA_RESULT_OK;
}
static int
@ -473,10 +473,9 @@ v4l2_on_fd_events (SpaPollNotifyData *data)
SpaV4l2Source *this = data->user_data;
SpaEvent event;
mmap_read (this);
if (mmap_read (this) < 0)
return 0;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_CAN_PULL_OUTPUT;
event.port_id = 0;
event.size = 0;
@ -487,14 +486,11 @@ v4l2_on_fd_events (SpaPollNotifyData *data)
}
static void
v4l2_buffer_recycle (void *data)
spa_v4l2_buffer_recycle (SpaV4l2Source *this, uint32_t buffer_id)
{
SpaBuffer *buf = data;
SpaV4l2Source *this = buf->user_data;
SpaV4l2State *state = &this->state[0];
V4l2Buffer *b = &state->buffers[buf->id];
V4l2Buffer *b = &state->buffers[buffer_id];
b->imported->refcount = 1;
b->outstanding = false;
if (xioctl (state->fd, VIDIOC_QBUF, &b->v4l2_buffer) < 0) {
@ -502,7 +498,7 @@ v4l2_buffer_recycle (void *data)
}
}
static int
static SpaResult
spa_v4l2_import_buffers (SpaV4l2Source *this, SpaBuffer **buffers, uint32_t n_buffers)
{
SpaV4l2State *state = &this->state[0];
@ -518,12 +514,12 @@ spa_v4l2_import_buffers (SpaV4l2Source *this, SpaBuffer **buffers, uint32_t n_bu
if (xioctl (state->fd, VIDIOC_REQBUFS, &reqbuf) < 0) {
perror ("VIDIOC_REQBUFS");
return -1;
return SPA_RESULT_ERROR;
}
fprintf (stderr, "got %d buffers\n", reqbuf.count);
if (reqbuf.count < 2) {
fprintf (stderr, "can't allocate enough buffers\n");
return -1;
return SPA_RESULT_ERROR;
}
state->reqbuf = reqbuf;
@ -532,14 +528,9 @@ spa_v4l2_import_buffers (SpaV4l2Source *this, SpaBuffer **buffers, uint32_t n_bu
b = &state->buffers[i];
buffers[i]->notify = v4l2_buffer_recycle;
buffers[i]->user_data = this;
fprintf (stderr, "import buffer %p\n", buffers[i]);
b->source = this;
b->buffer.refcount = 0;
b->buffer.notify = v4l2_buffer_recycle;
b->buffer.id = buffers[i]->id;
b->buffer.size = buffers[i]->size;
b->buffer.n_metas = buffers[i]->n_metas;
@ -556,14 +547,14 @@ spa_v4l2_import_buffers (SpaV4l2Source *this, SpaBuffer **buffers, uint32_t n_bu
b->v4l2_buffer.m.userptr = (unsigned long) b->buffer.datas[0].ptr;
b->v4l2_buffer.length = b->buffer.datas[0].size;
v4l2_buffer_recycle (buffers[i]);
spa_v4l2_buffer_recycle (this, buffers[i]->id);
}
state->have_buffers = true;
return 0;
return SPA_RESULT_OK;
}
static int
static SpaResult
mmap_init (SpaV4l2Source *this,
SpaAllocParam **params,
unsigned int n_params,
@ -583,7 +574,7 @@ mmap_init (SpaV4l2Source *this,
if (xioctl (state->fd, VIDIOC_REQBUFS, &reqbuf) < 0) {
perror ("VIDIOC_REQBUFS");
return -1;
return SPA_RESULT_ERROR;
}
fprintf (stderr, "got %d buffers\n", reqbuf.count);
@ -591,7 +582,7 @@ mmap_init (SpaV4l2Source *this,
if (reqbuf.count < 2) {
fprintf (stderr, "can't allocate enough buffers\n");
return -1;
return SPA_RESULT_ERROR;
}
if (state->export_buf)
fprintf (stderr, "using EXPBUF\n");
@ -609,7 +600,7 @@ mmap_init (SpaV4l2Source *this,
if (xioctl (state->fd, VIDIOC_QUERYBUF, &buf) < 0) {
perror ("VIDIOC_QUERYBUF");
return -1;
return SPA_RESULT_ERROR;
}
b = &state->buffers[i];
@ -617,9 +608,6 @@ mmap_init (SpaV4l2Source *this,
buffers[i] = &b->buffer;
b->source = this;
b->buffer.refcount = 0;
b->buffer.notify = v4l2_buffer_recycle;
b->buffer.user_data = this;
b->buffer.id = i;
b->buffer.size = buf.length;
b->buffer.n_metas = 1;
@ -679,47 +667,49 @@ mmap_init (SpaV4l2Source *this,
b->v4l2_buffer.memory = state->memtype;
b->v4l2_buffer.index = i;
v4l2_buffer_recycle (b);
spa_v4l2_buffer_recycle (this, i);
}
state->have_buffers = true;
return 0;
return SPA_RESULT_OK;
}
static int
static SpaResult
userptr_init (SpaV4l2Source *this)
{
return -1;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static int
static SpaResult
read_init (SpaV4l2Source *this)
{
return -1;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static int
static SpaResult
spa_v4l2_alloc_buffers (SpaV4l2Source *this,
SpaAllocParam **params,
unsigned int n_params,
SpaBuffer **buffers,
unsigned int *n_buffers)
{
SpaResult res;
SpaV4l2State *state = &this->state[0];
if (state->cap.capabilities & V4L2_CAP_STREAMING) {
if (mmap_init (this, params, n_params, buffers, n_buffers) < 0)
if (userptr_init (this) < 0)
return -1;
if ((res = mmap_init (this, params, n_params, buffers, n_buffers)) < 0)
if ((res = userptr_init (this)) < 0)
return res;
} else if (state->cap.capabilities & V4L2_CAP_READWRITE) {
if (read_init (this) < 0)
return -1;
if ((res = read_init (this)) < 0)
return res;
} else
return -1;
return 0;
return SPA_RESULT_ERROR;
return SPA_RESULT_OK;
}
static int
static SpaResult
spa_v4l2_start (SpaV4l2Source *this)
{
SpaV4l2State *state = &this->state[0];
@ -727,14 +717,11 @@ spa_v4l2_start (SpaV4l2Source *this)
SpaEvent event;
if (spa_v4l2_open (this) < 0)
return -1;
return SPA_RESULT_ERROR;
if (!state->have_buffers)
return -1;
return SPA_RESULT_NO_BUFFERS;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_ADD_POLL;
event.port_id = 0;
event.data = &state->poll;
@ -756,12 +743,12 @@ spa_v4l2_start (SpaV4l2Source *this)
type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
if (xioctl (state->fd, VIDIOC_STREAMON, &type) < 0) {
perror ("VIDIOC_STREAMON");
return -1;
return SPA_RESULT_ERROR;
}
return 0;
return SPA_RESULT_OK;
}
static int
static SpaResult
spa_v4l2_stop (SpaV4l2Source *this)
{
SpaV4l2State *state = &this->state[0];
@ -769,10 +756,13 @@ spa_v4l2_stop (SpaV4l2Source *this)
SpaEvent event;
int i;
if (!state->opened)
return SPA_RESULT_OK;
type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
if (xioctl (state->fd, VIDIOC_STREAMOFF, &type) < 0) {
perror ("VIDIOC_STREAMOFF");
return -1;
return SPA_RESULT_ERROR;
}
for (i = 0; i < state->reqbuf.count; i++) {
@ -781,7 +771,7 @@ spa_v4l2_stop (SpaV4l2Source *this)
b = &state->buffers[i];
if (b->outstanding) {
fprintf (stderr, "queueing outstanding buffer %p\n", b);
v4l2_buffer_recycle (b);
spa_v4l2_buffer_recycle (this, i);
}
if (state->export_buf) {
close (b->dmafd);
@ -791,8 +781,6 @@ spa_v4l2_stop (SpaV4l2Source *this)
}
state->have_buffers = false;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_REMOVE_POLL;
event.port_id = 0;
event.data = &state->poll;
@ -801,5 +789,5 @@ spa_v4l2_stop (SpaV4l2Source *this)
spa_v4l2_close (this);
return 0;
return SPA_RESULT_OK;
}

95
spa/plugins/volume/volume.c
View file

@ -159,13 +159,13 @@ spa_volume_node_send_command (SpaNode *node,
case SPA_COMMAND_START:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STARTED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_STREAMING;
this->event_cb (node, &event, this->user_data);
}
@ -174,13 +174,13 @@ spa_volume_node_send_command (SpaNode *node,
case SPA_COMMAND_STOP:
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STOPPED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_PAUSED;
this->event_cb (node, &event, this->user_data);
}
@ -398,6 +398,36 @@ spa_volume_node_port_set_props (SpaNode *node,
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_volume_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_volume_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_volume_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id,
off_t offset,
size_t size)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_volume_node_port_get_status (SpaNode *node,
uint32_t port_id,
@ -424,26 +454,6 @@ spa_volume_node_port_get_status (SpaNode *node,
return SPA_RESULT_OK;
}
static SpaResult
spa_volume_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
SpaBuffer **buffers,
uint32_t n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_volume_node_port_alloc_buffers (SpaNode *node,
uint32_t port_id,
SpaAllocParam **params,
uint32_t n_params,
SpaBuffer **buffers,
uint32_t *n_buffers)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_volume_node_port_push_input (SpaNode *node,
@ -471,7 +481,7 @@ spa_volume_node_port_push_input (SpaNode *node,
}
port = &this->ports[info[i].port_id];
buffer = port->buffers[info[i].id];
buffer = port->buffers[info[i].buffer_id];
if (buffer == NULL) {
info[i].status = SPA_RESULT_INVALID_ARGUMENTS;
@ -491,7 +501,7 @@ spa_volume_node_port_push_input (SpaNode *node,
have_enough = true;
continue;
}
port->buffer = spa_buffer_ref (buffer);
port->buffer = buffer;
this->ports[0].status.flags &= ~SPA_PORT_STATUS_FLAG_NEED_INPUT;
this->ports[1].status.flags |= SPA_PORT_STATUS_FLAG_HAVE_OUTPUT;
@ -512,6 +522,22 @@ find_free_buffer (SpaVolume *this, SpaVolumePort *port)
return NULL;
}
static void
release_buffer (SpaVolume *this, SpaBuffer *buffer)
{
SpaEvent event;
SpaEventReuseBuffer rb;
event.type = SPA_EVENT_TYPE_REUSE_BUFFER;
event.port_id = 0;
event.data = &rb;
event.size = sizeof (rb);
rb.buffer_id = buffer->id;
rb.offset = 0;
rb.size = -1;
this->event_cb (&this->node, &event, this->user_data);
}
static SpaResult
spa_volume_node_port_pull_output (SpaNode *node,
unsigned int n_info,
@ -586,10 +612,10 @@ spa_volume_node_port_pull_output (SpaNode *node,
}
if (sbuf != dbuf)
spa_buffer_unref (sbuf);
release_buffer (this, sbuf);
this->ports[0].buffer = NULL;
info->id = dbuf->id;
info->buffer_id = dbuf->id;
this->ports[0].status.flags |= SPA_PORT_STATUS_FLAG_NEED_INPUT;
this->ports[1].status.flags &= ~SPA_PORT_STATUS_FLAG_HAVE_OUTPUT;
@ -624,6 +650,7 @@ static const SpaNode volume_node = {
spa_volume_node_port_set_props,
spa_volume_node_port_use_buffers,
spa_volume_node_port_alloc_buffers,
spa_volume_node_port_reuse_buffer,
spa_volume_node_port_get_status,
spa_volume_node_port_push_input,
spa_volume_node_port_pull_output,

71
spa/plugins/xv/xv-sink.c
View file

@ -187,13 +187,13 @@ spa_xv_sink_node_send_command (SpaNode *node,
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STARTED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_STREAMING;
this->event_cb (node, &event, this->user_data);
}
@ -203,13 +203,13 @@ spa_xv_sink_node_send_command (SpaNode *node,
if (this->event_cb) {
SpaEvent event;
SpaEventStateChange sc;
event.refcount = 1;
event.notify = NULL;
event.type = SPA_EVENT_TYPE_STOPPED;
event.type = SPA_EVENT_TYPE_STATE_CHANGE;
event.port_id = -1;
event.data = NULL;
event.size = 0;
event.data = &sc;
event.size = sizeof (sc);
sc.state = SPA_NODE_STATE_PAUSED;
this->event_cb (node, &event, this->user_data);
}
@ -435,26 +435,6 @@ spa_xv_sink_node_port_set_props (SpaNode *node,
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_xv_sink_node_port_get_status (SpaNode *node,
uint32_t port_id,
const SpaPortStatus **status)
{
SpaXvSink *this;
if (node == NULL || node->handle == NULL || status == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = (SpaXvSink *) node->handle;
if (port_id != 0)
return SPA_RESULT_INVALID_PORT;
*status = &this->status;
return SPA_RESULT_OK;
}
static SpaResult
spa_xv_sink_node_port_use_buffers (SpaNode *node,
uint32_t port_id,
@ -475,6 +455,36 @@ spa_xv_sink_node_port_alloc_buffers (SpaNode *node,
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_xv_sink_node_port_reuse_buffer (SpaNode *node,
uint32_t port_id,
uint32_t buffer_id,
off_t offset,
size_t size)
{
return SPA_RESULT_NOT_IMPLEMENTED;
}
static SpaResult
spa_xv_sink_node_port_get_status (SpaNode *node,
uint32_t port_id,
const SpaPortStatus **status)
{
SpaXvSink *this;
if (node == NULL || node->handle == NULL || status == NULL)
return SPA_RESULT_INVALID_ARGUMENTS;
this = (SpaXvSink *) node->handle;
if (port_id != 0)
return SPA_RESULT_INVALID_PORT;
*status = &this->status;
return SPA_RESULT_OK;
}
static SpaResult
spa_xv_sink_node_port_push_input (SpaNode *node,
unsigned int n_info,
@ -518,6 +528,7 @@ static const SpaNode xvsink_node = {
spa_xv_sink_node_port_set_props,
spa_xv_sink_node_port_use_buffers,
spa_xv_sink_node_port_alloc_buffers,
spa_xv_sink_node_port_reuse_buffer,
spa_xv_sink_node_port_get_status,
spa_xv_sink_node_port_push_input,
spa_xv_sink_node_port_pull_output,