source: roaraudio/roard/streams.c @ 387:82bc53da6dd5

Last change on this file since 387:82bc53da6dd5 was 382:46b315ef4767, checked in by phi, 16 years ago

CF Vorbis: fixed a read() on closed() and a close() before read() bug

File size: 15.6 KB
Line 
1//streams.c:
2
3#include "roard.h"
4
5int streams_init (void) {
6 int i;
7
8 for (i = 0; i < ROAR_STREAMS_MAX; i++)
9  g_streams[i] = NULL;
10
11 return 0;
12}
13
14int streams_free (void) {
15 int i;
16
17 for (i = 0; i < ROAR_STREAMS_MAX; i++) {
18  if ( g_streams[i] != NULL ) {
19   streams_delete(i);
20  }
21 }
22
23 return 0;
24}
25
26
27int streams_new    (void) {
28 int i, j;
29 struct roar_stream * n = NULL;
30
31 for (i = 0; i < ROAR_STREAMS_MAX; i++) {
32  if ( g_streams[i] == NULL ) {
33   n = malloc(sizeof(struct roar_stream_server));
34   if ( n == NULL ) {
35    ROAR_ERR("streams_new(void): can not allocate memory for new stream: %s", strerror(errno));
36    ROAR_DBG("streams_new(void) = -1");
37    return -1;
38   }
39
40   n->id         = i;
41   n->fh         = -1;
42//   n->pos_rel_id = i;
43   n->database   = NULL;
44   n->dataoff    = NULL;
45   n->datalen    = 0;
46   n->offset     = 0;
47   n->pos        = 0;
48
49   ((struct roar_stream_server*)n)->client        = -1;
50   ((struct roar_stream_server*)n)->socktype      = ROAR_SOCKET_TYPE_UNKNOWN;
51   ((struct roar_stream_server*)n)->buffer        = NULL;
52   ((struct roar_stream_server*)n)->need_extra    =  0;
53   ((struct roar_stream_server*)n)->output        = NULL;
54   ((struct roar_stream_server*)n)->is_new        =  1;
55   ((struct roar_stream_server*)n)->codecfilter   = -1;
56   ((struct roar_stream_server*)n)->mixer.scale   = 65535;
57   ((struct roar_stream_server*)n)->mixer.rpg_mul = 1;
58   ((struct roar_stream_server*)n)->mixer.rpg_div = 1;
59   for (j = 0; j < ROAR_MAX_CHANNELS; j++)
60    ((struct roar_stream_server*)n)->mixer.mixer[j] = 65535;
61   for (j = 0; j < ROAR_META_MAX_PER_STREAM; j++) {
62    ((struct roar_stream_server*)n)->meta[j].type   = ROAR_META_TYPE_NONE;
63    ((struct roar_stream_server*)n)->meta[j].key[0] = 0;
64    ((struct roar_stream_server*)n)->meta[j].value  = NULL;
65   }
66
67   g_streams[i] = (struct roar_stream_server*)n;
68   ROAR_DBG("streams_new(void) = %i", i);
69   return i;
70  }
71 }
72
73 return -1;
74}
75
76int streams_delete (int id) {
77 if ( g_streams[id] == NULL )
78  return 0;
79
80 ROAR_DBG("streams_delete(id=%i) = ?", id);
81
82 if ( g_streams[id]->codecfilter != -1 ) {
83  codecfilter_close(g_streams[id]->codecfilter_inst, g_streams[id]->codecfilter);
84  g_streams[id]->codecfilter_inst = NULL;
85  g_streams[id]->codecfilter = -1;
86 }
87
88 if ( g_streams[id]->client != -1 ) {
89  ROAR_DBG("streams_delete(id=%i): Stream is owned by client %i", id, g_streams[id]->client);
90  client_stream_delete(g_streams[id]->client, id);
91 }
92
93 if ( g_streams[id]->buffer != NULL )
94  roar_buffer_free(g_streams[id]->buffer);
95
96 if ( g_streams[id]->output != NULL )
97  free(g_streams[id]->output);
98
99 if ( ROAR_STREAM(g_streams[id])->fh != -1 )
100  close(ROAR_STREAM(g_streams[id])->fh);
101
102 free(g_streams[id]);
103
104 g_streams[id] = NULL;
105
106 ROAR_DBG("streams_delete(id=%i) = 0", id);
107 return 0;
108}
109
110int streams_set_client (int id, int client) {
111 if ( g_streams[id] == NULL )
112  return -1;
113
114 g_streams[id]->client = client;
115
116 return 0;
117}
118
119int streams_set_fh     (int id, int fh) {
120 int dir;
121
122 if ( g_streams[id] == NULL )
123  return -1;
124
125 ((struct roar_stream *)g_streams[id])->fh = fh;
126
127 codecfilter_open(&(g_streams[id]->codecfilter_inst), &(g_streams[id]->codecfilter), NULL,
128                  ((struct roar_stream *)g_streams[id])->info.codec, g_streams[id]);
129
130 dir = ((struct roar_stream *)g_streams[id])->dir;
131
132 if ( dir == ROAR_DIR_MONITOR || dir == ROAR_DIR_RECORD ) {
133  shutdown(fh, SHUT_RD);
134 }
135
136 if ( dir == ROAR_DIR_FILTER ) {
137  return 0;
138 } else {
139  return roar_socket_nonblock(fh, ROAR_SOCKET_NONBLOCK);
140 }
141}
142
143int streams_get_fh     (int id) {
144 if ( id < 0 )
145  return -1;
146
147 if ( g_streams[id] == NULL )
148  return -1;
149
150 return ((struct roar_stream *)g_streams[id])->fh;
151}
152
153int streams_get    (int id, struct roar_stream_server ** stream) {
154 if ( g_streams[id] == NULL )
155  return -1;
156
157 *stream = g_streams[id];
158
159 return 0;
160}
161
162int streams_set_socktype (int id, int socktype) {
163 if ( g_streams[id] == NULL )
164  return -1;
165
166 g_streams[id]->socktype = socktype;
167
168 return 0;
169}
170
171int streams_get_socktype (int id) {
172 if ( g_streams[id] == NULL )
173  return -1;
174
175 return g_streams[id]->socktype;
176}
177
178int streams_get_outputbuffer  (int id, void ** buffer, size_t size) {
179 if ( g_streams[id] == NULL )
180  return -1;
181
182 // output buffer size does never change.
183 if ( g_streams[id]->output != NULL ) {
184  *buffer = g_streams[id]->output;
185  return 0;
186 }
187
188 if ( (g_streams[id]->output = malloc(size)) == NULL ) {
189  ROAR_ERR("streams_get_outputbuffer(*): Can not alloc: %s", strerror(errno));
190  return -1;
191 }
192
193 *buffer = g_streams[id]->output;
194
195 return 0;
196}
197
198int streams_fill_mixbuffer (int id, struct roar_audio_info * info) {
199 // TODO: decide: is this the most complex, hacked, un-understadable,
200 //               un-writeable and even worse: un-readable
201 //               function in the whole project?
202 size_t todo = ROAR_OUTPUT_CALC_OUTBUFSIZE(info);
203 size_t needed = todo;
204 size_t todo_in;
205 size_t len, outlen;
206 size_t mul = 1, div = 1;
207 void * rest = NULL;
208 void * in   = NULL;
209 struct roar_buffer     * buf;
210 struct roar_audio_info * stream_info;
211 int is_the_same = 0;
212
213 if ( g_streams[id] == NULL )
214  return -1;
215
216 if ( streams_get_outputbuffer(id, &rest, todo) == -1 ) {
217  return -1;
218 }
219
220 if ( rest == NULL ) {
221  return -1;
222 }
223
224 // set up stream_info
225
226 stream_info = &(((struct roar_stream*)g_streams[id])->info);
227
228 // calc todo_in
229 todo_in = ROAR_OUTPUT_CALC_OUTBUFSIZE(stream_info);
230
231 // calc mul and div:
232 mul = todo    / todo_in;
233 div = todo_in / todo;
234
235 if ( mul == 0 ) {
236  mul = 1;
237 } else {
238  div = 1;
239 }
240
241 ROAR_DBG("streams_fill_mixbuffer(*): mul=%i, div=%i", mul, div);
242
243 ROAR_DBG("streams_fill_mixbuffer(*): rest=%p, todo=%i->%i (in->out)", rest, todo_in, todo);
244 // are both (input and output) of same format?
245
246
247 ROAR_DBG("streams_fill_mixbuffer(*): stream_info:");
248 roar_debug_audio_info_print(stream_info);
249 ROAR_DBG("streams_fill_mixbuffer(*): info:");
250 roar_debug_audio_info_print(info);
251
252 is_the_same = stream_info->rate     == info->rate     && stream_info->bits  == info->bits &&
253               stream_info->channels == info->channels && stream_info->codec == info->codec;
254
255 ROAR_DBG("streams_fill_mixbuffer(*): is_the_same=%i", is_the_same);
256
257/* How it works:
258 *
259 * set a counter to the number of samples we need.
260 * loop until we have all samples done or no samples are
261 * left in our input buffer.
262 * If there a no samples left in the input buffer: fill the rest
263 * of the output buffer with zeros.
264 *
265 * The loop:
266 * get a buffer from the input.
267 * if it's bigger than needed, set an offset.
268 * The rest of the data:
269 * 0) convert endianness (codec) from remote to local...
270 * 1) change bits in of the samples
271 * 2) change sample rate
272 * 3) change the nummber of channels
273 * 4) insert into output buffer
274 */
275
276/*
277 // get our first buffer:
278
279 if ( stream_shift_buffer(id, &buf) == -1 ) {
280  return -1;
281 }
282
283 // first test for some basic simple cases...
284
285 if ( buf == NULL ) { // we habe nothing in input queue
286                      // we may memset() our output buffer OR
287                      // just return with -1 so we are going to
288                      // be ignored.
289  return -1;
290 }
291*/
292
293 while (todo) { // main loop
294  ROAR_DBG("streams_fill_mixbuffer(*): looping...");
295  // exit loop if nothing is left, even if we need more data..
296  if ( stream_shift_buffer(id, &buf) == -1 )
297   break;
298  if ( buf == NULL )
299   break;
300
301  // read the data for this loop...
302  roar_buffer_get_data(buf, &in);
303  roar_buffer_get_len(buf, &len);
304
305  ROAR_DBG("streams_fill_mixbuffer(*): len = %i", len);
306
307  if ( len > todo_in ) {
308   roar_buffer_set_offset(buf, todo_in);
309   len = todo_in;
310  } else {
311   roar_buffer_set_len(buf, 0); // queue for deletation
312  }
313
314  // we now have 'len' bytes in 'in'
315
316  // calc how much outlen this has...
317  outlen = (len * mul) / div;
318
319  ROAR_DBG("streams_fill_mixbuffer(*): outlen = %i, buf = %p, len = %i", outlen, in, len);
320
321  if ( is_the_same ) {
322/*
323 * 0) convert endianness (codec) from remote to local...
324 * 1) change bits in of the samples
325 * 2) change sample rate
326 * 3) change the nummber of channels
327   \\==> skiping,...
328 */
329   // * 4) insert into output buffer
330   ROAR_DBG("streams_fill_mixbuffer(*): memcpy: in->rest: %p -> %p", in, rest);
331   if ( memcpy(rest, in, len) != rest ) {
332    ROAR_ERR("streams_fill_mixbuffer(*): memcpy returned invalid pointer.");
333   }
334
335  } else {
336
337/*
338   // * 0) convert endianness (codec) from remote to local...
339   if ( stream_info->codec != info->codec ) {
340    // we neet to convert...
341    return -1;
342   }
343
344   // * 1) change bits in of the samples
345   if ( stream_info->bits != info->bits ) {
346    return -1;
347   }
348
349   // * 2) change sample rate
350   if ( stream_info->rate != info->rate ) {
351    return -1;
352   }
353
354   // * 3) change the nummber of channels
355   if ( stream_info->channels != info->channels ) {
356    return -1;
357   }
358
359   // * 4) insert into output buffer
360*/
361  // hey! we have roar_conv() :)
362
363  if ( roar_conv(rest, in, 8*len / stream_info->bits, stream_info, info) == -1 )
364   return -1;
365  }
366
367  if ( change_vol(rest, info->bits, rest, 8*outlen / info->bits, info->channels, &(((struct roar_stream_server*)g_streams[id])->mixer)) == -1 )
368   return -1;
369
370  // we habe outlen bytes more...
371  todo    -= outlen;
372  rest    += outlen;
373  todo_in -= len;
374
375  roar_buffer_get_len(buf, &len);
376  ROAR_DBG("streams_fill_mixbuffer(*): New length of buffer %p is %i", buf, len);
377  if ( len == 0 ) {
378   roar_buffer_delete(buf, NULL);
379  } else {
380   stream_unshift_buffer(id, buf);
381  }
382 }
383
384//len = 0;
385//roar_buffer_get_len(buf, &len);
386
387/*
388 if ( len > 0 ) // we still have some data in this buffer, re-inserting it to the input buffers...
389  stream_unshift_buffer(id, buf);
390 else
391  buffer_delete(buf, NULL);
392*/
393
394 ((struct roar_stream*)g_streams[id])->pos =
395      ROAR_MATH_OVERFLOW_ADD(((struct roar_stream*)g_streams[id])->pos,
396          ROAR_OUTPUT_CALC_OUTBUFSAMP(info, needed-todo));
397 //ROAR_WARN("stream=%i, pos=%u", id, ((struct roar_stream*)g_streams[id])->pos);
398
399 if ( todo > 0 ) { // zeroize the rest of teh buffer
400  memset(rest, 0, todo);
401
402  if ( todo != ROAR_OUTPUT_CALC_OUTBUFSIZE(info) ) {
403   if ( !g_streams[id]->is_new )
404    ROAR_WARN("streams_fill_mixbuffer(*): Underrun in stream: %i bytes missing, filling with zeros", todo);
405
406   g_streams[id]->is_new = 0;
407  }
408 } else {
409  g_streams[id]->is_new = 0;
410 }
411
412 return 0;
413}
414
415
416int streams_get_mixbuffers (void *** bufferlist, struct roar_audio_info * info, unsigned int pos) {
417 static void * bufs[ROAR_STREAMS_MAX+1];
418 int i;
419 int have = 0;
420
421 for (i = 0; i < ROAR_STREAMS_MAX; i++) {
422  if ( g_streams[i] != NULL ) {
423   if ( ((struct roar_stream *)g_streams[i])->dir != ROAR_DIR_PLAY )
424    continue;
425
426   if ( streams_get_outputbuffer(i, &bufs[have], ROAR_OUTPUT_CALC_OUTBUFSIZE(info)) == -1 ) {
427    ROAR_ERR("streams_get_mixbuffer(*): Can not alloc output buffer for stream %i, BAD!", i);
428    ROAR_ERR("streams_get_mixbuffer(*): Ignoring stream for this round.");
429    continue;
430   }
431   if ( streams_fill_mixbuffer(i, info) == -1 ) {
432    ROAR_ERR("streams_get_mixbuffer(*): Can not fill output buffer for stream %i, this should not happen", i);
433    continue;
434   }
435
436//   printf("D: bufs[have=%i] = %p\n", have, bufs[have]);
437
438   ROAR_DBG("streams_get_mixbuffers(*):  bufs[have] = %p", bufs[have]);
439   ROAR_DBG("streams_get_mixbuffers(*): *bufs[have] = 0x%08x...", *(uint32_t*)bufs[have]);
440
441   have++; // we have a new stream!
442  }
443 }
444
445 bufs[have] = NULL;
446 //printf("D: bufs[have=%i] = %p\n", have, bufs[have]);
447
448 ROAR_DBG("streams_get_mixbuffers(*): have = %i", have);
449
450 *bufferlist = bufs;
451 return 0;
452}
453
454
455int stream_add_buffer  (int id, struct roar_buffer * buf) {
456 ROAR_DBG("stream_add_buffer(id=%i, buf=%p) = ?", id, buf);
457
458 if ( g_streams[id] == NULL )
459  return -1;
460
461 if ( g_streams[id]->buffer == NULL ) {
462  g_streams[id]->buffer = buf;
463  ROAR_DBG("stream_add_buffer(id=%i, buf=%p) = 0", id, buf);
464  return 0;
465 }
466
467 ROAR_DBG("stream_add_buffer(id=%i, buf=%p) = ?", id, buf);
468 return roar_buffer_add(g_streams[id]->buffer, buf);
469}
470
471int stream_shift_buffer   (int id, struct roar_buffer ** buf) {
472 struct roar_buffer * next;
473
474 if ( g_streams[id] == NULL )
475  return -1;
476
477 if ( g_streams[id]->buffer == NULL ) {
478  *buf = NULL;
479  return 0;
480 }
481
482 roar_buffer_get_next(g_streams[id]->buffer, &next);
483
484 *buf                  = g_streams[id]->buffer;
485 g_streams[id]->buffer = next;
486
487 return 0;
488}
489int stream_unshift_buffer (int id, struct roar_buffer *  buf) {
490 if ( g_streams[id] == NULL )
491  return -1;
492
493 if ( g_streams[id]->buffer == NULL ) {
494  g_streams[id]->buffer = buf;
495  return 0;
496 }
497
498 buf->next = NULL;
499
500 roar_buffer_add(buf, g_streams[id]->buffer);
501
502 g_streams[id]->buffer = buf;
503
504 return 0;
505}
506
507int streams_check  (int id) {
508 int fh;
509 ssize_t req;
510 struct roar_stream        *   s;
511 struct roar_stream_server *  ss;
512 struct roar_buffer        *   b;
513 char                      * buf;
514
515 if ( g_streams[id] == NULL )
516  return -1;
517
518 ROAR_DBG("streams_check(id=%i) = ?", id);
519
520 s = (struct roar_stream *) (ss = g_streams[id]);
521
522 if ( (fh = s->fh) == -1 )
523  return 0;
524
525 if ( s->dir != ROAR_DIR_PLAY )
526  return 0;
527
528 ROAR_DBG("streams_check(id=%i): fh = %i", id, fh);
529
530 req  = ROAR_OUTPUT_BUFFER_SAMPLES * s->info.channels * s->info.bits / 8; // optimal size
531 req += ss->need_extra; // bytes left we sould get....
532
533 if ( roar_buffer_new(&b, req) == -1 ) {
534  ROAR_ERR("streams_check(*): Can not alloc buffer space!");
535  ROAR_DBG("streams_check(*) = -1");
536  return -1;
537 }
538
539 roar_buffer_get_data(b, (void **)&buf);
540
541 ROAR_DBG("streams_check(id=%i): buffer is up and ready ;)", id);
542
543 if ( ss->codecfilter == -1 ) {
544  req = read(fh, buf, req);
545 } else {
546  req = codecfilter_read(ss->codecfilter_inst, ss->codecfilter, buf, req);
547 }
548
549 if ( req > 0 ) {
550  ROAR_DBG("streams_check(id=%i): got %i bytes", id, req);
551
552  roar_buffer_set_len(b, req);
553
554  if ( stream_add_buffer(id, b) != -1 )
555   return 0;
556
557  ROAR_ERR("streams_check(id=%i): something is wrong, could not add buffer to stream!", id);
558  roar_buffer_free(b);
559 } else {
560  ROAR_DBG("streams_check(id=%i): read() = %i // errno: %s", id, req, strerror(errno));
561#ifdef ROAR_HAVE_LIBVORBISFILE
562  if ( errno != EAGAIN && errno != ESPIPE ) { // libvorbis file trys to seek a bit ofen :)
563#else
564  if ( errno != EAGAIN ) {
565#endif
566   ROAR_DBG("streams_check(id=%i): EOF!", id);
567   streams_delete(id);
568   ROAR_DBG("streams_check(id=%i) = 0", id);
569  }
570  roar_buffer_free(b);
571  return 0;
572 }
573
574
575 ROAR_DBG("streams_check(id=%i) = -1", id);
576 return -1;
577}
578
579
580int streams_send_mon   (int id) {
581 int fh;
582 struct roar_stream        *   s;
583 struct roar_stream_server *  ss;
584
585 if ( g_streams[id] == NULL )
586  return -1;
587
588 ROAR_DBG("streams_send_mon(id=%i) = ?", id);
589
590 s = (struct roar_stream *) (ss = g_streams[id]);
591
592 if ( (fh = s->fh) == -1 )
593  return 0;
594
595 if ( s->dir != ROAR_DIR_MONITOR )
596  return 0;
597
598 ROAR_DBG("streams_send_mon(id=%i): fh = %i", id, fh);
599
600 if ( write(fh, g_output_buffer, g_output_buffer_len) == g_output_buffer_len )
601  return 0;
602
603 if ( errno == EAGAIN ) {
604  // ok, the client blocks for a moment, we try to sleep a bit an retry in the hope not to
605  // make any gapes in any output because of this
606
607  usleep(100); // 0.1ms
608
609  if ( write(fh, g_output_buffer, g_output_buffer_len) == g_output_buffer_len )
610   return 0;
611 }
612
613 // ug... error... delete stream!
614
615 streams_delete(id);
616
617 return -1;
618}
619
620int streams_send_filter(int id) {
621 int fh;
622 int have = 0;
623 int len;
624 struct roar_stream        *   s;
625 struct roar_stream_server *  ss;
626
627 if ( g_streams[id] == NULL )
628  return -1;
629
630 ROAR_DBG("streams_send_filter(id=%i) = ?", id);
631
632 s = (struct roar_stream *) (ss = g_streams[id]);
633
634 if ( (fh = s->fh) == -1 )
635  return 0;
636
637 if ( s->dir != ROAR_DIR_FILTER )
638  return 0;
639
640 ROAR_DBG("streams_send_filter(id=%i): fh = %i", id, fh);
641
642 if ( write(fh, g_output_buffer, g_output_buffer_len) == g_output_buffer_len ) {
643  while ( have < g_output_buffer_len ) {
644   if ( (len = read(fh, g_output_buffer+have, g_output_buffer_len-have)) < 1 ) {
645    streams_delete(id);
646    return -1;
647   }
648   have += len;
649  }
650  return 0;
651 }
652
653 // ug... error... delete stream!
654
655 streams_delete(id);
656
657 return -1;
658}
659
660//ll
Note: See TracBrowser for help on using the repository browser.