与其他框架集成¶
在 1.0.8 版本加入.
在Kivy中使用Twisted¶
备注
您可以使用 kivy.support.install_twisted_reactor 函数来安装一个将在 Kivy 事件循环内运行的 Twisted reactor。
传递给此函数的任何参数或关键字参数都将传递给threadedselect反应器的interleave函数。这些参数通常是传递给Twisted的reactor.startRunning的参数。
警告
与默认的twisted reactor不同，安装的reactor不会处理任何信号，除非你明确将`installSignalHandlers`关键字参数设置为1。这样做是为了让kivy像往常一样处理信号，除非你特别希望twisted reactor来处理信号（例如SIGINT）。
kivy 示例中包含一个 twisted 服务器和客户端的小例子。服务器应用运行一个简单的 twisted 服务器并记录所有消息。客户端应用可以向服务器发送消息，并打印其消息及收到的响应。这些示例主要基于 twisted 文档中的简单 Echo 示例，你可以在以下链接找到：
要尝试该示例，请先运行echo_server_app.py，然后再启动echo_client_app.py。当你在文本框中输入内容并按回车后，服务器将对客户端应用发送的任何内容回复简单的回显消息。
服务端应用¶
# install_twisted_rector must be called before importing and using the reactor
from kivy.support import install_twisted_reactor
install_twisted_reactor()
from twisted.internet import reactor
from twisted.internet import protocol
class EchoServer(protocol.Protocol):
def dataReceived(self, data):
response = self.factory.app.handle_message(data)
if response:
self.transport.write(response)
class EchoServerFactory(protocol.Factory):
protocol = EchoServer
def __init__(self, app):
self.app = app
from kivy.app import App
from kivy.uix.label import Label
class TwistedServerApp(App):
label = None
def build(self):
self.label = Label(text="server started\n")
reactor.listenTCP(8000, EchoServerFactory(self))
return self.label
def handle_message(self, msg):
msg = msg.decode('utf-8')
self.label.text = "received: {}\n".format(msg)
if msg == "ping":
msg = "Pong"
if msg == "plop":
msg = "Kivy Rocks!!!"
self.label.text += "responded: {}\n".format(msg)
return msg.encode('utf-8')
if __name__ == '__main__':
TwistedServerApp().run()
客户端应用¶
# install_twisted_rector must be called before importing the reactor
from __future__ import unicode_literals
from kivy.support import install_twisted_reactor
install_twisted_reactor()
# A Simple Client that send messages to the Echo Server
from twisted.internet import reactor, protocol
class EchoClient(protocol.Protocol):
def connectionMade(self):
self.factory.app.on_connection(self.transport)
def dataReceived(self, data):
self.factory.app.print_message(data.decode('utf-8'))
class EchoClientFactory(protocol.ClientFactory):
protocol = EchoClient
def __init__(self, app):
self.app = app
def startedConnecting(self, connector):
self.app.print_message('Started to connect.')
def clientConnectionLost(self, connector, reason):
self.app.print_message('Lost connection.')
def clientConnectionFailed(self, connector, reason):
self.app.print_message('Connection failed.')
from kivy.app import App
from kivy.uix.label import Label
from kivy.uix.textinput import TextInput
from kivy.uix.boxlayout import BoxLayout
# A simple kivy App, with a textbox to enter messages, and
# a large label to display all the messages received from
# the server
class TwistedClientApp(App):
connection = None
textbox = None
label = None
def build(self):
root = self.setup_gui()
self.connect_to_server()
return root
def setup_gui(self):
self.textbox = TextInput(size_hint_y=.1, multiline=False)
self.textbox.bind(on_text_validate=self.send_message)
self.label = Label(text='connecting...\n')
layout = BoxLayout(orientation='vertical')
layout.add_widget(self.label)
layout.add_widget(self.textbox)
return layout
def connect_to_server(self):
reactor.connectTCP('localhost', 8000, EchoClientFactory(self))
def on_connection(self, connection):
self.print_message("Connected successfully!")
self.connection = connection
def send_message(self, *args):
msg = self.textbox.text
if msg and self.connection:
self.connection.write(msg.encode('utf-8'))
self.textbox.text = ""
def print_message(self, msg):
self.label.text += "{}\n".format(msg)
if __name__ == '__main__':
TwistedClientApp().run()