This is a work in progress abstraction class to handle freeswitch's eventsocket command lines.
Project description
This is a work in progress abstraction class to handle freeswitch’s eventsocket command lines.
Currently only inbound server is implemented.
Implementation under Tornado’s IOLoop
Here is a simple server implementation under Tornado:
import eventsocket from tornado.tcpserver import TCPServer import tornado.ioloop from tornado.gen import coroutine, Return class TestServer(eventsocket.BaseEventSocket, TCPServer): @coroutine def handle_stream(self, stream, address): self.stream = stream self.on_connect() @coroutine def send_and_receive(self, chunk): yield self.stream.write(chunk) _ = yield self.stream.read_until_regex(b'\n\n') raise Return(self._parse_event(_)) @coroutine def on_connect(self): print 'connected!' data = yield self.connect() print data yield self.my_events() yield self.answer() yield self.playback( '{loops=10}tone_stream://path=${base_dir}/conf/tetris.ttml', lock=True ) if __name__ == '__main__': server = TestServer() server.listen(8888, '0.0.0.0') tornado.ioloop.IOLoop.instance().start()
Implementation under Twisted’s Reactor
Here is a simple server implementation under Twisted:
import eventsocket from twisted.internet import defer, protocol from twisted.protocols import basic from cStringIO import StringIO class TestServer(eventsocket.BaseEventSocket, basic.LineReceiver): delimiter = '\n' def __init__(self, *args, **kwargs): self.stream = None self._io = StringIO() self._queue = [] super(TestServer, self).__init__(*args, **kwargs) def lineReceived(self, line): if len(line) > 0: self._io.write('%s\n' % line) else: self.dispatch_event() def dispatch_event(self): try: deferred = self._queue.pop(0) except IndexError: pass else: self._io.reset() _ = self._parse_event(self._io.read()) deferred.callback(_) self._io.reset() self._io.truncate() def send_and_receive(self, chunk): deferred = defer.Deferred() self._queue.append(deferred) self.transport.write(chunk) return deferred @defer.inlineCallbacks def connectionMade(self): print 'connected!' data = yield self.connect() print data yield self.my_events() yield self.answer() yield self.playback( '{loops=10}tone_stream://path=${base_dir}/conf/tetris.ttml', lock=True ) class PubFactory(protocol.Factory): def buildProtocol(self, addr): return TestServer() if __name__ == '__main__': from twisted.internet import reactor reactor.listenTCP(8888, PubFactory()) reactor.run()
Freeswitch EventSocket is available under the Apache License, Version 2.0.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
No source distribution files available for this release.See tutorial on generating distribution archives.
Built Distribution
Close
Hashes for freeswitch_eventsocket-0.5.dev-py2-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4d7aeadbae5f7df20b2a3aafe51f1d33be45f48576c2a37531a8fdd167b522c2 |
|
MD5 | 841296c83f987cccc1895ae584713f5d |
|
BLAKE2b-256 | 2404d87b93c6aa43e4bf16ab56417342feba079c77dcb5c266ca56c6943ef090 |