6 #include "protocol-native.h"
7 #include "protocol-native-spec.h"
10 #include "sourceoutput.h"
11 #include "sinkinput.h"
13 #include "tagstruct.h"
14 #include "pdispatch.h"
15 #include "pstream-util.h"
20 struct pa_protocol_native
;
22 struct record_stream
{
23 struct connection
*connection
;
25 struct pa_source_output
*source_output
;
26 struct pa_memblockq
*memblockq
;
29 struct playback_stream
{
30 struct connection
*connection
;
32 struct pa_sink_input
*sink_input
;
33 struct pa_memblockq
*memblockq
;
34 size_t requested_bytes
;
41 struct pa_protocol_native
*protocol
;
42 struct pa_client
*client
;
43 struct pa_pstream
*pstream
;
44 struct pa_pdispatch
*pdispatch
;
45 struct pa_idxset
*record_streams
, *playback_streams
;
48 struct pa_protocol_native
{
51 struct pa_socket_server
*server
;
52 struct pa_idxset
*connections
;
53 uint8_t auth_cookie
[PA_NATIVE_COOKIE_LENGTH
];
56 static int sink_input_peek_cb(struct pa_sink_input
*i
, struct pa_memchunk
*chunk
);
57 static void sink_input_drop_cb(struct pa_sink_input
*i
, size_t length
);
58 static void sink_input_kill_cb(struct pa_sink_input
*i
);
59 static uint32_t sink_input_get_latency_cb(struct pa_sink_input
*i
);
61 static void request_bytes(struct playback_stream
*s
);
63 static void command_exit(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
);
64 static void command_create_playback_stream(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
);
65 static void command_delete_playback_stream(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
);
66 static void command_drain_playback_stream(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
);
67 static void command_auth(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
);
68 static void command_set_name(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
);
69 static void command_lookup(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
);
71 static const struct pa_pdispatch_command command_table
[PA_COMMAND_MAX
] = {
72 [PA_COMMAND_ERROR
] = { NULL
},
73 [PA_COMMAND_TIMEOUT
] = { NULL
},
74 [PA_COMMAND_REPLY
] = { NULL
},
75 [PA_COMMAND_CREATE_PLAYBACK_STREAM
] = { command_create_playback_stream
},
76 [PA_COMMAND_DELETE_PLAYBACK_STREAM
] = { command_delete_playback_stream
},
77 [PA_COMMAND_DRAIN_PLAYBACK_STREAM
] = { command_drain_playback_stream
},
78 [PA_COMMAND_CREATE_RECORD_STREAM
] = { NULL
},
79 [PA_COMMAND_DELETE_RECORD_STREAM
] = { NULL
},
80 [PA_COMMAND_AUTH
] = { command_auth
},
81 [PA_COMMAND_REQUEST
] = { NULL
},
82 [PA_COMMAND_EXIT
] = { command_exit
},
83 [PA_COMMAND_SET_NAME
] = { command_set_name
},
84 [PA_COMMAND_LOOKUP_SINK
] = { command_lookup
},
85 [PA_COMMAND_LOOKUP_SOURCE
] = { command_lookup
},
88 /* structure management */
90 static void record_stream_free(struct record_stream
* r
) {
91 assert(r
&& r
->connection
);
93 pa_idxset_remove_by_data(r
->connection
->record_streams
, r
, NULL
);
94 pa_source_output_free(r
->source_output
);
95 pa_memblockq_free(r
->memblockq
);
99 static struct playback_stream
* playback_stream_new(struct connection
*c
, struct pa_sink
*sink
, struct pa_sample_spec
*ss
, const char *name
,
104 struct playback_stream
*s
;
105 assert(c
&& sink
&& ss
&& name
&& maxlength
);
107 s
= malloc(sizeof(struct playback_stream
));
111 s
->sink_input
= pa_sink_input_new(sink
, name
, ss
);
112 assert(s
->sink_input
);
113 s
->sink_input
->peek
= sink_input_peek_cb
;
114 s
->sink_input
->drop
= sink_input_drop_cb
;
115 s
->sink_input
->kill
= sink_input_kill_cb
;
116 s
->sink_input
->get_latency
= sink_input_get_latency_cb
;
117 s
->sink_input
->userdata
= s
;
119 s
->memblockq
= pa_memblockq_new(maxlength
, tlength
, pa_sample_size(ss
), prebuf
, minreq
);
120 assert(s
->memblockq
);
122 s
->requested_bytes
= 0;
123 s
->drain_request
= 0;
125 pa_idxset_put(c
->playback_streams
, s
, &s
->index
);
129 static void playback_stream_free(struct playback_stream
* p
) {
130 assert(p
&& p
->connection
);
132 if (p
->drain_request
)
133 pa_pstream_send_error(p
->connection
->pstream
, p
->drain_tag
, PA_ERROR_NOENTITY
);
135 pa_idxset_remove_by_data(p
->connection
->playback_streams
, p
, NULL
);
136 pa_sink_input_free(p
->sink_input
);
137 pa_memblockq_free(p
->memblockq
);
141 static void connection_free(struct connection
*c
) {
142 struct record_stream
*r
;
143 struct playback_stream
*p
;
144 assert(c
&& c
->protocol
);
146 pa_idxset_remove_by_data(c
->protocol
->connections
, c
, NULL
);
147 while ((r
= pa_idxset_first(c
->record_streams
, NULL
)))
148 record_stream_free(r
);
149 pa_idxset_free(c
->record_streams
, NULL
, NULL
);
151 while ((p
= pa_idxset_first(c
->playback_streams
, NULL
)))
152 playback_stream_free(p
);
153 pa_idxset_free(c
->playback_streams
, NULL
, NULL
);
155 pa_pdispatch_free(c
->pdispatch
);
156 pa_pstream_free(c
->pstream
);
157 pa_client_free(c
->client
);
161 static void request_bytes(struct playback_stream
*s
) {
162 struct pa_tagstruct
*t
;
166 if (!(l
= pa_memblockq_missing(s
->memblockq
)))
169 if (l
<= s
->requested_bytes
)
172 l
-= s
->requested_bytes
;
174 if (l
< pa_memblockq_get_minreq(s
->memblockq
))
177 s
->requested_bytes
+= l
;
179 t
= pa_tagstruct_new(NULL
, 0);
181 pa_tagstruct_putu32(t
, PA_COMMAND_REQUEST
);
182 pa_tagstruct_putu32(t
, (uint32_t) -1); /* tag */
183 pa_tagstruct_putu32(t
, s
->index
);
184 pa_tagstruct_putu32(t
, l
);
185 pa_pstream_send_tagstruct(s
->connection
->pstream
, t
);
187 /*fprintf(stderr, "Requesting %u bytes\n", l);*/
190 /*** sinkinput callbacks ***/
192 static int sink_input_peek_cb(struct pa_sink_input
*i
, struct pa_memchunk
*chunk
) {
193 struct playback_stream
*s
;
194 assert(i
&& i
->userdata
&& chunk
);
197 if (pa_memblockq_peek(s
->memblockq
, chunk
) < 0)
203 static void sink_input_drop_cb(struct pa_sink_input
*i
, size_t length
) {
204 struct playback_stream
*s
;
205 assert(i
&& i
->userdata
&& length
);
208 pa_memblockq_drop(s
->memblockq
, length
);
211 if (s
->drain_request
&& !pa_memblockq_is_readable(s
->memblockq
)) {
212 pa_pstream_send_simple_ack(s
->connection
->pstream
, s
->drain_tag
);
213 s
->drain_request
= 0;
217 static void sink_input_kill_cb(struct pa_sink_input
*i
) {
218 struct playback_stream
*s
;
219 assert(i
&& i
->userdata
);
222 playback_stream_free(s
);
225 static uint32_t sink_input_get_latency_cb(struct pa_sink_input
*i
) {
226 struct playback_stream
*s
;
227 assert(i
&& i
->userdata
);
230 return pa_samples_usec(pa_memblockq_get_length(s
->memblockq
), &s
->sink_input
->sample_spec
);
233 /*** pdispatch callbacks ***/
235 static void protocol_error(struct connection
*c
) {
236 fprintf(stderr
, __FILE__
": protocol error, kicking client\n");
240 static void command_create_playback_stream(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
) {
241 struct connection
*c
= userdata
;
242 struct playback_stream
*s
;
243 size_t maxlength
, tlength
, prebuf
, minreq
;
246 struct pa_sample_spec ss
;
247 struct pa_tagstruct
*reply
;
248 struct pa_sink
*sink
;
249 assert(c
&& t
&& c
->protocol
&& c
->protocol
->core
);
251 if (pa_tagstruct_gets(t
, &name
) < 0 ||
252 pa_tagstruct_get_sample_spec(t
, &ss
) < 0 ||
253 pa_tagstruct_getu32(t
, &sink_index
) < 0 ||
254 pa_tagstruct_getu32(t
, &maxlength
) < 0 ||
255 pa_tagstruct_getu32(t
, &tlength
) < 0 ||
256 pa_tagstruct_getu32(t
, &prebuf
) < 0 ||
257 pa_tagstruct_getu32(t
, &minreq
) < 0 ||
258 !pa_tagstruct_eof(t
)) {
263 if (!c
->authorized
) {
264 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_ACCESS
);
268 if (sink_index
== (uint32_t) -1)
269 sink
= pa_sink_get_default(c
->protocol
->core
);
271 sink
= pa_idxset_get_by_index(c
->protocol
->core
->sinks
, sink_index
);
274 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_NOENTITY
);
278 if (!(s
= playback_stream_new(c
, sink
, &ss
, name
, maxlength
, tlength
, prebuf
, minreq
))) {
279 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_INVALID
);
283 reply
= pa_tagstruct_new(NULL
, 0);
285 pa_tagstruct_putu32(reply
, PA_COMMAND_REPLY
);
286 pa_tagstruct_putu32(reply
, tag
);
287 pa_tagstruct_putu32(reply
, s
->index
);
288 assert(s
->sink_input
);
289 pa_tagstruct_putu32(reply
, s
->sink_input
->index
);
290 pa_pstream_send_tagstruct(c
->pstream
, reply
);
294 static void command_delete_playback_stream(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
) {
295 struct connection
*c
= userdata
;
297 struct playback_stream
*s
;
300 if (pa_tagstruct_getu32(t
, &channel
) < 0 ||
301 !pa_tagstruct_eof(t
)) {
306 if (!c
->authorized
) {
307 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_ACCESS
);
311 if (!(s
= pa_idxset_get_by_index(c
->playback_streams
, channel
))) {
312 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_EXIST
);
316 pa_pstream_send_simple_ack(c
->pstream
, tag
);
319 static void command_exit(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
) {
320 struct connection
*c
= userdata
;
323 if (!pa_tagstruct_eof(t
)) {
328 if (!c
->authorized
) {
329 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_ACCESS
);
333 assert(c
->protocol
&& c
->protocol
->core
&& c
->protocol
->core
->mainloop
);
334 c
->protocol
->core
->mainloop
->quit(c
->protocol
->core
->mainloop
, 0);
335 pa_pstream_send_simple_ack(c
->pstream
, tag
); /* nonsense */
339 static void command_auth(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
) {
340 struct connection
*c
= userdata
;
344 if (pa_tagstruct_get_arbitrary(t
, &cookie
, PA_NATIVE_COOKIE_LENGTH
) < 0 ||
345 !pa_tagstruct_eof(t
)) {
350 if (memcmp(c
->protocol
->auth_cookie
, cookie
, PA_NATIVE_COOKIE_LENGTH
) != 0) {
351 fprintf(stderr
, "protocol-native.c: Denied access to client with invalid authorization key.\n");
352 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_ACCESS
);
357 pa_pstream_send_simple_ack(c
->pstream
, tag
);
361 static void command_set_name(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
) {
362 struct connection
*c
= userdata
;
366 if (pa_tagstruct_gets(t
, &name
) < 0 ||
367 !pa_tagstruct_eof(t
)) {
372 pa_client_rename(c
->client
, name
);
373 pa_pstream_send_simple_ack(c
->pstream
, tag
);
377 static void command_lookup(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
) {
378 struct connection
*c
= userdata
;
380 uint32_t index
= PA_IDXSET_INVALID
;
383 if (pa_tagstruct_gets(t
, &name
) < 0 ||
384 !pa_tagstruct_eof(t
)) {
389 if (!c
->authorized
) {
390 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_ACCESS
);
394 if (command
== PA_COMMAND_LOOKUP_SINK
) {
395 struct pa_sink
*sink
;
396 if ((sink
= pa_namereg_get(c
->protocol
->core
, name
, PA_NAMEREG_SINK
)))
399 struct pa_source
*source
;
400 assert(command
== PA_COMMAND_LOOKUP_SOURCE
);
401 if ((source
= pa_namereg_get(c
->protocol
->core
, name
, PA_NAMEREG_SOURCE
)))
402 index
= source
->index
;
405 if (index
== PA_IDXSET_INVALID
)
406 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_NOENTITY
);
408 struct pa_tagstruct
*reply
;
409 reply
= pa_tagstruct_new(NULL
, 0);
411 pa_tagstruct_putu32(reply
, PA_COMMAND_REPLY
);
412 pa_tagstruct_putu32(reply
, tag
);
413 pa_tagstruct_putu32(reply
, index
);
414 pa_pstream_send_tagstruct(c
->pstream
, reply
);
418 static void command_drain_playback_stream(struct pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, struct pa_tagstruct
*t
, void *userdata
) {
419 struct connection
*c
= userdata
;
421 struct playback_stream
*s
;
424 if (pa_tagstruct_getu32(t
, &index
) < 0 ||
425 !pa_tagstruct_eof(t
)) {
430 if (!c
->authorized
) {
431 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_ACCESS
);
435 if (!(s
= pa_idxset_get_by_index(c
->playback_streams
, index
))) {
436 pa_pstream_send_error(c
->pstream
, tag
, PA_ERROR_NOENTITY
);
440 s
->drain_request
= 0;
442 if (!pa_memblockq_is_readable(s
->memblockq
))
443 pa_pstream_send_simple_ack(c
->pstream
, tag
);
445 s
->drain_request
= 1;
450 /*** pstream callbacks ***/
452 static void packet_callback(struct pa_pstream
*p
, struct pa_packet
*packet
, void *userdata
) {
453 struct connection
*c
= userdata
;
454 assert(p
&& packet
&& packet
->data
&& c
);
456 if (pa_pdispatch_run(c
->pdispatch
, packet
, c
) < 0) {
457 fprintf(stderr
, "protocol-native: invalid packet.\n");
462 static void memblock_callback(struct pa_pstream
*p
, uint32_t channel
, int32_t delta
, struct pa_memchunk
*chunk
, void *userdata
) {
463 struct connection
*c
= userdata
;
464 struct playback_stream
*stream
;
465 assert(p
&& chunk
&& userdata
);
467 if (!(stream
= pa_idxset_get_by_index(c
->playback_streams
, channel
))) {
468 fprintf(stderr
, "protocol-native: client sent block for invalid stream.\n");
473 if (chunk
->length
>= stream
->requested_bytes
)
474 stream
->requested_bytes
= 0;
476 stream
->requested_bytes
-= chunk
->length
;
478 pa_memblockq_push_align(stream
->memblockq
, chunk
, delta
);
479 assert(stream
->sink_input
);
480 pa_sink_notify(stream
->sink_input
->sink
);
482 /*fprintf(stderr, "Recieved %u bytes.\n", chunk->length);*/
485 static void die_callback(struct pa_pstream
*p
, void *userdata
) {
486 struct connection
*c
= userdata
;
490 fprintf(stderr
, "protocol-native: connection died.\n");
493 /*** client callbacks ***/
495 static void client_kill_cb(struct pa_client
*c
) {
496 assert(c
&& c
->userdata
);
497 connection_free(c
->userdata
);
500 /*** socket server callbacks ***/
502 static void on_connection(struct pa_socket_server
*s
, struct pa_iochannel
*io
, void *userdata
) {
503 struct pa_protocol_native
*p
= userdata
;
504 struct connection
*c
;
505 assert(s
&& io
&& p
);
507 c
= malloc(sizeof(struct connection
));
509 c
->authorized
= p
->public;
512 c
->client
= pa_client_new(p
->core
, "NATIVE", "Client");
514 c
->client
->kill
= client_kill_cb
;
515 c
->client
->userdata
= c
;
516 c
->pstream
= pa_pstream_new(p
->core
->mainloop
, io
);
519 pa_pstream_set_recieve_packet_callback(c
->pstream
, packet_callback
, c
);
520 pa_pstream_set_recieve_memblock_callback(c
->pstream
, memblock_callback
, c
);
521 pa_pstream_set_die_callback(c
->pstream
, die_callback
, c
);
523 c
->pdispatch
= pa_pdispatch_new(p
->core
->mainloop
, command_table
, PA_COMMAND_MAX
);
524 assert(c
->pdispatch
);
526 c
->record_streams
= pa_idxset_new(NULL
, NULL
);
527 c
->playback_streams
= pa_idxset_new(NULL
, NULL
);
528 assert(c
->record_streams
&& c
->playback_streams
);
530 pa_idxset_put(p
->connections
, c
, NULL
);
533 /*** module entry points ***/
535 struct pa_protocol_native
* pa_protocol_native_new(struct pa_core
*core
, struct pa_socket_server
*server
) {
536 struct pa_protocol_native
*p
;
537 assert(core
&& server
);
539 p
= malloc(sizeof(struct pa_protocol_native
));
542 if (pa_authkey_load_from_home(PA_NATIVE_COOKIE_FILE
, p
->auth_cookie
, sizeof(p
->auth_cookie
)) < 0) {
550 p
->connections
= pa_idxset_new(NULL
, NULL
);
551 assert(p
->connections
);
553 pa_socket_server_set_callback(p
->server
, on_connection
, p
);
558 void pa_protocol_native_free(struct pa_protocol_native
*p
) {
559 struct connection
*c
;
562 while ((c
= pa_idxset_first(p
->connections
, NULL
)))
564 pa_idxset_free(p
->connections
, NULL
, NULL
);
565 pa_socket_server_free(p
->server
);