-
Notifications
You must be signed in to change notification settings - Fork 88
/
rtmp.py
executable file
·1364 lines (1187 loc) · 72.1 KB
/
rtmp.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# Copyright (c) 2007-2009, Mamta Singh. All rights reserved. see README for details.
# Copyright (c) 2010-2011, Kundan Singh.
'''
This is a simple implementation of a Flash RTMP server to accept connections and stream requests. The module is organized as follows:
1. The FlashServer class is the main class to provide the server abstraction. It uses the multitask module for co-operative multitasking.
It also uses the App abstract class to implement the applications.
2. The Server class implements a simple server to receive new Client connections and inform the FlashServer application. The Client class
derived from Protocol implements the RTMP client functions. The Protocol class implements the base RTMP protocol parsing. A Client contains
various streams from the client, represented using the Stream class.
3. The Message, Header and Command represent RTMP message, header and command respectively. The FLV class implements functions to perform read
and write of FLV file format.
Typically an application can launch this server as follows:
$ python rtmp.py
To know the command line options use the -h option:
$ python rtmp.py -h
To start the server with a different directory for recording and playing FLV files from, use the following command.
$ python rtmp.py -r some-other-directory/
Note the terminal '/' in the directory name. Without this, it is just used as a prefix in FLV file names.
A test client is available in testClient directory, and can be compiled using Flex Builder. Alternatively, you can use the SWF file to launch
from testClient/bin-debug after starting the server. Once you have launched the client in the browser, you can connect to
local host by clicking on 'connect' button. Then click on publish button to publish a stream. Open another browser with
same URL and first connect then play the same stream name. If everything works fine you should be able to see the video
from first browser to the second browser. Similar, in the first browser, if you check the record box before publishing,
it will create a new FLV file for the recorded stream. You can close the publishing stream and play the recorded stream to
see your recording. Note that due to initial delay in timestamp (in case publish was clicked much later than connect),
your played video will start appearing after some initial delay.
If an application wants to use this module as a library, it can launch the server as follows:
>>> agent = FlashServer() # a new RTMP server instance
>>> agent.root = 'flvs/' # set the document root to be 'flvs' directory. Default is current './' directory.
>>> agent.start() # start the server
>>> multitask.run() # this is needed somewhere in the application to actually start the co-operative multitasking.
If an application wants to specify a different application other than the default App, it can subclass it and supply the application by
setting the server's apps property. The following example shows how to define "myapp" which invokes a 'connected()' method on client when
the client connects to the server.
class MyApp(App): # a new MyApp extends the default App in rtmp module.
def __init__(self): # constructor just invokes base class constructor
App.__init__(self)
def onConnect(self, client, *args):
result = App.onConnect(self, client, *args) # invoke base class method first
def invokeAdded(self, client): # define a method to invoke 'connected("some-arg")' on Flash client
yield client.call('connected', 'some-arg')
multitask.add(invokeAdded(self, client)) # need to invoke later so that connection is established before callback
return result # return True to accept, or None to postpone calling accept()
...
agent.apps = dict({'myapp': MyApp, 'someapp': MyApp, '*': App})
Now the client can connect to rtmp://server/myapp or rtmp://server/someapp and will get connected to this MyApp application.
If the client doesn't define "function connected(arg:String):void" in the NetConnection.client object then the server will
throw an exception and display the error message.
'''
import os, sys, time, struct, socket, traceback, multitask, amf, hashlib, hmac, random
_debug = False
class ConnectionClosed:
'raised when the client closed the connection'
def truncate(data, max=100):
return data and len(data)>max and data[:max] + '...(%d)'%(len(data),) or data
class SockStream(object):
'''A class that represents a socket as a stream'''
def __init__(self, sock):
self.sock, self.buffer = sock, ''
self.bytesWritten = self.bytesRead = 0
def close(self):
self.sock.close()
def read(self, count):
try:
while True:
if len(self.buffer) >= count: # do have enough data in buffer
data, self.buffer = self.buffer[:count], self.buffer[count:]
raise StopIteration(data)
if _debug: print 'socket.read[%d] calling recv()'%(count,)
data = (yield multitask.recv(self.sock, 4096)) # read more from socket
if not data: raise ConnectionClosed
if _debug: print 'socket.read[%d] %r'%(len(data), truncate(data))
self.bytesRead += len(data)
self.buffer += data
except StopIteration: raise
except: raise ConnectionClosed # anything else is treated as connection closed.
def unread(self, data):
self.buffer = data + self.buffer
def write(self, data):
while len(data) > 0: # write in 4K chunks each time
chunk, data = data[:4096], data[4096:]
self.bytesWritten += len(chunk)
if _debug: print 'socket.write[%d] %r'%(len(chunk), truncate(chunk))
try: yield multitask.send(self.sock, chunk)
except: raise ConnectionClosed
'''
NOTE: Here is a part of the documentation to understand how the Chunks' headers work.
To have a complete documentation, YOU HAVE TO READ rtmp_specification_1.0.pdf (from page 13)
This is the format of a chunk. Here, we store all except the chunk data:
------------------------------------------------------------------------
+-------------+----------------+-------------------+--------------+
| Basic header|Chunk Msg Header|Extended Time Stamp| Chunk Data |
+-------------+----------------+-------------------+--------------+
This are the formats of the basic header:
-----------------------------------------
0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3
+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|fmt| cs id | |fmt| 0 | cs id - 64 | |fmt| 1 | cs id - 64 |
+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
(cs id < 64) (64 <= cs id < 320) (320 <= cs id)
fmt store the format of the chunk message header. There are four different formats.
Type 0 (fmt=00):
----------------
This type MUST be used at the start of a chunk stream, and whenever the stream timestamp goes backward (e.g., because of a backward seek).
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| timestamp | message length |message type id| message stream id |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Type 1 (fmt=01):
----------------
Streams with variable-sized messages (for example, many video formats) SHOULD use this format for the first chunk of each new message after the first.
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| timestamp delta | message length |message type id|
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Type 2 (fmt=10):
----------------
Streams with constant-sized messages (for example, some audio and data formats) SHOULD use this format for the first chunk of each message after the first.
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| timestamp delta |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Type 3 (fmt=11):
----------------
Chunks of Type 3 have no header. Stream ID, message length and timestamp delta are not present; chunks of this type take values from
the preceding chunk. When a single message is split into chunks, all chunks of a message except the first one, SHOULD use this type.
Extended Timestamp:
-------------------
This field is transmitted only when the normal time stamp in the chunk message header is set to 0x00ffffff. If normal time stamp is
set to any value less than 0x00ffffff, this field MUST NOT be present. This field MUST NOT be present if the timestamp field is not
present. Type 3 chunks MUST NOT have this field. This field if transmitted is located immediately after the chunk message header
and before the chunk data.
'''
class Header(object):
# Chunk type 0 = FULL
# Chunk type 1 = MESSAGE
# Chunk type 2 = TIME
# Chunk type 3 = SEPARATOR
FULL, MESSAGE, TIME, SEPARATOR, MASK = 0x00, 0x40, 0x80, 0xC0, 0xC0
def __init__(self, channel=0, time=0, size=None, type=None, streamId=0):
self.channel = channel # in fact, this will be the fmt + cs id
self.time = time # timestamp[delta]
self.size = size # message length
self.type = type # message type id
self.streamId = streamId # message stream id
if (channel < 64): self.hdrdata = struct.pack('>B', channel)
elif (channel < 320): self.hdrdata = '\x00' + struct.pack('>B', channel-64)
else: self.hdrdata = '\x01' + struct.pack('>H', channel-64)
def toBytes(self, control):
data = chr(ord(self.hdrdata[0]) | control)
if len(self.hdrdata) >= 2: data += self.hdrdata[1:]
# if the chunk type is not 3
if control != Header.SEPARATOR:
data += struct.pack('>I', self.time if self.time < 0xFFFFFF else 0xFFFFFF)[1:] # add time in 3 bytes
# if the chunk type is not 2
if control != Header.TIME:
data += struct.pack('>I', self.size)[1:] # add size in 3 bytes
data += struct.pack('>B', self.type) # add type in 1 byte
# if the chunk type is not 1
if control != Header.MESSAGE:
data += struct.pack('<I', self.streamId) # add streamId in little-endian 4 bytes
# add the extended time part to the header if timestamp[delta] >= 16777215
if self.time >= 0xFFFFFF:
data += struct.pack('>I', self.time)
return data
def __repr__(self):
return ("<Header channel=%r time=%r size=%r type=%s (%r) streamId=%r>"
% (self.channel, self.time, self.size, Message.type_name.get(self.type, 'unknown'), self.type, self.streamId))
def dup(self):
return Header(channel=self.channel, time=self.time, size=self.size, type=self.type, streamId=self.streamId)
class Message(object):
# message types: RPC3, DATA3,and SHAREDOBJECT3 are used with AMF3
CHUNK_SIZE, ABORT, ACK, USER_CONTROL, WIN_ACK_SIZE, SET_PEER_BW, AUDIO, VIDEO, DATA3, SHAREDOBJ3, RPC3, DATA, SHAREDOBJ, RPC, AGGREGATE = \
0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x08, 0x09, 0x0F, 0x10, 0x11, 0x12, 0x13, 0x14, 0x16
type_name = dict(enumerate('unknown chunk-size abort ack user-control win-ack-size set-peer-bw unknown audio video unknown unknown unknown unknown unknown data3 sharedobj3 rpc3 data sharedobj rpc unknown aggregate'.split()))
def __init__(self, hdr=None, data=''):
self.header, self.data = hdr or Header(), data
# define properties type, streamId and time to access self.header.(property)
for p in ['type', 'streamId', 'time']:
exec 'def _g%s(self): return self.header.%s'%(p, p)
exec 'def _s%s(self, %s): self.header.%s = %s'%(p, p, p, p)
exec '%s = property(fget=_g%s, fset=_s%s)'%(p, p, p)
@property
def size(self): return len(self.data)
def __repr__(self):
return ("<Message header=%r data=%r>"% (self.header, truncate(self.data)))
def dup(self):
return Message(self.header.dup(), self.data[:])
class Protocol(object):
PING_SIZE, DEFAULT_CHUNK_SIZE, HIGH_WRITE_CHUNK_SIZE, PROTOCOL_CHANNEL_ID = 1536, 128, 4096, 2 # constants
READ_WIN_SIZE, WRITE_WIN_SIZE = 1000000L, 1073741824L
def __init__(self, sock):
self.stream = SockStream(sock)
self.lastReadHeaders, self.incompletePackets, self.lastWriteHeaders = dict(), dict(), dict()
self.readChunkSize = self.writeChunkSize = Protocol.DEFAULT_CHUNK_SIZE
self.readWinSize0, self.readWinSize, self.writeWinSize0, self.writeWinSize = 0L, self.READ_WIN_SIZE, 0L, self.WRITE_WIN_SIZE
self.nextChannelId = Protocol.PROTOCOL_CHANNEL_ID + 1
self._time0 = time.time()
self.writeQueue = multitask.Queue()
@property
def relativeTime(self):
return int(1000*(time.time() - self._time0))
def messageReceived(self, msg): # override in subclass
yield
def protocolMessage(self, msg):
if msg.type == Message.ACK: # respond to ACK requests
self.writeWinSize0 = struct.unpack('>L', msg.data)[0]
# response = Message()
# response.type, response.data = msg.type, msg.data
# yield self.writeMessage(response)
elif msg.type == Message.CHUNK_SIZE:
self.readChunkSize = struct.unpack('>L', msg.data)[0]
if _debug: print "set read chunk size to %d" % self.readChunkSize
elif msg.type == Message.WIN_ACK_SIZE:
self.readWinSize, self.readWinSize0 = struct.unpack('>L', msg.data)[0], self.stream.bytesRead
elif msg.type == Message.USER_CONTROL:
type, data = struct.unpack('>H', msg.data[:2])[0], msg.data[2:]
if type == 3: # client expects a response when it sends set buffer length
streamId, bufferTime = struct.unpack('>II', data)
response = Message()
response.time, response.type, response.data = self.relativeTime, Message.USER_CONTROL, struct.pack('>HI', 0, streamId)
yield self.writeMessage(response)
yield
def connectionClosed(self):
yield
def parse(self):
try:
yield self.parseCrossDomainPolicyRequest() # check for cross domain policy
yield self.parseHandshake() # parse rtmp handshake
yield self.parseMessages() # parse messages
except ConnectionClosed:
yield self.connectionClosed()
if _debug: print 'parse connection closed'
except:
if _debug: print 'exception, closing connection'
if _debug: traceback.print_exc()
yield self.connectionClosed()
def writeMessage(self, message):
yield self.writeQueue.put(message)
def parseCrossDomainPolicyRequest(self):
# read the request
REQUEST = '<policy-file-request/>\x00'
data = (yield self.stream.read(len(REQUEST)))
if data == REQUEST:
if _debug: print data
data = '''<!DOCTYPE cross-domain-policy SYSTEM "http://www.macromedia.com/xml/dtds/cross-domain-policy.dtd">
<cross-domain-policy>
<allow-access-from domain="*" to-ports="1935" secure='false'/>
</cross-domain-policy>'''
yield self.stream.write(data)
raise ConnectionClosed
else:
yield self.stream.unread(data)
SERVER_KEY = '\x47\x65\x6e\x75\x69\x6e\x65\x20\x41\x64\x6f\x62\x65\x20\x46\x6c\x61\x73\x68\x20\x4d\x65\x64\x69\x61\x20\x53\x65\x72\x76\x65\x72\x20\x30\x30\x31\xf0\xee\xc2\x4a\x80\x68\xbe\xe8\x2e\x00\xd0\xd1\x02\x9e\x7e\x57\x6e\xec\x5d\x2d\x29\x80\x6f\xab\x93\xb8\xe6\x36\xcf\xeb\x31\xae'
FLASHPLAYER_KEY = '\x47\x65\x6E\x75\x69\x6E\x65\x20\x41\x64\x6F\x62\x65\x20\x46\x6C\x61\x73\x68\x20\x50\x6C\x61\x79\x65\x72\x20\x30\x30\x31\xF0\xEE\xC2\x4A\x80\x68\xBE\xE8\x2E\x00\xD0\xD1\x02\x9E\x7E\x57\x6E\xEC\x5D\x2D\x29\x80\x6F\xAB\x93\xB8\xE6\x36\xCF\xEB\x31\xAE'
def parseHandshake(self):
'''Parses the rtmp handshake'''
data = (yield self.stream.read(Protocol.PING_SIZE + 1)) # bound version and first ping
data = Protocol.handshakeResponse(data)
yield self.stream.write(data)
data = (yield self.stream.read(Protocol.PING_SIZE))
@staticmethod
def handshakeResponse(data):
# send both data parts before reading next ping-size, to work with ffmpeg
if struct.unpack('>I', data[5:9])[0] == 0:
data = '\x03' + '\x00'*Protocol.PING_SIZE
return data + data[1:]
else:
type, data = ord(data[0]), data[1:] # first byte is ignored
scheme = None
for s in range(0, 2):
digest_offset = (sum([ord(data[i]) for i in range(772, 776)]) % 728 + 776) if s == 1 else (sum([ord(data[i]) for i in range(8, 12)]) % 728 + 12)
temp = data[0:digest_offset] + data[digest_offset+32:Protocol.PING_SIZE]
hash = Protocol._calculateHash(temp, Protocol.FLASHPLAYER_KEY[:30])
if hash == data[digest_offset:digest_offset+32]:
scheme = s
break
if scheme is None:
if _debug: print 'invalid RTMP connection data, assuming scheme 0'
scheme = 0
client_dh_offset = (sum([ord(data[i]) for i in range(768, 772)]) % 632 + 8) if scheme == 1 else (sum([ord(data[i]) for i in range(1532, 1536)]) % 632 + 772)
outgoingKp = data[client_dh_offset:client_dh_offset+128]
handshake = struct.pack('>IBBBB', 0, 1, 2, 3, 4) + ''.join([chr(random.randint(0, 255)) for i in xrange(Protocol.PING_SIZE-8)])
server_dh_offset = (sum([ord(handshake[i]) for i in range(768, 772)]) % 632 + 8) if scheme == 1 else (sum([ord(handshake[i]) for i in range(1532, 1536)]) % 632 + 772)
keys = Protocol._generateKeyPair() # (public, private)
handshake = handshake[:server_dh_offset] + keys[0][0:128] + handshake[server_dh_offset+128:]
if type > 0x03: raise Exception('encryption is not supported')
server_digest_offset = (sum([ord(handshake[i]) for i in range(772, 776)]) % 728 + 776) if scheme == 1 else (sum([ord(handshake[i]) for i in range(8, 12)]) % 728 + 12)
temp = handshake[0:server_digest_offset] + handshake[server_digest_offset+32:Protocol.PING_SIZE]
hash = Protocol._calculateHash(temp, Protocol.SERVER_KEY[:36])
handshake = handshake[:server_digest_offset] + hash + handshake[server_digest_offset+32:]
buffer = data[:Protocol.PING_SIZE-32]
key_challenge_offset = (sum([ord(buffer[i]) for i in range(772, 776)]) % 728 + 776) if scheme == 1 else (sum([ord(buffer[i]) for i in range(8, 12)]) % 728 + 12)
challenge_key = data[key_challenge_offset:key_challenge_offset+32]
hash = Protocol._calculateHash(challenge_key, Protocol.SERVER_KEY[:68])
rand_bytes = ''.join([chr(random.randint(0, 255)) for i in xrange(Protocol.PING_SIZE-32)])
last_hash = Protocol._calculateHash(rand_bytes, hash[:32])
output = chr(type) + handshake + rand_bytes + last_hash
return output
@staticmethod
def _calculateHash(msg, key): # Hmac-sha256
return hmac.new(key, msg, hashlib.sha256).digest()
@staticmethod
def _generateKeyPair(): # dummy key pair since we don't support encryption
return (''.join([chr(random.randint(0, 255)) for i in xrange(128)]), '')
def parseMessages(self):
'''Parses complete messages until connection closed. Raises ConnectionLost exception.'''
CHANNEL_MASK = 0x3F
while True:
hdrsize = ord((yield self.stream.read(1))[0]) # read header size byte
channel = hdrsize & CHANNEL_MASK
if channel == 0: # we need one more byte
channel = 64 + ord((yield self.stream.read(1))[0])
elif channel == 1: # we need two more bytes
data = (yield self.stream.read(2))
channel = 64 + ord(data[0]) + 256 * ord(data[1])
hdrtype = hdrsize & Header.MASK # read header type byte
if hdrtype == Header.FULL or not self.lastReadHeaders.has_key(channel):
header = Header(channel)
self.lastReadHeaders[channel] = header
else:
header = self.lastReadHeaders[channel]
if hdrtype < Header.SEPARATOR: # time or delta has changed
data = (yield self.stream.read(3))
header.time = struct.unpack('!I', '\x00' + data)[0]
if hdrtype < Header.TIME: # size and type also changed
data = (yield self.stream.read(3))
header.size = struct.unpack('!I', '\x00' + data)[0]
header.type = ord((yield self.stream.read(1))[0])
if hdrtype < Header.MESSAGE: # streamId also changed
data = (yield self.stream.read(4))
header.streamId = struct.unpack('<I', data)[0]
if header.time == 0xFFFFFF: # if we have extended timestamp, read it
data = (yield self.stream.read(4))
header.extendedTime = struct.unpack('!I', data)[0]
if _debug: print 'extended time stamp', '%x'%(header.extendedTime,)
else:
header.extendedTime = None
if hdrtype == Header.FULL:
header.currentTime = header.extendedTime or header.time
header.hdrtype = hdrtype
elif hdrtype in (Header.MESSAGE, Header.TIME):
header.hdrtype = hdrtype
#print header.type, '0x%02x'%(hdrtype,), header.time, header.currentTime
# if _debug: print 'R', header, header.currentTime, header.extendedTime, '0x%x'%(hdrsize,)
data = self.incompletePackets.get(channel, "") # are we continuing an incomplete packet?
count = min(header.size - (len(data)), self.readChunkSize) # how much more
data += (yield self.stream.read(count))
# check if we need to send Ack
if self.readWinSize is not None:
if self.stream.bytesRead > (self.readWinSize0 + self.readWinSize):
self.readWinSize0 = self.stream.bytesRead
ack = Message()
ack.time, ack.type, ack.data = self.relativeTime, Message.ACK, struct.pack('>L', self.readWinSize0)
yield self.writeMessage(ack)
if len(data) < header.size: # we don't have all data
self.incompletePackets[channel] = data
else: # we have all data
if hdrtype in (Header.MESSAGE, Header.TIME):
header.currentTime = header.currentTime + (header.extendedTime or header.time)
elif hdrtype == Header.SEPARATOR:
if header.hdrtype in (Header.MESSAGE, Header.TIME):
header.currentTime = header.currentTime + (header.extendedTime or header.time)
if len(data) == header.size:
if channel in self.incompletePackets:
del self.incompletePackets[channel]
if _debug:
print 'aggregated %r bytes message: readChunkSize(%r) x %r'%(len(data), self.readChunkSize, len(data) / self.readChunkSize)
else:
data, self.incompletePackets[channel] = data[:header.size], data[header.size:]
hdr = Header(channel=header.channel, time=header.currentTime, size=header.size, type=header.type, streamId=header.streamId)
msg = Message(hdr, data)
if hdr.type == Message.AGGREGATE:
''' see http://code.google.com/p/red5/source/browse/java/server/trunk/src/org/red5/server/net/rtmp/event/Aggregate.java / getParts()
'''
if _debug: print 'Protocol.parseMessages aggregated msg=', msg
aggdata = data;
while len(aggdata) > 0:
'''
type=1 byte
size=3 bytes
time=4 bytes
streamId= 4 bytes
data= size bytes
backPointer=4 bytes, value == size
'''
subtype = ord(aggdata[0])
subsize = struct.unpack('!I', '\x00' + aggdata[1:4])[0]
subtime = struct.unpack('!I', aggdata[4:8])[0]
substreamid = struct.unpack('<I', aggdata[8:12])[0]
subheader = Header(channel, time=subtime, size=subsize, type=subtype, streamId=substreamid) # TODO: set correct channel
aggdata = aggdata[11:] # skip header
submsgdata = aggdata[:subsize] # get message data
submsg = Message(subheader, submsgdata)
yield self.parseMessage(submsg)
aggdata = aggdata[subsize:] # skip message data
backpointer = struct.unpack('!I', aggdata[0:4])[0]
if backpointer != subsize:
print 'Warning aggregate submsg backpointer=%r != %r' % (backpointer, subsize)
aggdata = aggdata[4:] # skip back pointer, go to next message
else:
yield self.parseMessage(msg)
def parseMessage(self, msg):
try:
if _debug: print 'Protocol.parseMessage msg=', msg
if msg.header.channel == Protocol.PROTOCOL_CHANNEL_ID:
yield self.protocolMessage(msg)
else:
yield self.messageReceived(msg)
except:
if _debug: print 'Protocol.parseMessage exception', (traceback and traceback.print_exc() or None)
def write(self):
'''Writes messages to stream'''
while True:
# while self.writeQueue.empty(): (yield multitask.sleep(0.01))
# message = self.writeQueue.get() # TODO this should be used using multitask.Queue and remove previous wait.
message = yield self.writeQueue.get() # TODO this should be used using multitask.Queue and remove previous wait.
if _debug: print 'Protocol.write msg=', message
if message is None:
try: self.stream.close() # just in case TCP socket is not closed, close it.
except: pass
break
# get the header stored for the stream
if self.lastWriteHeaders.has_key(message.streamId):
header = self.lastWriteHeaders[message.streamId]
else:
if self.nextChannelId <= Protocol.PROTOCOL_CHANNEL_ID: self.nextChannelId = Protocol.PROTOCOL_CHANNEL_ID+1
header, self.nextChannelId = Header(self.nextChannelId), self.nextChannelId + 1
self.lastWriteHeaders[message.streamId] = header
if message.type < Message.AUDIO:
header = Header(Protocol.PROTOCOL_CHANNEL_ID)
# now figure out the header data bytes
if header.streamId != message.streamId or header.time == 0 or message.time <= header.time:
header.streamId, header.type, header.size, header.time, header.delta = message.streamId, message.type, message.size, message.time, message.time
control = Header.FULL
elif header.size != message.size or header.type != message.type:
header.type, header.size, header.time, header.delta = message.type, message.size, message.time, message.time-header.time
control = Header.MESSAGE
else:
header.time, header.delta = message.time, message.time-header.time
control = Header.TIME
hdr = Header(channel=header.channel, time=header.delta if control in (Header.MESSAGE, Header.TIME) else header.time, size=header.size, type=header.type, streamId=header.streamId)
assert message.size == len(message.data)
data = ''
while len(message.data) > 0:
data += hdr.toBytes(control) # gather header bytes
count = min(self.writeChunkSize, len(message.data))
data += message.data[:count]
message.data = message.data[count:]
control = Header.SEPARATOR # incomplete message continuation
try:
yield self.stream.write(data)
except ConnectionClosed:
yield self.connectionClosed()
except:
print traceback.print_exc()
class Command(object):
''' Class for command / data messages'''
def __init__(self, type=Message.RPC, name=None, id=None, tm=0, cmdData=None, args=[]):
'''Create a new command with given type, name, id, cmdData and args list.'''
self.type, self.name, self.id, self.time, self.cmdData, self.args = type, name, id, tm, cmdData, args[:]
def __repr__(self):
return ("<Command type=%r name=%r id=%r data=%r args=%r>" % (self.type, self.name, self.id, self.cmdData, self.args))
def setArg(self, arg):
self.args.append(arg)
def getArg(self, index):
return self.args[index]
@classmethod
def fromMessage(cls, message):
''' initialize from a parsed RTMP message'''
assert (message.type in [Message.RPC, Message.RPC3, Message.DATA, Message.DATA3])
length = len(message.data)
if length == 0: raise ValueError('zero length message data')
if message.type == Message.RPC3 or message.type == Message.DATA3:
assert message.data[0] == '\x00' # must be 0 in AMF3
data = message.data[1:]
else:
data = message.data
amfReader = amf.AMF0(data)
inst = cls()
inst.type = message.type
inst.time = message.time
inst.name = amfReader.read() # first field is command name
try:
if message.type == Message.RPC or message.type == Message.RPC3:
inst.id = amfReader.read() # second field *may* be message id
inst.cmdData = amfReader.read() # third is command data
else:
inst.id = 0
inst.args = [] # others are optional
while True:
inst.args.append(amfReader.read())
except EOFError:
pass
return inst
def toMessage(self):
msg = Message()
assert self.type
msg.type = self.type
msg.time = self.time
output = amf.BytesIO()
amfWriter = amf.AMF0(output)
amfWriter.write(self.name)
if msg.type == Message.RPC or msg.type == Message.RPC3:
amfWriter.write(self.id)
amfWriter.write(self.cmdData)
for arg in self.args:
amfWriter.write(arg)
output.seek(0)
#hexdump.hexdump(output)
#output.seek(0)
if msg.type == Message.RPC3 or msg.type == Message.DATA3:
data = '\x00' + output.read()
else:
data = output.read()
msg.data = data
output.close()
return msg
def getfilename(path, name, root):
'''return the file name for the given stream. The name is derived as root/scope/name.flv where scope is
the the path present in the path variable.'''
ignore, ignore, scope = path.partition('/')
if scope: scope = scope + '/'
result = root + scope + name + '.flv'
if _debug: print 'filename=', result
return result
class FLV(object):
'''An FLV file which converts between RTMP message and FLV tags.'''
def __init__(self):
self.fname = self.fp = self.type = None
self.tsp = self.tsr = 0; self.tsr0 = None
def open(self, path, type='read', mode=0775):
'''Open the file for reading (type=read) or writing (type=record or append).'''
if str(path).find('/../') >= 0 or str(path).find('\\..\\') >= 0: raise ValueError('Must not contain .. in name')
if _debug: print 'opening file', path
self.tsp = self.tsr = 0; self.tsr0 = None; self.tsr1 = 0; self.type = type
if type in ('record', 'append'):
try: os.makedirs(os.path.dirname(path), mode)
except: pass
if type == 'record' or not os.path.exists(path): # if file does not exist, use record mode
self.fp = open(path, 'w+b')
self.fp.write('FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00') # the header and first previousTagSize
self.writeDuration(0.0)
else:
self.fp = open(path, 'r+b')
self.fp.seek(-4, os.SEEK_END)
ptagsize, = struct.unpack('>I', self.fp.read(4))
self.fp.seek(-4-ptagsize, os.SEEK_END)
bytes = self.fp.read(ptagsize)
type, len0, len1, ts0, ts1, ts2, sid0, sid1 = struct.unpack('>BBHBHBBH', bytes[:11])
ts = (ts0 << 16) | (ts1 & 0x0ffff) | (ts2 << 24)
self.tsr1 = ts + 20; # some offset after the last packet
self.fp.seek(0, os.SEEK_END)
else:
self.fp = open(path, 'rb')
magic, version, flags, offset = struct.unpack('!3sBBI', self.fp.read(9))
if _debug: print 'FLV.open() hdr=', magic, version, flags, offset
if magic != 'FLV': raise ValueError('This is not a FLV file')
if version != 1: raise ValueError('Unsupported FLV file version')
if offset > 9: self.fp.seek(offset-9, os.SEEK_CUR)
self.fp.read(4) # ignore first previous tag size
return self
def close(self):
'''Close the underlying file for this object.'''
if _debug: print 'closing flv file'
if self.type in ('record', 'append') and self.tsr0 is not None:
self.writeDuration((self.tsr - self.tsr0)/1000.0)
if self.fp is not None:
try: self.fp.close()
except: pass
self.fp = None
def delete(self, path):
'''Delete the underlying file for this object.'''
try: os.unlink(path)
except: pass
def writeDuration(self, duration):
if _debug: print 'writing duration', duration
output = amf.BytesIO()
amfWriter = amf.AMF0(output) # TODO: use AMF3 if needed
amfWriter.write('onMetaData')
amfWriter.write({"duration": duration, "videocodecid": 2})
output.seek(0); data = output.read()
length, ts = len(data), 0
data = struct.pack('>BBHBHB', Message.DATA, (length >> 16) & 0xff, length & 0x0ffff, (ts >> 16) & 0xff, ts & 0x0ffff, (ts >> 24) & 0xff) + '\x00\x00\x00' + data
data += struct.pack('>I', len(data))
lastpos = self.fp.tell()
if lastpos != 13: self.fp.seek(13, os.SEEK_SET)
self.fp.write(data)
if lastpos != 13: self.fp.seek(lastpos, os.SEEK_SET)
def write(self, message):
'''Write a message to the file, assuming it was opened for writing or appending.'''
# if message.type == Message.VIDEO:
# self.videostarted = True
# elif not hasattr(self, "videostarted"): return
if message.type == Message.AUDIO or message.type == Message.VIDEO:
length, ts = message.size, message.time
#if _debug: print 'FLV.write()', message.type, ts
if self.tsr0 is None: self.tsr0 = ts - self.tsr1
self.tsr, ts = ts, ts - self.tsr0
# if message.type == Message.AUDIO: print 'w', message.type, ts
data = struct.pack('>BBHBHB', message.type, (length >> 16) & 0xff, length & 0x0ffff, (ts >> 16) & 0xff, ts & 0x0ffff, (ts >> 24) & 0xff) + '\x00\x00\x00' + message.data
data += struct.pack('>I', len(data))
self.fp.write(data)
def reader(self, stream):
'''A generator to periodically read the file and dispatch them to the stream. The supplied stream
object must have a send(Message) method and id and client properties.'''
if _debug: print 'reader started'
yield
try:
while self.fp is not None:
bytes = self.fp.read(11)
if len(bytes) == 0:
try: tm = stream.client.relativeTime
except: tm = 0
response = Command(name='onStatus', id=stream.id, tm=tm, args=[amf.Object(level='status',code='NetStream.Play.Stop', description='File ended', details=None)])
yield stream.send(response.toMessage())
break
type, len0, len1, ts0, ts1, ts2, sid0, sid1 = struct.unpack('>BBHBHBBH', bytes)
length = (len0 << 16) | len1; ts = (ts0 << 16) | (ts1 & 0x0ffff) | (ts2 << 24)
body = self.fp.read(length); ptagsize, = struct.unpack('>I', self.fp.read(4))
if ptagsize != (length+11):
if _debug: print 'invalid previous tag-size found:', ptagsize, '!=', (length+11),'ignored.'
if stream is None or stream.client is None: break # if it is closed
#hdr = Header(3 if type == Message.AUDIO else 4, ts if ts < 0xffffff else 0xffffff, length, type, stream.id)
hdr = Header(0, ts, length, type, stream.id)
msg = Message(hdr, body)
# if _debug: print 'FLV.read() length=', length, 'hdr=', hdr
# if hdr.type == Message.AUDIO: print 'r', hdr.type, hdr.time
if type == Message.DATA: # metadata
amfReader = amf.AMF0(body) # TODO: use AMF3 if needed
name = amfReader.read()
obj = amfReader.read()
if _debug: print 'FLV.read()', name, repr(obj)
yield stream.send(msg)
if ts > self.tsp:
diff, self.tsp = ts - self.tsp, ts
if _debug: print 'FLV.read() sleep', diff
yield multitask.sleep(diff / 1000.0)
except StopIteration: pass
except:
if _debug: print 'closing the reader', (sys and sys.exc_info() or None)
if self.fp is not None:
try: self.fp.close()
except: pass
self.fp = None
def seek(self, offset):
'''For file reader, try seek to the given time. The offset is in millisec'''
if self.type == 'read':
if _debug: print 'FLV.seek() offset=', offset, 'current tsp=', self.tsp
self.fp.seek(0, os.SEEK_SET)
magic, version, flags, length = struct.unpack('!3sBBI', self.fp.read(9))
if length > 9: self.fp.seek(length-9, os.SEEK_CUR)
self.fp.seek(4, os.SEEK_CUR) # ignore first previous tag size
self.tsp, ts = int(offset), 0
while self.tsp > 0 and ts < self.tsp:
bytes = self.fp.read(11)
if not bytes: break
type, len0, len1, ts0, ts1, ts2, sid0, sid1 = struct.unpack('>BBHBHBBH', bytes)
length = (len0 << 16) | len1; ts = (ts0 << 16) | (ts1 & 0x0ffff) | (ts2 << 24)
self.fp.seek(length, os.SEEK_CUR)
ptagsize, = struct.unpack('>I', self.fp.read(4))
if ptagsize != (length+11): break
if _debug: print 'FLV.seek() new ts=', ts, 'tell', self.fp.tell()
class Stream(object):
'''The stream object that is used for RTMP stream.'''
count = 0;
def __init__(self, client):
self.client, self.id, self.name = client, 0, ''
self.recordfile = self.playfile = None # so that it doesn't complain about missing attribute
self.queue = multitask.Queue()
self._name = 'Stream[' + str(Stream.count) + ']'; Stream.count += 1
if _debug: print self, 'created'
def close(self):
if _debug: print self, 'closing'
if self.recordfile is not None: self.recordfile.close(); self.recordfile = None
if self.playfile is not None: self.playfile.close(); self.playfile = None
self.client = None # to clear the reference
pass
def __repr__(self):
return self._name;
def recv(self):
'''Generator to receive new Message on this stream, or None if stream is closed.'''
return self.queue.get()
def send(self, msg):
'''Method to send a Message or Command on this stream.'''
if isinstance(msg, Command):
msg = msg.toMessage()
msg.streamId = self.id
# if _debug: print self,'send'
if self.client is not None: yield self.client.writeMessage(msg)
class Client(Protocol):
'''The client object represents a single connected client to the server.'''
def __init__(self, sock, server):
Protocol.__init__(self, sock)
self.server, self.agent, self.streams, self._nextCallId, self._nextStreamId, self.objectEncoding = \
server, None, {}, 2, 1, 0.0
self.queue = multitask.Queue() # receive queue used by application
multitask.add(self.parse()); multitask.add(self.write())
def recv(self):
'''Generator to receive new Message (msg, arg) on this stream, or (None,None) if stream is closed.'''
return self.queue.get()
def connectionClosed(self):
'''Called when the client drops the connection'''
if _debug: 'Client.connectionClosed'
yield self.writeMessage(None)
yield self.queue.put((None,None))
def messageReceived(self, msg):
if (msg.type == Message.RPC or msg.type == Message.RPC3) and msg.streamId == 0:
cmd = Command.fromMessage(msg)
# if _debug: print 'rtmp.Client.messageReceived cmd=', cmd
if cmd.name == 'connect':
self.agent = cmd.cmdData
if _debug: print 'connect', ', '.join(['%s=%r'%(x, getattr(self.agent, x)) for x in 'app flashVer swfUrl tcUrl fpad capabilities audioCodecs videoCodecs videoFunction pageUrl objectEncoding'.split() if hasattr(self.agent, x)])
self.objectEncoding = self.agent.objectEncoding if hasattr(self.agent, 'objectEncoding') else 0.0
yield self.server.queue.put((self, cmd.args)) # new connection
elif cmd.name == 'createStream':
response = Command(name='_result', id=cmd.id, tm=self.relativeTime, type=self.rpc, args=[self._nextStreamId])
yield self.writeMessage(response.toMessage())
stream = Stream(self) # create a stream object
stream.id = self._nextStreamId
self.streams[self._nextStreamId] = stream
self._nextStreamId += 1
yield self.queue.put(('stream', stream)) # also notify others of our new stream
elif cmd.name == 'closeStream':
assert msg.streamId in self.streams
yield self.streams[msg.streamId].queue.put(None) # notify closing to others
del self.streams[msg.streamId]
else:
# if _debug: print 'Client.messageReceived cmd=', cmd
yield self.queue.put(('command', cmd)) # RPC call
else: # this has to be a message on the stream
assert msg.streamId != 0
assert msg.streamId in self.streams
# if _debug: print self.streams[msg.streamId], 'recv'
stream = self.streams[msg.streamId]
if not stream.client: stream.client = self
yield stream.queue.put(msg) # give it to stream
@property
def rpc(self):
# TODO: reverting r141 since it causes exception in setting self.rpc
return Message.RPC if self.objectEncoding == 0.0 else Message.RPC3
def accept(self):
'''Method to accept an incoming client.'''
response = Command()
response.id, response.name, response.type = 1, '_result', self.rpc
if _debug: print 'Client.accept() objectEncoding=', self.objectEncoding
arg = amf.Object(level='status', code='NetConnection.Connect.Success',
description='Connection succeeded.', fmsVer='rtmplite/8,2')
if hasattr(self.agent, 'objectEncoding'):
arg.objectEncoding, arg.details = self.objectEncoding, None
response.setArg(arg)
yield self.writeMessage(response.toMessage())
def rejectConnection(self, reason=''):
'''Method to reject an incoming client.'''
response = Command()
response.id, response.name, response.type = 1, '_error', self.rpc
response.setArg(amf.Object(level='status', code='NetConnection.Connect.Rejected',
description=reason, fmsVer='rtmplite/8,2', details=None))
yield self.writeMessage(response.toMessage())
def redirectConnection(self, url, reason='Connection failed'):
'''Method to redirect an incoming client to the given url.'''
response = Command()
response.id, response.name, response.type = 1, '_error', self.rpc
extra = dict(code=302, redirect=url)
response.setArg(amf.Object(level='status', code='NetConnection.Connect.Rejected',
description=reason, fmsVer='rtmplite/8,2', details=None, ex=extra))
yield self.writeMessage(response.toMessage())
def call(self, method, *args):
'''Call a (callback) method on the client.'''
cmd = Command()
cmd.id, cmd.time, cmd.name, cmd.type = self._nextCallId, self.relativeTime, method, self.rpc
cmd.args, cmd.cmdData = args, None
self._nextCallId += 1
if _debug: print 'Client.call method=', method, 'args=', args, ' msg=', cmd.toMessage()
yield self.writeMessage(cmd.toMessage())
def createStream(self):
''' Create a stream on the server side'''
stream = Stream(self)
stream.id = self._nextStreamId
self.streams[stream.id] = stream
self._nextStreamId += 1
return stream
class Server(object):
'''A RTMP server listens for incoming connections and informs the app.'''
def __init__(self, sock):
'''Create an RTMP server on the given bound TCP socket. The server will terminate
when the socket is disconnected, or some other error occurs in listening.'''
self.sock = sock
self.queue = multitask.Queue() # queue to receive incoming client connections
multitask.add(self.run())
def recv(self):
'''Generator to wait for incoming client connections on this server and return
(client, args) or (None, None) if the socket is closed or some error.'''
return self.queue.get()
def run(self):
try:
while True:
sock, remote = (yield multitask.accept(self.sock)) # receive client TCP
if sock == None:
if _debug: print 'rtmp.Server accept(sock) returned None.'
break
if _debug: print 'connection received from', remote
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) # make it non-block
client = Client(sock, self)
except GeneratorExit: pass # terminate
except:
if _debug: print 'rtmp.Server exception ', (sys and sys.exc_info() or None)
if (self.sock):
try: self.sock.close(); self.sock = None
except: pass
if (self.queue):
yield self.queue.put((None, None))
self.queue = None
class App(object):
'''An application instance containing any number of streams. Except for constructor all methods are generators.'''
count = 0
def __init__(self):
self.name = str(self.__class__.__name__) + '[' + str(App.count) + ']'; App.count += 1
self.players, self.publishers, self._clients = {}, {}, [] # Streams indexed by stream name, and list of clients
if _debug: print self.name, 'created'
def __del__(self):
if _debug: print self.name, 'destroyed'
@property
def clients(self):
'''everytime this property is accessed it returns a new list of clients connected to this instance.'''
return self._clients[1:] if self._clients is not None else []
def onConnect(self, client, *args):
if _debug: print self.name, 'onConnect', client.path
return True
def onDisconnect(self, client):
if _debug: print self.name, 'onDisconnect', client.path
def onPublish(self, client, stream):
if _debug: print self.name, 'onPublish', client.path, stream.name
def onClose(self, client, stream):
if _debug: print self.name, 'onClose', client.path, stream.name
def onPlay(self, client, stream):
if _debug: print self.name, 'onPlay', client.path, stream.name
def onStop(self, client, stream):
if _debug: print self.name, 'onStop', client.path, stream.name
def onCommand(self, client, cmd, *args):
if _debug: print self.name, 'onCommand', cmd, args
def onStatus(self, client, info):
if _debug: print self.name, 'onStatus', info
def onResult(self, client, result):
if _debug: print self.name, 'onResult', result
def onPublishData(self, client, stream, message): # this is invoked every time some media packet is received from published stream.
return True # should return True so that the data is actually published in that stream
def onPlayData(self, client, stream, message):
return True # should return True so that data will be actually played in that stream
def getfile(self, path, name, root, mode):
if mode == 'play':
path = getfilename(path, name, root)
if not os.path.exists(path): return None
return FLV().open(path)
elif mode in ('record', 'append'):