标签: autobahn

我正在遵循github代码中的基本wamp pubsub示例:

此示例从类中发布消息:

class Component(ApplicationSession):
   """
An application component that publishes an event every second.
"""

   def __init__(self, realm = "realm1"):
      ApplicationSession.__init__(self)
      self._realm = realm


   def onConnect(self):
      self.join(self._realm)


   @inlineCallbacks
   def onJoin(self, details):
      counter = 0
      while True:
         self.publish('com.myapp.topic1', counter)
         counter += 1
         yield sleep(1)

我想创建一个引用,以便我可以从代码中的其他地方通过此连接发布消息,即 myobject.myconnection.publish('com.myapp.topic1', 'My message')

从这个类似的问题来看,答案似乎是在连接时,我需要设置类似的东西self.factory.myconnection = self.我已经尝试了多次这种排列而没有成功.

出厂设置部分如下:

   ## create a WAMP application session factory
   ##
   from autobahn.twisted.wamp import ApplicationSessionFactory
   session_factory = ApplicationSessionFactory()


   ## .. and set the session class …

我想sendMessage从MyServerProtocol类外部调用方法并向连接的客户端发送消息.我习惯threading这样做.

当我使用这段代码时:

from autobahn.twisted.websocket import WebSocketServerProtocol, WebSocketServerFactory
from twisted.internet import reactor
import threading

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')))

        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__()

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

    def send(self, data): …

我有一个使用Twisted的Autobahn Python客户端应用程序,它连接到Crossbar.io服务器.使用ReconnectingClientFactory丢失网络连接后,客户端应用程序可以成功重新连接.客户端在连接时注册被叫方名称,以便其他应用程序可以调用它.这始终适用于初始连接.

但是,从丢失的连接恢复时,无法重新注册被叫方名称,因为仍然从先前丢失的连接注册了被叫方名称.这会导致错误'wamp.error.procedure_already_exists'.由于被叫方名称注册仍与先前丢失的连接相关联,因此我将取消注册旧的被叫方名称.

我能看到的唯一解决方案是在每个连接上生成并注册唯一的新被叫方名称,以避免与先前注册的被叫方名称冲突.

有没有更好或更简单的方法来处理这个？似乎WAMP协议允许使用注册ID从另一个连接取消注册被调用者名称,但是高速公路Python客户端库似乎不允许这样做.

我想在高速公路客户端工具中添加基本的auth标头.我怎样才能做到这一点？

目前的要求

GET / HTTP/1.1  
User-Agent: AutobahnPython/0.5.2  
Host: 10.35.34.172:9000  
Upgrade: WebSocket  
Connection: Upgrade  
Sec-WebSocket-Key: 1U4OeBs60qkmk1V/8voLOw==  
Sec-WebSocket-Version: 8  

要求我需要:

GET / HTTP/1.1  
User-Agent: AutobahnPython/0.5.2  
Host: 10.35.34.172:9000  
Authorization: Basic TXlMb2NhdGlvbkFwcDpNeUxvY2F0aW9uQXBwMTIz  
Upgrade: WebSocket  
Connection: Upgrade  
Sec-WebSocket-Key: 1U4OeBs60qkmk1V/8voLOw==  
Sec-WebSocket-Version: 8  

注意:我不希望高速公路服务器对客户端进行身份验证.
我的方案是高速公路客户端 - >我的服务器 - >高速公路服务器.
我的服务器将负责提取授权标头,然后调用高速公路服务器.

我正在使用高速公路运行websocket服务器来补充我的Django应用程序.有时,我需要将消息从Django发送到websocket服务器,使用websocket-client模块可以正常工作.我很乐意使用WAMP协议,因为RPC/PubSub API看起来很棒,但python客户端代码是作为一个扭曲的协议实现的,我无法弄清楚如何使用它,即函数调用,而不是来自一些外国事件循环.有什么我在文档中没有看到的,或者我的架构应该有所不同？

我在python中编写一个Web套接字服务器.我尝试了下面的方法,包括txws,autobahn和tornado,都有类似的结果.

