bd00ba4f928f390a340ccfb2ef34214fdb21f923
[profile/ivi/pulseaudio.git] / src / polypcore / pstream.c
1 /* $Id$ */
2
3 /***
4   This file is part of polypaudio.
5  
6   polypaudio is free software; you can redistribute it and/or modify
7   it under the terms of the GNU Lesser General Public License as
8   published by the Free Software Foundation; either version 2.1 of the
9   License, or (at your option) any later version.
10  
11   polypaudio is distributed in the hope that it will be useful, but
12   WITHOUT ANY WARRANTY; without even the implied warranty of
13   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14   Lesser General Public License for more details.
15  
16   You should have received a copy of the GNU Lesser General Public
17   License along with polypaudio; if not, write to the Free Software
18   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
19   USA.
20 ***/
21
22 #ifdef HAVE_CONFIG_H
23 #include <config.h>
24 #endif
25
26 #include <stdio.h>
27 #include <stdlib.h>
28 #include <assert.h>
29 #include <unistd.h>
30
31 #ifdef HAVE_NETINET_IN_H
32 #include <netinet/in.h>
33 #endif
34
35 #include "winsock.h"
36
37 #include "pstream.h"
38 #include "queue.h"
39 #include "xmalloc.h"
40 #include "log.h"
41
42 typedef enum pa_pstream_descriptor_index {
43     PA_PSTREAM_DESCRIPTOR_LENGTH,
44     PA_PSTREAM_DESCRIPTOR_CHANNEL,
45     PA_PSTREAM_DESCRIPTOR_DELTA,
46     PA_PSTREAM_DESCRIPTOR_MAX
47 } pa_pstream_descriptor_index;
48
49 typedef uint32_t pa_pstream_descriptor[PA_PSTREAM_DESCRIPTOR_MAX];
50
51 #define PA_PSTREAM_DESCRIPTOR_SIZE (PA_PSTREAM_DESCRIPTOR_MAX*sizeof(uint32_t))
52 #define FRAME_SIZE_MAX (1024*500) /* half a megabyte */
53
54 struct item_info {
55     enum { PA_PSTREAM_ITEM_PACKET, PA_PSTREAM_ITEM_MEMBLOCK } type;
56
57     /* memblock info */
58     pa_memchunk chunk;
59     uint32_t channel;
60     uint32_t delta;
61
62     /* packet info */
63     pa_packet *packet;
64 };
65
66 struct pa_pstream {
67     int ref;
68     
69     pa_mainloop_api *mainloop;
70     pa_defer_event *defer_event;
71     pa_iochannel *io;
72     pa_queue *send_queue;
73
74     int dead;
75     void (*die_callback) (pa_pstream *p, void *userdata);
76     void *die_callback_userdata;
77
78     struct {
79         struct item_info* current;
80         pa_pstream_descriptor descriptor;
81         void *data;
82         size_t index;
83     } write;
84
85     struct {
86         pa_memblock *memblock;
87         pa_packet *packet;
88         pa_pstream_descriptor descriptor;
89         void *data;
90         size_t index;
91     } read;
92
93     void (*recieve_packet_callback) (pa_pstream *p, pa_packet *packet, void *userdata);
94     void *recieve_packet_callback_userdata;
95
96     void (*recieve_memblock_callback) (pa_pstream *p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk, void *userdata);
97     void *recieve_memblock_callback_userdata;
98
99     void (*drain_callback)(pa_pstream *p, void *userdata);
100     void *drain_userdata;
101
102     pa_memblock_stat *memblock_stat;
103 };
104
105 static void do_write(pa_pstream *p);
106 static void do_read(pa_pstream *p);
107
108 static void do_something(pa_pstream *p) {
109     assert(p);
110
111     p->mainloop->defer_enable(p->defer_event, 0);
112
113     pa_pstream_ref(p);
114     
115     if (!p->dead && pa_iochannel_is_readable(p->io))
116         do_read(p);
117
118     if (!p->dead && pa_iochannel_is_writable(p->io))
119         do_write(p);
120
121     /* In case the line was hungup, make sure to rerun this function
122        as soon as possible, until all data has been read. */
123     
124     if (!p->dead && pa_iochannel_is_hungup(p->io))
125         p->mainloop->defer_enable(p->defer_event, 1);
126     
127     pa_pstream_unref(p);
128 }
129
130 static void io_callback(pa_iochannel*io, void *userdata) {
131     pa_pstream *p = userdata;
132     assert(p && p->io == io);
133     do_something(p);
134 }
135
136 static void defer_callback(pa_mainloop_api *m, pa_defer_event *e, void*userdata) {
137     pa_pstream *p = userdata;
138     assert(p && p->defer_event == e && p->mainloop == m);
139     do_something(p);
140 }
141
142 pa_pstream *pa_pstream_new(pa_mainloop_api *m, pa_iochannel *io, pa_memblock_stat *s) {
143     pa_pstream *p;
144     assert(io);
145
146     p = pa_xmalloc(sizeof(pa_pstream));
147     p->ref = 1;
148     p->io = io;
149     pa_iochannel_set_callback(io, io_callback, p);
150
151     p->dead = 0;
152     p->die_callback = NULL;
153     p->die_callback_userdata = NULL;
154
155     p->mainloop = m;
156     p->defer_event = m->defer_new(m, defer_callback, p);
157     m->defer_enable(p->defer_event, 0);
158     
159     p->send_queue = pa_queue_new();
160     assert(p->send_queue);
161
162     p->write.current = NULL;
163     p->write.index = 0;
164
165     p->read.memblock = NULL;
166     p->read.packet = NULL;
167     p->read.index = 0;
168
169     p->recieve_packet_callback = NULL;
170     p->recieve_packet_callback_userdata = NULL;
171     
172     p->recieve_memblock_callback = NULL;
173     p->recieve_memblock_callback_userdata = NULL;
174
175     p->drain_callback = NULL;
176     p->drain_userdata = NULL;
177
178     p->memblock_stat = s;
179
180     pa_iochannel_socket_set_rcvbuf(io, 1024*8); 
181     pa_iochannel_socket_set_sndbuf(io, 1024*8); 
182
183     return p;
184 }
185
186 static void item_free(void *item, PA_GCC_UNUSED void *p) {
187     struct item_info *i = item;
188     assert(i);
189
190     if (i->type == PA_PSTREAM_ITEM_MEMBLOCK) {
191         assert(i->chunk.memblock);
192         pa_memblock_unref(i->chunk.memblock);
193     } else {
194         assert(i->type == PA_PSTREAM_ITEM_PACKET);
195         assert(i->packet);
196         pa_packet_unref(i->packet);
197     }
198
199     pa_xfree(i);
200 }
201
202 static void pstream_free(pa_pstream *p) {
203     assert(p);
204
205     pa_pstream_close(p);
206     
207     pa_queue_free(p->send_queue, item_free, NULL);
208
209     if (p->write.current)
210         item_free(p->write.current, NULL);
211
212     if (p->read.memblock)
213         pa_memblock_unref(p->read.memblock);
214     
215     if (p->read.packet)
216         pa_packet_unref(p->read.packet);
217
218     pa_xfree(p);
219 }
220
221 void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet) {
222     struct item_info *i;
223     assert(p && packet && p->ref >= 1);
224
225     if (p->dead)
226         return;
227     
228 /*     pa_log(__FILE__": push-packet %p\n", packet); */
229     
230     i = pa_xmalloc(sizeof(struct item_info));
231     i->type = PA_PSTREAM_ITEM_PACKET;
232     i->packet = pa_packet_ref(packet);
233
234     pa_queue_push(p->send_queue, i);
235     p->mainloop->defer_enable(p->defer_event, 1);
236 }
237
238 void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk) {
239     struct item_info *i;
240     assert(p && channel != (uint32_t) -1 && chunk && p->ref >= 1);
241
242     if (p->dead)
243         return;
244     
245 /*     pa_log(__FILE__": push-memblock %p\n", chunk); */
246     
247     i = pa_xmalloc(sizeof(struct item_info));
248     i->type = PA_PSTREAM_ITEM_MEMBLOCK;
249     i->chunk = *chunk;
250     i->channel = channel;
251     i->delta = delta;
252
253     pa_memblock_ref(i->chunk.memblock);
254
255     pa_queue_push(p->send_queue, i);
256     p->mainloop->defer_enable(p->defer_event, 1);
257 }
258
259 void pa_pstream_set_recieve_packet_callback(pa_pstream *p, void (*callback) (pa_pstream *p, pa_packet *packet, void *userdata), void *userdata) {
260     assert(p && callback);
261
262     p->recieve_packet_callback = callback;
263     p->recieve_packet_callback_userdata = userdata;
264 }
265
266 void pa_pstream_set_recieve_memblock_callback(pa_pstream *p, void (*callback) (pa_pstream *p, uint32_t channel, uint32_t delta, const pa_memchunk *chunk, void *userdata), void *userdata) {
267     assert(p && callback);
268
269     p->recieve_memblock_callback = callback;
270     p->recieve_memblock_callback_userdata = userdata;
271 }
272
273 static void prepare_next_write_item(pa_pstream *p) {
274     assert(p);
275
276     if (!(p->write.current = pa_queue_pop(p->send_queue)))
277         return;
278     
279     p->write.index = 0;
280     
281     if (p->write.current->type == PA_PSTREAM_ITEM_PACKET) {
282         /*pa_log(__FILE__": pop-packet %p\n", p->write.current->packet);*/
283         
284         assert(p->write.current->packet);
285         p->write.data = p->write.current->packet->data;
286         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->packet->length);
287         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl((uint32_t) -1);
288         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_DELTA] = 0;
289     } else {
290         assert(p->write.current->type == PA_PSTREAM_ITEM_MEMBLOCK && p->write.current->chunk.memblock);
291         p->write.data = (uint8_t*) p->write.current->chunk.memblock->data + p->write.current->chunk.index;
292         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->chunk.length);
293         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl(p->write.current->channel);
294         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_DELTA] = htonl(p->write.current->delta);
295     }
296 }
297
298 static void do_write(pa_pstream *p) {
299     void *d;
300     size_t l;
301     ssize_t r;
302     assert(p);
303
304     if (!p->write.current)
305         prepare_next_write_item(p);
306
307     if (!p->write.current)
308         return;
309
310     assert(p->write.data);
311
312     if (p->write.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
313         d = (uint8_t*) p->write.descriptor + p->write.index;
314         l = PA_PSTREAM_DESCRIPTOR_SIZE - p->write.index;
315     } else {
316         d = (uint8_t*) p->write.data + p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE;
317         l = ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE);
318     }
319
320     if ((r = pa_iochannel_write(p->io, d, l)) < 0)
321         goto die;
322
323     p->write.index += r;
324
325     if (p->write.index >= PA_PSTREAM_DESCRIPTOR_SIZE+ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH])) {
326         assert(p->write.current);
327         item_free(p->write.current, (void *) 1);
328         p->write.current = NULL;
329
330         if (p->drain_callback && !pa_pstream_is_pending(p))
331             p->drain_callback(p, p->drain_userdata);
332     }
333
334     return;
335     
336 die:
337     p->dead = 1;
338     if (p->die_callback)
339         p->die_callback(p, p->die_callback_userdata);
340 }
341
342 static void do_read(pa_pstream *p) {
343     void *d;
344     size_t l; 
345     ssize_t r;
346     assert(p);
347
348     if (p->read.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
349         d = (uint8_t*) p->read.descriptor + p->read.index;
350         l = PA_PSTREAM_DESCRIPTOR_SIZE - p->read.index;
351     } else {
352         assert(p->read.data);
353         d = (uint8_t*) p->read.data + p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE;
354         l = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE);
355     }
356
357     if ((r = pa_iochannel_read(p->io, d, l)) <= 0)
358         goto die;
359     
360     p->read.index += r;
361
362     if (p->read.index == PA_PSTREAM_DESCRIPTOR_SIZE) {
363         /* Reading of frame descriptor complete */
364
365         /* Frame size too large */
366         if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) > FRAME_SIZE_MAX) {
367             pa_log(__FILE__": Frame size too large\n");
368             goto die;
369         }
370         
371         assert(!p->read.packet && !p->read.memblock);
372
373         if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]) == (uint32_t) -1) {
374             /* Frame is a packet frame */
375             p->read.packet = pa_packet_new(ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]));
376             assert(p->read.packet);
377             p->read.data = p->read.packet->data;
378         } else {
379             /* Frame is a memblock frame */
380             p->read.memblock = pa_memblock_new(ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]), p->memblock_stat);
381             assert(p->read.memblock);
382             p->read.data = p->read.memblock->data;
383         }
384             
385     } else if (p->read.index > PA_PSTREAM_DESCRIPTOR_SIZE) {
386         /* Frame payload available */
387         
388         if (p->read.memblock && p->recieve_memblock_callback) { /* Is this memblock data? Than pass it to the user */
389             l = (p->read.index - r) < PA_PSTREAM_DESCRIPTOR_SIZE ? p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE : (size_t) r;
390                 
391             if (l > 0) {
392                 pa_memchunk chunk;
393                 
394                 chunk.memblock = p->read.memblock;
395                 chunk.index = p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE - l;
396                 chunk.length = l;
397
398                 if (p->recieve_memblock_callback)
399                     p->recieve_memblock_callback(
400                         p,
401                         ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]),
402                         ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_DELTA]),
403                         &chunk,
404                         p->recieve_memblock_callback_userdata);
405             }
406         }
407
408         /* Frame complete */
409         if (p->read.index >= ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) + PA_PSTREAM_DESCRIPTOR_SIZE) {
410             if (p->read.memblock) {
411                 assert(!p->read.packet);
412                 
413                 pa_memblock_unref(p->read.memblock);
414                 p->read.memblock = NULL;
415             } else {
416                 assert(p->read.packet);
417                 
418                 if (p->recieve_packet_callback)
419                     p->recieve_packet_callback(p, p->read.packet, p->recieve_packet_callback_userdata);
420
421                 pa_packet_unref(p->read.packet);
422                 p->read.packet = NULL;
423             }
424
425             p->read.index = 0;
426         }
427     }
428
429     return;
430
431 die:
432     p->dead = 1;
433     if (p->die_callback)
434         p->die_callback(p, p->die_callback_userdata);
435    
436 }
437
438 void pa_pstream_set_die_callback(pa_pstream *p, void (*callback)(pa_pstream *p, void *userdata), void *userdata) {
439     assert(p && callback);
440     p->die_callback = callback;
441     p->die_callback_userdata = userdata;
442 }
443
444 int pa_pstream_is_pending(pa_pstream *p) {
445     assert(p);
446
447     if (p->dead)
448         return 0;
449
450     return p->write.current || !pa_queue_is_empty(p->send_queue);
451 }
452
453 void pa_pstream_set_drain_callback(pa_pstream *p, void (*cb)(pa_pstream *p, void *userdata), void *userdata) {
454     assert(p);
455
456     p->drain_callback = cb;
457     p->drain_userdata = userdata;
458 }
459
460 void pa_pstream_unref(pa_pstream*p) {
461     assert(p && p->ref >= 1);
462
463     if (!(--(p->ref)))
464         pstream_free(p);
465 }
466
467 pa_pstream* pa_pstream_ref(pa_pstream*p) {
468     assert(p && p->ref >= 1);
469     p->ref++;
470     return p;
471 }
472
473 void pa_pstream_close(pa_pstream *p) {
474     assert(p);
475
476     p->dead = 1;
477
478     if (p->io) {
479         pa_iochannel_free(p->io);
480         p->io = NULL;
481     }
482
483     if (p->defer_event) {
484         p->mainloop->defer_free(p->defer_event);
485         p->defer_event = NULL;
486     }
487
488     p->die_callback = NULL;
489     p->drain_callback = NULL;
490     p->recieve_packet_callback = NULL;
491     p->recieve_memblock_callback = NULL;
492 }