source: roaraudio/roard/streams.c @ 934:0277459c7e79

Last change on this file since 934:0277459c7e79 was 934:0277459c7e79, checked in by phi, 15 years ago

got output streams basicly working, we should clean up some code anyway

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