Statistics
| Branch: | Revision:

chunker-player / streamers_interface / input-http.c @ 21aba08b

History | View | Annotate | Download (2.22 KB)

1
/*
2
 *  Copyright (c) 2010 Csaba Kiraly
3
 *
4
 *  This is free software; see gpl-3.0.txt
5
 */
6
#include <platform.h>
7
#include <microhttpd.h>
8

    
9
#include <chunk.h>
10
#include <http_default_urls.h>
11
#include "external_chunk_transcoding.h"
12

    
13
#include "input.h"
14

    
15
extern struct chunk_buffer *cb;
16
pthread_mutex_t cb_mutex;
17

    
18
struct input_desc {
19
  int dummy;
20
};
21

    
22
struct MHD_Daemon *httpd;
23

    
24
struct input_desc *input_open(const char *fname, uint16_t flags)
25
{
26
  struct input_desc *res;
27

    
28
  res = malloc(sizeof(struct input_desc));
29
  if (res == NULL) {
30
    return NULL;
31
  }
32

    
33
  res->dummy = 0;
34
dprintf("BEFORE INIT! %d\n", res->dummy);
35
  pthread_mutex_init(&cb_mutex, NULL);
36
  //this daemon will listen the network for incoming chunks from a streaming source
37
  //on the following path and port
38
  httpd = initChunkPuller(UL_DEFAULT_CHUNKBUFFER_PATH, UL_DEFAULT_CHUNKBUFFER_PORT);
39
dprintf("AFTER INIT! %d\n", res->dummy);
40

    
41
  return res;
42
}
43

    
44
void input_close(struct input_desc *s)
45
{
46
  free(s);
47
  finalizeChunkPuller(httpd);
48
}
49

    
50
//this one is not used, just routinely called by the firging thread
51
int input_get(struct input_desc *s, struct chunk *c)
52
{
53
  c->data = NULL;
54
  return 0;
55
}
56

    
57
//this is the real one, called by the http receiver thread
58
int enqueueBlock(const uint8_t *block, const int block_size) {
59
        static int ExternalChunk_header_size = 5*CHUNK_TRANSCODING_INT_SIZE + 2*CHUNK_TRANSCODING_INT_SIZE + 2*CHUNK_TRANSCODING_INT_SIZE + 1*CHUNK_TRANSCODING_INT_SIZE*2;
60
  int decoded_size = 0;
61
        int res = -1;
62
  struct chunk gchunk;
63

    
64
  decoded_size = decodeChunk(&gchunk, block, block_size);
65

    
66
  if(decoded_size < 0 || decoded_size != GRAPES_ENCODED_CHUNK_HEADER_SIZE + ExternalChunk_header_size + gchunk.size) {
67
            fprintf(stderr, "chunk %d probably corrupted!\n", gchunk.id);
68
                return -1;
69
        }
70

    
71
  if(cb) {
72
          pthread_mutex_lock(&cb_mutex);
73
          res = cb_add_chunk(cb, &gchunk);
74
          pthread_mutex_unlock(&cb_mutex);
75
  }
76
  if (res < 0) { //chunk sequence is older than previous chunk (SHOULD SEND ANYWAY!!!)
77
    free(gchunk.data);
78
    free(gchunk.attributes);
79
    fprintf(stderr, "Chunk %d of %d bytes FAIL res %d\n", gchunk.id, gchunk.size, res);
80
  }
81
  else {
82
    source_push_chunk(1); //push it one time
83
    dprintf("Chunk %d of %d bytes PUSHED res %d\n", gchunk.id, gchunk.size, res);
84
  }
85

    
86
  return 0;
87
}
88