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 def process_cookies(headers
, remote
, cookie_key
="Cookie", add_sess
=True):
105 ch
= headers
.getheaders(cookie_key
)
106 print "messageReceived cookieheaders=", '; '.join(ch
)
113 c
= filter(lambda x
: x
, c
)
116 response_cookies
= SimpleCookie()
118 print "found cookie", repr(c
)
119 name
, value
= c
.split("=")
120 response_cookies
[name
] = value
121 #response_cookies[name]['path'] = "/"
122 #response_cookies[name]['domain'] = remote[0]
123 #response_cookies[name]['version'] = 0
124 if name
== "session":
125 response_cookies
[name
]['expires'] = 50000
128 return response_cookies
130 response_cookies
['session'] = uuid
.uuid4().hex
131 response_cookies
['session']['expires'] = 50000
132 #response_cookies['session']['path'] = '/'
133 #response_cookies['session']['domain'] = remote[0]
134 #response_cookies['session']['version'] = 0
135 return response_cookies
137 class ConnectionClosed
:
138 'raised when the client closed the connection'
140 class SockStream(object):
141 '''A class that represents a socket as a stream'''
142 def __init__(self
, sock
):
143 self
.sock
, self
.buffer = sock
, ''
144 self
.bytesWritten
= self
.bytesRead
= 0
147 self
.sock
.shutdown(1) # can't just close it.
153 nl
= self
.buffer.find("\n")
154 if nl
>= 0: # do not have newline in buffer
155 data
, self
.buffer = self
.buffer[:nl
+1], self
.buffer[nl
+1:]
156 raise StopIteration(data
)
157 data
= (yield multitask
.recv(self
.sock
, 4096)) # read more from socket
158 if not data
: raise ConnectionClosed
159 if _debug
: print 'socket.read[%d] %r'%(len(data
), data
)
160 self
.bytesRead
+= len(data
)
162 except StopIteration: raise
163 except: raise ConnectionClosed
# anything else is treated as connection closed.
165 def read(self
, count
):
168 if len(self
.buffer) >= count
: # do not have data in buffer
169 data
, self
.buffer = self
.buffer[:count
], self
.buffer[count
:]
170 raise StopIteration(data
)
171 data
= (yield multitask
.recv(self
.sock
, 4096)) # read more from socket
172 if not data
: raise ConnectionClosed
173 # if _debug: print 'socket.read[%d] %r'%(len(data), data)
174 self
.bytesRead
+= len(data
)
176 except StopIteration: raise
177 except: raise ConnectionClosed
# anything else is treated as connection closed.
179 def unread(self
, data
):
180 self
.buffer = data
+ self
.buffer
182 def write(self
, data
):
183 while len(data
) > 0: # write in 4K chunks each time
184 chunk
, data
= data
[:4096], data
[4096:]
185 self
.bytesWritten
+= len(chunk
)
186 if _debug
: print 'socket.write[%d] %r'%(len(chunk
), chunk
)
187 try: yield multitask
.send(self
.sock
, chunk
)
188 except: raise ConnectionClosed
189 if _debug
: print 'socket.written'
192 class Header(object):
193 FULL
, MESSAGE
, TIME
, SEPARATOR
, MASK
= 0x00, 0x40, 0x80, 0xC0, 0xC0
195 def __init__(self
, channel
=0, time
=0, size
=None, type=None, streamId
=0):
196 self
.channel
, self
.time
, self
.size
, self
.type, self
.streamId
= channel
, time
, size
, type, streamId
197 self
.extendedtime
= 0
198 if channel
<64: self
.hdrdata
= chr(channel
)
199 elif channel
<(64+256): self
.hdrdata
= '\x00'+chr(channel
-64)
200 else: self
.hdrdata
= '\x01'+chr((channel
-64)%256)+chr((channel
-64)/256)
202 def _appendExtendedTimestamp(self
, data
):
203 if self
.time
== 0xFFFFFF:
204 data
+= struct
.pack('>I', self
.extendedtime
)
207 def toBytes(self
, control
):
208 data
= chr(ord(self
.hdrdata
[0]) | control
) + self
.hdrdata
[1:]
209 if control
== Header
.SEPARATOR
: return self
._appendExtendedTimestamp
(data
)
211 data
+= struct
.pack('>I', self
.time
& 0xFFFFFF)[1:] # add time
212 if control
== Header
.TIME
: return self
._appendExtendedTimestamp
(data
)
214 data
+= struct
.pack('>I', self
.size
)[1:] # size
215 data
+= chr(self
.type) # type
216 if control
== Header
.MESSAGE
: return self
._appendExtendedTimestamp
(data
)
218 data
+= struct
.pack('<I', self
.streamId
) # add streamId
219 return self
._appendExtendedTimestamp
(data
)
222 return ("<Header channel=%r time=%r size=%r type=%r (0x%02x) streamId=%r>"
223 % (self
.channel
, self
.time
, self
.size
, self
.type, self
.type or 0, self
.streamId
))
225 class Message(object):
226 # message types: RPC3, DATA3,and SHAREDOBJECT3 are used with AMF3
227 RPC
, RPC3
, DATA
, DATA3
, SHAREDOBJ
, SHAREDOBJ3
, AUDIO
, VIDEO
, ACK
, CHUNK_SIZE
= \
228 0x14, 0x11, 0x12, 0x0F, 0x13, 0x10, 0x08, 0x09, 0x03, 0x01
230 def __init__(self
, hdr
=None, data
=''):
231 if hdr
is None: hdr
= Header()
232 self
.header
, self
.data
= hdr
, data
234 # define properties type, streamId and time to access self.header.(property)
235 for p
in ['type', 'streamId', 'time']:
236 exec 'def _g%s(self): return self.header.%s'%(p
, p
)
237 exec 'def _s%s(self, %s): self.header.%s = %s'%(p
, p
, p
, p
)
238 exec '%s = property(fget=_g%s, fset=_s%s)'%(p
, p
, p
)
240 def size(self
): return len(self
.data
)
243 return ("<Message header=%r data=%r>"% (self
.header
, self
.data
))
245 class Protocol(object):
248 DEFAULT_CHUNK_SIZE
= 128
250 PROTOCOL_CHANNEL_ID
= 2
252 def __init__(self
, sock
):
253 self
.stream
= SockStream(sock
)
254 self
.lastReadHeaders
= dict() # indexed by channelId
255 self
.incompletePackets
= dict() #indexed by channelId
256 self
.readChunkSize
= self
.writeChunkSize
= Protocol
.DEFAULT_CHUNK_SIZE
257 self
.lastWriteHeaders
= dict() # indexed by streamId
258 self
.nextChannelId
= Protocol
.MIN_CHANNEL_ID
259 self
.writeLock
= threading
.Lock()
260 self
.writeQueue
= Queue
.Queue()
262 def messageReceived(self
, msg
):
265 def protocolMessage(self
, msg
):
266 if msg
.type == Message
.ACK
: # respond to ACK requests
268 response
.type, response
.data
= msg
.type, msg
.data
269 self
.writeMessage(response
)
270 elif msg
.type == Message
.CHUNK_SIZE
:
271 self
.readChunkSize
= struct
.unpack('>L', msg
.data
)[0]
273 def connectionClosed(self
):
278 yield self
.parseRequests() # parse http requests
279 except ConnectionClosed
:
280 yield self
.connectionClosed()
281 if _debug
: print 'parse connection closed'
283 def writeMessage(self
, message
):
284 self
.writeQueue
.put(message
)
286 def parseHandshake(self
):
287 '''Parses the rtmp handshake'''
288 data
= (yield self
.stream
.read(Protocol
.PING_SIZE
+ 1)) # bound version and first ping
289 yield self
.stream
.write(data
)
290 data
= (yield self
.stream
.read(Protocol
.PING_SIZE
)) # bound second ping
291 yield self
.stream
.write(data
)
293 def parseRequests(self
):
294 '''Parses complete messages until connection closed. Raises ConnectionLost exception.'''
295 self
.hr
= MultitaskHTTPRequestHandler(self
.stream
, self
.remote
, None)
296 self
.hr
.close_connection
= 1
297 self
.cookies
= CookieJar()
301 # prepare reading the request so that when it's handed
302 # over to "standard" HTTPRequestHandler, the data's already
304 print "parseRequests"
305 raw_requestline
= (yield self
.stream
.readline())
306 if _debug
: print "parseRequests, line", raw_requestline
307 if not raw_requestline
:
308 raise ConnectionClosed
312 line
= (yield self
.stream
.readline())
314 if line
in ['\n', '\r\n']:
316 except StopIteration:
317 if _debug
: print "parseRequests, stopiter"
321 print 'parseRequests', \
322 (traceback
and traceback
.print_exc() or None)
324 raise ConnectionClosed
326 self
.hr
.raw_requestline
= raw_requestline
327 pos
= self
.hr
.rfile
.tell()
328 #self.hr.rfile.truncate(0)
329 self
.hr
.rfile
.write(data
)
330 print "parseRequests write after"
331 self
.hr
.rfile
.seek(pos
)
332 print "parseRequests seek after"
334 if not self
.hr
.parse_request():
335 raise ConnectionClosed
336 print "parseRequests parse_req after"
337 print "parseRequest headers", repr(self
.hr
.headers
), str(self
.hr
.headers
)
339 yield self
.messageReceived(self
.hr
)
342 print 'messageReceived', \
343 (traceback
and traceback
.print_exc() or None)
346 '''Writes messages to stream'''
348 while self
.writeQueue
.empty(): (yield multitask
.sleep(0.01))
349 data
= self
.writeQueue
.get() # TODO this should be used using multitask.Queue and remove previous wait.
350 if _debug
: print "write to stream", repr(data
)
352 # just in case TCP socket is not closed, close it.
354 print "stream closing"
356 yield self
.stream
.close()
357 print "stream closed"
362 yield self
.stream
.write(data
)
363 except ConnectionClosed
:
364 yield self
.connectionClosed()
366 print traceback
.print_exc()
368 class Command(object):
369 ''' Class for command / data messages'''
370 def __init__(self
, type=Message
.RPC
, name
=None, id=None, cmdData
=None, args
=[]):
371 '''Create a new command with given type, name, id, cmdData and args list.'''
372 self
.type, self
.name
, self
.id, self
.cmdData
, self
.args
= type, name
, id, cmdData
, args
[:]
375 return ("<Command type=%r name=%r id=%r data=%r args=%r>" % (self
.type, self
.name
, self
.id, self
.cmdData
, self
.args
))
377 def setArg(self
, arg
):
378 self
.args
.append(arg
)
380 def getArg(self
, index
):
381 return self
.args
[index
]
384 def fromMessage(cls
, message
):
385 ''' initialize from a parsed RTMP message'''
386 assert (message
.type in [Message
.RPC
, Message
.RPC3
, Message
.DATA
, Message
.DATA3
])
388 length
= len(message
.data
)
389 if length
== 0: raise ValueError('zero length message data')
391 if message
.type == Message
.RPC3
or message
.type == Message
.DATA3
:
392 assert message
.data
[0] == '\x00' # must be 0 in AMD3
393 data
= message
.data
[1:]
397 amfReader
= amf
.AMF0(data
)
400 inst
.name
= amfReader
.read() # first field is command name
403 if message
.type == Message
.RPC
:
404 inst
.id = amfReader
.read() # second field *may* be message id
405 inst
.cmdData
= amfReader
.read() # third is command data
408 inst
.args
= [] # others are optional
410 inst
.args
.append(amfReader
.read())
419 output
= amf
.BytesIO()
420 amfWriter
= amf
.AMF0(output
)
421 amfWriter
.write(self
.name
)
422 if msg
.type == Message
.RPC
or msg
.type == Message
.RPC3
:
423 amfWriter
.write(self
.id)
424 amfWriter
.write(self
.cmdData
)
425 for arg
in self
.args
:
428 #hexdump.hexdump(output)
430 if msg
.type == Message
.RPC3
or msg
.type == Message
.DATA3
:
431 data
= '\x00' + output
.read()
438 def getfilename(path
, name
, root
):
439 '''return the file name for the given stream. The name is derived as root/scope/name.flv where scope is
440 the the path present in the path variable.'''
441 ignore
, ignore
, scope
= path
.partition('/')
442 if scope
: scope
= scope
+ '/'
443 result
= root
+ scope
+ name
+ '.flv'
444 if _debug
: print 'filename=', result
447 class Stream(object):
448 '''The stream object that is used for RTMP stream.'''
450 def __init__(self
, client
):
451 self
.client
, self
.id, self
.name
= client
, 0, ''
452 self
.recordfile
= self
.playfile
= None # so that it doesn't complain about missing attribute
453 self
.queue
= multitask
.Queue()
454 self
._name
= 'Stream[' + str(Stream
.count
) + ']'; Stream
.count
+= 1
455 if _debug
: print self
, 'created'
458 if _debug
: print self
, 'closing'
459 if self
.recordfile
is not None: self
.recordfile
.close(); self
.recordfile
= None
460 if self
.playfile
is not None: self
.playfile
.close(); self
.playfile
= None
461 self
.client
= None # to clear the reference
468 '''Generator to receive new Message on this stream, or None if stream is closed.'''
469 return self
.queue
.get()
472 '''Method to send a Message or Command on this stream.'''
473 if isinstance(msg
, Command
):
474 msg
= msg
.toMessage()
475 msg
.streamId
= self
.id
476 # if _debug: print self,'send'
477 if self
.client
is not None: self
.client
.writeMessage(msg
)
479 class Client(Protocol
):
480 '''The client object represents a single connected client to the server.'''
481 def __init__(self
, sock
, server
, remote
):
482 Protocol
.__init
__(self
, sock
)
488 self
._nextStreamId
= 1
489 self
.objectEncoding
= 0.0
490 self
.queue
= multitask
.Queue() # receive queue used by application
491 multitask
.add(self
.parse())
492 multitask
.add(self
.write())
495 '''Generator to receive new Message (msg, arg) on this stream, or (None,None) if stream is closed.'''
496 return self
.queue
.get()
498 def connectionClosed(self
):
499 '''Called when the client drops the connection'''
500 if _debug
: 'Client.connectionClosed'
501 self
.writeMessage(None)
502 yield self
.queue
.put((None,None))
504 def messageReceived(self
, msg
):
505 if _debug
: print 'messageReceived cmd=', msg
.command
, msg
.path
506 msg
.response_cookies
= process_cookies(msg
.headers
, self
.remote
)
508 # slightly bad, this: read everything, put it into memory...
509 if msg
.headers
.has_key('content-length'):
510 max_chunk_size
= 10*1024*1024
511 size_remaining
= int(msg
.headers
["content-length"])
513 while size_remaining
:
514 chunk_size
= min(size_remaining
, max_chunk_size
)
515 data
= (yield self
.stream
.read(chunk_size
))
517 size_remaining
-= len(L
[-1])
519 pos
= msg
.rfile
.tell()
520 msg
.rfile
.write(''.join(L
))
523 yield self
.server
.queue
.put((self
, msg
)) # new connection
526 '''Method to accept an incoming client.'''
528 response
.id, response
.name
, response
.type = 1, '_result', Message
.RPC
529 if _debug
: print 'Client.accept() objectEncoding=', self
.objectEncoding
530 response
.setArg(dict(level
='status', code
='NetConnection.Connect.Success',
531 description
='Connection succeeded.', details
=None,
532 objectEncoding
=self
.objectEncoding
))
533 self
.writeMessage(response
.toMessage())
535 def rejectConnection(self
, reason
=''):
536 '''Method to reject an incoming client.'''
538 response
.id, response
.name
, response
.type = 1, '_error', Message
.RPC
539 response
.setArg(dict(level
='status', code
='NetConnection.Connect.Rejected',
540 description
=reason
, details
=None))
541 self
.writeMessage(response
.toMessage())
543 def call(self
, method
, *args
):
544 '''Call a (callback) method on the client.'''
546 cmd
.id, cmd
.name
, cmd
.type = self
._nextCallId
, method
, (self
.objectEncoding
== 0.0 and Message
.RPC
or Message
.RPC3
)
547 cmd
.args
, cmd
.cmdData
= args
, None
548 self
._nextCallId
+= 1
549 if _debug
: print 'Client.call method=', method
, 'args=', args
, ' msg=', cmd
.toMessage()
550 self
.writeMessage(cmd
.toMessage())
552 def createStream(self
):
553 ''' Create a stream on the server side'''
554 stream
= Stream(self
)
555 stream
.id = self
._nextStreamId
556 self
.streams
[stream
.id] = stream
557 self
._nextStreamId
+= 1
561 class Server(object):
562 '''A RTMP server listens for incoming connections and informs the app.'''
563 def __init__(self
, sock
):
564 '''Create an RTMP server on the given bound TCP socket. The server will terminate
565 when the socket is disconnected, or some other error occurs in listening.'''
567 self
.queue
= multitask
.Queue() # queue to receive incoming client connections
568 multitask
.add(self
.run())
571 '''Generator to wait for incoming client connections on this server and return
572 (client, args) or (None, None) if the socket is closed or some error.'''
573 return self
.queue
.get()
578 sock
, remote
= (yield multitask
.accept(self
.sock
)) # receive client TCP
580 if _debug
: print 'rtmp.Server accept(sock) returned None.'
582 if _debug
: print 'connection received from', remote
583 sock
.setsockopt(socket
.IPPROTO_TCP
, socket
.TCP_NODELAY
, 1) # make it non-block
584 client
= Client(sock
, self
, remote
)
586 if _debug
: print 'rtmp.Server exception ', (sys
and sys
.exc_info() or None)
587 traceback
.print_exc()
590 try: self
.sock
.close(); self
.sock
= None
593 yield self
.queue
.put((None, None))
596 class BaseApp(object):
597 '''An application instance containing any number of streams. Except for constructor all methods are generators.'''
600 self
.name
= str(self
.__class
__.__name
__) + '[' + str(App
.count
) + ']'; App
.count
+= 1
601 self
.players
, self
.publishers
, self
._clients
= {}, {}, [] # Streams indexed by stream name, and list of clients
602 if _debug
: print self
.name
, 'created'
604 if _debug
: print self
.name
, 'destroyed'
607 '''everytime this property is accessed it returns a new list of clients connected to this instance.'''
608 return self
._clients
[1:] if self
._clients
is not None else []
610 class App(BaseApp
, SimpleAppHTTPRequestHandler
):
613 class HTTPServer(object):
614 '''A RTMP server to record and stream Flash video.'''
616 '''Construct a new HttpServer. It initializes the local members.'''
617 self
.sock
= self
.server
= None
618 self
.apps
= dict({'*': App
}) # supported applications: * means any as in {'*': App}
619 self
.clients
= dict() # list of clients indexed by scope. First item in list is app instance.
622 def start(self
, host
='0.0.0.0', port
=8080):
623 '''This should be used to start listening for RTMP connections on the given port, which defaults to 8080.'''
624 if _debug
: print 'start', host
, port
626 sock
= self
.sock
= socket
.socket(type=socket
.SOCK_STREAM
)
627 sock
.setsockopt(socket
.SOL_SOCKET
, socket
.SO_REUSEADDR
, 1)
628 sock
.bind((host
, port
))
629 if _debug
: print 'listening on ', sock
.getsockname()
631 server
= self
.server
= Server(sock
) # start rtmp server on that socket
632 multitask
.add(self
.serverlistener())
635 if _debug
: print 'stopping Flash server'
636 if self
.server
and self
.sock
:
637 try: self
.sock
.close(); self
.sock
= None
641 def serverlistener(self
):
642 '''Server listener (generator). It accepts all connections and invokes client listener'''
644 while True: # main loop to receive new connections on the server
645 client
, msg
= (yield self
.server
.recv()) # receive an incoming client connection.
646 # TODO: we should reject non-localhost client connections.
647 if not client
: # if the server aborted abnormally,
648 break # hence close the listener.
649 if _debug
: print 'client connection received', client
, msg
650 # if client.objectEncoding != 0 and client.objectEncoding != 3:
651 if client
.objectEncoding
!= 0:
652 yield client
.rejectConnection(reason
='Unsupported encoding ' + str(client
.objectEncoding
) + '. Please use NetConnection.defaultObjectEncoding=ObjectEncoding.AMF0')
653 yield client
.connectionClosed()
655 print "cookies", str(msg
.response_cookies
)
656 session
= msg
.response_cookies
['session'].value
658 print "serverlistener", name
659 if '*' not in self
.apps
and name
not in self
.apps
:
660 yield client
.rejectConnection(reason
='Application not found: ' + name
)
661 else: # create application instance as needed and add in our list
662 if _debug
: print 'name=', name
, 'name in apps', str(name
in self
.apps
)
663 app
= self
.apps
[name
] if name
in self
.apps
else self
.apps
['*'] # application class
664 print "clients", self
.clients
.keys()
665 if session
in self
.clients
:
666 inst
= self
.clients
[session
][0]
669 msg
.server
= inst
# whew! just in time!
671 methodname
= "on%s" % msg
.command
672 print methodname
, dir(inst
)
673 method
= getattr(inst
, methodname
, None)
674 result
= method(client
, msg
)
675 close_connection
= msg
.close_connection
676 print "close connection", close_connection
678 if _debug
: traceback
.print_exc()
679 yield client
.rejectConnection(reason
='Exception on %s' % methodname
)
681 if result
is True or result
is None:
682 if session
not in self
.clients
:
683 self
.clients
[session
] = [inst
]; inst
._clients
=self
.clients
[session
]
686 data
= msg
.wfile
.read()
689 yield client
.writeMessage(data
)
692 print 'close_connection requested'
694 yield client
.connectionClosed()
695 except ConnectionClosed
:
697 print 'close_connection done'
700 print "result", result
701 yield client
.rejectConnection(reason
='Rejected in onConnect')
702 except StopIteration: raise
705 print 'serverlistener exception', \
706 (sys
and sys
.exc_info() or None)
707 traceback
.print_exc()
709 def clientlistener(self
, client
):
710 '''Client listener (generator). It receives a command and invokes client handler, or receives a new stream and invokes streamlistener.'''
713 msg
, arg
= (yield client
.recv()) # receive new message from client
714 if not msg
: # if the client disconnected,
715 if _debug
: print 'connection closed from client'
716 break # come out of listening loop.
717 if msg
== 'command': # handle a new command
718 multitask
.add(self
.clienthandler(client
, arg
))
719 elif msg
== 'stream': # a new stream is created, handle the stream.
721 multitask
.add(self
.streamlistener(arg
))
722 except StopIteration: raise
724 if _debug
: print 'clientlistener exception', (sys
and sys
.exc_info() or None)
725 traceback
.print_exc()
727 # client is disconnected, clear our state for application instance.
728 if _debug
: print 'cleaning up client', client
.path
730 if client
.path
in self
.clients
:
731 inst
= self
.clients
[client
.path
][0]
732 self
.clients
[client
.path
].remove(client
)
733 for stream
in client
.streams
.values(): # for all streams of this client
734 self
.closehandler(stream
)
735 client
.streams
.clear() # and clear the collection of streams
736 if client
.path
in self
.clients
and len(self
.clients
[client
.path
]) == 1: # no more clients left, delete the instance.
737 if _debug
: print 'removing the application instance'
738 inst
= self
.clients
[client
.path
][0]
740 del self
.clients
[client
.path
]
741 if inst
is not None: inst
.onDisconnect(client
)
743 def closehandler(self
, stream
):
744 '''A stream is closed explicitly when a closeStream command is received from given client.'''
745 if stream
.client
is not None:
746 inst
= self
.clients
[stream
.client
.path
][0]
747 if stream
.name
in inst
.publishers
and inst
.publishers
[stream
.name
] == stream
: # clear the published stream
748 inst
.onClose(stream
.client
, stream
)
749 del inst
.publishers
[stream
.name
]
750 if stream
.name
in inst
.players
and stream
in inst
.players
[stream
.name
]:
751 inst
.onStop(stream
.client
, stream
)
752 inst
.players
[stream
.name
].remove(stream
)
753 if len(inst
.players
[stream
.name
]) == 0:
754 del inst
.players
[stream
.name
]
757 def clienthandler(self
, client
, cmd
):
758 '''A generator to handle a single command on the client.'''
759 inst
= self
.clients
[client
.path
][0]
761 if cmd
.name
== '_error':
762 if hasattr(inst
, 'onStatus'):
763 result
= inst
.onStatus(client
, cmd
.args
[0])
764 elif cmd
.name
== '_result':
765 if hasattr(inst
, 'onResult'):
766 result
= inst
.onResult(client
, cmd
.args
[0])
768 res
, code
, args
= Command(), '_result', dict()
769 try: result
= inst
.onCommand(client
, cmd
.name
, *cmd
.args
)
771 if _debug
: print 'Client.call exception', (sys
and sys
.exc_info() or None)
772 code
, args
= '_error', dict()
773 res
.id, res
.name
, res
.type = cmd
.id, code
, (client
.objectEncoding
== 0.0 and Message
.RPC
or Message
.RPC3
)
774 res
.args
, res
.cmdData
= args
, None
775 if _debug
: print 'Client.call method=', code
, 'args=', args
, ' msg=', res
.toMessage()
776 client
.writeMessage(res
.toMessage())
777 # TODO return result to caller
780 def streamlistener(self
, stream
):
781 '''Stream listener (generator). It receives stream message and invokes streamhandler.'''
782 stream
.recordfile
= None # so that it doesn't complain about missing attribute
784 msg
= (yield stream
.recv())
786 if _debug
: print 'stream closed'
787 self
.closehandler(stream
)
790 multitask
.add(self
.streamhandler(stream
, msg
))
792 def streamhandler(self
, stream
, message
):
793 '''A generator to handle a single message on the stream.'''
795 if message
.type == Message
.RPC
:
796 cmd
= Command
.fromMessage(message
)
797 if _debug
: print 'streamhandler received cmd=', cmd
798 if cmd
.name
== 'publish':
799 yield self
.publishhandler(stream
, cmd
)
800 elif cmd
.name
== 'play':
801 yield self
.playhandler(stream
, cmd
)
802 elif cmd
.name
== 'closeStream':
803 self
.closehandler(stream
)
804 else: # audio or video message
805 yield self
.mediahandler(stream
, message
)
806 except GeneratorExit
: pass
807 except StopIteration: pass
809 if _debug
: print 'exception in streamhandler', (sys
and sys
.exc_info())
811 def publishhandler(self
, stream
, cmd
):
812 '''A new stream is published. Store the information in the application instance.'''
814 stream
.mode
= 'live' if len(cmd
.args
) < 2 else cmd
.args
[1] # live, record, append
815 stream
.name
= cmd
.args
[0]
816 if _debug
: print 'publishing stream=', stream
.name
, 'mode=', stream
.mode
817 inst
= self
.clients
[stream
.client
.path
][0]
818 if (stream
.name
in inst
.publishers
):
819 raise ValueError, 'Stream name already in use'
820 inst
.publishers
[stream
.name
] = stream
# store the client for publisher
821 result
= inst
.onPublish(stream
.client
, stream
)
823 if stream
.mode
== 'record' or stream
.mode
== 'append':
824 stream
.recordfile
= FLV().open(getfilename(stream
.client
.path
, stream
.name
, self
.root
), stream
.mode
)
825 response
= Command(name
='onStatus', id=cmd
.id, args
=[dict(level
='status', code
='NetStream.Publish.Start', description
='', details
=None)])
826 yield stream
.send(response
)
827 except ValueError, E
: # some error occurred. inform the app.
828 if _debug
: print 'error in publishing stream', str(E
)
829 response
= Command(name
='onStatus', id=cmd
.id, args
=[dict(level
='error',code
='NetStream.Publish.BadName',description
=str(E
),details
=None)])
830 yield stream
.send(response
)
832 def playhandler(self
, stream
, cmd
):
833 '''A new stream is being played. Just updated the players list with this stream.'''
834 inst
= self
.clients
[stream
.client
.path
][0]
835 name
= stream
.name
= cmd
.args
[0] # store the stream's name
836 start
= cmd
.args
[1] if len(cmd
.args
) >= 2 else -2
837 if name
not in inst
.players
:
838 inst
.players
[name
] = [] # initialize the players for this stream name
839 if stream
not in inst
.players
[name
]: # store the stream as players of this name
840 inst
.players
[name
].append(stream
)
841 path
= getfilename(stream
.client
.path
, stream
.name
, self
.root
)
842 if os
.path
.exists(path
):
843 stream
.playfile
= FLV().open(path
)
844 multitask
.add(stream
.playfile
.reader(stream
))
845 if _debug
: print 'playing stream=', name
, 'start=', start
846 result
= inst
.onPlay(stream
.client
, stream
)
847 response
= Command(name
='onStatus', id=cmd
.id, args
=[dict(level
='status',code
='NetStream.Play.Start', description
=stream
.name
, details
=None)])
848 yield stream
.send(response
)
850 def mediahandler(self
, stream
, message
):
851 '''Handle incoming media on the stream, by sending to other stream in this application instance.'''
852 if stream
.client
is not None:
853 inst
= self
.clients
[stream
.client
.path
][0]
854 result
= inst
.onPublishData(stream
.client
, stream
, message
)
856 client
= stream
.client
857 for s
in (inst
.players
.get(stream
.name
, [])):
858 # if _debug: print 'D', stream.name, s.name
859 result
= inst
.onPlayData(s
.client
, s
, message
)
861 yield s
.send(message
)
862 if stream
.recordfile
is not None:
863 stream
.recordfile
.write(message
)
865 # The main routine to start, run and stop the service
866 if __name__
== '__main__':
868 from optparse
import OptionParser
869 parser
= OptionParser()
870 parser
.add_option('-i', '--host', dest
='host', default
='0.0.0.0', help="listening IP address. Default '0.0.0.0'")
871 parser
.add_option('-p', '--port', dest
='port', default
=8080, type="int", help='listening port number. Default 8080')
872 parser
.add_option('-r', '--root', dest
='root', default
='./', help="document root directory. Default './'")
873 parser
.add_option('-d', '--verbose', dest
='verbose', default
=False, action
='store_true', help='enable debug trace')
874 (options
, args
) = parser
.parse_args()
876 _debug
= options
.verbose
878 if _debug
: print time
.asctime(), 'Options - %s:%d' % (options
.host
, options
.port
)
880 agent
.root
= options
.root
881 agent
.start(options
.host
, options
.port
)
882 if _debug
: print time
.asctime(), 'Flash Server Starts - %s:%d' % (options
.host
, options
.port
)
884 except KeyboardInterrupt:
885 print traceback
.print_exc()
889 print traceback
.print_exc()
890 if _debug
: print time
.asctime(), 'Flash Server Stops'