]> git.vomp.tv Git - vompserver.git/blob - mvpreceiver.c
Merge branch '0-4-1-dev' of git.vomp.tv:/vompserver
[vompserver.git] / mvpreceiver.c
1 #include "mvpreceiver.h"
2
3 int MVPReceiver::numMVPReceivers = 0;
4
5 MVPReceiver* MVPReceiver::create(cChannel* channel, int priority)
6 {
7 #if VDRVERSNUM < 10500
8   bool NeedsDetachReceivers;
9   cDevice* device = cDevice::GetDevice(channel, priority, &NeedsDetachReceivers);
10 #else
11   cDevice* device = cDevice::GetDevice(channel, priority, true); // last param is live-view
12 #endif
13
14   if (!device)
15   {
16     Log::getInstance()->log("MVPReceiver", Log::INFO, "No device found to receive this channel at this priority");
17     return NULL;
18   }
19
20 #if VDRVERSNUM < 10500
21   if (NeedsDetachReceivers)
22   {
23     Log::getInstance()->log("MVPReceiver", Log::WARN, "Needs detach receivers");
24
25     // Need to detach other receivers or VDR will shut down??
26   }
27 #endif
28
29   MVPReceiver* m = new MVPReceiver(channel, device);
30
31   numMVPReceivers++;
32   Log::getInstance()->log("MVPReceiver", Log::DEBUG, "num mvp receivers now up to %i", numMVPReceivers);
33   
34   return m;
35 }
36
37 MVPReceiver::MVPReceiver(cChannel* channel, cDevice* device)
38 #if VDRVERSNUM < 10300
39 : cReceiver(channel->Ca(), 0, 7, channel->Vpid(), channel->Ppid(), channel->Apid1(), channel->Apid2(), channel->Dpid1(), channel->Dpid2(), channel->Tpid())
40 #elif VDRVERSNUM < 10500
41 : cReceiver(channel->Ca(), 0, channel->Vpid(), channel->Apids(), channel->Dpids(), mergeSpidsTpid(channel->Spids(),channel->Tpid()))
42 #elif VDRVERSNUM < 10712
43 : cReceiver(channel->GetChannelID(), 0, channel->Vpid(), channel->Apids(), channel->Dpids(), mergeSpidsTpid(channel->Spids(),channel->Tpid()))
44 #else
45 : cReceiver(channel, 0)
46 #endif
47 {
48   logger = Log::getInstance();
49   vdrActivated = false;
50   inittedOK = 0;
51   streamID = 0;
52   tcp = NULL;
53
54 #if VDRVERSNUM >= 10712
55   AddPid(channel->Tpid()); 
56 #endif
57
58 //  logger->log("MVPReceiver", Log::DEBUG, "Channel has VPID %i APID %i", channel->Vpid(), channel->Apid(0));
59
60   if (!processed.init(1000000)) return;
61   pthread_mutex_init(&processedRingLock, NULL);
62
63   // OK
64   
65   // Detect whether this is video or radio and set an appropriate stream chunk size
66   // 50k for video, 5k for radio
67   // Perhaps move this client side?
68   if (channel->Vpid()) streamChunkSize = 50000;
69   else streamChunkSize = 5000;
70   
71   inittedOK = 1;
72   device->SwitchChannel(channel, false);
73   device->AttachReceiver(this);
74 }
75
76 int MVPReceiver::init(TCP* ttcp, ULONG tstreamID)
77 {
78   tcp = ttcp;
79   streamID = tstreamID;
80   return inittedOK;
81 }
82
83 MVPReceiver::~MVPReceiver()
84 {
85   numMVPReceivers--;
86   Log::getInstance()->log("MVPReceiver", Log::DEBUG, "num mvp receivers now down to %i", numMVPReceivers);
87 }
88
89 void MVPReceiver::Activate(bool on)
90 {
91   vdrActivated = on;
92   if (on) 
93   {
94     logger->log("MVPReceiver", Log::DEBUG, "VDR active");
95     threadStart();
96   }
97   else
98   {
99     logger->log("MVPReceiver", Log::DEBUG, "VDR inactive, sending stream end message");
100     threadStop();
101     sendStreamEnd();
102   }
103 }
104
105 bool MVPReceiver::isVdrActivated()
106 {
107   return vdrActivated;
108 }
109
110 void MVPReceiver::detachMVPReceiver()
111 {
112   threadStop();
113   Detach();
114 }
115
116 void MVPReceiver::Receive(UCHAR* data, int length)
117 {
118   pthread_mutex_lock(&processedRingLock);
119   processed.put(data, length);
120   if (processed.getContent() > streamChunkSize) threadSignal();
121   pthread_mutex_unlock(&processedRingLock);
122 }
123
124 void MVPReceiver::threadMethod()
125 {
126   ULONG *p;
127   ULONG headerLength = sizeof(ULONG) * 4;
128   UCHAR buffer[streamChunkSize + headerLength];
129   int amountReceived;
130
131 //   threadSetKillable(); ??
132
133   while(1)
134   {
135     threadLock();
136     threadWaitForSignal();
137     threadUnlock();
138     threadCheckExit();
139     
140     do
141     {
142       pthread_mutex_lock(&processedRingLock);
143       amountReceived = processed.get(buffer+headerLength, streamChunkSize);
144       pthread_mutex_unlock(&processedRingLock);
145     
146       p = (ULONG*)&buffer[0]; *p = htonl(2); // stream channel
147       p = (ULONG*)&buffer[4]; *p = htonl(streamID);
148       p = (ULONG*)&buffer[8]; *p = htonl(0); // here insert flag: 0 = ok, data follows
149       p = (ULONG*)&buffer[12]; *p = htonl(amountReceived);
150
151       tcp->sendPacket(buffer, amountReceived + headerLength);
152     } while(processed.getContent() >= streamChunkSize);
153   }  
154 }
155
156 void MVPReceiver::sendStreamEnd()
157 {
158   ULONG *p;
159   ULONG bufferLength = sizeof(ULONG) * 4;
160   UCHAR buffer[bufferLength];
161   p = (ULONG*)&buffer[0]; *p = htonl(2); // stream channel
162   p = (ULONG*)&buffer[4]; *p = htonl(streamID);
163   p = (ULONG*)&buffer[8]; *p = htonl(1); // stream end
164   p = (ULONG*)&buffer[12]; *p = htonl(0); // zero length, no more data
165   tcp->sendPacket(buffer, bufferLength);
166 }
167
168
169 int *MVPReceiver::mergeSpidsTpid(const int *spids,int tpid)
170 {
171   int *destpids;
172   const int *runspid=spids;
173   for (runspid=spids,destpids=mergedSpidsTpid;*runspid;runspid++,destpids++) {
174        *destpids=*runspid;
175   }
176   *destpids=tpid;
177   destpids++;
178   *destpids=0;
179   return mergedSpidsTpid;
180 }
181