我似乎有大量的内存消耗和安全的websockets,我无法弄清楚这可能发生在何处或为何.以下是龙卷风的一个例子,但我可以提供高速公路或txws的例子.

import tornado.httpserver
import tornado.websocket
import tornado.ioloop
import tornado.web
import json 

class AuthHandler(tornado.websocket.WebSocketHandler):

    def open(self):
        print 'new connection for auth'

    def on_message(self, message):
        message = json.loads(message)
        client_id = message['client_id']
        if client_id not in app.clients:
            app.clients[client_id] = self
        self.write_message('Agent Recorded')

    def on_close(self):
        print 'auth connection closed'


class MsgHandler(tornado.websocket.WebSocketHandler):

    def open(self):
        print 'new connection for msg'

    def on_message(self, message):
        message = json.loads(message)
        to_client = message['client_id']
        if to_client in app.clients:
            app.clients[to_client].write_message('You got a message')

    def on_close(self):
        print 'msg connection closed'


app = …

我试图在本地主机上运行websocket服务器并使用ngrok将其转发到web.但无法弄清楚如何.这些是来自AutobahnPython git repository https://github.com/tavendo/AutobahnPython的原始代码.

服务器代码:

from autobahn.twisted.websocket import WebSocketServerProtocol, \
                                       WebSocketServerFactory


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))



if __name__ == '__main__':

   import sys

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

   log.startLogging(sys.stdout)

   factory = WebSocketServerFactory("ws://localhost:9000", debug = False)
   factory.protocol = …

我想ReconnectingClientFactory用 asyncio做一个。特别是处理客户端启动时服务器不可用的情况，在这种情况下ReconnectingClientFactory将继续尝试。这是asyncio.events.create_connection不做的事情。

具体来说：

该EchoClient例子就可以了。关键是如何建立连接。

factory = EchoClientFactory('ws://127.0.0.1:5678')
connectWS(factory)

在扭曲版本的情况下ReconnectingClientFactory。

对比

factory = EchoClientFactory(u"ws://127.0.0.1:5678")
factory.protocol = SecureServerClientProtocol

loop = asyncio.get_event_loop()

# coro = loop.create_connection(factory, 'ws_server', 5678)
coro = loop.create_connection(factory, '127.0.0.1', 5678)

loop.run_until_complete(asyncio.wait([
    alive(), coro
]))
loop.run_forever()
loop.close()

或与asycnio版本类似。

问题在于，在 asyncio 版本中，asyncio.events.create_connection如果服务器不可用，连接是通过它建立的。

我该如何调和两者？

非常感谢

我尝试使用python连接到poloniex中的Push API,并按照答案中的说明操作:

如何使用python库连接到poloniex.com websocket api

但是我一直收到这个错误:

2017-06-25T04:07:04断开与peer tcp的连接:104.20.13.48:443 with abort = True:WebSocket打开握手超时(对等没有及时完成打开握手)

谁知道这里发生了什么？我无法从在线文档中找到它.谢谢!

我正在尝试通过以下方式连接到币安服务：

wss://stream.binance.com:9443/ws/bnbbtc@kline_1m

我知道它有效，因为已经尝试过使用在线网络服务检查器，它注册以侦听服务器并毫无问题地接收 100 万支蜡烛。

正如我所见，当我将路径添加到主机时会出现问题。如果我不添加路径“/ws/bnbbtc@kline_1m”，它会连接但会立即出现错误：

WebSocket connection closed: connection was closed uncleanly (WebSocket connection upgrade failed (400 - BadRequest))

这是我正在使用的代码，主要是从示例中提取的：

from autobahn.asyncio.websocket import WebSocketClientProtocol, WebSocketClientFactory

class MyClientProtocol(WebSocketClientProtocol):

    def onConnect(self, response):
        print("Server connected: {0}".format(response.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')))

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


if __name__ == '__main__':

    import asyncio

    factory = WebSocketClientFactory()
    factory.protocol = MyClientProtocol

    loop = asyncio.get_event_loop()
    coro …