Skip to content

Instantly share code, notes, and snippets.

@ahmedfgad
Created June 30, 2020 18:28
Show Gist options
  • Save ahmedfgad/aa0cc7945c1dc4312fa9e49659e5851e to your computer and use it in GitHub Desktop.
Save ahmedfgad/aa0cc7945c1dc4312fa9e49659e5851e to your computer and use it in GitHub Desktop.
import threading
class ListenThread(threading.Thread):
def __init__(self, kivy_app):
threading.Thread.__init__(self)
self.kivy_app = kivy_app
def run(self):
while True:
try:
connection, client_info = self.kivy_app.soc.accept()
self.kivy_app.label.text = "New Connection from {client_info}".format(client_info=client_info)
socket_thread = SocketThread(connection=connection,
client_info=client_info,
kivy_app=self.kivy_app,
buffer_size=1024,
recv_timeout=10)
socket_thread.start()
except BaseException as e:
self.kivy_app.soc.close()
print(e)
self.kivy_app.label.text = "Socket is No Longer Accepting Connections"
self.kivy_app.create_socket_btn.disabled = False
self.kivy_app.close_socket_btn.disabled = True
break
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment