source: roaraudio/roard/streams.c @ 971:302305b809d4

Last change on this file since 971:302305b809d4 was 971:302305b809d4, checked in by phi, 15 years ago

commented out change of recv buffer

File size: 20.2 KB
Line 
1//streams.c:
2
3/*
4 *      Copyright (C) Philipp 'ph3-der-loewe' Schafft - 2008
5 *
6 *  This file is part of roard a part of RoarAudio,
7 *  a cross-platform sound system for both, home and professional use.
8 *  See README for details.
9 *
10 *  This file is free software; you can redistribute it and/or modify
11 *  it under the terms of the GNU General Public License version 3
12 *  as published by the Free Software Foundation.
13 *
14 *  RoarAudio is distributed in the hope that it will be useful,
15 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
16 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
17 *  GNU General Public License for more details.
18 *
19 *  You should have received a copy of the GNU General Public License
20 *  along with this software; see the file COPYING.  If not, write to
21 *  the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
22 *
23 */
24
25#include "roard.h"
26
27int streams_init (void) {
28 int i;
29
30 for (i = 0; i < ROAR_STREAMS_MAX; i++)
31  g_streams[i] = NULL;
32
33 return 0;
34}
35
36int streams_free (void) {
37 int i;
38
39 for (i = 0; i < ROAR_STREAMS_MAX; i++) {
40  if ( g_streams[i] != NULL ) {
41   streams_delete(i);
42  }
43 }
44
45 return 0;
46}
47
48
49int streams_new    (void) {
50 int i, j;
51 struct roar_stream        * n = NULL;
52 struct roar_stream_server * s = NULL;
53
54 for (i = 0; i < ROAR_STREAMS_MAX; i++) {
55  if ( g_streams[i] == NULL ) {
56   s = ROAR_STREAM_SERVER(n = ROAR_STREAM(malloc(sizeof(struct roar_stream_server))));
57   if ( n == NULL ) {
58    ROAR_ERR("streams_new(void): can not allocate memory for new stream: %s", strerror(errno));
59    ROAR_DBG("streams_new(void) = -1");
60    return -1;
61   }
62
63   n->id         = i;
64   n->fh         = -1;
65//   n->pos_rel_id = i;
66   n->database   = NULL;
67   n->dataoff    = NULL;
68   n->datalen    = 0;
69   n->offset     = 0;
70   n->pos        = 0;
71
72   s->client          = -1;
73   s->socktype        = ROAR_SOCKET_TYPE_UNKNOWN;
74   s->buffer          = NULL;
75   s->need_extra      =  0;
76   s->output          = NULL;
77   s->is_new          =  1;
78   s->codecfilter     = -1;
79   s->pre_underruns   =  0;
80   s->post_underruns  =  0;
81   s->codec_orgi      = -1;
82   s->primary         =  0;
83
84   s->mixer.scale     = 65535;
85   s->mixer.rpg_mul   = 1;
86   s->mixer.rpg_div   = 1;
87   for (j = 0; j < ROAR_MAX_CHANNELS; j++)
88    s->mixer.mixer[j] = 65535;
89
90   for (j = 0; j < ROAR_META_MAX_PER_STREAM; j++) {
91    s->meta[j].type   = ROAR_META_TYPE_NONE;
92    s->meta[j].key[0] = 0;
93    s->meta[j].value  = NULL;
94   }
95
96   roar_vio_init_calls(&(s->vio));
97   s->driver_id = -1;
98
99   g_streams[i] = s;
100   ROAR_DBG("streams_new(void): n->id=%i", n->id);
101   ROAR_DBG("streams_new(void) = %i", i);
102   return i;
103  }
104 }
105
106 return -1;
107}
108
109int streams_delete (int id) {
110 struct roar_stream_server * s;
111 int prim;
112
113 if ( (s = g_streams[id]) == NULL )
114  return 0;
115
116 ROAR_DBG("streams_delete(id=%i) = ?", id);
117 ROAR_DBG("streams_delete(id=%i): g_streams[id]->id=%i", id, ROAR_STREAM(s)->id);
118
119 if ( s->codecfilter != -1 ) {
120  codecfilter_close(s->codecfilter_inst, s->codecfilter);
121  s->codecfilter_inst = NULL;
122  s->codecfilter = -1;
123 }
124
125 if ( s->driver_id != -1 ) {
126  driver_closevio(&(s->vio), s->driver_id);
127  roar_vio_init_calls(&(s->vio));
128  s->driver_id = -1;
129 }
130
131 if ( s->client != -1 ) {
132  ROAR_DBG("streams_delete(id=%i): Stream is owned by client %i", id, g_streams[id]->client);
133  client_stream_delete(s->client, id);
134 }
135
136 if ( s->buffer != NULL )
137  roar_buffer_free(s->buffer);
138
139 if ( s->output != NULL )
140  free(s->output);
141
142 if ( ROAR_STREAM(s)->fh != -1 )
143  close(ROAR_STREAM(s)->fh);
144
145 prim = s->primary;
146
147 free(s);
148
149 g_streams[id] = NULL;
150
151 if ( prim ) {
152  alive = 0;
153  clean_quit();
154 }
155
156 ROAR_DBG("streams_delete(id=%i) = 0", id);
157 return 0;
158}
159
160int streams_set_client (int id, int client) {
161 if ( g_streams[id] == NULL )
162  return -1;
163
164 ROAR_DBG("streams_set_client(id=%i): g_streams[id]->id=%i", id, ROAR_STREAM(g_streams[id])->id);
165 g_streams[id]->client = client;
166
167 return 0;
168}
169
170int streams_get_client (int id) {
171 if ( g_streams[id] == NULL )
172  return -1;
173
174 return g_streams[id]->client;
175}
176
177
178int streams_set_fh     (int id, int fh) {
179 int dir;
180
181 if ( g_streams[id] == NULL )
182  return -1;
183
184 ROAR_DBG("streams_set_fh(id=%i): g_streams[id]->id=%i", id, ROAR_STREAM(g_streams[id])->id);
185
186 ROAR_STREAM(g_streams[id])->fh = fh;
187
188 if ( codecfilter_open(&(g_streams[id]->codecfilter_inst), &(g_streams[id]->codecfilter), NULL,
189                  ROAR_STREAM(g_streams[id])->info.codec, g_streams[id]) == -1 ) {
190  return streams_delete(id);
191 }
192
193 if ( fh == -1 ) { // yes, this is valid, indecats full vio!
194  return 0;
195 }
196
197// roar_socket_recvbuf(fh, ROAR_OUTPUT_CALC_OUTBUFSIZE( &(ROAR_STREAM(g_streams[id])->info) )); // set recv buffer to minimum
198
199 dir = ROAR_STREAM(g_streams[id])->dir;
200
201 if ( dir == ROAR_DIR_MONITOR || dir == ROAR_DIR_RECORD || dir == ROAR_DIR_OUTPUT ) {
202  shutdown(fh, SHUT_RD);
203 }
204
205 if ( dir == ROAR_DIR_FILTER ) {
206  return 0;
207 } else {
208  return roar_socket_nonblock(fh, ROAR_SOCKET_NONBLOCK);
209 }
210}
211
212int streams_get_fh     (int id) {
213 if ( id < 0 )
214  return -1;
215
216 if ( g_streams[id] == NULL )
217  return -1;
218
219 return ROAR_STREAM(g_streams[id])->fh;
220}
221
222int streams_get    (int id, struct roar_stream_server ** stream) {
223 if ( g_streams[id] == NULL )
224  return -1;
225
226 *stream = g_streams[id];
227
228 return 0;
229}
230
231int streams_set_socktype (int id, int socktype) {
232 if ( g_streams[id] == NULL )
233  return -1;
234
235 g_streams[id]->socktype = socktype;
236
237 return 0;
238}
239
240int streams_get_socktype (int id) {
241 if ( g_streams[id] == NULL )
242  return -1;
243
244 return g_streams[id]->socktype;
245}
246
247int streams_set_primary (int id, int prim) {
248 if ( g_streams[id] == NULL )
249  return -1;
250
251 g_streams[id]->primary = prim;
252
253 return 0;
254}
255
256int streams_mark_primary (int id) {
257 return streams_set_primary(id, 1);
258}
259int streams_get_outputbuffer  (int id, void ** buffer, size_t size) {
260 if ( g_streams[id] == NULL )
261  return -1;
262
263 // output buffer size does never change.
264 if ( g_streams[id]->output != NULL ) {
265  *buffer = g_streams[id]->output;
266  return 0;
267 }
268
269 if ( (g_streams[id]->output = malloc(size)) == NULL ) {
270  ROAR_ERR("streams_get_outputbuffer(*): Can not alloc: %s", strerror(errno));
271  return -1;
272 }
273
274 *buffer = g_streams[id]->output;
275
276 return 0;
277}
278
279int streams_fill_mixbuffer (int id, struct roar_audio_info * info) {
280 // TODO: decide: is this the most complex, hacked, un-understadable,
281 //               un-writeable and even worse: un-readable
282 //               function in the whole project?
283 size_t todo = ROAR_OUTPUT_CALC_OUTBUFSIZE(info);
284 size_t needed = todo;
285 size_t todo_in;
286 size_t len, outlen;
287 size_t mul = 1, div = 1;
288 void * rest = NULL;
289 void * in   = NULL;
290 struct roar_buffer     * buf;
291 struct roar_audio_info * stream_info;
292 struct roar_stream_server * stream = g_streams[id];
293 int is_the_same = 0;
294
295 if ( g_streams[id] == NULL )
296  return -1;
297
298 if ( streams_get_outputbuffer(id, &rest, todo) == -1 ) {
299  return -1;
300 }
301
302 if ( rest == NULL ) {
303  return -1;
304 }
305
306 // set up stream_info
307
308 stream_info = &(ROAR_STREAM(stream)->info);
309
310 // calc todo_in
311 todo_in = ROAR_OUTPUT_CALC_OUTBUFSIZE(stream_info);
312
313 // calc mul and div:
314 mul = todo    / todo_in;
315 div = todo_in / todo;
316
317 if ( mul == 0 ) {
318  mul = 1;
319 } else {
320  div = 1;
321 }
322
323 ROAR_DBG("streams_fill_mixbuffer(*): mul=%i, div=%i", mul, div);
324
325 ROAR_DBG("streams_fill_mixbuffer(*): rest=%p, todo=%i->%i (in->out)", rest, todo_in, todo);
326 // are both (input and output) of same format?
327
328
329 ROAR_DBG("streams_fill_mixbuffer(*): stream_info:");
330 roar_debug_audio_info_print(stream_info);
331 ROAR_DBG("streams_fill_mixbuffer(*): info:");
332 roar_debug_audio_info_print(info);
333
334 is_the_same = stream_info->rate     == info->rate     && stream_info->bits  == info->bits &&
335               stream_info->channels == info->channels && stream_info->codec == info->codec;
336
337 ROAR_DBG("streams_fill_mixbuffer(*): is_the_same=%i", is_the_same);
338
339/* How it works:
340 *
341 * set a counter to the number of samples we need.
342 * loop until we have all samples done or no samples are
343 * left in our input buffer.
344 * If there a no samples left in the input buffer: fill the rest
345 * of the output buffer with zeros.
346 *
347 * The loop:
348 * get a buffer from the input.
349 * if it's bigger than needed, set an offset.
350 * The rest of the data:
351 * 0) convert endianness (codec) from remote to local...
352 * 1) change bits in of the samples
353 * 2) change sample rate
354 * 3) change the nummber of channels
355 * 4) insert into output buffer
356 */
357
358/*
359 // get our first buffer:
360
361 if ( stream_shift_buffer(id, &buf) == -1 ) {
362  return -1;
363 }
364
365 // first test for some basic simple cases...
366
367 if ( buf == NULL ) { // we habe nothing in input queue
368                      // we may memset() our output buffer OR
369                      // just return with -1 so we are going to
370                      // be ignored.
371  return -1;
372 }
373*/
374
375 while (todo) { // main loop
376  ROAR_DBG("streams_fill_mixbuffer(*): looping...");
377  // exit loop if nothing is left, even if we need more data..
378  if ( stream_shift_buffer(id, &buf) == -1 )
379   break;
380  if ( buf == NULL )
381   break;
382
383  // read the data for this loop...
384  roar_buffer_get_data(buf, &in);
385  roar_buffer_get_len(buf, &len);
386
387  ROAR_DBG("streams_fill_mixbuffer(*): len = %i", len);
388
389  if ( len > todo_in ) {
390   roar_buffer_set_offset(buf, todo_in);
391   len = todo_in;
392  } else {
393   roar_buffer_set_len(buf, 0); // queue for deletation
394  }
395
396  // we now have 'len' bytes in 'in'
397
398  // calc how much outlen this has...
399  outlen = (len * mul) / div;
400
401  ROAR_DBG("streams_fill_mixbuffer(*): outlen = %i, buf = %p, len = %i", outlen, in, len);
402
403  if ( is_the_same ) {
404/*
405 * 0) convert endianness (codec) from remote to local...
406 * 1) change bits in of the samples
407 * 2) change sample rate
408 * 3) change the nummber of channels
409   \\==> skiping,...
410 */
411   // * 4) insert into output buffer
412   ROAR_DBG("streams_fill_mixbuffer(*): memcpy: in->rest: %p -> %p", in, rest);
413   if ( memcpy(rest, in, len) != rest ) {
414    ROAR_ERR("streams_fill_mixbuffer(*): memcpy returned invalid pointer.");
415   }
416
417  } else {
418
419/*
420   // * 0) convert endianness (codec) from remote to local...
421   if ( stream_info->codec != info->codec ) {
422    // we neet to convert...
423    return -1;
424   }
425
426   // * 1) change bits in of the samples
427   if ( stream_info->bits != info->bits ) {
428    return -1;
429   }
430
431   // * 2) change sample rate
432   if ( stream_info->rate != info->rate ) {
433    return -1;
434   }
435
436   // * 3) change the nummber of channels
437   if ( stream_info->channels != info->channels ) {
438    return -1;
439   }
440
441   // * 4) insert into output buffer
442*/
443  // hey! we have roar_conv() :)
444
445  if ( roar_conv(rest, in, 8*len / stream_info->bits, stream_info, info) == -1 )
446   return -1;
447  }
448
449  if ( change_vol(rest, info->bits, rest, 8*outlen / info->bits, info->channels, &(stream->mixer)) == -1 )
450   return -1;
451
452  // we habe outlen bytes more...
453  todo    -= outlen;
454  rest    += outlen;
455  todo_in -= len;
456
457  roar_buffer_get_len(buf, &len);
458  ROAR_DBG("streams_fill_mixbuffer(*): New length of buffer %p is %i", buf, len);
459  if ( len == 0 ) {
460   roar_buffer_delete(buf, NULL);
461  } else {
462   stream_unshift_buffer(id, buf);
463  }
464 }
465
466//len = 0;
467//roar_buffer_get_len(buf, &len);
468
469/*
470 if ( len > 0 ) // we still have some data in this buffer, re-inserting it to the input buffers...
471  stream_unshift_buffer(id, buf);
472 else
473  buffer_delete(buf, NULL);
474*/
475
476 ROAR_STREAM(g_streams[id])->pos =
477      ROAR_MATH_OVERFLOW_ADD(ROAR_STREAM(g_streams[id])->pos,
478          ROAR_OUTPUT_CALC_OUTBUFSAMP(info, needed-todo));
479 //ROAR_WARN("stream=%i, pos=%u", id, ((struct roar_stream*)g_streams[id])->pos);
480
481 if ( todo > 0 ) { // zeroize the rest of the buffer
482  memset(rest, 0, todo);
483
484  if ( todo != ROAR_OUTPUT_CALC_OUTBUFSIZE(info) ) {
485   if ( g_streams[id]->is_new ) {
486    stream->pre_underruns++;
487   } else {
488    ROAR_WARN("streams_fill_mixbuffer(*): Underrun in stream: %u bytes missing, filling with zeros", (unsigned int)todo);
489    stream->post_underruns++;
490   }
491
492   stream->is_new = 0;
493  }
494 } else {
495  stream->is_new = 0;
496 }
497
498 return 0;
499}
500
501
502int streams_get_mixbuffers (void *** bufferlist, struct roar_audio_info * info, unsigned int pos) {
503 static void * bufs[ROAR_STREAMS_MAX+1];
504 int i;
505 int have = 0;
506
507 for (i = 0; i < ROAR_STREAMS_MAX; i++) {
508  if ( g_streams[i] != NULL ) {
509   if ( ROAR_STREAM(g_streams[i])->dir != ROAR_DIR_PLAY && ROAR_STREAM(g_streams[i])->dir != ROAR_DIR_BIDIR )
510    continue;
511
512   if ( streams_get_outputbuffer(i, &bufs[have], ROAR_OUTPUT_CALC_OUTBUFSIZE(info)) == -1 ) {
513    ROAR_ERR("streams_get_mixbuffer(*): Can not alloc output buffer for stream %i, BAD!", i);
514    ROAR_ERR("streams_get_mixbuffer(*): Ignoring stream for this round.");
515    continue;
516   }
517   if ( streams_fill_mixbuffer(i, info) == -1 ) {
518    ROAR_ERR("streams_get_mixbuffer(*): Can not fill output buffer for stream %i, this should not happen", i);
519    continue;
520   }
521
522//   printf("D: bufs[have=%i] = %p\n", have, bufs[have]);
523
524   ROAR_DBG("streams_get_mixbuffers(*):  bufs[have] = %p", bufs[have]);
525   ROAR_DBG("streams_get_mixbuffers(*): *bufs[have] = 0x%08x...", *(uint32_t*)bufs[have]);
526
527   have++; // we have a new stream!
528  }
529 }
530
531 bufs[have] = NULL;
532 //printf("D: bufs[have=%i] = %p\n", have, bufs[have]);
533
534 ROAR_DBG("streams_get_mixbuffers(*): have = %i", have);
535
536 *bufferlist = bufs;
537 return have;
538}
539
540
541int stream_add_buffer  (int id, struct roar_buffer * buf) {
542 ROAR_DBG("stream_add_buffer(id=%i, buf=%p) = ?", id, buf);
543
544 if ( g_streams[id] == NULL )
545  return -1;
546
547 if ( g_streams[id]->buffer == NULL ) {
548  g_streams[id]->buffer = buf;
549  ROAR_DBG("stream_add_buffer(id=%i, buf=%p) = 0", id, buf);
550  return 0;
551 }
552
553 ROAR_DBG("stream_add_buffer(id=%i, buf=%p) = ?", id, buf);
554 return roar_buffer_add(g_streams[id]->buffer, buf);
555}
556
557int stream_shift_buffer   (int id, struct roar_buffer ** buf) {
558 struct roar_buffer * next;
559
560 if ( g_streams[id] == NULL )
561  return -1;
562
563 if ( g_streams[id]->buffer == NULL ) {
564  *buf = NULL;
565  return 0;
566 }
567
568 roar_buffer_get_next(g_streams[id]->buffer, &next);
569
570 *buf                  = g_streams[id]->buffer;
571 g_streams[id]->buffer = next;
572
573 return 0;
574}
575int stream_unshift_buffer (int id, struct roar_buffer *  buf) {
576 if ( g_streams[id] == NULL )
577  return -1;
578
579 if ( g_streams[id]->buffer == NULL ) {
580  g_streams[id]->buffer = buf;
581  return 0;
582 }
583
584 buf->next = NULL;
585
586 roar_buffer_add(buf, g_streams[id]->buffer);
587
588 g_streams[id]->buffer = buf;
589
590 return 0;
591}
592
593int streams_check  (int id) {
594 int fh;
595 ssize_t req, realreq, done;
596 struct roar_stream        *   s;
597 struct roar_stream_server *  ss;
598 struct roar_buffer        *   b;
599 char                      * buf;
600
601 if ( g_streams[id] == NULL )
602  return -1;
603
604 ROAR_DBG("streams_check(id=%i) = ?", id);
605
606 s = ROAR_STREAM(ss = g_streams[id]);
607
608 if ( (fh = s->fh) == -1 )
609  return 0;
610
611 if ( s->dir != ROAR_DIR_PLAY && s->dir != ROAR_DIR_BIDIR )
612  return 0;
613
614 ROAR_DBG("streams_check(id=%i): fh = %i", id, fh);
615
616 req  = ROAR_OUTPUT_BUFFER_SAMPLES * s->info.channels * s->info.bits / 8; // optimal size
617 req += ss->need_extra; // bytes left we sould get....
618
619 if ( roar_buffer_new(&b, req) == -1 ) {
620  ROAR_ERR("streams_check(*): Can not alloc buffer space!");
621  ROAR_DBG("streams_check(*) = -1");
622  return -1;
623 }
624
625 roar_buffer_get_data(b, (void **)&buf);
626
627 ROAR_DBG("streams_check(id=%i): buffer is up and ready ;)", id);
628
629 if ( ss->codecfilter == -1 ) {
630  realreq = req;
631/*
632  req = read(fh, buf, req);
633  if ( req < realreq ) { // we can do this as the stream is in nonblocking mode!
634   if ( (realreq = read(fh, buf+req, realreq-req)) > 0 )
635    req += realreq;
636  }
637*/
638  done = 0;
639  while (req > 0 && done != realreq) {
640   if ( (req = stream_vio_s_read(ss, buf+done, realreq-done)) > 0 )
641    done += req;
642  }
643  req = done;
644 } else {
645  req = codecfilter_read(ss->codecfilter_inst, ss->codecfilter, buf, req);
646 }
647
648 if ( req > 0 ) {
649  ROAR_DBG("streams_check(id=%i): got %i bytes", id, req);
650
651  roar_buffer_set_len(b, req);
652
653  if ( stream_add_buffer(id, b) != -1 )
654   return 0;
655
656  ROAR_ERR("streams_check(id=%i): something is wrong, could not add buffer to stream!", id);
657  roar_buffer_free(b);
658 } else {
659  ROAR_DBG("streams_check(id=%i): read() = %i // errno: %s", id, req, strerror(errno));
660#ifdef ROAR_HAVE_LIBVORBISFILE
661  if ( errno != EAGAIN && errno != ESPIPE ) { // libvorbis file trys to seek a bit ofen :)
662#else
663  if ( errno != EAGAIN ) {
664#endif
665   ROAR_DBG("streams_check(id=%i): EOF!", id);
666   streams_delete(id);
667   ROAR_DBG("streams_check(id=%i) = 0", id);
668  }
669  roar_buffer_free(b);
670  return 0;
671 }
672
673
674 ROAR_DBG("streams_check(id=%i) = -1", id);
675 return -1;
676}
677
678
679int streams_send_mon   (int id) {
680// int fh;
681 struct roar_stream        *   s;
682 struct roar_stream_server *  ss;
683 void * obuf;
684 int    olen;
685 int    need_to_free = 0;
686
687 if ( g_streams[id] == NULL )
688  return -1;
689
690 ROAR_DBG("streams_send_mon(id=%i) = ?", id);
691
692 s = ROAR_STREAM((ss = g_streams[id]));
693
694/*
695 if ( (fh = s->fh) == -1 )
696  return 0;
697*/
698
699 if ( s->dir != ROAR_DIR_MONITOR && s->dir != ROAR_DIR_OUTPUT && s->dir != ROAR_DIR_BIDIR )
700  return 0;
701
702 if ( s->dir == ROAR_DIR_OUTPUT && g_standby )
703  return 0;
704
705 ROAR_DBG("streams_send_mon(id=%i): fh = %i", id, fh);
706
707 if ( s->info.channels != g_sa->channels || s->info.bits  != g_sa->bits ||
708      s->info.rate     != g_sa->rate     || s->info.codec != g_sa->codec  ) {
709  olen = ROAR_OUTPUT_CALC_OUTBUFSIZE(&(s->info)); // we hope g_output_buffer_len
710                                                  // is ROAR_OUTPUT_CALC_OUTBUFSIZE(g_sa) here
711  if ( (obuf = malloc(olen)) == NULL )
712   return -1;
713
714  need_to_free = 1;
715
716  ROAR_DBG("streams_send_mon(id=%i): obuf=%p, olen=%i", id, obuf, olen);
717
718  if ( roar_conv(obuf, g_output_buffer, ROAR_OUTPUT_BUFFER_SAMPLES*g_sa->channels, g_sa, &(s->info)) == -1 ) {
719   free(obuf);
720   return -1;
721  }
722 } else {
723  obuf = g_output_buffer;
724  olen = g_output_buffer_len;
725 }
726
727 errno = 0;
728
729 if ( ss->codecfilter == -1 ) {
730  if ( stream_vio_s_write(ss, obuf, olen) == olen ) {
731   if ( need_to_free ) free(obuf);
732   return 0;
733  }
734 } else {
735  if ( codecfilter_write(ss->codecfilter_inst, ss->codecfilter, obuf, olen)
736            == olen ) {
737   if ( need_to_free ) free(obuf);
738   return 0;
739  } else { // we cann't retry on codec filetered streams
740   if ( need_to_free ) free(obuf);
741   streams_delete(id);
742   return -1;
743  }
744 }
745
746 if ( errno == EAGAIN ) {
747  // ok, the client blocks for a moment, we try to sleep a bit an retry in the hope not to
748  // make any gapes in any output because of this
749
750  usleep(100); // 0.1ms
751
752  if ( stream_vio_s_write(ss, obuf, olen) == olen ) {
753   if ( need_to_free ) free(obuf);
754   return 0;
755  }
756 }
757
758 // ug... error... delete stream!
759
760 if ( need_to_free ) free(obuf);
761 streams_delete(id);
762
763 return -1;
764}
765
766int streams_send_filter(int id) {
767 int fh;
768 int have = 0;
769 int len;
770 struct roar_stream        *   s;
771 struct roar_stream_server *  ss;
772
773 if ( g_streams[id] == NULL )
774  return -1;
775
776 ROAR_DBG("streams_send_filter(id=%i) = ?", id);
777
778 s = ROAR_STREAM(ss = g_streams[id]);
779
780 if ( (fh = s->fh) == -1 )
781  return 0;
782
783 if ( s->dir != ROAR_DIR_FILTER )
784  return 0;
785
786 ROAR_DBG("streams_send_filter(id=%i): fh = %i", id, fh);
787
788 if ( stream_vio_s_write(ss, g_output_buffer, g_output_buffer_len) == g_output_buffer_len ) {
789  while ( have < g_output_buffer_len ) {
790   if ( (len = stream_vio_s_read(ss, g_output_buffer+have, g_output_buffer_len-have)) < 1 ) {
791    streams_delete(id);
792    return -1;
793   }
794   have += len;
795  }
796  return 0;
797 }
798
799 // ug... error... delete stream!
800
801 streams_delete(id);
802
803 return -1;
804}
805
806
807// VIO:
808
809ssize_t stream_vio_read (int stream, void *buf, size_t count) {
810 struct roar_stream_server * s = g_streams[stream];
811
812 if ( !s )
813  return -1;
814
815 return stream_vio_s_read(s, buf, count);
816}
817
818ssize_t stream_vio_write(int stream, void *buf, size_t count) {
819 struct roar_stream_server * s = g_streams[stream];
820
821 if ( !s )
822  return -1;
823
824 return stream_vio_s_write(s, buf, count);
825}
826
827
828ssize_t stream_vio_s_read (struct roar_stream_server * stream, void *buf, size_t count) {
829  size_t len =  0;
830 ssize_t r   = -1;
831
832 errno = 0;
833
834 if ( !stream )
835  return -1;
836
837 roar_vio_set_fh(&(stream->vio), ROAR_STREAM(stream)->fh);
838
839 if ( ! stream->vio.read )
840  return -1;
841
842 while ( (r = roar_vio_read(&(stream->vio), buf, count)) > 0 ) {
843  len   += r;
844  buf   += r;
845  count -= r;
846  if ( count == 0 )
847   break;
848 }
849
850 if ( len == 0 && r == -1 )
851  return -1;
852
853 return len;
854}
855
856ssize_t stream_vio_s_write(struct roar_stream_server * stream, void *buf, size_t count) {
857 errno = 0;
858
859 if ( !stream )
860  return -1;
861
862 if ( roar_vio_get_fh(&(stream->vio)) == -1 && ROAR_STREAM(stream)->fh != -1 )
863  roar_vio_set_fh(&(stream->vio), ROAR_STREAM(stream)->fh);
864
865// ROAR_WARN("stream_vio_s_write(*): writing...");
866
867 return roar_vio_write(&(stream->vio), buf, count);
868}
869
870//ll
Note: See TracBrowser for help on using the repository browser.