1 # Copyright (c) 2007-2009, Mamta Singh. All rights reserved. see README for details.
4 This is a simple implementation of a Flash RTMP server to accept connections and stream requests. The module is organized as follows:
5 1. The FlashServer class is the main class to provide the server abstraction. It uses the multitask module for co-operative multitasking.
6 It also uses the App abstract class to implement the applications.
7 2. The Server class implements a simple server to receive new Client connections and inform the FlashServer application. The Client class
8 derived from Protocol implements the RTMP client functions. The Protocol class implements the base RTMP protocol parsing. A Client contains
9 various streams from the client, represented using the Stream class.
10 3. The Message, Header and Command represent RTMP message, header and command respectively. The FLV class implements functions to perform read
11 and write of FLV file format.
14 Typically an application can launch this server as follows:
17 To know the command line options use the -h option:
20 To start the server with a different directory for recording and playing FLV files from, use the following command.
21 $ python rtmp.py -r some-other-directory/
22 Note the terminal '/' in the directory name. Without this, it is just used as a prefix in FLV file names.
24 A test client is available in testClient directory, and can be compiled using Flex Builder. Alternatively, you can use the SWF file to launch
25 from testClient/bin-debug after starting the server. Once you have launched the client in the browser, you can connect to
26 local host by clicking on 'connect' button. Then click on publish button to publish a stream. Open another browser with
27 same URL and first connect then play the same stream name. If everything works fine you should be able to see the video
28 from first browser to the second browser. Similar, in the first browser, if you check the record box before publishing,
29 it will create a new FLV file for the recorded stream. You can close the publishing stream and play the recorded stream to
30 see your recording. Note that due to initial delay in timestamp (in case publish was clicked much later than connect),
31 your played video will start appearing after some initial delay.
34 If an application wants to use this module as a library, it can launch the server as follows:
35 >>> agent = FlashServer() # a new RTMP server instance
36 >>> agent.root = 'flvs/' # set the document root to be 'flvs' directory. Default is current './' directory.
37 >>> agent.start() # start the server
38 >>> multitask.run() # this is needed somewhere in the application to actually start the co-operative multitasking.
41 If an application wants to specify a different application other than the default App, it can subclass it and supply the application by
42 setting the server's apps property. The following example shows how to define "myapp" which invokes a 'connected()' method on client when
43 the client connects to the server.
45 class MyApp(App): # a new MyApp extends the default App in rtmp module.
46 def __init__(self): # constructor just invokes base class constructor
48 def onConnect(self, client, *args):
49 result = App.onConnect(self, client, *args) # invoke base class method first
50 def invokeAdded(self, client): # define a method to invoke 'connected("some-arg")' on Flash client
51 client.call('connected', 'some-arg')
53 multitask.add(invokeAdded(self, client)) # need to invoke later so that connection is established before callback
55 agent.apps = dict({'myapp': MyApp, 'someapp': MyApp, '*': App})
57 Now the client can connect to rtmp://server/myapp or rtmp://server/someapp and will get connected to this MyApp application.
58 If the client doesn't define "function connected(arg:String):void" in the NetConnection.client object then the server will
59 throw an exception and display the error message.
63 import os
, sys
, time
, struct
, socket
, traceback
, multitask
64 import threading
, Queue
66 from string
import strip
68 from BaseHTTPServer
import BaseHTTPRequestHandler
69 from SimpleAppHTTPServer
import SimpleAppHTTPRequestHandler
70 from cStringIO
import StringIO
71 from cookielib
import parse_ns_headers
, CookieJar
72 from Cookie
import SimpleCookie
80 class MultitaskHTTPRequestHandler(BaseHTTPRequestHandler
):
83 self
.connection
= self
.request
84 self
.rfile
= StringIO()
85 self
.wfile
= StringIO()
93 def get_full_url(self
):
96 def get_header(self
, hdr
, default
):
97 return self
.headers
.getheader(hdr
, default
)
99 def add_cookies(self
):
100 for k
, v
in self
.response_cookies
.items():
101 val
= v
.OutputString()
102 self
.send_header("Set-Cookie", val
)
104 class ConnectionClosed
:
105 'raised when the client closed the connection'
107 class SockStream(object):
108 '''A class that represents a socket as a stream'''
109 def __init__(self
, sock
):
110 self
.sock
, self
.buffer = sock
, ''
111 self
.bytesWritten
= self
.bytesRead
= 0
114 self
.sock
.shutdown(1) # can't just close it.
120 nl
= self
.buffer.find("\n")
121 if nl
>= 0: # do not have newline in buffer
122 data
, self
.buffer = self
.buffer[:nl
+1], self
.buffer[nl
+1:]
123 raise StopIteration(data
)
124 data
= (yield multitask
.recv(self
.sock
, 4096)) # read more from socket
125 if not data
: raise ConnectionClosed
126 if _debug
: print 'socket.read[%d] %r'%(len(data
), data
)
127 self
.bytesRead
+= len(data
)
129 except StopIteration: raise
130 except: raise ConnectionClosed
# anything else is treated as connection closed.
132 def read(self
, count
):
135 if len(self
.buffer) >= count
: # do not have data in buffer
136 data
, self
.buffer = self
.buffer[:count
], self
.buffer[count
:]
137 raise StopIteration(data
)
138 data
= (yield multitask
.recv(self
.sock
, 4096)) # read more from socket
139 if not data
: raise ConnectionClosed
140 # if _debug: print 'socket.read[%d] %r'%(len(data), data)
141 self
.bytesRead
+= len(data
)
143 except StopIteration: raise
144 except: raise ConnectionClosed
# anything else is treated as connection closed.
146 def unread(self
, data
):
147 self
.buffer = data
+ self
.buffer
149 def write(self
, data
):
150 while len(data
) > 0: # write in 4K chunks each time
151 chunk
, data
= data
[:4096], data
[4096:]
152 self
.bytesWritten
+= len(chunk
)
153 if _debug
: print 'socket.write[%d] %r'%(len(chunk
), chunk
)
154 try: yield multitask
.send(self
.sock
, chunk
)
155 except: raise ConnectionClosed
156 if _debug
: print 'socket.written'
159 class Header(object):
160 FULL
, MESSAGE
, TIME
, SEPARATOR
, MASK
= 0x00, 0x40, 0x80, 0xC0, 0xC0
162 def __init__(self
, channel
=0, time
=0, size
=None, type=None, streamId
=0):
163 self
.channel
, self
.time
, self
.size
, self
.type, self
.streamId
= channel
, time
, size
, type, streamId
164 self
.extendedtime
= 0
165 if channel
<64: self
.hdrdata
= chr(channel
)
166 elif channel
<(64+256): self
.hdrdata
= '\x00'+chr(channel
-64)
167 else: self
.hdrdata
= '\x01'+chr((channel
-64)%256)+chr((channel
-64)/256)
169 def _appendExtendedTimestamp(self
, data
):
170 if self
.time
== 0xFFFFFF:
171 data
+= struct
.pack('>I', self
.extendedtime
)
174 def toBytes(self
, control
):
175 data
= chr(ord(self
.hdrdata
[0]) | control
) + self
.hdrdata
[1:]
176 if control
== Header
.SEPARATOR
: return self
._appendExtendedTimestamp
(data
)
178 data
+= struct
.pack('>I', self
.time
& 0xFFFFFF)[1:] # add time
179 if control
== Header
.TIME
: return self
._appendExtendedTimestamp
(data
)
181 data
+= struct
.pack('>I', self
.size
)[1:] # size
182 data
+= chr(self
.type) # type
183 if control
== Header
.MESSAGE
: return self
._appendExtendedTimestamp
(data
)
185 data
+= struct
.pack('<I', self
.streamId
) # add streamId
186 return self
._appendExtendedTimestamp
(data
)
189 return ("<Header channel=%r time=%r size=%r type=%r (0x%02x) streamId=%r>"
190 % (self
.channel
, self
.time
, self
.size
, self
.type, self
.type or 0, self
.streamId
))
192 class Message(object):
193 # message types: RPC3, DATA3,and SHAREDOBJECT3 are used with AMF3
194 RPC
, RPC3
, DATA
, DATA3
, SHAREDOBJ
, SHAREDOBJ3
, AUDIO
, VIDEO
, ACK
, CHUNK_SIZE
= \
195 0x14, 0x11, 0x12, 0x0F, 0x13, 0x10, 0x08, 0x09, 0x03, 0x01
197 def __init__(self
, hdr
=None, data
=''):
198 if hdr
is None: hdr
= Header()
199 self
.header
, self
.data
= hdr
, data
201 # define properties type, streamId and time to access self.header.(property)
202 for p
in ['type', 'streamId', 'time']:
203 exec 'def _g%s(self): return self.header.%s'%(p
, p
)
204 exec 'def _s%s(self, %s): self.header.%s = %s'%(p
, p
, p
, p
)
205 exec '%s = property(fget=_g%s, fset=_s%s)'%(p
, p
, p
)
207 def size(self
): return len(self
.data
)
210 return ("<Message header=%r data=%r>"% (self
.header
, self
.data
))
212 class Protocol(object):
215 DEFAULT_CHUNK_SIZE
= 128
217 PROTOCOL_CHANNEL_ID
= 2
219 def __init__(self
, sock
):
220 self
.stream
= SockStream(sock
)
221 self
.lastReadHeaders
= dict() # indexed by channelId
222 self
.incompletePackets
= dict() #indexed by channelId
223 self
.readChunkSize
= self
.writeChunkSize
= Protocol
.DEFAULT_CHUNK_SIZE
224 self
.lastWriteHeaders
= dict() # indexed by streamId
225 self
.nextChannelId
= Protocol
.MIN_CHANNEL_ID
226 self
.writeLock
= threading
.Lock()
227 self
.writeQueue
= Queue
.Queue()
229 def messageReceived(self
, msg
):
232 def protocolMessage(self
, msg
):
233 if msg
.type == Message
.ACK
: # respond to ACK requests
235 response
.type, response
.data
= msg
.type, msg
.data
236 self
.writeMessage(response
)
237 elif msg
.type == Message
.CHUNK_SIZE
:
238 self
.readChunkSize
= struct
.unpack('>L', msg
.data
)[0]
240 def connectionClosed(self
):
245 yield self
.parseRequests() # parse http requests
246 except ConnectionClosed
:
247 yield self
.connectionClosed()
248 if _debug
: print 'parse connection closed'
250 def writeMessage(self
, message
):
251 self
.writeQueue
.put(message
)
253 def parseHandshake(self
):
254 '''Parses the rtmp handshake'''
255 data
= (yield self
.stream
.read(Protocol
.PING_SIZE
+ 1)) # bound version and first ping
256 yield self
.stream
.write(data
)
257 data
= (yield self
.stream
.read(Protocol
.PING_SIZE
)) # bound second ping
258 yield self
.stream
.write(data
)
260 def parseRequests(self
):
261 '''Parses complete messages until connection closed. Raises ConnectionLost exception.'''
262 self
.hr
= MultitaskHTTPRequestHandler(self
.stream
, self
.remote
, None)
263 self
.hr
.close_connection
= 1
264 self
.cookies
= CookieJar()
268 # prepare reading the request so that when it's handed
269 # over to "standard" HTTPRequestHandler, the data's already
271 print "parseRequests"
272 raw_requestline
= (yield self
.stream
.readline())
273 if _debug
: print "parseRequests, line", raw_requestline
274 if not raw_requestline
:
275 raise ConnectionClosed
279 line
= (yield self
.stream
.readline())
281 if line
in ['\n', '\r\n']:
283 except StopIteration:
284 if _debug
: print "parseRequests, stopiter"
288 print 'parseRequests', \
289 (traceback
and traceback
.print_exc() or None)
291 raise ConnectionClosed
293 self
.hr
.raw_requestline
= raw_requestline
294 self
.hr
.rfile
.write(data
)
295 print "parseRequests write after"
296 self
.hr
.rfile
.seek(0)
297 print "parseRequests seek after"
299 if not self
.hr
.parse_request():
300 raise ConnectionClosed
301 print "parseRequests parse_req after"
303 yield self
.messageReceived(self
.hr
)
306 print 'messageReceived', \
307 (traceback
and traceback
.print_exc() or None)
310 '''Writes messages to stream'''
312 while self
.writeQueue
.empty(): (yield multitask
.sleep(0.01))
313 data
= self
.writeQueue
.get() # TODO this should be used using multitask.Queue and remove previous wait.
314 if _debug
: print "write to stream", repr(data
)
316 # just in case TCP socket is not closed, close it.
318 print "stream closing"
320 yield self
.stream
.close()
321 print "stream closed"
326 yield self
.stream
.write(data
)
327 except ConnectionClosed
:
328 yield self
.connectionClosed()
330 print traceback
.print_exc()
332 class Command(object):
333 ''' Class for command / data messages'''
334 def __init__(self
, type=Message
.RPC
, name
=None, id=None, cmdData
=None, args
=[]):
335 '''Create a new command with given type, name, id, cmdData and args list.'''
336 self
.type, self
.name
, self
.id, self
.cmdData
, self
.args
= type, name
, id, cmdData
, args
[:]
339 return ("<Command type=%r name=%r id=%r data=%r args=%r>" % (self
.type, self
.name
, self
.id, self
.cmdData
, self
.args
))
341 def setArg(self
, arg
):
342 self
.args
.append(arg
)
344 def getArg(self
, index
):
345 return self
.args
[index
]
348 def fromMessage(cls
, message
):
349 ''' initialize from a parsed RTMP message'''
350 assert (message
.type in [Message
.RPC
, Message
.RPC3
, Message
.DATA
, Message
.DATA3
])
352 length
= len(message
.data
)
353 if length
== 0: raise ValueError('zero length message data')
355 if message
.type == Message
.RPC3
or message
.type == Message
.DATA3
:
356 assert message
.data
[0] == '\x00' # must be 0 in AMD3
357 data
= message
.data
[1:]
361 amfReader
= amf
.AMF0(data
)
364 inst
.name
= amfReader
.read() # first field is command name
367 if message
.type == Message
.RPC
:
368 inst
.id = amfReader
.read() # second field *may* be message id
369 inst
.cmdData
= amfReader
.read() # third is command data
372 inst
.args
= [] # others are optional
374 inst
.args
.append(amfReader
.read())
383 output
= amf
.BytesIO()
384 amfWriter
= amf
.AMF0(output
)
385 amfWriter
.write(self
.name
)
386 if msg
.type == Message
.RPC
or msg
.type == Message
.RPC3
:
387 amfWriter
.write(self
.id)
388 amfWriter
.write(self
.cmdData
)
389 for arg
in self
.args
:
392 #hexdump.hexdump(output)
394 if msg
.type == Message
.RPC3
or msg
.type == Message
.DATA3
:
395 data
= '\x00' + output
.read()
402 def getfilename(path
, name
, root
):
403 '''return the file name for the given stream. The name is derived as root/scope/name.flv where scope is
404 the the path present in the path variable.'''
405 ignore
, ignore
, scope
= path
.partition('/')
406 if scope
: scope
= scope
+ '/'
407 result
= root
+ scope
+ name
+ '.flv'
408 if _debug
: print 'filename=', result
411 class Stream(object):
412 '''The stream object that is used for RTMP stream.'''
414 def __init__(self
, client
):
415 self
.client
, self
.id, self
.name
= client
, 0, ''
416 self
.recordfile
= self
.playfile
= None # so that it doesn't complain about missing attribute
417 self
.queue
= multitask
.Queue()
418 self
._name
= 'Stream[' + str(Stream
.count
) + ']'; Stream
.count
+= 1
419 if _debug
: print self
, 'created'
422 if _debug
: print self
, 'closing'
423 if self
.recordfile
is not None: self
.recordfile
.close(); self
.recordfile
= None
424 if self
.playfile
is not None: self
.playfile
.close(); self
.playfile
= None
425 self
.client
= None # to clear the reference
432 '''Generator to receive new Message on this stream, or None if stream is closed.'''
433 return self
.queue
.get()
436 '''Method to send a Message or Command on this stream.'''
437 if isinstance(msg
, Command
):
438 msg
= msg
.toMessage()
439 msg
.streamId
= self
.id
440 # if _debug: print self,'send'
441 if self
.client
is not None: self
.client
.writeMessage(msg
)
443 class Client(Protocol
):
444 '''The client object represents a single connected client to the server.'''
445 def __init__(self
, sock
, server
, remote
):
446 Protocol
.__init
__(self
, sock
)
452 self
._nextStreamId
= 1
453 self
.objectEncoding
= 0.0
454 self
.queue
= multitask
.Queue() # receive queue used by application
455 multitask
.add(self
.parse())
456 multitask
.add(self
.write())
459 '''Generator to receive new Message (msg, arg) on this stream, or (None,None) if stream is closed.'''
460 return self
.queue
.get()
462 def connectionClosed(self
):
463 '''Called when the client drops the connection'''
464 if _debug
: 'Client.connectionClosed'
465 self
.writeMessage(None)
466 yield self
.queue
.put((None,None))
468 def messageReceived(self
, msg
):
469 if _debug
: print 'messageReceived cmd=', msg
.command
, msg
.path
470 ch
= msg
.headers
.getheaders("Cookie")
471 print "messageReceived cookieheaders=", '; '.join(ch
)
478 msg
.response_cookies
= SimpleCookie()
480 print "found cookie", str(c
)
481 name
, value
= c
.split("=")
482 msg
.response_cookies
[name
] = value
483 msg
.response_cookies
[name
]['path'] = "/"
484 msg
.response_cookies
[name
]['domain'] = self
.remote
[0]
485 #msg.response_cookies[name]['expires'] = 'None'
486 msg
.response_cookies
[name
]['version'] = 0
487 if name
== "session":
490 msg
.response_cookies
['session'] = uuid
.uuid4().hex
491 #msg.response_cookies['session']['expires'] = 'None'
492 msg
.response_cookies
['session']['path'] = '/'
493 msg
.response_cookies
['session']['domain'] = self
.remote
[0]
494 msg
.response_cookies
['session']['version'] = 0
496 if msg
.headers
.has_key('content-length'):
497 max_chunk_size
= 10*1024*1024
498 size_remaining
= int(msg
.headers
["content-length"])
500 while size_remaining
:
501 chunk_size
= min(size_remaining
, max_chunk_size
)
502 data
= (yield self
.stream
.read(chunk_size
))
504 size_remaining
-= len(L
[-1])
506 pos
= msg
.rfile
.tell()
507 msg
.rfile
.write(''.join(L
))
510 yield self
.server
.queue
.put((self
, msg
)) # new connection
513 '''Method to accept an incoming client.'''
515 response
.id, response
.name
, response
.type = 1, '_result', Message
.RPC
516 if _debug
: print 'Client.accept() objectEncoding=', self
.objectEncoding
517 response
.setArg(dict(level
='status', code
='NetConnection.Connect.Success',
518 description
='Connection succeeded.', details
=None,
519 objectEncoding
=self
.objectEncoding
))
520 self
.writeMessage(response
.toMessage())
522 def rejectConnection(self
, reason
=''):
523 '''Method to reject an incoming client.'''
525 response
.id, response
.name
, response
.type = 1, '_error', Message
.RPC
526 response
.setArg(dict(level
='status', code
='NetConnection.Connect.Rejected',
527 description
=reason
, details
=None))
528 self
.writeMessage(response
.toMessage())
530 def call(self
, method
, *args
):
531 '''Call a (callback) method on the client.'''
533 cmd
.id, cmd
.name
, cmd
.type = self
._nextCallId
, method
, (self
.objectEncoding
== 0.0 and Message
.RPC
or Message
.RPC3
)
534 cmd
.args
, cmd
.cmdData
= args
, None
535 self
._nextCallId
+= 1
536 if _debug
: print 'Client.call method=', method
, 'args=', args
, ' msg=', cmd
.toMessage()
537 self
.writeMessage(cmd
.toMessage())
539 def createStream(self
):
540 ''' Create a stream on the server side'''
541 stream
= Stream(self
)
542 stream
.id = self
._nextStreamId
543 self
.streams
[stream
.id] = stream
544 self
._nextStreamId
+= 1
548 class Server(object):
549 '''A RTMP server listens for incoming connections and informs the app.'''
550 def __init__(self
, sock
):
551 '''Create an RTMP server on the given bound TCP socket. The server will terminate
552 when the socket is disconnected, or some other error occurs in listening.'''
554 self
.queue
= multitask
.Queue() # queue to receive incoming client connections
555 multitask
.add(self
.run())
558 '''Generator to wait for incoming client connections on this server and return
559 (client, args) or (None, None) if the socket is closed or some error.'''
560 return self
.queue
.get()
565 sock
, remote
= (yield multitask
.accept(self
.sock
)) # receive client TCP
567 if _debug
: print 'rtmp.Server accept(sock) returned None.'
569 if _debug
: print 'connection received from', remote
570 sock
.setsockopt(socket
.IPPROTO_TCP
, socket
.TCP_NODELAY
, 1) # make it non-block
571 client
= Client(sock
, self
, remote
)
573 if _debug
: print 'rtmp.Server exception ', (sys
and sys
.exc_info() or None)
574 traceback
.print_exc()
577 try: self
.sock
.close(); self
.sock
= None
580 yield self
.queue
.put((None, None))
583 class BaseApp(object):
584 '''An application instance containing any number of streams. Except for constructor all methods are generators.'''
587 self
.name
= str(self
.__class
__.__name
__) + '[' + str(App
.count
) + ']'; App
.count
+= 1
588 self
.players
, self
.publishers
, self
._clients
= {}, {}, [] # Streams indexed by stream name, and list of clients
589 if _debug
: print self
.name
, 'created'
591 if _debug
: print self
.name
, 'destroyed'
594 '''everytime this property is accessed it returns a new list of clients connected to this instance.'''
595 return self
._clients
[1:] if self
._clients
is not None else []
597 class App(BaseApp
, SimpleAppHTTPRequestHandler
):
600 class HTTPServer(object):
601 '''A RTMP server to record and stream Flash video.'''
603 '''Construct a new HttpServer. It initializes the local members.'''
604 self
.sock
= self
.server
= None
605 self
.apps
= dict({'*': App
}) # supported applications: * means any as in {'*': App}
606 self
.clients
= dict() # list of clients indexed by scope. First item in list is app instance.
609 def start(self
, host
='0.0.0.0', port
=8080):
610 '''This should be used to start listening for RTMP connections on the given port, which defaults to 8080.'''
611 if _debug
: print 'start', host
, port
613 sock
= self
.sock
= socket
.socket(type=socket
.SOCK_STREAM
)
614 sock
.setsockopt(socket
.SOL_SOCKET
, socket
.SO_REUSEADDR
, 1)
615 sock
.bind((host
, port
))
616 if _debug
: print 'listening on ', sock
.getsockname()
618 server
= self
.server
= Server(sock
) # start rtmp server on that socket
619 multitask
.add(self
.serverlistener())
622 if _debug
: print 'stopping Flash server'
623 if self
.server
and self
.sock
:
624 try: self
.sock
.close(); self
.sock
= None
628 def serverlistener(self
):
629 '''Server listener (generator). It accepts all connections and invokes client listener'''
631 while True: # main loop to receive new connections on the server
632 client
, msg
= (yield self
.server
.recv()) # receive an incoming client connection.
633 # TODO: we should reject non-localhost client connections.
634 if not client
: # if the server aborted abnormally,
635 break # hence close the listener.
636 if _debug
: print 'client connection received', client
, msg
637 # if client.objectEncoding != 0 and client.objectEncoding != 3:
638 if client
.objectEncoding
!= 0:
639 yield client
.rejectConnection(reason
='Unsupported encoding ' + str(client
.objectEncoding
) + '. Please use NetConnection.defaultObjectEncoding=ObjectEncoding.AMF0')
640 yield client
.connectionClosed()
642 print "cookies", str(msg
.response_cookies
)
643 session
= msg
.response_cookies
['session'].value
645 print "serverlistener", name
646 if '*' not in self
.apps
and name
not in self
.apps
:
647 yield client
.rejectConnection(reason
='Application not found: ' + name
)
648 else: # create application instance as needed and add in our list
649 if _debug
: print 'name=', name
, 'name in apps', str(name
in self
.apps
)
650 app
= self
.apps
[name
] if name
in self
.apps
else self
.apps
['*'] # application class
651 print "clients", self
.clients
.keys()
652 if session
in self
.clients
:
653 inst
= self
.clients
[session
][0]
656 msg
.server
= inst
# whew! just in time!
658 methodname
= "on%s" % msg
.command
659 print methodname
, dir(inst
)
660 method
= getattr(inst
, methodname
, None)
661 result
= method(client
, msg
)
662 close_connection
= msg
.close_connection
663 print "close connection", close_connection
665 if _debug
: traceback
.print_exc()
666 yield client
.rejectConnection(reason
='Exception on %s' % methodname
)
668 if result
is True or result
is None:
669 if session
not in self
.clients
:
670 self
.clients
[session
] = [inst
]; inst
._clients
=self
.clients
[session
]
671 self
.clients
[session
].append(client
)
673 data
= msg
.wfile
.read()
676 yield client
.writeMessage(data
)
679 print 'close_connection requested'
681 yield client
.connectionClosed()
684 print 'close_connection done'
687 yield client
.rejectConnection(reason
='Rejected in onConnect')
688 except StopIteration: raise
691 print 'serverlistener exception', \
692 (sys
and sys
.exc_info() or None)
693 traceback
.print_exc()
695 def clientlistener(self
, client
):
696 '''Client listener (generator). It receives a command and invokes client handler, or receives a new stream and invokes streamlistener.'''
699 msg
, arg
= (yield client
.recv()) # receive new message from client
700 if not msg
: # if the client disconnected,
701 if _debug
: print 'connection closed from client'
702 break # come out of listening loop.
703 if msg
== 'command': # handle a new command
704 multitask
.add(self
.clienthandler(client
, arg
))
705 elif msg
== 'stream': # a new stream is created, handle the stream.
707 multitask
.add(self
.streamlistener(arg
))
708 except StopIteration: raise
710 if _debug
: print 'clientlistener exception', (sys
and sys
.exc_info() or None)
711 traceback
.print_exc()
713 # client is disconnected, clear our state for application instance.
714 if _debug
: print 'cleaning up client', client
.path
716 if client
.path
in self
.clients
:
717 inst
= self
.clients
[client
.path
][0]
718 self
.clients
[client
.path
].remove(client
)
719 for stream
in client
.streams
.values(): # for all streams of this client
720 self
.closehandler(stream
)
721 client
.streams
.clear() # and clear the collection of streams
722 if client
.path
in self
.clients
and len(self
.clients
[client
.path
]) == 1: # no more clients left, delete the instance.
723 if _debug
: print 'removing the application instance'
724 inst
= self
.clients
[client
.path
][0]
726 del self
.clients
[client
.path
]
727 if inst
is not None: inst
.onDisconnect(client
)
729 def closehandler(self
, stream
):
730 '''A stream is closed explicitly when a closeStream command is received from given client.'''
731 if stream
.client
is not None:
732 inst
= self
.clients
[stream
.client
.path
][0]
733 if stream
.name
in inst
.publishers
and inst
.publishers
[stream
.name
] == stream
: # clear the published stream
734 inst
.onClose(stream
.client
, stream
)
735 del inst
.publishers
[stream
.name
]
736 if stream
.name
in inst
.players
and stream
in inst
.players
[stream
.name
]:
737 inst
.onStop(stream
.client
, stream
)
738 inst
.players
[stream
.name
].remove(stream
)
739 if len(inst
.players
[stream
.name
]) == 0:
740 del inst
.players
[stream
.name
]
743 def clienthandler(self
, client
, cmd
):
744 '''A generator to handle a single command on the client.'''
745 inst
= self
.clients
[client
.path
][0]
747 if cmd
.name
== '_error':
748 if hasattr(inst
, 'onStatus'):
749 result
= inst
.onStatus(client
, cmd
.args
[0])
750 elif cmd
.name
== '_result':
751 if hasattr(inst
, 'onResult'):
752 result
= inst
.onResult(client
, cmd
.args
[0])
754 res
, code
, args
= Command(), '_result', dict()
755 try: result
= inst
.onCommand(client
, cmd
.name
, *cmd
.args
)
757 if _debug
: print 'Client.call exception', (sys
and sys
.exc_info() or None)
758 code
, args
= '_error', dict()
759 res
.id, res
.name
, res
.type = cmd
.id, code
, (client
.objectEncoding
== 0.0 and Message
.RPC
or Message
.RPC3
)
760 res
.args
, res
.cmdData
= args
, None
761 if _debug
: print 'Client.call method=', code
, 'args=', args
, ' msg=', res
.toMessage()
762 client
.writeMessage(res
.toMessage())
763 # TODO return result to caller
766 def streamlistener(self
, stream
):
767 '''Stream listener (generator). It receives stream message and invokes streamhandler.'''
768 stream
.recordfile
= None # so that it doesn't complain about missing attribute
770 msg
= (yield stream
.recv())
772 if _debug
: print 'stream closed'
773 self
.closehandler(stream
)
776 multitask
.add(self
.streamhandler(stream
, msg
))
778 def streamhandler(self
, stream
, message
):
779 '''A generator to handle a single message on the stream.'''
781 if message
.type == Message
.RPC
:
782 cmd
= Command
.fromMessage(message
)
783 if _debug
: print 'streamhandler received cmd=', cmd
784 if cmd
.name
== 'publish':
785 yield self
.publishhandler(stream
, cmd
)
786 elif cmd
.name
== 'play':
787 yield self
.playhandler(stream
, cmd
)
788 elif cmd
.name
== 'closeStream':
789 self
.closehandler(stream
)
790 else: # audio or video message
791 yield self
.mediahandler(stream
, message
)
792 except GeneratorExit
: pass
793 except StopIteration: pass
795 if _debug
: print 'exception in streamhandler', (sys
and sys
.exc_info())
797 def publishhandler(self
, stream
, cmd
):
798 '''A new stream is published. Store the information in the application instance.'''
800 stream
.mode
= 'live' if len(cmd
.args
) < 2 else cmd
.args
[1] # live, record, append
801 stream
.name
= cmd
.args
[0]
802 if _debug
: print 'publishing stream=', stream
.name
, 'mode=', stream
.mode
803 inst
= self
.clients
[stream
.client
.path
][0]
804 if (stream
.name
in inst
.publishers
):
805 raise ValueError, 'Stream name already in use'
806 inst
.publishers
[stream
.name
] = stream
# store the client for publisher
807 result
= inst
.onPublish(stream
.client
, stream
)
809 if stream
.mode
== 'record' or stream
.mode
== 'append':
810 stream
.recordfile
= FLV().open(getfilename(stream
.client
.path
, stream
.name
, self
.root
), stream
.mode
)
811 response
= Command(name
='onStatus', id=cmd
.id, args
=[dict(level
='status', code
='NetStream.Publish.Start', description
='', details
=None)])
812 yield stream
.send(response
)
813 except ValueError, E
: # some error occurred. inform the app.
814 if _debug
: print 'error in publishing stream', str(E
)
815 response
= Command(name
='onStatus', id=cmd
.id, args
=[dict(level
='error',code
='NetStream.Publish.BadName',description
=str(E
),details
=None)])
816 yield stream
.send(response
)
818 def playhandler(self
, stream
, cmd
):
819 '''A new stream is being played. Just updated the players list with this stream.'''
820 inst
= self
.clients
[stream
.client
.path
][0]
821 name
= stream
.name
= cmd
.args
[0] # store the stream's name
822 start
= cmd
.args
[1] if len(cmd
.args
) >= 2 else -2
823 if name
not in inst
.players
:
824 inst
.players
[name
] = [] # initialize the players for this stream name
825 if stream
not in inst
.players
[name
]: # store the stream as players of this name
826 inst
.players
[name
].append(stream
)
827 path
= getfilename(stream
.client
.path
, stream
.name
, self
.root
)
828 if os
.path
.exists(path
):
829 stream
.playfile
= FLV().open(path
)
830 multitask
.add(stream
.playfile
.reader(stream
))
831 if _debug
: print 'playing stream=', name
, 'start=', start
832 result
= inst
.onPlay(stream
.client
, stream
)
833 response
= Command(name
='onStatus', id=cmd
.id, args
=[dict(level
='status',code
='NetStream.Play.Start', description
=stream
.name
, details
=None)])
834 yield stream
.send(response
)
836 def mediahandler(self
, stream
, message
):
837 '''Handle incoming media on the stream, by sending to other stream in this application instance.'''
838 if stream
.client
is not None:
839 inst
= self
.clients
[stream
.client
.path
][0]
840 result
= inst
.onPublishData(stream
.client
, stream
, message
)
842 client
= stream
.client
843 for s
in (inst
.players
.get(stream
.name
, [])):
844 # if _debug: print 'D', stream.name, s.name
845 result
= inst
.onPlayData(s
.client
, s
, message
)
847 yield s
.send(message
)
848 if stream
.recordfile
is not None:
849 stream
.recordfile
.write(message
)
851 # The main routine to start, run and stop the service
852 if __name__
== '__main__':
854 from optparse
import OptionParser
855 parser
= OptionParser()
856 parser
.add_option('-i', '--host', dest
='host', default
='0.0.0.0', help="listening IP address. Default '0.0.0.0'")
857 parser
.add_option('-p', '--port', dest
='port', default
=8080, type="int", help='listening port number. Default 8080')
858 parser
.add_option('-r', '--root', dest
='root', default
='./', help="document root directory. Default './'")
859 parser
.add_option('-d', '--verbose', dest
='verbose', default
=False, action
='store_true', help='enable debug trace')
860 (options
, args
) = parser
.parse_args()
862 _debug
= options
.verbose
864 if _debug
: print time
.asctime(), 'Options - %s:%d' % (options
.host
, options
.port
)
866 agent
.root
= options
.root
867 agent
.start(options
.host
, options
.port
)
868 if _debug
: print time
.asctime(), 'Flash Server Starts - %s:%d' % (options
.host
, options
.port
)
870 except KeyboardInterrupt:
871 print traceback
.print_exc()
875 print traceback
.print_exc()
876 if _debug
: print time
.asctime(), 'Flash Server Stops'