Can anybody explain, or better yet provide a working example, of how to write a Python client and server where the server can initiate a send via WebSocket at any arbitrary time based on processing logic, and the client can receive at any time? Both have to be freed up to perform other processing activities as their majority purpose.
I've Googled extensively for this, including looking into Tornado, which seems to be the classic Python option and websockets, and asyncio which seem to be the newer way. In either case, every working example I can find is based on the client initiating the communication. Some cover sending from server to client as an echo where the client's logic initiates the series of communication, but I need to get an example working where the server initiates the communication.
Many Tornado examples are timeout based which won't work in my case. The websockets and asyncio examples all are either client to server and/or have a line like somethingOrOther.run() which blocks execution on the client or server or both. Also, many examples have one of either the client or the server written in Python but the other written in something else like JavaScript (I need both in Python).
I don't need the connection to be full-duplex, that is to say, the client never has to send to the server on the same socket, except perhaps for sending back an acknowledgment.
This question seems to be similar, however, the poster's code is not complete, there is no accepted answer, and the one answer provided (not accepted) does not contain complete code, and the client is in JavaScript (I need both in Python).
Here is an example of what I'm trying to do using regular sockets:
server:
# server.py
import socket
import time
import random
# module-level variables
HOST='127.0.0.1'
PORT=65439
ACK_TEXT = 'ack_text'
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind((HOST, PORT))
sock.listen()
conn, addr = sock.accept()
while True:
# In an actual program some significant processing/logic would happen here that would take time
# and also produce a certain result, depending on which we may or may not need to send a message.
# To simulate this, we'll pause for a short and random amount of time, then get a random integer.
time.sleep(random.uniform(0.5, 1.0))
myRandInt = random.randint(0, 10)
# if the ranom integer is even, send a message
if myRandInt % 2 == 0:
message = str(myRandInt)
print('sending message: ' + message)
sendTextViaSocket(conn, message)
# end if
# end while
# end function
def sendTextViaSocket(conn, message):
# encode the message
encodedMessage = bytes(message, 'utf-8')
# send the encoded message
conn.sendall(encodedMessage)
# receive the encoded acknowledgement
encodedAck = conn.recv(1024)
# decode the acknowledgement
ack = encodedAck.decode('utf-8')
# check the received acknowledgement is correct, if not log an error
if ack == ACK_TEXT:
pass
else:
print('error: acknowledgement was received as ' + str(ack))
# end if
# end function
if __name__ == '__main__':
main()
client:
# client
import socket
import select
import time
import random
# module-level variables
HOST='127.0.0.1'
PORT=65439
ACK_TEXT = 'ack_text'
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# if the client is started 1st connect will crash, so continually try to connect
# in a try-catch until the server is available
connectionSuccessful = False
while not connectionSuccessful:
try:
sock.connect((HOST, PORT))
connectionSuccessful = True
except:
pass
# end try
# end while
socks = [ sock ]
# endlessly receive messages
while True:
readySocks, _, _ = select.select(socks, [], [], 0)
for readySock in readySocks:
message = receiveTextViaSocket(readySock)
print('received message = ' + str(message))
# end for
# In an actual program client would have other significant tasks to do
# To simulate this we'll pause for a short but random amount of time
time.sleep(random.uniform(0.5, 1.0))
# end while
# end function
def receiveTextViaSocket(sock):
# receive the encoded message
encodedMessage = sock.recv(1024)
# decode the message
message = encodedMessage.decode('utf-8')
# now send the acknowledgement
# encode the acknowledgement
encodedAck = bytes(ACK_TEXT, 'utf-8')
# send the encoded acknowledgement
sock.sendall(encodedAck)
return message
# end function
if __name__ == '__main__':
main()
How can I do the same thing, but with WebSockets ??
--- Edit ---
After spending most of today on this, this is the best I have done so far:
server:
# server.py
import threading
import asyncio
import websockets
import collections
import random
import time
sendMessageQueue = collections.deque()
def main():
webSocketServer = websockets.serve(sendMessages, 'localhost', 8765)
myLoop = asyncio.get_event_loop()
webSockThread = threading.Thread(target=webSockStart, args=(webSocketServer, myLoop,))
webSockThread.start()
while True:
myRandInt = random.randint(1, 10)
print('appending ' + str(myRandInt))
sendMessageQueue.append(str(myRandInt))
print('sendMessageQueue = ' + str(sendMessageQueue))
time.sleep(random.uniform(1.0, 2.0))
# end while
# end function
def webSockStart(webSocketServer, myLoop):
myLoop.run_until_complete(webSocketServer)
myLoop.run_forever()
# end function
async def sendMessages(websocket, path):
while len(sendMessageQueue) > 0:
await websocket.send(sendMessageQueue.popleft())
# end while
# end function
if __name__ == '__main__':
main()
client:
# client.py
import threading
import asyncio
import websockets
import collections
import random
import time
receiveMessageQueue = collections.deque()
def main():
receiveWebSockThread = threading.Thread(target=receiveWebSockStart)
receiveWebSockThread.start()
while True:
print('doing other stuff')
time.sleep(1.0)
while len(receiveMessageQueue) > 0:
message = receiveMessageQueue.popleft()
print('message = ' + str(message))
# end while
# end while
# end function
def receiveWebSockStart():
loop = asyncio.new_event_loop()
loop.run_until_complete(receiveMessages())
loop.run_forever()
# end function
async def receiveMessages():
while True:
uri = 'ws://localhost:8765'
async with websockets.connect(uri) as webSockConn:
message = await webSockConn.recv()
receiveMessageQueue.append(str(message))
# while True:
# end while
# end with
# end function
if __name__ == '__main__':
main()
server output:
$ python3 server.py
appending 3
sendMessageQueue = deque(['3'])
appending 8
sendMessageQueue = deque(['8'])
appending 8
sendMessageQueue = deque(['8', '8'])
appending 6
sendMessageQueue = deque(['8', '8', '6'])
appending 1
sendMessageQueue = deque(['8', '8', '6', '1'])
client output:
$ python3 client.py
doing other stuff
Exception in thread Thread-1:
Traceback (most recent call last):
File "/usr/lib/python3.6/threading.py", line 916, in _bootstrap_inner
self.run()
File "/usr/lib/python3.6/threading.py", line 864, in run
self._target(*self._args, **self._kwargs)
File "client.py", line 30, in receiveWebSockStart
loop.run_until_complete(receiveMessages())
File "/usr/lib/python3.6/asyncio/base_events.py", line 484, in run_until_complete
return future.result()
File "client.py", line 38, in receiveMessages
message = await webSockConn.recv()
File "/usr/local/lib/python3.6/dist-packages/websockets/protocol.py", line 509, in recv
await self.ensure_open()
File "/usr/local/lib/python3.6/dist-packages/websockets/protocol.py", line 812, in ensure_open
raise self.connection_closed_exc()
websockets.exceptions.ConnectionClosedOK: code = 1000 (OK), no reason
message = 3
doing other stuff
doing other stuff
doing other stuff
doing other stuff
The server seems to be working as intended, however the client is disconnecting after receiving one message and I'm not sure why or how to resolve it. I'm not sure if threads are even the correct approach, or if I should be using run_in_executor from concurrent.futures, but I couldn't get that to work at all. Anybody with specific knowledge on this topic please help.
--- Edit2 ---
Well, at least I found an answer that works (tested with Python 3.6.9 on Ubuntu 18.04), but I'm really not happy with it:
server:
# server.py
import threading
import asyncio
import websockets
import collections
import random
import time
sendMessageQueue = collections.deque()
def main():
# start the WebSocket sending on a separate thread so it doesn't block main
webSockSendThread = threading.Thread(target=sendWebSockStart)
webSockSendThread.start()
while True:
# make up a random integer and append it to the send queue
myRandInt = random.randint(1, 10)
print('appending ' + str(myRandInt))
sendMessageQueue.append(str(myRandInt))
# an actual program would have many other activities to do here, use a random sleep to simulate this
print('doing other stuff')
time.sleep(random.uniform(1.0, 2.0))
# end while
# end function
def sendWebSockStart():
# since we're in a separate thread now, call new_event_loop() (rather than the usual get_event_loop())
# and set the returned loop as the current loop
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# instantiate the WebSocket server, note this also connects it to the sendMessages function
webSocketServer = websockets.serve(sendMessages, 'localhost', 8765)
# run the webSocketServer forever, which runs the sendMessages function forever
loop.run_until_complete(webSocketServer)
loop.run_forever() # note execution of this separate thread stays on this line forever
# end function
async def sendMessages(websocket, path):
while True:
while len(sendMessageQueue) > 0:
await websocket.send(sendMessageQueue.popleft())
# end while
# end while
# end function
if __name__ == '__main__':
main()
client:
# client.py
import threading
import asyncio
import websockets
import collections
import random
import time
receiveMessageQueue = collections.deque()
def main():
# start the WebSocket receiving on a separate thread so it doesn't block main
receiveWebSockThread = threading.Thread(target=receiveWebSockStart)
receiveWebSockThread.start()
while True:
# dequeue and print out all the messages in the receive queue currently
while len(receiveMessageQueue) > 0:
message = receiveMessageQueue.popleft()
print('message = ' + str(message))
# end while
# an actual program would have many other activities to do here, use a sleep to simulate this
print('doing other stuff')
time.sleep(1.0)
# end while
# end function
def receiveWebSockStart():
# since we're in a separate thread now, call new_event_loop() rather than the usual get_event_loop()
loop = asyncio.new_event_loop()
# run receiveMessages() forever
loop.run_until_complete(receiveMessages())
loop.run_forever() # note execution of this separate thread stays on this line forever
# end function
async def receiveMessages():
# set up the connection
uri = 'ws://localhost:8765'
async with websockets.connect(uri) as webSockConn:
# endlessly receive messages and append to the queue when received
while True:
message = await webSockConn.recv()
receiveMessageQueue.append(str(message))
# end while
# end with
# end function
if __name__ == '__main__':
main()
server output:
$ python3 server.py
appending 10
doing other stuff
appending 6
doing other stuff
appending 2
doing other stuff
appending 8
doing other stuff
appending 2
doing other stuff
appending 3
doing other stuff
appending 9
doing other stuff
client output:
$ python3 client.py
doing other stuff
message = 10
message = 6
doing other stuff
doing other stuff
message = 2
doing other stuff
doing other stuff
message = 8
doing other stuff
doing other stuff
message = 2
doing other stuff
doing other stuff
message = 3
doing other stuff
message = 9
doing other stuff
I'm using threading as an intermediary between main and the asyncio websockets function. Can anybody explain or better yet provide a working example of how to re-work this to use run_in_executor to eliminate the intermediary thread ??