По сути, я скопировал и вставил этот пример кода из плагина locust,
/>https://github.com/SvenskaSpel/locust-p ... etio_ex.py
но он вызывает исключение WebsocketConnectionClosedException при отправке 2 (ping). (нет ответа, когда 2 отправлено)
Кроме того, я создаю соединение ws с реальным сервером, и когда я отлаживаю полученные сообщения, он успешно печатает идентификатор сокета. Однако на стороне сервера журналы подключений не печатаются. Я здесь в замешательстве. Это вообще устанавливает связь? если нет, то как он может создать идентификатор сокета?
Другой вопрос касается прослушивателя событий. Мой первоначальный клиент написан с использованием Socket.io, и я мог прослушивать определенные события с помощью приведенного ниже кода. Поддерживает ли Python такую функцию? Как реализовать прослушиватель событий в Python?
Код: Выделить всё
socket.on("message", (data) => console.log(data))
Код: Выделить всё
class MySocketIOUser(SocketIOUser):
@task
def my_task(self):
self.my_value = None
self.connect("wss:///socket.io/?EIO=4&transport=websocket", [])
# example of subscribe
self.send('42["subscribe",{"url":"/namespace","sendInitialUpdate": true}]')
print('receive: ' + self.ws.recv())
# wait until I get a push message to on_message
while not self.my_value:
time.sleep(0.1)
# wait for additional pushes, while occasionally sending heartbeats, like a real client would
self.sleep_with_heartbeat(10)
def on_message(self, message):
print('message: ' + message)
self.my_value = message
if __name__ == "__main__":
host = ""
Код: Выделить всё
raise WebSocketConnectionClosedException(
websocket._exceptions.WebSocketConnectionClosedException: Connection to remote host was lost.
Любая помощь приветствуется.>
Подробнее здесь: https://stackoverflow.com/questions/703 ... ith-locust
Мобильная версия