+/* Called from input thread context */
+static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
+ struct userdata *u;
+ size_t rlen, plen, to_skip;
+ pa_memchunk rchunk;
+
+ pa_source_output_assert_ref(o);
+ pa_source_output_assert_io_context(o);
+ pa_assert_se(u = o->userdata);
+
+ if (!PA_SOURCE_OUTPUT_IS_LINKED(pa_source_output_get_state(u->source_output))) {
+ pa_log("push when no link?");
+ return;
+ }
+
+ if (PA_UNLIKELY(u->source->thread_info.state != PA_SOURCE_RUNNING ||
+ u->sink->thread_info.state != PA_SINK_RUNNING)) {
+ pa_source_post(u->source, chunk);
+ return;
+ }
+
+ /* handle queued messages, do any message sending of our own */
+ while (pa_asyncmsgq_process_one(u->asyncmsgq) > 0)
+ ;
+
+ pa_memblockq_push_align(u->source_memblockq, chunk);
+
+ rlen = pa_memblockq_get_length(u->source_memblockq);
+ plen = pa_memblockq_get_length(u->sink_memblockq);
+
+ /* Let's not do anything else till we have enough data to process */
+ if (rlen < u->blocksize)
+ return;
+
+ /* See if we need to drop samples in order to sync */
+ if (pa_atomic_cmpxchg (&u->request_resync, 1, 0)) {
+ do_resync(u);
+ }
+
+ /* Okay, skip cancellation for skipped source samples if needed. */
+ if (PA_UNLIKELY(u->source_skip)) {
+ /* The slightly tricky bit here is that we drop all but modulo
+ * blocksize bytes and then adjust for that last bit on the sink side.
+ * We do this because the source data is coming at a fixed rate, which
+ * means the only way to try to catch up is drop sink samples and let
+ * the canceller cope up with this. */
+ to_skip = rlen >= u->source_skip ? u->source_skip : rlen;
+ to_skip -= to_skip % u->blocksize;
+
+ if (to_skip) {
+ pa_memblockq_peek_fixed_size(u->source_memblockq, to_skip, &rchunk);
+ pa_source_post(u->source, &rchunk);
+
+ pa_memblock_unref(rchunk.memblock);
+ pa_memblockq_drop(u->source_memblockq, u->blocksize);
+
+ rlen -= to_skip;
+ u->source_skip -= to_skip;
+ }
+
+ if (rlen && u->source_skip % u->blocksize) {
+ u->sink_skip += u->blocksize - (u->source_skip % u->blocksize);
+ u->source_skip -= (u->source_skip % u->blocksize);
+ }
+ }
+
+ /* And for the sink, these samples have been played back already, so we can
+ * just drop them and get on with it. */
+ if (PA_UNLIKELY(u->sink_skip)) {
+ to_skip = plen >= u->sink_skip ? u->sink_skip : plen;
+
+ pa_memblockq_drop(u->sink_memblockq, to_skip);
+
+ plen -= to_skip;
+ u->sink_skip -= to_skip;
+ }
+
+ /* process and push out samples */
+ if (u->ec->params.drift_compensation)
+ do_push_drift_comp(u);
+ else
+ do_push(u);
+}
+