mirrors/pulseaudio
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pulseaudio/pulseaudio.git synced 2025-11-06 13:29:56 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

A lot of updates, all necessary to get the native protocol ported:

Browse source 
* add an int64_t argument to pa_asyncmsgq because it is very difficult to pass 64 values otherwise
* simplify subclassing in pa_object
* s/drop/unlink/ at some places
* port the native protocol to the lock-free core (not tested, compiles fine)
* move synchronisation of playback streams into pa_sink_input
* add "start_corked" field to pa_sink_input_new_data
* allow casting of NULL values in pa_object


git-svn-id: file:///home/lennart/svn/public/pulseaudio/branches/lennart@1562 fefdeb5f-60dc-0310-8127-8f9354f1896f
This commit is contained in:
Lennart Poettering 2007-07-31 22:44:53 +00:00
parent a82505e72f
commit 0defdfb560
27 changed files with 1229 additions and 770 deletions
Download patch file Download diff file Expand all files Collapse all files

43
src/Makefile.am
View file

@ -722,10 +722,10 @@ modlibexec_LTLIBRARIES = \
libauthkey-prop.la \
libstrlist.la \
libprotocol-simple.la \
libprotocol-http.la
libprotocol-http.la \
libprotocol-native.la
# libprotocol-esound.la
# libprotocol-native.la
# We need to emulate sendmsg/recvmsg to support this on Win32
if !OS_IS_WIN32
@ -879,11 +879,10 @@ modlibexec_LTLIBRARIES += \
module-volume-restore.la \
module-rescue-streams.la \
module-http-protocol-tcp.la \
module-sine.la
module-sine.la \
module-native-protocol-tcp.la \
module-native-protocol-fd.la
# module-esound-protocol-tcp.la \
# module-native-protocol-tcp.la \
# module-native-protocol-fd.la \
# module-combine.la \
# module-tunnel-sink.la \
# module-tunnel-source.la \
@ -899,10 +898,10 @@ modlibexec_LTLIBRARIES += \
if HAVE_AF_UNIX
modlibexec_LTLIBRARIES += \
module-cli-protocol-unix.la \
module-simple-protocol-unix.la
module-http-protocol-unix.la
# module-esound-protocol-unix.la \
# module-native-protocol-unix.la
module-simple-protocol-unix.la \
module-http-protocol-unix.la \
module-native-protocol-unix.la
# module-esound-protocol-unix.la
endif
if HAVE_MKFIFO
@ -1083,20 +1082,20 @@ module_http_protocol_unix_la_LIBADD = $(AM_LIBADD) libpulsecore.la libprotocol-h
# Native protocol
#module_native_protocol_tcp_la_SOURCES = modules/module-protocol-stub.c
#module_native_protocol_tcp_la_CFLAGS = -DUSE_TCP_SOCKETS -DUSE_PROTOCOL_NATIVE $(AM_CFLAGS)
#module_native_protocol_tcp_la_LDFLAGS = -module -avoid-version
#module_native_protocol_tcp_la_LIBADD = $(AM_LIBADD) libpulsecore.la libprotocol-native.la libsocket-server.la
module_native_protocol_tcp_la_SOURCES = modules/module-protocol-stub.c
module_native_protocol_tcp_la_CFLAGS = -DUSE_TCP_SOCKETS -DUSE_PROTOCOL_NATIVE $(AM_CFLAGS)
module_native_protocol_tcp_la_LDFLAGS = -module -avoid-version
module_native_protocol_tcp_la_LIBADD = $(AM_LIBADD) libpulsecore.la libprotocol-native.la libsocket-server.la
#module_native_protocol_unix_la_SOURCES = modules/module-protocol-stub.c
#module_native_protocol_unix_la_CFLAGS = -DUSE_UNIX_SOCKETS -DUSE_PROTOCOL_NATIVE $(AM_CFLAGS)
#module_native_protocol_unix_la_LDFLAGS = -module -avoid-version
#module_native_protocol_unix_la_LIBADD = $(AM_LIBADD) libpulsecore.la libprotocol-native.la libsocket-server.la libsocket-util.la
module_native_protocol_unix_la_SOURCES = modules/module-protocol-stub.c
module_native_protocol_unix_la_CFLAGS = -DUSE_UNIX_SOCKETS -DUSE_PROTOCOL_NATIVE $(AM_CFLAGS)
module_native_protocol_unix_la_LDFLAGS = -module -avoid-version
module_native_protocol_unix_la_LIBADD = $(AM_LIBADD) libpulsecore.la libprotocol-native.la libsocket-server.la libsocket-util.la
#module_native_protocol_fd_la_SOURCES = modules/module-native-protocol-fd.c
#module_native_protocol_fd_la_CFLAGS = $(AM_CFLAGS)
#module_native_protocol_fd_la_LDFLAGS = -module -avoid-version
#module_native_protocol_fd_la_LIBADD = $(AM_LIBADD) libpulsecore.la libprotocol-native.la libsocket-server.la libsocket-util.la libiochannel.la
module_native_protocol_fd_la_SOURCES = modules/module-native-protocol-fd.c
module_native_protocol_fd_la_CFLAGS = $(AM_CFLAGS)
module_native_protocol_fd_la_LDFLAGS = -module -avoid-version
module_native_protocol_fd_la_LIBADD = $(AM_LIBADD) libpulsecore.la libprotocol-native.la libsocket-server.la libsocket-util.la libiochannel.la
# EsounD protocol

13
src/modules/module-alsa-sink.c
View file

@ -302,7 +302,7 @@ fail:
return -1;
}
static int sink_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *chunk) {
static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u = PA_SINK(o)->userdata;
switch (code) {
@ -347,7 +347,7 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *
break;
}
return pa_sink_process_msg(o, code, data, chunk);
return pa_sink_process_msg(o, code, data, offset, chunk);
}
static int mixer_callback(snd_mixer_elem_t *elem, unsigned int mask) {
@ -510,12 +510,13 @@ static void thread_func(void *userdata) {
int code;
void *data;
pa_memchunk chunk;
int64_t offset;
int r;
/* pa_log("loop"); */
/* Check whether there is a message for us to process */
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &chunk, 0) == 0) {
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &offset, &chunk, 0) == 0) {
int ret;
/* pa_log("processing msg"); */
@ -525,7 +526,7 @@ static void thread_func(void *userdata) {
goto finish;
}
ret = pa_asyncmsgq_dispatch(object, code, data, &chunk);
ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
pa_asyncmsgq_done(u->asyncmsgq, ret);
continue;
}
@ -660,7 +661,7 @@ static void thread_func(void *userdata) {
fail:
/* We have to continue processing messages until we receive the
* SHUTDOWN message */
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, NULL, NULL);
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
pa_asyncmsgq_wait_for(u->asyncmsgq, PA_MESSAGE_SHUTDOWN);
finish:
@ -893,7 +894,7 @@ void pa__done(pa_core *c, pa_module*m) {
pa_sink_disconnect(u->sink);
if (u->thread) {
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, NULL);
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
pa_thread_free(u->thread);
}

13
src/modules/module-alsa-source.c
View file

@ -290,7 +290,7 @@ fail:
return -1;
}
static int source_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *chunk) {
static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u = PA_SOURCE(o)->userdata;
switch (code) {
@ -335,7 +335,7 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk
break;
}
return pa_source_process_msg(o, code, data, chunk);
return pa_source_process_msg(o, code, data, offset, chunk);
}
static int mixer_callback(snd_mixer_elem_t *elem, unsigned int mask) {
@ -498,12 +498,13 @@ static void thread_func(void *userdata) {
int code;
void *data;
int r;
int64_t offset;
pa_memchunk chunk;
/* pa_log("loop"); */
/* Check whether there is a message for us to process */
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &chunk, 0) == 0) {
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &offset, &chunk, 0) == 0) {
int ret;
/* pa_log("processing msg"); */
@ -513,7 +514,7 @@ static void thread_func(void *userdata) {
goto finish;
}
ret = pa_asyncmsgq_dispatch(object, code, data, &chunk);
ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
pa_asyncmsgq_done(u->asyncmsgq, ret);
continue;
}
@ -634,7 +635,7 @@ static void thread_func(void *userdata) {
fail:
/* We have to continue processing messages until we receive the
* SHUTDOWN message */
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, NULL, NULL);
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
pa_asyncmsgq_wait_for(u->asyncmsgq, PA_MESSAGE_SHUTDOWN);
finish:
@ -864,7 +865,7 @@ void pa__done(pa_core *c, pa_module*m) {
pa_source_disconnect(u->source);
if (u->thread) {
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, NULL);
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
pa_thread_free(u->thread);
}

