Autobahn with sendMessage using threading and twisted

Question

From this thread (sendMessage from outside in autobahn running in separate thread) I am trying to get the code below working.

But I am getting the following error:

  File "/usr/local/lib/python2.7/dist-packages/autobahn/websocket/protocol.py", line 2421, in sendMessage
    assert(type(payload) == bytes)
exceptions.AssertionError:

Can anybody tell me where I am going wrong?

from autobahn.twisted.websocket import WebSocketServerProtocol, WebSocketServerFactory

import threading
from twisted.python import log
from twisted.internet import reactor
import sys
import time

class MyServerProtocol(WebSocketServerProtocol):

    def onConnect(self, request):
        print("Client connecting: {0}".format(request.peer))

    def onOpen(self):
        print("WebSocket connection open.")

    def onMessage(self, payload, isBinary):
        if isBinary:
            print("Binary message received: {0} bytes".format(len(payload)))
        else:
            print("Text message received: {0}".format(payload.decode('utf8')))

        # echo back message verbatim
        self.sendMessage(payload, isBinary)

    def onClose(self, wasClean, code, reason):
        print("WebSocket connection closed: {0}".format(reason))


class Connection(threading.Thread):
    def __init__(self):
        super(Connection, self).__init__()
        self.factory=WebSocketServerFactory("ws://localhost:9000", debug=False)
    def run(self):
        log.startLogging(sys.stdout)
        self.factory.protocol = MyServerProtocol()
        reactor.listenTCP(9000, self.factory)
        reactor.run(installSignalHandlers=0)

    def send(self, data):

        data = format(data.encode('utf8'))
        protocol = MyServerProtocol()
        reactor.callFromThread(protocol.sendMessage, protocol, data) 


connection = Connection()
connection.daemon = True
connection.start()
print "Running"

for x in range(1, 100):

    connection.send("hello")
    time.sleep(2)

Thanks

EDIT 1:

Full Error:

2015-04-30 07:27:09+0930 [-] Log opened.
2015-04-30 07:27:09+0930 [-] WebSocketServerFactory starting on 9000
2015-04-30 07:27:09+0930 [-] Running
2015-04-30 07:27:09+0930 [-] Starting factory <autobahn.twisted.websocket.WebSocketServerFactory object at 0x7fa97f888a50>
2015-04-30 07:27:09+0930 [-] Unhandled Error
    Traceback (most recent call last):
      File "/usr/lib/python2.7/threading.py", line 810, in __bootstrap_inner
        self.run()
      File "/home/me/Development/eclipse/new_workspace/Engine2/websockets/maybe.py", line 39, in run
        reactor.run(installSignalHandlers=0)
      File "/usr/local/lib/python2.7/dist-packages/twisted/internet/base.py", line 1192, in run
        self.mainLoop()
      File "/usr/local/lib/python2.7/dist-packages/twisted/internet/base.py", line 1201, in mainLoop
        self.runUntilCurrent()
    --- <exception caught here> ---
      File "/usr/local/lib/python2.7/dist-packages/twisted/internet/base.py", line 797, in runUntilCurrent
        f(*a, **kw)
      File "/usr/local/lib/python2.7/dist-packages/autobahn/websocket/protocol.py", line 2421, in sendMessage
        assert(type(payload) == bytes)
    exceptions.AssertionError:

EDIT 2:

Removed threading:

from autobahn.twisted.websocket import WebSocketServerProtocol, WebSocketServerFactory

from twisted.python import log
from twisted.internet import reactor
import sys

class MyServerProtocol(WebSocketServerProtocol):

    def onConnect(self, request):
        print("Client connecting: {0}".format(request.peer))

    def onOpen(self):
        print("WebSocket connection open.")

    def onMessage(self, payload, isBinary):
        if isBinary:
            print("Binary message received: {0} bytes".format(len(payload)))
        else:
            print("Text message received: {0}".format(payload.decode('utf8')))

        # echo back message verbatim
        self.sendMessage(payload, isBinary)

    def onClose(self, wasClean, code, reason):
        print("WebSocket connection closed: {0}".format(reason))

    def send(self, data):

        data = format(data.encode('utf8'))
        protocol = MyServerProtocol()
        reactor.callFromThread(protocol.sendMessage, protocol, data) 

class Test(MyServerProtocol):

    def __init__(self):
        self.factory=WebSocketServerFactory("ws://localhost:9000", debug=False)
        log.startLogging(sys.stdout)
        self.factory.protocol = MyServerProtocol
        reactor.listenTCP(9000, self.factory)
        reactor.run(installSignalHandlers=0)


Test = Test()
Test.send("hello")

print "Running"

score 3 · Answer 1 · edited Oct 01 '15 at 06:42

3

you can't start thread like that in twisted

from autobahn.twisted.websocket import WebSocketServerProtocol, WebSocketServerFactory

from twisted.python import log
from twisted.internet import reactor
import sys

class MyServerProtocol(WebSocketServerProtocol):

    def onConnect(self, request):
        print("Client connecting: {0}".format(request.peer))

    def onOpen(self):
        print("WebSocket connection open.")

    def onMessage(self, payload, isBinary):
        if isBinary:
            print("Binary message received: {0} bytes".format(len(payload)))
        else:
            print("Text message received: {0}".format(payload.decode('utf8')))

        # echo back message verbatim
        self.sendMessage(payload, isBinary)

    def onClose(self, wasClean, code, reason):
        print("WebSocket connection closed: {0}".format(reason))

    def send(self, data):

        data = format(data.encode('utf8'))
        protocol = MyServerProtocol()
        reactor.callFromThread(protocol.sendMessage, protocol, data) 

class Test(MyServerProtocol):

    def __init__(self):
        self.factory=WebSocketServerFactory("ws://localhost:9000", debug=False)
        log.startLogging(sys.stdout)
        self.factory.protocol = MyServerProtocol
        reactor.listenTCP(9000, self.factory)
        reactor.run(installSignalHandlers=0)


Test = Test()


print "Running"

edited Oct 01 '15 at 06:42

coelhudo

4,710
7
38
57

answered Apr 29 '15 at 22:59

rawinput

132
1
7

the reactor must be started in the main thread and to use thread in twisted you have to use twisted threading module but why you want to use thread ? – rawinput Apr 29 '15 at 23:41
This explains what I am trying to do: http://stackoverflow.com/questions/29951718/autobahn-sending-user-specific-and-broadcast-messages-from-external-application – someuser Apr 30 '15 at 00:08
if it's an external application that send messages to twisted so why class Connection(threading.Thread): use the reactor directly in the main thread and time.sleep() can not be applied – rawinput Apr 30 '15 at 00:19
Using your code I made a new version without threading but the process never gets to: Test.send("hello") – someuser Apr 30 '15 at 00:46
you are making a server so you can't send a message to a client that doesn't exist yet so remove the Test.send("hello") – rawinput Apr 30 '15 at 01:05
Yes, I know, but what I cannot figure out is how to connect to the server to broadcast messages.. Do I just connect as a regular client and send, for example, specially formed JSON which triggers the server to respond in a particular manner. – someuser Apr 30 '15 at 01:44
if you want to send a message to another server just use ClientFactory (tcp or udp or http etc ... ) and send your message to the server but if this is the server you have to wait for a connection wich will hold the message so can can trigger something – rawinput Apr 30 '15 at 02:00

Autobahn with sendMessage using threading and twisted

1 Answers1

Linked