2

我已经实现了以下类：

class WampServer(ApplicationSession):

    @inlineCallbacks
    def onJoin(self, details):
        self.connectedBoard = {}
        self.topic_connection = self.config.extra['topic_connection']
        self.topic_command = self.config.extra['topic_command']

        print("Session attached [Connect to WAMP Router]")

        def onMessage(*args):
            #DEBUG Message
            print args

            if args[1] == 'connection':
                self.connectedBoard[args[0]] = args[0]
                print self.connectedBoard

            if args[1] == 'disconnect':
                del self.connectedBoard[args[0]]
                print self.connectedBoard

        try:            
            yield self.subscribe(onMessage, self.topic_connection)
            print ("Subscribed to topic: "+self.topic_connection)

        except Exception as e:
            print("could not subscribe to topic:" +self.topic_connection)

使用另一个类来设置连接的一些配置细节，该类如下所示：

class s4_wamp_server:
    def __init__(self, t_connection, t_command):
        self.topic_connection = t_connection
        self.topic_command = t_command

    def start(self):        
        self.runner = ApplicationRunner(url = urlWampRouter, realm = realmWampRouter, extra={'topic_connection':self.topic_connection, 'topic_command':self.topic_command}) 
        self.runner.run(WampServer)

我有必要定义一种新方法来在特定主题中发布消息，如下所示：

...
...
def pubB(msg, topic):
        yield yelf.publish(topic,msg)
...
...

在使用方法 s4_wamp_serv.stard() 启动“高速公路反应堆”后，我会调用此方法。

- - - - - - - - 编辑 - - - - - - - - - - - - - -

经过一些研究，我已经实现了我需要的 websocket 协议，代码如下：

# ----- twisted ----------
class _WebSocketClientProtocol(WebSocketClientProtocol):
    def __init__(self, factory):
        self.factory = factory

    def onOpen(self):
        #log.debug("Client connected")
        self.factory.protocol_instance = self
        self.factory.base_client._connected_event.set()

class _WebSocketClientFactory(WebSocketClientFactory):
    def __init__(self, *args, **kwargs):
        WebSocketClientFactory.__init__(self, *args, **kwargs)
        self.protocol_instance = None
        self.base_client = None

    def buildProtocol(self, addr):
        return _WebSocketClientProtocol(self)
# ------ end twisted -------
lass BaseWBClient(object):

    def __init__(self, websocket_settings):
        #self.settings = websocket_settings
        # instance to be set by the own factory
        self.factory = None
        # this event will be triggered on onOpen()
        self._connected_event = threading.Event()
        # queue to hold not yet dispatched messages
        self._send_queue = Queue.Queue()
        self._reactor_thread = None

    def connect(self):

        log.msg("Connecting to host:port")
        self.factory = _WebSocketClientFactory(
                                "ws://host:port",
                                debug=True)
        self.factory.base_client = self

        c = connectWS(self.factory)

        self._reactor_thread = threading.Thread(target=reactor.run,
                                               args=(False,))
        self._reactor_thread.daemon = True
        self._reactor_thread.start()

    def send_message(self, body):
        if not self._check_connection():
            return
        log.msg("Queing send")
        self._send_queue.put(body)
        reactor.callFromThread(self._dispatch)

    def _check_connection(self):
        if not self._connected_event.wait(timeout=10):
            log.err("Unable to connect to server")
            self.close()
            return False
        return True

    def _dispatch(self):
        log.msg("Dispatching")
        while True:
            try:
                body = self._send_queue.get(block=False)
            except Queue.Empty:
                break
            self.factory.protocol_instance.sendMessage(body)

    def close(self):
        reactor.callFromThread(reactor.stop)

import time
def Ppippo(coda):
        while True:
            coda.send_message('YOOOOOOOO')
            time.sleep(5)

if __name__ == '__main__':

    ws_setting = {'host':'', 'port':}

    client = BaseWBClient(ws_setting)

    t1 = threading.Thread(client.connect())
    t11 = threading.Thread(Ppippo(client))
    t11.start()
    t1.start()

前面的代码工作正常，但我需要翻译它以在 WAMP 协议 insted websocket 上运行。

有谁知道我如何解决？

4

1 回答 1

0

您可以将您的发布放在 onJoin() 方法的底部，一旦客户端连接到路由器，它将触发，因此在反应器启动之后：

class WampServer(ApplicationSession):

    @inlineCallbacks
    def onJoin(self, details):
        self.connectedBoard = {}
        self.topic_connection = self.config.extra['topic_connection']
        self.topic_command = self.config.extra['topic_command']

        print("Session attached [Connect to WAMP Router]")

        def onMessage(*args):
            #DEBUG Message
            print args

            if args[1] == 'connection':
                self.connectedBoard[args[0]] = args[0]
                print self.connectedBoard

            if args[1] == 'disconnect':
                del self.connectedBoard[args[0]]
                print self.connectedBoard

        try:            
            yield self.subscribe(onMessage, self.topic_connection)
            print ("Subscribed to topic: "+self.topic_connection)

        except Exception as e:
            print("could not subscribe to topic:" +self.topic_connection)

        yield self.pubB(your, args)

    def pubB(msg, topic):
        return self.publish(topic, msg)
于 2015-06-23T13:05:45.527 回答