13
src/modules/module-null-sink.c
View file

@ -83,7 +83,7 @@ static const char* const valid_modargs[] = {
NULL
};
static int sink_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *chunk) {
static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u = PA_SINK(o)->userdata;
switch (code) {
@ -107,7 +107,7 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *
}
}
return pa_sink_process_msg(o, code, data, chunk);
return pa_sink_process_msg(o, code, data, offset, chunk);
}
static void thread_func(void *userdata) {
@ -131,9 +131,10 @@ static void thread_func(void *userdata) {
pa_memchunk chunk;
int r, timeout;
struct timeval now;
int64_t offset;
/* Check whether there is a message for us to process */
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &chunk, 0) == 0) {
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &offset, &chunk, 0) == 0) {
int ret;
if (!object && code == PA_MESSAGE_SHUTDOWN) {
@ -141,7 +142,7 @@ static void thread_func(void *userdata) {
goto finish;
}
ret = pa_asyncmsgq_dispatch(object, code, data, &chunk);
ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
pa_asyncmsgq_done(u->asyncmsgq, ret);
continue;
}
@ -190,7 +191,7 @@ static void thread_func(void *userdata) {
fail:
/* We have to continue processing messages until we receive the
* SHUTDOWN message */
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, NULL, NULL);
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
pa_asyncmsgq_wait_for(u->asyncmsgq, PA_MESSAGE_SHUTDOWN);
finish:
@ -271,7 +272,7 @@ void pa__done(pa_core *c, pa_module*m) {
pa_sink_disconnect(u->sink);
if (u->thread) {
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, NULL);
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
pa_thread_free(u->thread);
}

21
src/modules/module-oss.c
View file

@ -581,7 +581,7 @@ fail:
return -1;
}
static int sink_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *chunk) {
static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u = PA_SINK(o)->userdata;
int do_trigger = 0, ret, quick = 1;
@ -673,7 +673,7 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *
break;
}
ret = pa_sink_process_msg(o, code, data, chunk);
ret = pa_sink_process_msg(o, code, data, offset, chunk);
if (do_trigger)
trigger(u, quick);
@ -681,7 +681,7 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *
return ret;
}
static int source_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *chunk) {
static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u = PA_SOURCE(o)->userdata;
int do_trigger = 0, ret, quick = 1;
@ -770,7 +770,7 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk
break;
}
ret = pa_source_process_msg(o, code, data, chunk);
ret = pa_source_process_msg(o, code, data, offset, chunk);
if (do_trigger)
trigger(u, quick);
@ -807,11 +807,12 @@ static void thread_func(void *userdata) {
void *data;
pa_memchunk chunk;
int r;
int64_t offset;
/* pa_log("loop"); */
/* Check whether there is a message for us to process */
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &chunk, 0) == 0) {
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &offset, &chunk, 0) == 0) {
int ret;
/* pa_log("processing msg"); */
@ -821,7 +822,7 @@ static void thread_func(void *userdata) {
goto finish;
}
ret = pa_asyncmsgq_dispatch(object, code, data, &chunk);
ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
pa_asyncmsgq_done(u->asyncmsgq, ret);
continue;
}
@ -1051,7 +1052,7 @@ static void thread_func(void *userdata) {
fail:
/* We have to continue processing messages until we receive the
* SHUTDOWN message */
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, NULL, NULL);
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
pa_asyncmsgq_wait_for(u->asyncmsgq, PA_MESSAGE_SHUTDOWN);
finish:
@ -1300,9 +1301,9 @@ go_on:
/* Read mixer settings */
if (u->source)
pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->source), PA_SOURCE_MESSAGE_GET_VOLUME, &u->source->volume, NULL, NULL);
pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->source), PA_SOURCE_MESSAGE_GET_VOLUME, &u->source->volume, 0, NULL, NULL);
if (u->sink)
pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->sink), PA_SINK_MESSAGE_GET_VOLUME, &u->sink->volume, NULL, NULL);
pa_asyncmsgq_post(u->asyncmsgq, PA_MSGOBJECT(u->sink), PA_SINK_MESSAGE_GET_VOLUME, &u->sink->volume, 0, NULL, NULL);
return 0;
@ -1335,7 +1336,7 @@ void pa__done(pa_core *c, pa_module*m) {
pa_source_disconnect(u->source);
if (u->thread) {
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, NULL);
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
pa_thread_free(u->thread);
}

13
src/modules/module-pipe-sink.c
View file

