Additional thread for sending
This commit is contained in:
parent
7a1ec5d0ba
commit
4c12ffc375
@ -21,6 +21,7 @@ sp = pUp(s)
|
|||||||
import threading
|
import threading
|
||||||
import socket
|
import socket
|
||||||
import struct
|
import struct
|
||||||
|
import time
|
||||||
|
|
||||||
addr = ("127.0.0.1",21779)
|
addr = ("127.0.0.1",21779)
|
||||||
|
|
||||||
@ -79,20 +80,48 @@ def addEventHandler(event,func):
|
|||||||
def sendResponse(connection,data):
|
def sendResponse(connection,data):
|
||||||
connection.sendall(len(data).to_bytes(4,"big") + data)
|
connection.sendall(len(data).to_bytes(4,"big") + data)
|
||||||
|
|
||||||
class connectionThread(threading.Thread):
|
class senderThread(threading.Thread):
|
||||||
global threadsLock
|
def __init__(self,connectionThread):
|
||||||
|
threading.Thread.__init__(self)
|
||||||
|
self.lock = threading.Lock()
|
||||||
|
with self.lock:
|
||||||
|
self.connectionThread = connectionThread
|
||||||
|
self.queue = []
|
||||||
|
|
||||||
|
def closeThread(self):
|
||||||
|
with self.lock:
|
||||||
|
self.queue = [["close"]]
|
||||||
|
|
||||||
|
def addToQueue(self,entry):
|
||||||
|
with self.lock:
|
||||||
|
self.queue.append(entry)
|
||||||
|
|
||||||
|
def run(self):
|
||||||
|
while True:
|
||||||
|
time.sleep(0.0333)
|
||||||
|
with self.lock:
|
||||||
|
if len(self.queue) == 0: continue
|
||||||
|
for entry in self.queue:
|
||||||
|
if entry[0] == "close": return
|
||||||
|
entry[0](*entry[1],**entry[2])
|
||||||
|
self.queue = []
|
||||||
|
|
||||||
|
class connectionThread(threading.Thread):
|
||||||
def __init__(self,threadId,connection,address):
|
def __init__(self,threadId,connection,address):
|
||||||
threading.Thread.__init__(self)
|
threading.Thread.__init__(self)
|
||||||
|
self.lock = threading.Lock()
|
||||||
|
with self.lock:
|
||||||
self.threadId = threadId
|
self.threadId = threadId
|
||||||
self.connection = connection
|
self.connection = connection
|
||||||
self.address = address
|
self.address = address
|
||||||
self.closed = False
|
self.closed = False
|
||||||
self.user = False
|
self.user = False
|
||||||
self.lock = threading.Lock()
|
self.senderThread = senderThread(self)
|
||||||
|
self.senderThread.start()
|
||||||
|
|
||||||
def closeThread(self):
|
def closeThread(self):
|
||||||
with self.lock, threadsLock:
|
with self.lock, threadsLock:
|
||||||
|
self.senderThread.closeThread()
|
||||||
try:
|
try:
|
||||||
self.connection.close()
|
self.connection.close()
|
||||||
except:
|
except:
|
||||||
@ -103,8 +132,11 @@ class connectionThread(threading.Thread):
|
|||||||
print("thread closed: " +str(self.threadId)+ " (open: " +str(len(threads))+ ")")
|
print("thread closed: " +str(self.threadId)+ " (open: " +str(len(threads))+ ")")
|
||||||
self.closed = True
|
self.closed = True
|
||||||
|
|
||||||
|
def sendResponse(self,data):
|
||||||
|
with self.lock:
|
||||||
|
self.senderThread.addToQueue([sendResponse,[self.connection,data],{}])
|
||||||
|
|
||||||
def run(self):
|
def run(self):
|
||||||
# inform about connection
|
|
||||||
with self.lock:
|
with self.lock:
|
||||||
print("thread opened: " +", ".join((str(self.threadId),str(self.address))))
|
print("thread opened: " +", ".join((str(self.threadId),str(self.address))))
|
||||||
|
|
||||||
@ -179,6 +211,8 @@ def main():
|
|||||||
global close
|
global close
|
||||||
while True:
|
while True:
|
||||||
connection, address = socketServer.accept()
|
connection, address = socketServer.accept()
|
||||||
|
|
||||||
|
# inform about connection
|
||||||
with threadsLock:
|
with threadsLock:
|
||||||
if close: break
|
if close: break
|
||||||
cancel = triggerEvent("onConnect",connection,address)
|
cancel = triggerEvent("onConnect",connection,address)
|
||||||
|
@ -43,7 +43,7 @@ def textOnRequest(event,self,requestLength):
|
|||||||
|
|
||||||
response = textCommandRun(self,textCommandToList(text))
|
response = textCommandRun(self,textCommandToList(text))
|
||||||
print("response: " +textListToCommand(response))
|
print("response: " +textListToCommand(response))
|
||||||
sendResponse(self.connection,textListToCommand(response).encode("utf-8"))
|
self.sendResponse(textListToCommand(response).encode("utf-8"))
|
||||||
|
|
||||||
self.connection.settimeout(textKeepAliveTimeout)
|
self.connection.settimeout(textKeepAliveTimeout)
|
||||||
addEventHandler("onRequest",textOnRequest)
|
addEventHandler("onRequest",textOnRequest)
|
||||||
@ -52,8 +52,8 @@ global textOnException
|
|||||||
def textOnException(event,self,exc):
|
def textOnException(event,self,exc):
|
||||||
self.connection.settimeout(textTimeout)
|
self.connection.settimeout(textTimeout)
|
||||||
if type(exc) == socket.timeout:
|
if type(exc) == socket.timeout:
|
||||||
sendResponse(self.connection,textListToCommand(["error","fatal","timeout"]).encode("utf-8"))
|
self.sendResponse(textListToCommand(["error","fatal","timeout"]).encode("utf-8"))
|
||||||
return
|
return
|
||||||
|
|
||||||
sendResponse(self.connection,textListToCommand(["error","fatal","unhandled",str(exc)]).encode("utf-8"))
|
self.sendResponse(textListToCommand(["error","fatal","unhandled",str(exc)]).encode("utf-8"))
|
||||||
addEventHandler("onException",textOnException)
|
addEventHandler("onException",textOnException)
|
Loading…
Reference in New Issue
Block a user