mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-03 09:01:54 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions
pipewire/pinos/client/stream.c
David Svensson Fors dac9755765 stream: free formats and transport in destroy()
2017-02-15 11:56:45 +01:00

1209 lines
32 KiB
C
Raw Blame History

/* Pinos
* Copyright (C) 2015 Wim Taymans <wim.taymans@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 <unistd.h>
#include <sys/socket.h>
#include <string.h>
#include <sys/mman.h>
#include <errno.h>
#include <time.h>
#include "spa/lib/debug.h"
#include "pinos/client/pinos.h"
#include "pinos/client/array.h"
#include "pinos/client/connection.h"
#include "pinos/client/context.h"
#include "pinos/client/stream.h"
#include "pinos/client/serialize.h"
#include "pinos/client/transport.h"
#define MAX_BUFFER_SIZE 4096
#define MAX_FDS 32
#define MAX_INPUTS 64
#define MAX_OUTPUTS 64
typedef struct {
uint32_t id;
int fd;
uint32_t flags;
void *ptr;
off_t offset;
size_t size;
} MemId;
typedef struct {
uint32_t id;
bool used;
void *buf_ptr;
SpaBuffer *buf;
} BufferId;
typedef struct
{
PinosStream this;
SpaNodeState node_state;
uint32_t seq;
unsigned int n_possible_formats;
SpaFormat **possible_formats;
SpaFormat *format;
SpaPortInfo port_info;
SpaDirection direction;
uint32_t port_id;
uint32_t pending_seq;
PinosStreamFlags flags;
PinosStreamMode mode;
int rtfd;
SpaSource *rtsocket_source;
PinosProxy *node_proxy;
bool disconnecting;
PinosListener node_proxy_destroy;
PinosTransport *trans;
SpaSource *timeout_source;
PinosArray mem_ids;
PinosArray buffer_ids;
bool in_order;
int64_t last_ticks;
int32_t last_rate;
int64_t last_monotonic;
} PinosStreamImpl;
static void
clear_memid (MemId *mid)
{
if (mid->ptr != NULL)
munmap (mid->ptr, mid->size + mid->offset);
mid->ptr = NULL;
close (mid->fd);
}
static void
clear_mems (PinosStream *stream)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
MemId *mid;
pinos_array_for_each (mid, &impl->mem_ids)
clear_memid (mid);
impl->mem_ids.size = 0;
}
static void
clear_buffers (PinosStream *stream)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
BufferId *bid;
pinos_array_for_each (bid, &impl->buffer_ids) {
pinos_signal_emit (&stream->remove_buffer, stream, bid->id);
free (bid->buf);
bid->buf = NULL;
bid->used = false;
}
impl->buffer_ids.size = 0;
impl->in_order = true;
}
static void
stream_set_state (PinosStream *stream,
PinosStreamState state,
char *error)
{
if (stream->state != state) {
if (stream->error)
free (stream->error);
stream->error = error;
stream->state = state;
pinos_signal_emit (&stream->state_changed, stream);
}
}
/**
* pinos_stream_state_as_string:
* @state: a #PinosStreamState
*
* Return the string representation of @state.
*
* Returns: the string representation of @state.
*/
const char *
pinos_stream_state_as_string (PinosStreamState state)
{
switch (state) {
case PINOS_STREAM_STATE_ERROR:
return "error";
case PINOS_STREAM_STATE_UNCONNECTED:
return "unconnected";
case PINOS_STREAM_STATE_CONNECTING:
return "connecting";
case PINOS_STREAM_STATE_CONFIGURE:
return "configure";
case PINOS_STREAM_STATE_READY:
return "ready";
case PINOS_STREAM_STATE_PAUSED:
return "paused";
case PINOS_STREAM_STATE_STREAMING:
return "streaming";
}
return "invalid-state";
}
/**
* pinos_stream_new:
* @context: a #PinosContext
* @name: a stream name
* @properties: (transfer full): stream properties
*
* Make a new unconnected #PinosStream
*
* Returns: a new unconnected #PinosStream
*/
PinosStream *
pinos_stream_new (PinosContext *context,
const char *name,
PinosProperties *props)
{
PinosStreamImpl *impl;
PinosStream *this;
impl = calloc (1, sizeof (PinosStreamImpl));
if (impl == NULL)
return NULL;
this = &impl->this;
pinos_log_debug ("stream %p: new", impl);
if (props == NULL) {
props = pinos_properties_new ("media.name", name, NULL);
} else if (!pinos_properties_get (props, "media.name")) {
pinos_properties_set (props, "media.name", name);
}
if (props == NULL)
goto no_mem;
this->properties = props;
this->context = context;
this->name = strdup (name);
pinos_signal_init (&this->destroy_signal);
pinos_signal_init (&this->state_changed);
pinos_signal_init (&this->format_changed);
pinos_signal_init (&this->add_buffer);
pinos_signal_init (&this->remove_buffer);
pinos_signal_init (&this->new_buffer);
pinos_signal_init (&this->need_buffer);
this->state = PINOS_STREAM_STATE_UNCONNECTED;
impl->node_state = SPA_NODE_STATE_INIT;
pinos_array_init (&impl->mem_ids);
pinos_array_ensure_size (&impl->mem_ids, sizeof (MemId) * 64);
pinos_array_init (&impl->buffer_ids);
pinos_array_ensure_size (&impl->buffer_ids, sizeof (BufferId) * 64);
impl->pending_seq = SPA_ID_INVALID;
spa_list_insert (&context->stream_list, &this->link);
return this;
no_mem:
free (impl);
return NULL;
}
static void
unhandle_socket (PinosStream *stream)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
if (impl->rtsocket_source) {
pinos_loop_destroy_source (stream->context->loop, impl->rtsocket_source);
impl->rtsocket_source = NULL;
}
if (impl->timeout_source) {
pinos_loop_destroy_source (stream->context->loop, impl->timeout_source);
impl->timeout_source = NULL;
}
}
void
pinos_stream_destroy (PinosStream *stream)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
int i;
pinos_log_debug ("stream %p: destroy", stream);
pinos_signal_emit (&stream->destroy_signal, stream);
unhandle_socket (stream);
spa_list_remove (&stream->link);
if (impl->node_proxy)
pinos_signal_remove (&impl->node_proxy_destroy);
if (impl->possible_formats) {
for (i = 0; i < impl->n_possible_formats; i++) {
free (impl->possible_formats[i]);
}
free (impl->possible_formats);
}
if (impl->format)
free (impl->format);
if (stream->error)
free (stream->error);
clear_buffers (stream);
pinos_array_clear (&impl->buffer_ids);
clear_mems (stream);
pinos_array_clear (&impl->mem_ids);
if (stream->properties)
pinos_properties_free (stream->properties);
if (impl->trans)
pinos_transport_destroy (impl->trans);
if (stream->name)
free (stream->name);
free (impl);
}
static void
add_node_update (PinosStream *stream, uint32_t change_mask, bool flush)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
PinosMessageNodeUpdate nu = { 0, };
nu.change_mask = change_mask;
if (change_mask & PINOS_MESSAGE_NODE_UPDATE_MAX_INPUTS)
nu.max_input_ports = impl->direction == SPA_DIRECTION_INPUT ? 1 : 0;
if (change_mask & PINOS_MESSAGE_NODE_UPDATE_MAX_OUTPUTS)
nu.max_output_ports = impl->direction == SPA_DIRECTION_OUTPUT ? 1 : 0;
nu.props = NULL;
pinos_proxy_send_message (impl->node_proxy,
PINOS_MESSAGE_NODE_UPDATE,
&nu,
flush);
}
static void
add_state_change (PinosStream *stream, SpaNodeState state, bool flush)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
PinosMessageNodeStateChange sc;
if (impl->node_state != state) {
sc.state = impl->node_state = state;
pinos_proxy_send_message (impl->node_proxy,
PINOS_MESSAGE_NODE_STATE_CHANGE,
&sc,
flush);
}
}
static void
add_port_update (PinosStream *stream, uint32_t change_mask, bool flush)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
PinosMessagePortUpdate pu = { 0, };;
pu.direction = impl->direction;
pu.port_id = impl->port_id;
pu.change_mask = change_mask;
if (change_mask & PINOS_MESSAGE_PORT_UPDATE_POSSIBLE_FORMATS) {
pu.n_possible_formats = impl->n_possible_formats;
pu.possible_formats = impl->possible_formats;
}
if (change_mask & PINOS_MESSAGE_PORT_UPDATE_FORMAT) {
pu.format = impl->format;
}
pu.props = NULL;
if (change_mask & PINOS_MESSAGE_PORT_UPDATE_INFO) {
pu.info = &impl->port_info;
}
pinos_proxy_send_message (impl->node_proxy,
PINOS_MESSAGE_PORT_UPDATE,
&pu,
flush);
}
static inline void
send_need_input (PinosStream *stream)
{
#if 0
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
SpaNodeEventNeedInput ni;
uint64_t cmd = 1;
pinos_log_debug ("stream %p: need input", stream);
ni.event.type = SPA_NODE_EVENT_TYPE_NEED_INPUT;
ni.event.size = sizeof (ni);
pinos_transport_add_event (impl->trans, &ni.event);
write (impl->rtfd, &cmd, 8);
#endif
}
static inline void
send_have_output (PinosStream *stream)
{
#if 0
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
SpaNodeEventHaveOutput ho;
uint64_t cmd = 1;
pinos_log_debug ("stream %p: have output", stream);
ho.event.type = SPA_NODE_EVENT_TYPE_HAVE_OUTPUT;
ho.event.size = sizeof (ho);
pinos_transport_add_event (impl->trans, &ho.event);
write (impl->rtfd, &cmd, 8);
#endif
}
static void
add_request_clock_update (PinosStream *stream, bool flush)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
PinosMessageNodeEvent cne;
SpaNodeEventRequestClockUpdate rcu;
cne.event = &rcu.event;
rcu.event.type = SPA_NODE_EVENT_TYPE_REQUEST_CLOCK_UPDATE;
rcu.event.size = sizeof (rcu);
rcu.update_mask = SPA_NODE_EVENT_REQUEST_CLOCK_UPDATE_TIME;
rcu.timestamp = 0;
rcu.offset = 0;
pinos_proxy_send_message (impl->node_proxy,
PINOS_MESSAGE_NODE_EVENT,
&cne,
flush);
}
static void
add_async_complete (PinosStream *stream,
uint32_t seq,
SpaResult res,
bool flush)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
PinosMessageNodeEvent cne;
SpaNodeEventAsyncComplete ac;
cne.event = &ac.event;
ac.event.type = SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE;
ac.event.size = sizeof (ac);
ac.seq = seq;
ac.res = res;
pinos_proxy_send_message (impl->node_proxy,
PINOS_MESSAGE_NODE_EVENT,
&cne,
flush);
}
static void
do_node_init (PinosStream *stream)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
add_node_update (stream, PINOS_MESSAGE_NODE_UPDATE_MAX_INPUTS |
PINOS_MESSAGE_NODE_UPDATE_MAX_OUTPUTS,
false);
impl->port_info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
add_port_update (stream, PINOS_MESSAGE_PORT_UPDATE_POSSIBLE_FORMATS |
PINOS_MESSAGE_PORT_UPDATE_INFO,
false);
add_state_change (stream, SPA_NODE_STATE_CONFIGURE, true);
}
static void
on_timeout (SpaSource *source,
void *data)
{
PinosStream *stream = data;
add_request_clock_update (stream, true);
}
static MemId *
find_mem (PinosStream *stream, uint32_t id)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
MemId *mid;
pinos_array_for_each (mid, &impl->mem_ids) {
if (mid->id == id)
return mid;
}
return NULL;
}
static BufferId *
find_buffer (PinosStream *stream, uint32_t id)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
if (impl->in_order && pinos_array_check_index (&impl->buffer_ids, id, BufferId)) {
return pinos_array_get_unchecked (&impl->buffer_ids, id, BufferId);
} else {
BufferId *bid;
pinos_array_for_each (bid, &impl->buffer_ids) {
if (bid->id == id)
return bid;
}
}
return NULL;
}
static void
handle_rtnode_event (PinosStream *stream,
SpaNodeEvent *event)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
switch (event->type) {
case SPA_NODE_EVENT_TYPE_HAVE_OUTPUT:
{
int i;
//pinos_log_debug ("stream %p: have output", stream);
for (i = 0; i < impl->trans->area->n_inputs; i++) {
SpaPortInput *input = &impl->trans->inputs[i];
if (input->buffer_id == SPA_ID_INVALID)
continue;
pinos_signal_emit (&stream->new_buffer, stream, input->buffer_id);
input->buffer_id = SPA_ID_INVALID;
}
send_need_input (stream);
break;
}
case SPA_NODE_EVENT_TYPE_NEED_INPUT:
//pinos_log_debug ("stream %p: need input", stream);
pinos_signal_emit (&stream->need_buffer, stream);
break;
case SPA_NODE_EVENT_TYPE_REUSE_BUFFER:
{
SpaNodeEventReuseBuffer *p = (SpaNodeEventReuseBuffer *) event;
BufferId *bid;
if (p->port_id != impl->port_id)
break;
if (impl->direction != SPA_DIRECTION_OUTPUT)
break;
if ((bid = find_buffer (stream, p->buffer_id)) && bid->used) {
bid->used = false;
pinos_signal_emit (&stream->new_buffer, stream, p->buffer_id);
}
break;
}
default:
pinos_log_warn ("unexpected node event %d", event->type);
break;
}
}
static void
on_rtsocket_condition (SpaSource *source,
int fd,
SpaIO mask,
void *data)
{
PinosStream *stream = data;
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
if (mask & (SPA_IO_ERR | SPA_IO_HUP)) {
pinos_log_warn ("got error");
unhandle_socket (stream);
return;
}
if (mask & SPA_IO_IN) {
SpaNodeEvent event;
uint64_t cmd;
read (impl->rtfd, &cmd, 8);
while (pinos_transport_next_event (impl->trans, &event) == SPA_RESULT_OK) {
SpaNodeEvent *ev = alloca (event.size);
pinos_transport_parse_event (impl->trans, ev);
handle_rtnode_event (stream, ev);
}
}
}
static void
handle_socket (PinosStream *stream, int rtfd)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
struct timespec interval;
impl->rtfd = rtfd;
impl->rtsocket_source = pinos_loop_add_io (stream->context->loop,
impl->rtfd,
SPA_IO_IN | SPA_IO_ERR | SPA_IO_HUP,
true,
on_rtsocket_condition,
stream);
impl->timeout_source = pinos_loop_add_timer (stream->context->loop,
on_timeout,
stream);
interval.tv_sec = 0;
interval.tv_nsec = 100000000;
pinos_loop_update_timer (stream->context->loop,
impl->timeout_source,
NULL,
&interval,
false);
return;
}
static void
handle_node_event (PinosStream *stream,
SpaNodeEvent *event)
{
switch (event->type) {
case SPA_NODE_EVENT_TYPE_INVALID:
case SPA_NODE_EVENT_TYPE_HAVE_OUTPUT:
case SPA_NODE_EVENT_TYPE_NEED_INPUT:
case SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE:
case SPA_NODE_EVENT_TYPE_REUSE_BUFFER:
case SPA_NODE_EVENT_TYPE_ERROR:
case SPA_NODE_EVENT_TYPE_BUFFERING:
case SPA_NODE_EVENT_TYPE_REQUEST_REFRESH:
case SPA_NODE_EVENT_TYPE_REQUEST_CLOCK_UPDATE:
pinos_log_warn ("unhandled node event %d", event->type);
break;
}
}
static bool
handle_node_command (PinosStream *stream,
uint32_t seq,
SpaNodeCommand *command)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
switch (command->type) {
case SPA_NODE_COMMAND_INVALID:
break;
case SPA_NODE_COMMAND_PAUSE:
{
pinos_log_debug ("stream %p: pause %d", stream, seq);
add_state_change (stream, SPA_NODE_STATE_PAUSED, false);
add_async_complete (stream, seq, SPA_RESULT_OK, true);
stream_set_state (stream, PINOS_STREAM_STATE_PAUSED, NULL);
break;
}
case SPA_NODE_COMMAND_START:
{
pinos_log_debug ("stream %p: start %d", stream, seq);
add_state_change (stream, SPA_NODE_STATE_STREAMING, false);
add_async_complete (stream, seq, SPA_RESULT_OK, true);
if (impl->direction == SPA_DIRECTION_INPUT)
send_need_input (stream);
stream_set_state (stream, PINOS_STREAM_STATE_STREAMING, NULL);
break;
}
case SPA_NODE_COMMAND_FLUSH:
case SPA_NODE_COMMAND_DRAIN:
case SPA_NODE_COMMAND_MARKER:
{
pinos_log_warn ("unhandled node command %d", command->type);
add_async_complete (stream, seq, SPA_RESULT_NOT_IMPLEMENTED, true);
break;
}
case SPA_NODE_COMMAND_CLOCK_UPDATE:
{
SpaNodeCommandClockUpdate *cu = (SpaNodeCommandClockUpdate *) command;
if (cu->flags & SPA_NODE_COMMAND_CLOCK_UPDATE_FLAG_LIVE) {
pinos_properties_set (stream->properties,
"pinos.latency.is-live", "1");
pinos_properties_setf (stream->properties,
"pinos.latency.min", "%"PRId64, cu->latency);
}
impl->last_ticks = cu->ticks;
impl->last_rate = cu->rate;
impl->last_monotonic = cu->monotonic_time;
break;
}
}
return true;
}
static SpaResult
stream_dispatch_func (void *object,
PinosMessageType type,
void *message,
void *data)
{
PinosStream *stream = data;
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
switch (type) {
case PINOS_MESSAGE_CREATE_NODE_DONE:
pinos_log_warn ("create node done %d", type);
break;
case PINOS_MESSAGE_CREATE_CLIENT_NODE_DONE:
{
PinosMessageCreateClientNodeDone *cnd = message;
pinos_log_warn ("create client node done %d", type);
handle_socket (stream, cnd->datafd);
do_node_init (stream);
break;
}
case PINOS_MESSAGE_ADD_PORT:
case PINOS_MESSAGE_REMOVE_PORT:
pinos_log_warn ("add/remove port not supported");
break;
case PINOS_MESSAGE_SET_FORMAT:
{
PinosMessageSetFormat *p = message;
void *mem;
if (impl->format)
free (impl->format);
mem = malloc (pinos_serialize_format_get_size (p->format));
impl->format = pinos_serialize_format_copy_into (mem, p->format);
impl->pending_seq = p->seq;
pinos_signal_emit (&stream->format_changed, stream, impl->format);
stream_set_state (stream, PINOS_STREAM_STATE_READY, NULL);
break;
}
case PINOS_MESSAGE_SET_PROPERTY:
pinos_log_warn ("set property not implemented");
break;
case PINOS_MESSAGE_ADD_MEM:
{
PinosMessageAddMem *p = message;
MemId *m;
m = find_mem (stream, p->mem_id);
if (m) {
pinos_log_debug ("update mem %u, fd %d, flags %d, off %zd, size %zd",
p->mem_id, p->memfd, p->flags, p->offset, p->size);
clear_memid (m);
} else {
m = pinos_array_add (&impl->mem_ids, sizeof (MemId));
pinos_log_debug ("add mem %u, fd %d, flags %d, off %zd, size %zd",
p->mem_id, p->memfd, p->flags, p->offset, p->size);
}
m->id = p->mem_id;
m->fd = p->memfd;
m->flags = p->flags;
m->ptr = NULL;
m->offset = p->offset;
m->size = p->size;
break;
}
case PINOS_MESSAGE_USE_BUFFERS:
{
PinosMessageUseBuffers *p = message;
BufferId *bid;
unsigned int i, j, len;
SpaBuffer *b;
/* clear previous buffers */
clear_buffers (stream);
for (i = 0; i < p->n_buffers; i++) {
off_t offset = 0;
MemId *mid = find_mem (stream, p->buffers[i].mem_id);
if (mid == NULL) {
pinos_log_warn ("unknown memory id %u", p->buffers[i].mem_id);
continue;
}
if (mid->ptr == NULL) {
mid->ptr = mmap (NULL, mid->size + mid->offset, PROT_READ | PROT_WRITE, MAP_SHARED, mid->fd, 0);
if (mid->ptr == MAP_FAILED) {
mid->ptr = NULL;
pinos_log_warn ("Failed to mmap memory %zd %p: %s", mid->size, mid, strerror (errno));
continue;
}
mid->ptr = SPA_MEMBER (mid->ptr, mid->offset, void);
}
len = pinos_array_get_len (&impl->buffer_ids, BufferId);
bid = pinos_array_add (&impl->buffer_ids, sizeof (BufferId));
bid->used = false;
b = p->buffers[i].buffer;
bid->buf_ptr = SPA_MEMBER (mid->ptr, p->buffers[i].offset, void);
{
size_t size;
size = pinos_serialize_buffer_get_size (p->buffers[i].buffer);
b = bid->buf = malloc (size);
pinos_serialize_buffer_copy_into (b, p->buffers[i].buffer);
}
bid->id = b->id;
if (bid->id != len) {
pinos_log_warn ("unexpected id %u found, expected %u", bid->id, len);
impl->in_order = false;
}
pinos_log_debug ("add buffer %d %d %zd", mid->id, bid->id, p->buffers[i].offset);
for (j = 0; j < b->n_metas; j++) {
SpaMeta *m = &b->metas[j];
m->data = SPA_MEMBER (bid->buf_ptr, offset, void);
offset += m->size;
}
for (j = 0; j < b->n_datas; j++) {
SpaData *d = &b->datas[j];
d->chunk = SPA_MEMBER (bid->buf_ptr, offset + sizeof (SpaChunk) * j, SpaChunk);
switch (d->type) {
case SPA_DATA_TYPE_ID:
{
MemId *bmid = find_mem (stream, SPA_PTR_TO_UINT32 (d->data));
d->type = SPA_DATA_TYPE_MEMFD;
d->data = NULL;
d->fd = bmid->fd;
pinos_log_debug (" data %d %u -> fd %d", j, bmid->id, bmid->fd);
break;
}
case SPA_DATA_TYPE_MEMPTR:
{
d->data = SPA_MEMBER (bid->buf_ptr, SPA_PTR_TO_INT (d->data), void);
d->fd = -1;
pinos_log_debug (" data %d %u -> mem %p", j, bid->id, d->data);
break;
}
default:
pinos_log_warn ("unknown buffer data type %d", d->type);
break;
}
}
pinos_signal_emit (&stream->add_buffer, stream, bid->id);
}
if (p->n_buffers) {
add_state_change (stream, SPA_NODE_STATE_PAUSED, false);
} else {
clear_mems (stream);
add_state_change (stream, SPA_NODE_STATE_READY, false);
}
add_async_complete (stream, p->seq, SPA_RESULT_OK, true);
if (p->n_buffers)
stream_set_state (stream, PINOS_STREAM_STATE_PAUSED, NULL);
else
stream_set_state (stream, PINOS_STREAM_STATE_READY, NULL);
break;
}
case PINOS_MESSAGE_NODE_EVENT:
{
PinosMessageNodeEvent *p = message;
handle_node_event (stream, p->event);
break;
}
case PINOS_MESSAGE_NODE_COMMAND:
{
PinosMessageNodeCommand *p = message;
handle_node_command (stream, p->seq, p->command);
break;
}
case PINOS_MESSAGE_PORT_COMMAND:
{
break;
}
case PINOS_MESSAGE_TRANSPORT_UPDATE:
{
PinosMessageTransportUpdate *p = message;
PinosTransportInfo info;
info.memfd = p->memfd;
if (info.memfd == -1)
break;
info.offset = p->offset;
info.size = p->size;
if (impl->trans)
pinos_transport_destroy (impl->trans);
impl->trans = pinos_transport_new_from_info (&info);
pinos_log_debug ("transport update %d %p", impl->rtfd, impl->trans);
break;
}
default:
case PINOS_MESSAGE_INVALID:
pinos_log_warn ("unhandled message %d", type);
break;
}
return SPA_RESULT_OK;
}
static void
on_node_proxy_destroy (PinosListener *listener,
PinosProxy *proxy)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (listener, PinosStreamImpl, node_proxy_destroy);
PinosStream *this = &impl->this;
impl->disconnecting = false;
impl->node_proxy = NULL;
pinos_signal_remove (&impl->node_proxy_destroy);
stream_set_state (this, PINOS_STREAM_STATE_UNCONNECTED, NULL);
}
/**
* pinos_stream_connect:
* @stream: a #PinosStream
* @direction: the stream direction
* @mode: a #PinosStreamMode
* @port_path: the port path to connect to or %NULL to get the default port
* @flags: a #PinosStreamFlags
* @possible_formats: (transfer full): a #GPtrArray with possible accepted formats
*
* Connect @stream for input or output on @port_path.
*
* When @mode is #PINOS_STREAM_MODE_BUFFER, you should connect to the new-buffer
* signal and use pinos_stream_capture_buffer() to get the latest metadata and
* data.
*
* Returns: %true on success.
*/
bool
pinos_stream_connect (PinosStream *stream,
PinosDirection direction,
PinosStreamMode mode,
const char *port_path,
PinosStreamFlags flags,
unsigned int n_possible_formats,
SpaFormat **possible_formats)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
PinosMessageCreateClientNode ccn;
impl->direction = direction == PINOS_DIRECTION_INPUT ? SPA_DIRECTION_INPUT : SPA_DIRECTION_OUTPUT;
impl->port_id = 0;
impl->mode = mode;
impl->flags = flags;
impl->n_possible_formats = n_possible_formats;
impl->possible_formats = possible_formats;
stream_set_state (stream, PINOS_STREAM_STATE_CONNECTING, NULL);
if (stream->properties == NULL)
stream->properties = pinos_properties_new (NULL, NULL);
if (port_path)
pinos_properties_set (stream->properties,
"pinos.target.node", port_path);
impl->node_proxy = pinos_proxy_new (stream->context,
SPA_ID_INVALID,
stream->context->uri.client_node);
if (impl->node_proxy == NULL)
return false;
pinos_signal_add (&impl->node_proxy->destroy_signal,
&impl->node_proxy_destroy,
on_node_proxy_destroy);
pinos_proxy_set_dispatch (impl->node_proxy,
stream_dispatch_func,
impl);
ccn.seq = ++impl->seq;
ccn.name = "client-node";
ccn.props = &stream->properties->dict;
ccn.new_id = impl->node_proxy->id;
pinos_proxy_send_message (stream->context->core_proxy,
PINOS_MESSAGE_CREATE_CLIENT_NODE,
&ccn,
true);
return true;
}
/**
* pinos_stream_finish_format:
* @stream: a #PinosStream
* @res: a #SpaResult
* @params: an array of pointers to #SpaAllocParam
* @n_params: number of elements in @params
*
* Complete the negotiation process with result code @res.
*
* This function should be called after notification of the format.
* When @res indicates success, @params contain the parameters for the
* allocation state.
*
* Returns: %true on success
*/
bool
pinos_stream_finish_format (PinosStream *stream,
SpaResult res,
SpaAllocParam **params,
unsigned int n_params)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
impl->port_info.params = params;
impl->port_info.n_params = n_params;
if (SPA_RESULT_IS_OK (res)) {
add_port_update (stream, PINOS_MESSAGE_PORT_UPDATE_INFO |
PINOS_MESSAGE_PORT_UPDATE_FORMAT,
false);
if (impl->format) {
add_state_change (stream, SPA_NODE_STATE_READY, false);
} else {
clear_buffers (stream);
add_state_change (stream, SPA_NODE_STATE_CONFIGURE, false);
}
}
impl->port_info.params = NULL;
impl->port_info.n_params = 0;
add_async_complete (stream, impl->pending_seq, res, true);
impl->pending_seq = SPA_ID_INVALID;
return true;
}
/**
* pinos_stream_start:
* @stream: a #PinosStream
*
* Start capturing from @stream.
*
* Returns: %true on success.
*/
bool
pinos_stream_start (PinosStream *stream)
{
return true;
}
/**
* pinos_stream_stop:
* @stream: a #PinosStream
*
* Stop capturing from @stream.
*
* Returns: %true on success.
*/
bool
pinos_stream_stop (PinosStream *stream)
{
return true;
}
/**
* pinos_stream_disconnect:
* @stream: a #PinosStream
*
* Disconnect @stream.
*
* Returns: %true on success
*/
bool
pinos_stream_disconnect (PinosStream *stream)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
PinosMessageDestroy msg;
impl->disconnecting = true;
unhandle_socket (stream);
msg.seq = ++impl->seq;
pinos_proxy_send_message (impl->node_proxy,
PINOS_MESSAGE_DESTROY,
&msg,
true);
return true;
}
bool
pinos_stream_get_time (PinosStream *stream,
PinosTime *time)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
int64_t now, elapsed;
struct timespec ts;
clock_gettime (CLOCK_MONOTONIC, &ts);
now = SPA_TIMESPEC_TO_TIME (&ts);
elapsed = (now - impl->last_monotonic) / 1000;
time->ticks = impl->last_ticks + (elapsed * impl->last_rate) / SPA_USEC_PER_SEC;
time->rate = impl->last_rate;
return true;
}
/**
* pinos_stream_get_empty_buffer:
* @stream: a #PinosStream
*
* Get the id of an empty buffer that can be filled
*
* Returns: the id of an empty buffer or #SPA_ID_INVALID when no buffer is
* available.
*/
uint32_t
pinos_stream_get_empty_buffer (PinosStream *stream)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
BufferId *bid;
pinos_array_for_each (bid, &impl->buffer_ids) {
if (!bid->used)
return bid->id;
}
return SPA_ID_INVALID;
}
/**
* pinos_stream_recycle_buffer:
* @stream: a #PinosStream
* @id: a buffer id
*
* Recycle the buffer with @id.
*
* Returns: %true on success.
*/
bool
pinos_stream_recycle_buffer (PinosStream *stream,
uint32_t id)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
SpaNodeEventReuseBuffer rb;
uint64_t cmd = 1;
rb.event.type = SPA_NODE_EVENT_TYPE_REUSE_BUFFER;
rb.event.size = sizeof (rb);
rb.port_id = impl->port_id;
rb.buffer_id = id;
pinos_transport_add_event (impl->trans, &rb.event);
write (impl->rtfd, &cmd, 8);
return true;
}
/**
* pinos_stream_peek_buffer:
* @stream: a #PinosStream
* @id: the buffer id
*
* Get the buffer with @id from @stream. This function should be called from
* the new-buffer signal callback.
*
* Returns: a #SpaBuffer or %NULL when there is no buffer.
*/
SpaBuffer *
pinos_stream_peek_buffer (PinosStream *stream,
uint32_t id)
{
BufferId *bid;
if ((bid = find_buffer (stream, id)))
return bid->buf;
return NULL;
}
/**
* pinos_stream_send_buffer:
* @stream: a #PinosStream
* @id: a buffer id
* @offset: the offset in the buffer
* @size: the size in the buffer
*
* Send a buffer with @id to @stream.
*
* For provider streams, this function should be called whenever there is a new frame
* available.
*
* Returns: %true when @id was handled
*/
bool
pinos_stream_send_buffer (PinosStream *stream,
uint32_t id)
{
PinosStreamImpl *impl = SPA_CONTAINER_OF (stream, PinosStreamImpl, this);
BufferId *bid;
if (impl->trans->outputs[0].buffer_id != SPA_ID_INVALID) {
pinos_log_debug ("can't send %u, pending buffer %u", id, impl->trans->outputs[0].buffer_id);
return false;
}
if ((bid = find_buffer (stream, id)) && !bid->used) {
bid->used = true;
impl->trans->outputs[0].buffer_id = id;
impl->trans->outputs[0].status = SPA_RESULT_OK;
send_have_output (stream);
} else {
pinos_log_debug ("stream %p: output %u was used", stream, id);
}
return true;
}
Reference in a new issue View git blame Copy permalink