testsuite/rtmpy-echo-server.py

   1 from rtmpy import server
   2
   3 from twisted.internet import reactor
   4
   5
   6 class LiveApplication(server.Application):
   7     """
   8     The simplest application possible.
   9     """
  10
  11     def startup(self):
  12         print("Startup")
  13
  14     def onAppStart(self):
  15         print("App start")
  16
  17     def onAppStop(self):
  18         print("App stop")
  19
  20     def onConnect(self, client, **args):
  21         print("Connection")
  22         client.call("initial", [ "connection attempt received", args ])
  23         print(args)
  24         return True
  25
  26     def onConnectAccept(self, client, **args):
  27         print("Connection accepted", client)
  28         client.call("welcome", [ "You have connected!", args ])
  29         return True
  30
  31     def onDownstreamBandwidth(self, interval, bandwidth):
  32         print("Downstream Bandwidth:", interval, bandwidth)
  33         return True
  34
  35     def echo(self, *args, **kw):
  36         print("echo", args, kw)
  37         return args
  38
  39     def onDisconnect(self, client):
  40         # This is probably never actually sent.
  41         client.call("disconnected", "arg1");
  42         return True;
  43
  44
  45 app = LiveApplication()
  46
  47 reactor.listenTCP(9984, server.ServerFactory({
  48     'rtmpyecho': app
  49 }))
  50
  51 reactor.run()