@ -84,7 +84,7 @@ static const char* const valid_modargs[] = {
NULL
};
static int sink_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *chunk) {
static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u = PA_SINK(o)->userdata;
switch (code) {
@ -103,7 +103,7 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, pa_memchunk *
}
}
return pa_sink_process_msg(o, code, data, chunk);
return pa_sink_process_msg(o, code, data, offset, chunk);
}
static void thread_func(void *userdata) {
@ -133,9 +133,10 @@ static void thread_func(void *userdata) {
void *data;
pa_memchunk chunk;
int r;
int64_t offset;
/* Check whether there is a message for us to process */
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &chunk, 0) == 0) {
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &offset, &chunk, 0) == 0) {
int ret;
if (!object && code == PA_MESSAGE_SHUTDOWN) {
@ -143,7 +144,7 @@ static void thread_func(void *userdata) {
goto finish;
}
ret = pa_asyncmsgq_dispatch(object, code, data, &chunk);
ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
pa_asyncmsgq_done(u->asyncmsgq, ret);
continue;
}
@ -224,7 +225,7 @@ static void thread_func(void *userdata) {
fail:
/* We have to continue processing messages until we receive the
* SHUTDOWN message */
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, NULL, NULL);
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
pa_asyncmsgq_wait_for(u->asyncmsgq, PA_MESSAGE_SHUTDOWN);
finish:
@ -326,7 +327,7 @@ void pa__done(pa_core *c, pa_module*m) {
pa_sink_disconnect(u->sink);
if (u->thread) {
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, NULL);
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
pa_thread_free(u->thread);
}

9
src/modules/module-pipe-source.c
View file

@ -111,9 +111,10 @@ static void thread_func(void *userdata) {
void *data;
pa_memchunk chunk;
int r;
int64_t offset;
/* Check whether there is a message for us to process */
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &chunk, 0) == 0) {
if (pa_asyncmsgq_get(u->asyncmsgq, &object, &code, &data, &offset, &chunk, 0) == 0) {
int ret;
if (!object && code == PA_MESSAGE_SHUTDOWN) {
@ -121,7 +122,7 @@ static void thread_func(void *userdata) {
goto finish;
}
ret = pa_asyncmsgq_dispatch(object, code, data, &chunk);
ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
pa_asyncmsgq_done(u->asyncmsgq, ret);
continue;
}
@ -202,7 +203,7 @@ static void thread_func(void *userdata) {
fail:
/* We have to continue processing messages until we receive the
* SHUTDOWN message */
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, NULL, NULL);
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
pa_asyncmsgq_wait_for(u->asyncmsgq, PA_MESSAGE_SHUTDOWN);
finish:
@ -303,7 +304,7 @@ void pa__done(pa_core *c, pa_module*m) {
pa_source_disconnect(u->source);
if (u->thread) {
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, NULL);
pa_asyncmsgq_send(u->asyncmsgq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
pa_thread_free(u->thread);
}

20
src/pulsecore/asyncmsgq.c
View file

@ -46,6 +46,7 @@ struct asyncmsgq_item {
pa_msgobject *object;
void *userdata;
pa_free_cb_t free_cb;
int64_t offset;
pa_memchunk memchunk;
pa_semaphore *semaphore;
int ret;
@ -96,7 +97,7 @@ void pa_asyncmsgq_free(pa_asyncmsgq *a) {
pa_xfree(a);
}
void pa_asyncmsgq_post(pa_asyncmsgq *a, pa_msgobject *object, int code, const void *userdata, const pa_memchunk *chunk, pa_free_cb_t free_cb) {
void pa_asyncmsgq_post(pa_asyncmsgq *a, pa_msgobject *object, int code, const void *userdata, int64_t offset, const pa_memchunk *chunk, pa_free_cb_t free_cb) {
struct asyncmsgq_item *i;
pa_assert(a);
@ -107,6 +108,7 @@ void pa_asyncmsgq_post(pa_asyncmsgq *a, pa_msgobject *object, int code, const vo
i->object = object ? pa_msgobject_ref(object) : NULL;
i->userdata = (void*) userdata;
i->free_cb = free_cb;
i->offset = offset;
if (chunk) {
pa_assert(chunk->memblock);
i->memchunk = *chunk;
@ -121,7 +123,7 @@ void pa_asyncmsgq_post(pa_asyncmsgq *a, pa_msgobject *object, int code, const vo
pa_mutex_unlock(a->mutex);
}
int pa_asyncmsgq_send(pa_asyncmsgq *a, pa_msgobject *object, int code, const void *userdata, const pa_memchunk *chunk) {
int pa_asyncmsgq_send(pa_asyncmsgq *a, pa_msgobject *object, int code, const void *userdata, int64_t offset, const pa_memchunk *chunk) {
struct asyncmsgq_item i;
pa_assert(a);
@ -130,6 +132,7 @@ int pa_asyncmsgq_send(pa_asyncmsgq *a, pa_msgobject *object, int code, const voi
i.userdata = (void*) userdata;
i.free_cb = NULL;
i.ret = -1;
i.offset = offset;
if (chunk) {
pa_assert(chunk->memblock);
i.memchunk = *chunk;
@ -148,7 +151,7 @@ int pa_asyncmsgq_send(pa_asyncmsgq *a, pa_msgobject *object, int code, const voi
return i.ret;
}
int pa_asyncmsgq_get(pa_asyncmsgq *a, pa_msgobject **object, int *code, void **userdata, pa_memchunk *chunk, int wait) {
int pa_asyncmsgq_get(pa_asyncmsgq *a, pa_msgobject **object, int *code, void **userdata, int64_t *offset, pa_memchunk *chunk, int wait) {
pa_assert(a);
pa_assert(code);
pa_assert(!a->current);
@ -163,6 +166,8 @@ int pa_asyncmsgq_get(pa_asyncmsgq *a, pa_msgobject **object, int *code, void **u
*code = a->current->code;
if (userdata)
*userdata = a->current->userdata;
if (offset)
*offset = a->current->offset;
if (object) {
if ((*object = a->current->object))
pa_msgobject_assert_ref(*object);
@ -207,13 +212,14 @@ int pa_asyncmsgq_wait_for(pa_asyncmsgq *a, int code) {
do {
pa_msgobject *o;
void *data;
int64_t offset;
pa_memchunk chunk;
int ret;
if (pa_asyncmsgq_get(a, &o, &c, &data, &chunk, 1) < 0)
if (pa_asyncmsgq_get(a, &o, &c, &data, &offset, &chunk, 1) < 0)
return -1;
ret = pa_asyncmsgq_dispatch(o, c, data, &chunk);
ret = pa_asyncmsgq_dispatch(o, c, data, offset, &chunk);
pa_asyncmsgq_done(a, ret);
} while (c != code);
@ -239,10 +245,10 @@ void pa_asyncmsgq_after_poll(pa_asyncmsgq *a) {
pa_asyncq_after_poll(a->asyncq);
}
int pa_asyncmsgq_dispatch(pa_msgobject *object, int code, void *userdata, pa_memchunk *memchunk) {
int pa_asyncmsgq_dispatch(pa_msgobject *object, int code, void *userdata, int64_t offset, pa_memchunk *memchunk) {
if (object)
return object->process_msg(object, code, userdata, memchunk);
return object->process_msg(object, code, userdata, offset, memchunk);
return 0;
}

8
src/pulsecore/asyncmsgq.h
View file

@ -57,11 +57,11 @@ typedef struct pa_asyncmsgq pa_asyncmsgq;
pa_asyncmsgq* pa_asyncmsgq_new(size_t size);
void pa_asyncmsgq_free(pa_asyncmsgq* q);
void pa_asyncmsgq_post(pa_asyncmsgq *q, pa_msgobject *object, int code, const void *userdata, const pa_memchunk *memchunk, pa_free_cb_t userdata_free_cb);
int pa_asyncmsgq_send(pa_asyncmsgq *q, pa_msgobject *object, int code, const void *userdata, const pa_memchunk *memchunk);
void pa_asyncmsgq_post(pa_asyncmsgq *q, pa_msgobject *object, int code, const void *userdata, int64_t offset, const pa_memchunk *memchunk, pa_free_cb_t userdata_free_cb);
int pa_asyncmsgq_send(pa_asyncmsgq *q, pa_msgobject *object, int code, const void *userdata, int64_t offset, const pa_memchunk *memchunk);
int pa_asyncmsgq_get(pa_asyncmsgq *q, pa_msgobject **object, int *code, void **userdata, pa_memchunk *memchunk, int wait);
int pa_asyncmsgq_dispatch(pa_msgobject *object, int code, void *userdata, pa_memchunk *memchunk);
int pa_asyncmsgq_get(pa_asyncmsgq *q, pa_msgobject **object, int *code, void **userdata, int64_t *offset, pa_memchunk *memchunk, int wait);
int pa_asyncmsgq_dispatch(pa_msgobject *object, int code, void *userdata, int64_t offset, pa_memchunk *memchunk);
void pa_asyncmsgq_done(pa_asyncmsgq *q, int ret);
int pa_asyncmsgq_wait_for(pa_asyncmsgq *a, int code);

11
src/pulsecore/core.c
View file

@ -49,9 +49,9 @@
#include "core.h"
static PA_DEFINE_CHECK_TYPE(pa_core, core_check_type, pa_msgobject_check_type);
static PA_DEFINE_CHECK_TYPE(pa_core, pa_msgobject);
static int core_process_msg(pa_msgobject *o, int code, void *userdata, pa_memchunk *chunk) {
static int core_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
pa_core *c = PA_CORE(o);
pa_core_assert_ref(c);
@ -79,13 +79,14 @@ static void asyncmsgq_cb(pa_mainloop_api*api, pa_io_event* e, int fd, pa_io_even
pa_msgobject *object;
int code;
void *data;
int64_t offset;
pa_memchunk chunk;
/* Check whether there is a message for us to process */
while (pa_asyncmsgq_get(c->asyncmsgq, &object, &code, &data, &chunk, 0) == 0) {
while (pa_asyncmsgq_get(c->asyncmsgq, &object, &code, &data, &offset, &chunk, 0) == 0) {
int ret;
ret = pa_asyncmsgq_dispatch(object, code, data, &chunk);
ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
pa_asyncmsgq_done(c->asyncmsgq, ret);
}
@ -116,7 +117,7 @@ pa_core* pa_core_new(pa_mainloop_api *m, int shared) {
}
}
c = pa_msgobject_new(pa_core, core_check_type);
c = pa_msgobject_new(pa_core);
c->parent.parent.free = core_free;
c->parent.process_msg = core_process_msg;

13
src/pulsecore/msgobject.c
View file

@ -28,15 +28,22 @@
#include "msgobject.h"
PA_DEFINE_CHECK_TYPE(pa_msgobject, pa_msgobject_check_type, pa_object_check_type);
PA_DEFINE_CHECK_TYPE(pa_msgobject, pa_object);
pa_msgobject *pa_msgobject_new_internal(size_t size, const char *type_name, int (*check_type)(pa_object *o, const char *type_name)) {
pa_msgobject *pa_msgobject_new_internal(size_t size, const char *type_name, int (*check_type)(const char *type_name)) {
pa_msgobject *o;
pa_assert(size > sizeof(pa_msgobject));
pa_assert(type_name);
o = PA_MSGOBJECT(pa_object_new_internal(size, type_name, check_type ? check_type : pa_msgobject_check_type));
if (!check_type)
check_type = pa_msgobject_check_type;
pa_assert(check_type(type_name));
pa_assert(check_type("pa_object"));
pa_assert(check_type("pa_msgobject"));
o = PA_MSGOBJECT(pa_object_new_internal(size, type_name, check_type));
o->process_msg = NULL;
return o;
}

8
src/pulsecore/msgobject.h
View file

@ -37,14 +37,14 @@ typedef struct pa_msgobject pa_msgobject;
struct pa_msgobject {
pa_object parent;
int (*process_msg)(pa_msgobject *o, int code, void *userdata, pa_memchunk *chunk);
int (*process_msg)(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
};
pa_msgobject *pa_msgobject_new_internal(size_t size, const char *type_name, int (*check_type)(pa_object *o, const char *type_name));
pa_msgobject *pa_msgobject_new_internal(size_t size, const char *type_name, int (*check_type)(const char *type_name));
int pa_msgobject_check_type(pa_object *o, const char *type);
int pa_msgobject_check_type(const char *type);
#define pa_msgobject_new(type, check_type) ((type*) pa_msgobject_new_internal(sizeof(type), #type, check_type))
#define pa_msgobject_new(type) ((type*) pa_msgobject_new_internal(sizeof(type), #type, type##_check_type))
#define pa_msgobject_free ((void (*) (pa_msgobject* o)) pa_object_free)
#define PA_MSGOBJECT(o) pa_msgobject_cast(o)

2
src/pulsecore/native-common.h
View file

@ -115,6 +115,8 @@ enum {
PA_COMMAND_MOVE_SINK_INPUT,
PA_COMMAND_MOVE_SOURCE_OUTPUT,
PA_COMMAND_SET_SINK_INPUT_MUTE,
PA_COMMAND_MAX
};

13
src/pulsecore/object.c
View file

@ -28,17 +28,23 @@
#include "object.h"
pa_object *pa_object_new_internal(size_t size, const char *type_name, int (*check_type)(pa_object *o, const char *type_name)) {
pa_object *pa_object_new_internal(size_t size, const char *type_name, int (*check_type)(const char *type_name)) {
pa_object *o;
pa_assert(size > sizeof(pa_object));
pa_assert(type_name);
if (!check_type)
check_type = pa_object_check_type;
pa_assert(check_type(type_name));
pa_assert(check_type("pa_object"));
o = pa_xmalloc(size);
PA_REFCNT_INIT(o);
o->type_name = type_name;
o->free = pa_object_free;
o->check_type = check_type ? check_type : pa_object_check_type;
o->check_type = check_type;
return o;
}
@ -59,8 +65,7 @@ void pa_object_unref(pa_object *o) {
}
}
int pa_object_check_type(pa_object *o, const char *type_name) {
pa_assert(o);
int pa_object_check_type(const char *type_name) {
pa_assert(type_name);
return type_name == "pa_object" || strcmp(type_name, "pa_object") == 0;

25
src/pulsecore/object.h
View file

@ -38,20 +38,19 @@ struct pa_object {
PA_REFCNT_DECLARE;
const char *type_name;
void (*free)(pa_object *o);
int (*check_type)(pa_object *o, const char *type_name);
int (*check_type)(const char *type_name);
};
pa_object *pa_object_new_internal(size_t size, const char *type_name, int (*check_type)(pa_object *o, const char *type_name));
#define pa_object_new(type, check_type) ((type*) pa_object_new_internal(sizeof(type), #type, check_type)
pa_object *pa_object_new_internal(size_t size, const char *type_name, int (*check_type)(const char *type_name));
#define pa_object_new(type) ((type*) pa_object_new_internal(sizeof(type), #type, type##_check_type)
#define pa_object_free ((void (*) (pa_object* o)) pa_xfree)
int pa_object_check_type(pa_object *o, const char *type);
int pa_object_check_type(const char *type);
static inline int pa_object_isinstance(void *o) {
pa_object *obj = (pa_object*) o;
pa_assert(obj);
return obj->check_type(obj, "pa_object");
return obj ? obj->check_type("pa_object") : 0;
}
pa_object *pa_object_ref(pa_object *o);
@ -63,19 +62,18 @@ static inline int pa_object_refcnt(pa_object *o) {
static inline pa_object* pa_object_cast(void *o) {
pa_object *obj = (pa_object*) o;
pa_assert(obj->check_type(obj, "pa_object"));
pa_assert(!obj || obj->check_type("pa_object"));
return obj;
}
#define pa_object_assert_ref(o) pa_assert(pa_object_refcnt(o))
#define pa_object_assert_ref(o) pa_assert(pa_object_refcnt(o) > 0)
#define PA_OBJECT(o) pa_object_cast(o)
#define PA_DECLARE_CLASS(c) \
static inline int c##_isinstance(void *o) { \
pa_object *obj = (pa_object*) o; \
pa_assert(obj); \
return obj->check_type(obj, #c); \
return obj ? obj->check_type(#c) : 1; \
} \
static inline c* c##_cast(void *o) { \
pa_assert(c##_isinstance(o)); \
@ -95,14 +93,13 @@ static inline pa_object* pa_object_cast(void *o) {
} \
struct __stupid_useless_struct_to_allow_trailing_semicolon
#define PA_DEFINE_CHECK_TYPE(c, func, parent) \
int func(pa_object *o, const char *type) { \
pa_assert(o); \
#define PA_DEFINE_CHECK_TYPE(c, parent) \
int c##_check_type(const char *type) { \
pa_assert(type); \
if (type == #c || \
strcmp(type, #c) == 0) \
return 1; \
return parent(o, type); \
return parent##_check_type(type); \
} \
struct __stupid_useless_struct_to_allow_trailing_semicolon

1453
src/pulsecore/protocol-native.c
View file

File diff suppressed because it is too large Load diff

112
src/pulsecore/protocol-simple.c
View file

@ -67,7 +67,7 @@ typedef struct connection {
PA_DECLARE_CLASS(connection);
#define CONNECTION(o) (connection_cast(o))
static PA_DEFINE_CHECK_TYPE(connection, connection_check_type, pa_msgobject_check_type);
static PA_DEFINE_CHECK_TYPE(connection, pa_msgobject);
struct pa_protocol_simple {
pa_module *module;
@ -91,9 +91,9 @@ enum {
};
enum {
MESSAGE_REQUEST_DATA, /* data requested from sink input from the main loop */
MESSAGE_POST_DATA, /* data from source output to main loop */
MESSAGE_DROP_CONNECTION /* Please drop a aconnection now */
CONNECTION_MESSAGE_REQUEST_DATA, /* data requested from sink input from the main loop */
CONNECTION_MESSAGE_POST_DATA, /* data from source output to main loop */
CONNECTION_MESSAGE_DROP_CONNECTION /* Please drop a aconnection now */
};
@ -102,27 +102,10 @@ enum {
#define RECORD_BUFFER_SECONDS (5)
#define RECORD_BUFFER_FRAGMENTS (100)
static void connection_free(pa_object *o) {
connection *c = CONNECTION(o);
static void connection_unlink(connection *c) {
pa_assert(c);
if (c->playback.current_memblock)
pa_memblock_unref(c->playback.current_memblock);
if (c->io)
pa_iochannel_free(c->io);
if (c->input_memblockq)
pa_memblockq_free(c->input_memblockq);
if (c->output_memblockq)
pa_memblockq_free(c->output_memblockq);
pa_xfree(c);
}
static void connection_drop(connection *c) {
pa_assert(c);
if (!pa_idxset_remove_by_data(c->protocol->connections, c, NULL))
if (!c->protocol)
return;
if (c->sink_input) {
@ -142,9 +125,30 @@ static void connection_drop(connection *c) {
c->client = NULL;
}
pa_assert_se(pa_idxset_remove_by_data(c->protocol->connections, c, NULL) == c);
c->protocol = NULL;
connection_unref(c);
}
static void connection_free(pa_object *o) {
connection *c = CONNECTION(o);
pa_assert(c);
connection_unref(c);
if (c->playback.current_memblock)
pa_memblock_unref(c->playback.current_memblock);
if (c->io)
pa_iochannel_free(c->io);
if (c->input_memblockq)
pa_memblockq_free(c->input_memblockq);
if (c->output_memblockq)
pa_memblockq_free(c->output_memblockq);
pa_xfree(c);
}
static int do_read(connection *c) {
pa_memchunk chunk;
ssize_t r;
@ -190,7 +194,7 @@ static int do_read(connection *c) {
c->playback.memblock_index += r;
pa_asyncmsgq_post(c->sink_input->sink->asyncmsgq, PA_MSGOBJECT(c->sink_input), SINK_INPUT_MESSAGE_POST_DATA, NULL, &chunk, NULL);
pa_asyncmsgq_post(c->sink_input->sink->asyncmsgq, PA_MSGOBJECT(c->sink_input), SINK_INPUT_MESSAGE_POST_DATA, NULL, 0, &chunk, NULL);
pa_atomic_sub(&c->playback.missing, r);
return 0;
@ -263,28 +267,28 @@ fail:
pa_iochannel_free(c->io);
c->io = NULL;
pa_asyncmsgq_post(c->sink_input->sink->asyncmsgq, PA_MSGOBJECT(c->sink_input), SINK_INPUT_MESSAGE_DISABLE_PREBUF, NULL, NULL, NULL);
pa_asyncmsgq_post(c->sink_input->sink->asyncmsgq, PA_MSGOBJECT(c->sink_input), SINK_INPUT_MESSAGE_DISABLE_PREBUF, NULL, 0, NULL, NULL);
} else
connection_drop(c);
connection_unlink(c);
}
static int connection_process_msg(pa_msgobject *o, int code, void*userdata, pa_memchunk *chunk) {
static int connection_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
connection *c = CONNECTION(o);
connection_assert_ref(c);
switch (code) {
case MESSAGE_REQUEST_DATA:
case CONNECTION_MESSAGE_REQUEST_DATA:
do_work(c);
break;
case MESSAGE_POST_DATA:
case CONNECTION_MESSAGE_POST_DATA:
/* pa_log("got data %u", chunk->length); */
pa_memblockq_push_align(c->output_memblockq, chunk);
do_work(c);
break;
case MESSAGE_DROP_CONNECTION:
connection_drop(c);
case CONNECTION_MESSAGE_DROP_CONNECTION:
connection_unlink(c);
break;
}
@ -294,13 +298,13 @@ static int connection_process_msg(pa_msgobject *o, int code, void*userdata, pa_m
/*** sink_input callbacks ***/
/* Called from thread context */
static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, pa_memchunk *chunk) {
static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
pa_sink_input *i = PA_SINK_INPUT(o);
connection*c;
pa_assert(i);
c = i->userdata;
pa_assert(c);
pa_sink_input_assert_ref(i);
c = CONNECTION(i->userdata);
connection_assert_ref(c);
switch (code) {
@ -330,7 +334,7 @@ static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, pa_
}
default:
return pa_sink_input_process_msg(o, code, userdata, chunk);
return pa_sink_input_process_msg(o, code, userdata, offset, chunk);
}
}
@ -349,7 +353,7 @@ static int sink_input_peek_cb(pa_sink_input *i, pa_memchunk *chunk) {
/* pa_log("peeked %u %i", r >= 0 ? chunk->length: 0, r); */
if (c->dead && r < 0)
pa_asyncmsgq_post(c->protocol->core->asyncmsgq, PA_MSGOBJECT(c), MESSAGE_DROP_CONNECTION, NULL, NULL, NULL);
pa_asyncmsgq_post(c->protocol->core->asyncmsgq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_DROP_CONNECTION, NULL, 0, NULL, NULL);
return r;
}
@ -369,19 +373,20 @@ static void sink_input_drop_cb(pa_sink_input *i, size_t length) {
if (new > old) {
if (pa_atomic_add(&c->playback.missing, new - old) <= 0)
pa_asyncmsgq_post(c->protocol->core->asyncmsgq, PA_MSGOBJECT(c), MESSAGE_REQUEST_DATA, NULL, NULL, NULL);
pa_asyncmsgq_post(c->protocol->core->asyncmsgq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_REQUEST_DATA, NULL, 0, NULL, NULL);
}
}
/* Called from main context */
static void sink_input_kill_cb(pa_sink_input *i) {
pa_assert(i);
pa_sink_input_assert_ref(i);
connection_drop(CONNECTION(i->userdata));
connection_unlink(CONNECTION(i->userdata));
}
/*** source_output callbacks ***/
/* Called from thread context */
static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
connection *c;
@ -390,24 +395,22 @@ static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk)
pa_assert(c);
pa_assert(chunk);
pa_asyncmsgq_post(c->protocol->core->asyncmsgq, PA_MSGOBJECT(c), MESSAGE_POST_DATA, NULL, chunk, NULL);
pa_asyncmsgq_post(c->protocol->core->asyncmsgq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
}
/* Called from main context */
static void source_output_kill_cb(pa_source_output *o) {
connection*c;
pa_source_output_assert_ref(o);
pa_assert(o);
c = o->userdata;
pa_assert(c);
connection_drop(c);
connection_unlink(CONNECTION(o->userdata));
}
/* Called from main context */
static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
connection*c;
pa_assert(o);
c = o->userdata;
c = CONNECTION(o->userdata);
pa_assert(c);
return pa_bytes_to_usec(pa_memblockq_get_length(c->output_memblockq), &c->source_output->sample_spec);
@ -419,16 +422,16 @@ static void client_kill_cb(pa_client *client) {
connection*c;
pa_assert(client);
c = client->userdata;
c = CONNECTION(client->userdata);
pa_assert(c);
connection_drop(c);
connection_unlink(c);
}
/*** pa_iochannel callbacks ***/
static void io_callback(pa_iochannel*io, void *userdata) {
connection *c = userdata;
connection *c = CONNECTION(userdata);
pa_assert(io);
pa_assert(c);
@ -453,7 +456,7 @@ static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata)
return;
}
c = pa_msgobject_new(connection, connection_check_type);
c = pa_msgobject_new(connection);
c->parent.parent.free = connection_free;
c->parent.process_msg = connection_process_msg;
c->io = io;
@ -547,7 +550,6 @@ static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata)
pa_source_output_put(c->source_output);
}
pa_iochannel_set_callback(c->io, io_callback, c);
pa_idxset_put(p->connections, c, NULL);
@ -555,7 +557,7 @@ static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata)
fail:
if (c)
connection_drop(c);
connection_unlink(c);
}
pa_protocol_simple* pa_protocol_simple_new(pa_core *core, pa_socket_server *server, pa_module *m, pa_modargs *ma) {
@ -618,7 +620,7 @@ void pa_protocol_simple_free(pa_protocol_simple *p) {
if (p->connections) {
while((c = pa_idxset_first(p->connections, NULL)))
connection_drop(c);
connection_unlink(c);
pa_idxset_free(p->connections, NULL, NULL);
}

64
src/pulsecore/sink-input.c
View file

@ -45,7 +45,7 @@
#define MOVE_BUFFER_LENGTH (1024*1024)
#define SILENCE_BUFFER_LENGTH (64*1024)
static PA_DEFINE_CHECK_TYPE(pa_sink_input, sink_input_check_type, pa_msgobject_check_type);
static PA_DEFINE_CHECK_TYPE(pa_sink_input, pa_msgobject);
static void sink_input_free(pa_object *o);
@ -110,6 +110,7 @@ pa_sink_input* pa_sink_input_new(
pa_return_null_if_fail(data->sink);
pa_return_null_if_fail(pa_sink_get_state(data->sink) != PA_SINK_DISCONNECTED);
pa_return_null_if_fail(!data->sync_base || (data->sync_base->sink == data->sink && pa_sink_input_get_state(data->sync_base) == PA_SINK_INPUT_CORKED));
if (!data->sample_spec_is_set)
data->sample_spec = data->sink->sample_spec;
@ -161,12 +162,12 @@ pa_sink_input* pa_sink_input_new(
data->resample_method = pa_resampler_get_method(resampler);
}
i = pa_msgobject_new(pa_sink_input, sink_input_check_type);
i = pa_msgobject_new(pa_sink_input);
i->parent.parent.free = sink_input_free;
i->parent.process_msg = pa_sink_input_process_msg;
i->core = core;
i->state = PA_SINK_INPUT_RUNNING;
i->state = data->start_corked ? PA_SINK_INPUT_CORKED : PA_SINK_INPUT_RUNNING;
i->flags = flags;
i->name = pa_xstrdup(data->name);
i->driver = pa_xstrdup(data->driver);
@ -181,6 +182,16 @@ pa_sink_input* pa_sink_input_new(
i->volume = data->volume;
i->muted = data->muted;
if (data->sync_base) {
i->sync_next = data->sync_base->sync_next;
i->sync_prev = data->sync_base;
if (data->sync_base->sync_next)
data->sync_base->sync_next->sync_prev = i;
data->sync_base->sync_next = i;
} else
i->sync_next = i->sync_prev = NULL;
i->peek = NULL;
i->drop = NULL;
i->kill = NULL;
@ -213,6 +224,7 @@ pa_sink_input* pa_sink_input_new(
}
static int sink_input_set_state(pa_sink_input *i, pa_sink_input_state_t state) {
pa_sink_input *ssync;
pa_assert(i);
if (state == PA_SINK_INPUT_DRAINED)
@ -221,10 +233,15 @@ static int sink_input_set_state(pa_sink_input *i, pa_sink_input_state_t state) {
if (i->state == state)
return 0;
if (pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), NULL) < 0)
if (pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) < 0)
return -1;
i->state = state;
for (ssync = i->sync_prev; ssync; ssync = ssync->sync_prev)
ssync->state = state;
for (ssync = i->sync_next; ssync; ssync = ssync->sync_next)
ssync->state = state;
return 0;
}
@ -232,10 +249,16 @@ void pa_sink_input_disconnect(pa_sink_input *i) {
pa_assert(i);
pa_return_if_fail(i->state != PA_SINK_INPUT_DISCONNECTED);
pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_REMOVE_INPUT, i, NULL);
if (i->sync_prev)
i->sync_prev->sync_next = i->sync_next;
if (i->sync_next)
i->sync_next->sync_prev = i->sync_prev;
i->sync_prev = i->sync_next = NULL;
pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_REMOVE_INPUT, i, 0, NULL);
pa_idxset_remove_by_data(i->sink->core->sink_inputs, i, NULL);
pa_idxset_remove_by_data(i->sink->inputs, i, NULL);
pa_sink_input_unref(i);
pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_REMOVE, i->index);
@ -248,6 +271,7 @@ void pa_sink_input_disconnect(pa_sink_input *i) {
i->kill = NULL;
i->get_latency = NULL;
i->underrun = NULL;
pa_sink_input_unref(i);
}
static void sink_input_free(pa_object *o) {
@ -281,7 +305,7 @@ void pa_sink_input_put(pa_sink_input *i) {
i->thread_info.volume = i->volume;
i->thread_info.muted = i->muted;
pa_asyncmsgq_post(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_ADD_INPUT, pa_sink_input_ref(i), NULL, (pa_free_cb_t) pa_sink_input_unref);
pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_ADD_INPUT, i, 0, NULL);
pa_sink_update_status(i->sink);
pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_NEW, i->index);
@ -299,7 +323,7 @@ pa_usec_t pa_sink_input_get_latency(pa_sink_input *i) {
pa_sink_input_assert_ref(i);
if (pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_GET_LATENCY, &r, NULL) < 0)
if (pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_GET_LATENCY, &r, 0, NULL) < 0)
r = 0;
if (i->get_latency)
@ -509,7 +533,7 @@ void pa_sink_input_set_volume(pa_sink_input *i, const pa_cvolume *volume) {
i->volume = *volume;
pa_asyncmsgq_post(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_VOLUME, pa_xnewdup(struct pa_cvolume, volume, 1), NULL, pa_xfree);
pa_asyncmsgq_post(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_VOLUME, pa_xnewdup(struct pa_cvolume, volume, 1), 0, NULL, pa_xfree);
pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE, i->index);
}
@ -528,7 +552,7 @@ void pa_sink_input_set_mute(pa_sink_input *i, int mute) {
i->muted = mute;
pa_asyncmsgq_post(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_MUTE, PA_UINT_TO_PTR(mute), NULL, NULL);
pa_asyncmsgq_post(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_MUTE, PA_UINT_TO_PTR(mute), 0, NULL, NULL);
pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE, i->index);
}
@ -553,7 +577,7 @@ int pa_sink_input_set_rate(pa_sink_input *i, uint32_t rate) {
i->sample_spec.rate = rate;
pa_asyncmsgq_post(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_RATE, PA_UINT_TO_PTR(rate), NULL, NULL);
pa_asyncmsgq_post(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_SET_RATE, PA_UINT_TO_PTR(rate), 0, NULL, NULL);
pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE, i->index);
return 0;
@ -741,7 +765,7 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, int immediately) {
/* return 0; */
}
int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, pa_memchunk *chunk) {
int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
pa_sink_input *i = PA_SINK_INPUT(o);
pa_sink_input_assert_ref(i);
@ -776,12 +800,28 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, pa_memc
}
case PA_SINK_INPUT_MESSAGE_SET_STATE: {
pa_sink_input *ssync;
if ((PA_PTR_TO_UINT(userdata) == PA_SINK_INPUT_DRAINED || PA_PTR_TO_UINT(userdata) == PA_SINK_INPUT_RUNNING) &&
(i->thread_info.state != PA_SINK_INPUT_DRAINED) && (i->thread_info.state != PA_SINK_INPUT_RUNNING))
pa_atomic_store(&i->thread_info.drained, 1);
i->thread_info.state = PA_PTR_TO_UINT(userdata);
for (ssync = i->thread_info.sync_prev; ssync; ssync = ssync->thread_info.sync_prev) {
if ((PA_PTR_TO_UINT(userdata) == PA_SINK_INPUT_DRAINED || PA_PTR_TO_UINT(userdata) == PA_SINK_INPUT_RUNNING) &&
(ssync->thread_info.state != PA_SINK_INPUT_DRAINED) && (ssync->thread_info.state != PA_SINK_INPUT_RUNNING))
pa_atomic_store(&ssync->thread_info.drained, 1);
ssync->thread_info.state = PA_PTR_TO_UINT(userdata);
}
for (ssync = i->thread_info.sync_next; ssync; ssync = ssync->thread_info.sync_next) {
if ((PA_PTR_TO_UINT(userdata) == PA_SINK_INPUT_DRAINED || PA_PTR_TO_UINT(userdata) == PA_SINK_INPUT_RUNNING) &&
(ssync->thread_info.state != PA_SINK_INPUT_DRAINED) && (ssync->thread_info.state != PA_SINK_INPUT_RUNNING))
pa_atomic_store(&ssync->thread_info.drained, 1);
ssync->thread_info.state = PA_PTR_TO_UINT(userdata);
}
return 0;
}
}

9
src/pulsecore/sink-input.h
View file

@ -71,6 +71,8 @@ struct pa_sink_input {
pa_sample_spec sample_spec;
pa_channel_map channel_map;
pa_sink_input *sync_prev, *sync_next;
pa_cvolume volume;
int muted;
@ -97,6 +99,8 @@ struct pa_sink_input {
/* size_t move_silence; */
pa_memblock *silence_memblock; /* may be NULL */
pa_sink_input *sync_prev, *sync_next;
pa_cvolume volume;
int muted;
} thread_info;
@ -133,6 +137,9 @@ typedef struct pa_sink_input_new_data {
int muted_is_set;
pa_resample_method_t resample_method;
int start_corked;
pa_sink_input *sync_base;
} pa_sink_input_new_data;
pa_sink_input_new_data* pa_sink_input_new_data_init(pa_sink_input_new_data *data);
@ -179,6 +186,6 @@ pa_sink_input_state_t pa_sink_input_get_state(pa_sink_input *i);
int pa_sink_input_peek(pa_sink_input *i, pa_memchunk *chunk, pa_cvolume *volume);
void pa_sink_input_drop(pa_sink_input *i, size_t length);
int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, pa_memchunk *chunk);
int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
#endif

58
src/pulsecore/sink.c
View file

@ -48,7 +48,7 @@
#define MAX_MIX_CHANNELS 32
#define SILENCE_BUFFER_LENGTH (64*1024)
static PA_DEFINE_CHECK_TYPE(pa_sink, sink_check_type, pa_msgobject_check_type);
static PA_DEFINE_CHECK_TYPE(pa_sink, pa_msgobject);
static void sink_free(pa_object *s);
@ -80,7 +80,7 @@ pa_sink* pa_sink_new(
pa_return_null_if_fail(!driver || pa_utf8_valid(driver));
pa_return_null_if_fail(name && pa_utf8_valid(name) && *name);
s = pa_msgobject_new(pa_sink, sink_check_type);
s = pa_msgobject_new(pa_sink);
if (!(name = pa_namereg_register(core, name, PA_NAMEREG_SINK, s, fail))) {
pa_xfree(s);
@ -161,7 +161,7 @@ static int sink_set_state(pa_sink *s, pa_sink_state_t state) {
if ((ret = s->set_state(s, state)) < 0)
return -1;
if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), NULL) < 0)
if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) < 0)
return -1;
s->state = state;
@ -264,7 +264,7 @@ int pa_sink_suspend(pa_sink *s, int suspend) {
void pa_sink_ping(pa_sink *s) {
pa_sink_assert_ref(s);
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_PING, NULL, NULL, NULL);
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_PING, NULL, 0, NULL, NULL);
}
static unsigned fill_mix_info(pa_sink *s, pa_mix_info *info, unsigned maxinfo) {
@ -530,7 +530,7 @@ pa_usec_t pa_sink_get_latency(pa_sink *s) {
if (s->get_latency)
return s->get_latency(s);
if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, NULL) < 0)
if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
return 0;
return usec;
@ -549,7 +549,7 @@ void pa_sink_set_volume(pa_sink *s, const pa_cvolume *volume) {
s->set_volume = NULL;
if (!s->set_volume)
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_VOLUME, pa_xnewdup(struct pa_cvolume, volume, 1), NULL, pa_xfree);
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_VOLUME, pa_xnewdup(struct pa_cvolume, volume, 1), 0, NULL, pa_xfree);
if (changed)
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
@ -566,7 +566,7 @@ const pa_cvolume *pa_sink_get_volume(pa_sink *s) {
s->get_volume = NULL;
if (!s->get_volume && s->refresh_volume)
pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_VOLUME, &s->volume, NULL);
pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_VOLUME, &s->volume, 0, NULL);
if (!pa_cvolume_equal(&old_volume, &s->volume))
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
@ -585,7 +585,7 @@ void pa_sink_set_mute(pa_sink *s, int mute) {
s->set_mute = NULL;
if (!s->set_mute)
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_MUTE, PA_UINT_TO_PTR(mute), NULL, NULL);
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_SET_MUTE, PA_UINT_TO_PTR(mute), 0, NULL, NULL);
if (changed)
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
@ -602,7 +602,7 @@ int pa_sink_get_mute(pa_sink *s) {
s->get_mute = NULL;
if (!s->get_mute && s->refresh_mute)
pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_MUTE, &s->muted, NULL);
pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_MUTE, &s->muted, 0, NULL);
if (old_muted != s->muted)
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
@ -660,21 +660,58 @@ unsigned pa_sink_used_by(pa_sink *s) {
return ret;
}
int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, pa_memchunk *chunk) {
int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
pa_sink *s = PA_SINK(o);
pa_sink_assert_ref(s);
switch ((pa_sink_message_t) code) {
case PA_SINK_MESSAGE_ADD_INPUT: {
pa_sink_input *i = userdata;
pa_hashmap_put(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index), pa_sink_input_ref(i));
/* Since the caller sleeps in pa_sink_input_put(), we can
* safely access data outside of thread_info even though
* it is mutable */
if ((i->thread_info.sync_prev = i->sync_prev)) {
pa_assert(i->sink == i->thread_info.sync_prev->sink);
pa_assert(i->sync_prev->sync_next == i);
i->thread_info.sync_prev->thread_info.sync_next = i;
}
if ((i->thread_info.sync_next = i->sync_next)) {
pa_assert(i->sink == i->thread_info.sync_next->sink);
pa_assert(i->sync_next->sync_prev == i);
i->thread_info.sync_next->thread_info.sync_prev = i;
}
return 0;
}
case PA_SINK_MESSAGE_REMOVE_INPUT: {
pa_sink_input *i = userdata;
/* Since the caller sleeps in pa_sink_input_disconnect(),
* we can safely access data outside of thread_info even
* though it is mutable */
pa_assert(!i->thread_info.sync_prev);
pa_assert(!i->thread_info.sync_next);
if (i->thread_info.sync_prev) {
i->thread_info.sync_prev->thread_info.sync_next = i->thread_info.sync_prev->sync_next;
i->thread_info.sync_prev = NULL;
}
if (i->thread_info.sync_next) {
i->thread_info.sync_next->thread_info.sync_prev = i->thread_info.sync_next->sync_prev;
i->thread_info.sync_next = NULL;
}
if (pa_hashmap_remove(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index)))
pa_sink_input_unref(i);
return 0;
}
@ -698,6 +735,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, pa_memchunk *
return 0;
case PA_SINK_MESSAGE_SET_STATE:
s->thread_info.state = PA_PTR_TO_UINT(userdata);
return 0;

2
src/pulsecore/sink.h
View file

@ -156,7 +156,7 @@ void pa_sink_render_full(pa_sink *s, size_t length, pa_memchunk *result);
void pa_sink_render_into(pa_sink*s, pa_memchunk *target);
void pa_sink_render_into_full(pa_sink *s, pa_memchunk *target);
int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, pa_memchunk *chunk);
int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
static inline int PA_SINK_OPENED(pa_sink_state_t x) {
return x == PA_SINK_RUNNING || x == PA_SINK_IDLE;

8
src/pulsecore/sound-file-stream.c
View file

@ -56,7 +56,7 @@ enum {
PA_DECLARE_CLASS(file_stream);
#define FILE_STREAM(o) (file_stream_cast(o))
static PA_DEFINE_CHECK_TYPE(file_stream, file_stream_check_type, pa_msgobject_check_type);
static PA_DEFINE_CHECK_TYPE(file_stream, pa_msgobject);
static void file_stream_free(pa_object *o) {
file_stream *u = FILE_STREAM(o);
@ -85,7 +85,7 @@ static void file_stream_drop(file_stream *u) {
}
}
static int file_stream_process_msg(pa_msgobject *o, int code, void*userdata, pa_memchunk *chunk) {
static int file_stream_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
file_stream *u = FILE_STREAM(o);
file_stream_assert_ref(u);
@ -154,7 +154,7 @@ static int sink_input_peek_cb(pa_sink_input *i, pa_memchunk *chunk) {
pa_memblock_unref(u->memchunk.memblock);
pa_memchunk_reset(&u->memchunk);
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u), MESSAGE_DROP_FILE_STREAM, NULL, NULL, NULL);
pa_asyncmsgq_post(u->core->asyncmsgq, PA_MSGOBJECT(u), MESSAGE_DROP_FILE_STREAM, NULL, 0, NULL, NULL);
sf_close(u->sndfile);
u->sndfile = NULL;
@ -224,7 +224,7 @@ int pa_play_file(
pa_assert(sink);
pa_assert(fname);
u = pa_msgobject_new(file_stream, file_stream_check_type);
u = pa_msgobject_new(file_stream);
u->parent.parent.free = file_stream_free;
u->parent.process_msg = file_stream_process_msg;
u->core = sink->core;

18
src/pulsecore/source-output.c
View file

@ -38,7 +38,7 @@
#include "source-output.h"
static PA_DEFINE_CHECK_TYPE(pa_source_output, source_output_check_type, pa_msgobject_check_type);
static PA_DEFINE_CHECK_TYPE(pa_source_output, pa_msgobject);
static void source_output_free(pa_object* mo);
@ -130,12 +130,12 @@ pa_source_output* pa_source_output_new(
data->resample_method = pa_resampler_get_method(resampler);
}
o = pa_msgobject_new(pa_source_output, source_output_check_type);
o = pa_msgobject_new(pa_source_output);
o->parent.parent.free = source_output_free;
o->parent.process_msg = pa_source_output_process_msg;
o->core = core;
o->state = PA_SOURCE_OUTPUT_RUNNING;
o->state = data->corked ? PA_SOURCE_OUTPUT_CORKED : PA_SOURCE_OUTPUT_RUNNING;
o->flags = flags;
o->name = pa_xstrdup(data->name);
o->driver = pa_xstrdup(data->driver);
@ -176,7 +176,7 @@ static int source_output_set_state(pa_source_output *o, pa_source_output_state_t
if (o->state == state)
return 0;
if (pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), NULL) < 0)
if (pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) < 0)
return -1;
o->state = state;
@ -187,7 +187,7 @@ void pa_source_output_disconnect(pa_source_output*o) {
pa_assert(o);
pa_return_if_fail(o->state != PA_SOURCE_OUTPUT_DISCONNECTED);
pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_REMOVE_OUTPUT, o, NULL);
pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_REMOVE_OUTPUT, o, 0, NULL);
pa_idxset_remove_by_data(o->source->core->source_outputs, o, NULL);
pa_idxset_remove_by_data(o->source->outputs, o, NULL);
@ -225,7 +225,7 @@ static void source_output_free(pa_object* mo) {
void pa_source_output_put(pa_source_output *o) {
pa_source_output_assert_ref(o);
pa_asyncmsgq_post(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, pa_source_output_ref(o), NULL, (pa_free_cb_t) pa_source_output_unref);
pa_asyncmsgq_post(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, pa_source_output_ref(o), 0, NULL, (pa_free_cb_t) pa_source_output_unref);
pa_source_update_status(o->source);
pa_subscription_post(o->source->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_NEW, o->index);
@ -243,7 +243,7 @@ pa_usec_t pa_source_output_get_latency(pa_source_output *o) {
pa_source_output_assert_ref(o);
if (pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY, &r, NULL) < 0)
if (pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY, &r, 0, NULL) < 0)
r = 0;
if (o->get_latency)
@ -293,7 +293,7 @@ int pa_source_output_set_rate(pa_source_output *o, uint32_t rate) {
o->sample_spec.rate = rate;
pa_asyncmsgq_post(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_SET_RATE, PA_UINT_TO_PTR(rate), NULL, NULL);
pa_asyncmsgq_post(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_SET_RATE, PA_UINT_TO_PTR(rate), 0, NULL, NULL);
pa_subscription_post(o->source->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_CHANGE, o->index);
return 0;
@ -380,7 +380,7 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) {
/* return 0; */
}
int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, pa_memchunk* chunk) {
int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int64_t offset, pa_memchunk* chunk) {
pa_source_output *o = PA_SOURCE_OUTPUT(mo);
pa_source_output_assert_ref(o);

4
src/pulsecore/source-output.h
View file

@ -103,6 +103,8 @@ typedef struct pa_source_output_new_data {
int channel_map_is_set;
pa_resample_method_t resample_method;
int corked;
} pa_source_output_new_data;
pa_source_output_new_data* pa_source_output_new_data_init(pa_source_output_new_data *data);
@ -142,6 +144,6 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest);
/* To be used exclusively by the source driver thread */
void pa_source_output_push(pa_source_output *o, const pa_memchunk *chunk);
int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, pa_memchunk *chunk);
int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
#endif

20
src/pulsecore/source.c
View file

@ -42,7 +42,7 @@
#include "source.h"
static PA_DEFINE_CHECK_TYPE(pa_source, source_check_type, pa_msgobject_check_type);
static PA_DEFINE_CHECK_TYPE(pa_source, pa_msgobject);
static void source_free(pa_object *o);
@ -73,7 +73,7 @@ pa_source* pa_source_new(
pa_return_null_if_fail(!driver || pa_utf8_valid(driver));
pa_return_null_if_fail(pa_utf8_valid(name) && *name);
s = pa_msgobject_new(pa_source, source_check_type);
s = pa_msgobject_new(pa_source);
if (!(name = pa_namereg_register(core, name, PA_NAMEREG_SOURCE, s, fail))) {
pa_xfree(s);
@ -140,7 +140,7 @@ static int source_set_state(pa_source *s, pa_source_state_t state) {
if ((ret = s->set_state(s, state)) < 0)
return -1;
if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), NULL) < 0)
if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_STATE, PA_UINT_TO_PTR(state), 0, NULL) < 0)
return -1;
s->state = state;
@ -222,7 +222,7 @@ int pa_source_suspend(pa_source *s, int suspend) {
void pa_source_ping(pa_source *s) {
pa_source_assert_ref(s);
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_PING, NULL, NULL, NULL);
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_PING, NULL, 0, NULL, NULL);
}
void pa_source_post(pa_source*s, const pa_memchunk *chunk) {
@ -266,7 +266,7 @@ pa_usec_t pa_source_get_latency(pa_source *s) {
if (s->get_latency)
return s->get_latency(s);
if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY, &usec, NULL) < 0)
if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0)
return 0;
return usec;
@ -285,7 +285,7 @@ void pa_source_set_volume(pa_source *s, const pa_cvolume *volume) {
s->set_volume = NULL;
if (!s->set_volume)
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, pa_xnewdup(struct pa_cvolume, volume, 1), NULL, pa_xfree);
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_VOLUME, pa_xnewdup(struct pa_cvolume, volume, 1), 0, NULL, pa_xfree);
if (changed)
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
@ -301,7 +301,7 @@ const pa_cvolume *pa_source_get_volume(pa_source *s) {
s->get_volume = NULL;
if (!s->get_volume && s->refresh_volume)
pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_VOLUME, &s->volume, NULL);
pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_VOLUME, &s->volume, 0, NULL);
if (!pa_cvolume_equal(&old_volume, &s->volume))
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
@ -320,7 +320,7 @@ void pa_source_set_mute(pa_source *s, int mute) {
s->set_mute = NULL;
if (!s->set_mute)
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MUTE, PA_UINT_TO_PTR(mute), NULL, NULL);
pa_asyncmsgq_post(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_SET_MUTE, PA_UINT_TO_PTR(mute), 0, NULL, NULL);
if (changed)
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
@ -337,7 +337,7 @@ int pa_source_get_mute(pa_source *s) {
s->get_mute = NULL;
if (!s->get_mute && s->refresh_muted)
pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MUTE, &s->muted, NULL);
pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_MUTE, &s->muted, 0, NULL);
if (old_muted != s->muted)
pa_subscription_post(s->core, PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE, s->index);
@ -384,7 +384,7 @@ unsigned pa_source_used_by(pa_source *s) {
return pa_idxset_size(s->outputs);
}
int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, pa_memchunk *chunk) {
int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
pa_source *s = PA_SOURCE(object);
pa_source_assert_ref(s);

2
src/pulsecore/source.h
View file

@ -146,7 +146,7 @@ unsigned pa_source_used_by(pa_source *s);
/* To be used exclusively by the source driver thread */
void pa_source_post(pa_source*s, const pa_memchunk *b);
int pa_source_process_msg(pa_msgobject *o, int code, void *userdata, pa_memchunk *chunk);
int pa_source_process_msg(pa_msgobject *o, int code, void *userdata, int64_t, pa_memchunk *chunk);
static inline int PA_SOURCE_OPENED(pa_source_state_t x) {
return x == PA_SOURCE_RUNNING || x == PA_SOURCE_IDLE;

10
src/tests/asyncmsgq-test.c
View file

@ -49,7 +49,7 @@ static void the_thread(void *_q) {
do {
int code = 0;
pa_assert_se(pa_asyncmsgq_get(q, NULL, &code, NULL, NULL, 1) == 0);
pa_assert_se(pa_asyncmsgq_get(q, NULL, &code, NULL, NULL, NULL, 1) == 0);
switch (code) {
@ -85,22 +85,22 @@ int main(int argc, char *argv[]) {
pa_assert_se(t = pa_thread_new(the_thread, q));
printf("Operation A post\n");
pa_asyncmsgq_post(q, NULL, OPERATION_A, NULL, NULL, NULL);
pa_asyncmsgq_post(q, NULL, OPERATION_A, NULL, 0, NULL, NULL);
pa_thread_yield();
printf("Operation B post\n");
pa_asyncmsgq_post(q, NULL, OPERATION_B, NULL, NULL, NULL);
pa_asyncmsgq_post(q, NULL, OPERATION_B, NULL, 0, NULL, NULL);
pa_thread_yield();
printf("Operation C send\n");
pa_asyncmsgq_send(q, NULL, OPERATION_C, NULL, NULL);
pa_asyncmsgq_send(q, NULL, OPERATION_C, NULL, 0, NULL);
pa_thread_yield();
printf("Quit post\n");
pa_asyncmsgq_post(q, NULL, QUIT, NULL, NULL, NULL);
pa_asyncmsgq_post(q, NULL, QUIT, NULL, 0, NULL, NULL);
pa_thread_free(t);