Merge branch 'dev' into 'master'

Dev

See merge request server/openapi/openapi-python-sdk!113

Author: tigersk

CHANGELOG.md

@@ -1,3 +1,10 @@
+## 2.1.2 (2022-06-14)
+### Modify
+- 升级 stomp.py 版本, 将之前的 4.1.24 升级到 8.0.1
+### Breaking
+- PushClient 去除 `auto_reconnect` 参数，如需自定义重连方式，可自定义方法并绑定 `disconnect_callback` 进行重连
+- 处理连接被踢的情况，如果有多台设备使用相同 tiger_id 连接, 新连接会将较早的连接踢掉，较早连接会抛出异常，停止接收消息
+
 ## 2.1.1 (2022-05-25)
 ### New
 - 新增批量分页获取k线接口

requirements.txt

@@ -1,9 +1,10 @@
-simplejson==3.17.3
+simplejson==3.17.6
 delorean==1.0.0
 pandas
 python-dateutil==2.8.2
-pytz==2021.1
+pytz==2022.1
 pyasn1==0.4.8
-rsa==4.7.2
-stomp.py==4.1.24
-getmac==0.8.2
+rsa==4.8
+stomp.py==8.0.1
+getmac==0.8.3
+cryptography==37.0.2

tigeropen/__init__.py

@@ -4,4 +4,4 @@
 
 @author: gaoan
 """
-__VERSION__ = '2.1.1'
+__VERSION__ = '2.1.2'

tigeropen/examples/push_client_demo.py

@@ -120,12 +120,12 @@ def unsubscribe_callback(destination, content):
     print('subscribe:{}, callback content:{}'.format(destination, content))
 
 
-def error_callback(content):
+def error_callback(frame):
     """错误回调"""
-    print(content)
+    print(frame)
 
 
-def connect_callback():
+def connect_callback(frame):
     """连接建立回调"""
     print('connected')
 

tigeropen/push/__init__.py

@@ -3,4 +3,26 @@
 Created on 2018/10/29
 
 @author: gaoan
-"""
+"""
+import ssl
+import time
+
+
+def _patch_ssl(wait=0.01):
+    def new_wrap_socket(self, sock, server_side=False,
+                        do_handshake_on_connect=True,
+                        suppress_ragged_eofs=True,
+                        server_hostname=None, session=None):
+        time.sleep(wait)
+        return self.sslsocket_class._create(
+            sock=sock,
+            server_side=server_side,
+            do_handshake_on_connect=do_handshake_on_connect,
+            suppress_ragged_eofs=suppress_ragged_eofs,
+            server_hostname=server_hostname,
+            context=self,
+            session=session
+        )
+    ssl.SSLContext.old_wrap_socket = ssl.SSLContext.wrap_socket
+    ssl.SSLContext.wrap_socket = new_wrap_socket
+

tigeropen/push/push_client.py

@@ -7,6 +7,7 @@
 import json
 import logging
 import sys
+import os
 from collections import defaultdict
 
 import stomp
@@ -22,10 +23,12 @@
     SUBSCRIPTION_TRADE_ORDER
 from tigeropen.common.consts.push_types import RequestType, ResponseType
 from tigeropen.common.consts.quote_keys import QuoteChangeKey, QuoteKeyType
+from tigeropen.common.exceptions import ApiException
 from tigeropen.common.util.string_utils import camel_to_underline
 from tigeropen.common.util.common_utils import get_enum_value
 from tigeropen.common.util.order_utils import get_order_status
 from tigeropen.common.util.signature_utils import sign_with_rsa
+from tigeropen.push import _patch_ssl
 
 HOUR_TRADING_QUOTE_KEYS_MAPPINGS = {'hourTradingLatestPrice': 'latest_price', 'hourTradingPreClose': 'pre_close',
                                     'hourTradingLatestTime': 'latest_time', 'hourTradingVolume': 'volume',
@@ -67,14 +70,12 @@
 
 
 class PushClient(stomp.ConnectionListener):
-    def __init__(self, host, port, use_ssl=True, connection_timeout=120, auto_reconnect=True,
-                 heartbeats=(30 * 1000, 30 * 1000)):
+    def __init__(self, host, port, use_ssl=True, connection_timeout=120, heartbeats=(30 * 1000, 30 * 1000)):
         """
         :param host:
         :param port:
         :param use_ssl:
-        :param connection_timeout: second
-        :param auto_reconnect:
+        :param connection_timeout: unit: second. The timeout value should be greater the heartbeats interval
         :param heartbeats: tuple of millisecond
         """
         self.host = host
@@ -97,8 +98,8 @@ def __init__(self, host, port, use_ssl=True, connection_timeout=120, auto_reconn
         self.unsubscribe_callback = None
         self.error_callback = None
         self._connection_timeout = connection_timeout
-        self._auto_reconnect = auto_reconnect
         self._heartbeats = heartbeats
+        _patch_ssl()
 
     def _connect(self):
         try:
@@ -108,12 +109,13 @@ def _connect(self):
         except:
             pass
 
-        self._stomp_connection = stomp.Connection12(host_and_ports=[(self.host, self.port), ], use_ssl=self.use_ssl,
+        self._stomp_connection = stomp.Connection12(host_and_ports=[(self.host, self.port)],
                                                     keepalive=KEEPALIVE, timeout=self._connection_timeout,
                                                     heartbeats=self._heartbeats)
         self._stomp_connection.set_listener('push', self)
-        self._stomp_connection.start()
         try:
+            if self.use_ssl:
+                self._stomp_connection.set_ssl([(self.host, self.port)])
             self._stomp_connection.connect(self._tiger_id, self._sign, wait=True, headers=self._generate_headers())
         except ConnectFailedException as e:
             raise e
@@ -128,23 +130,26 @@ def disconnect(self):
         if self._stomp_connection:
             self._stomp_connection.disconnect()
 
-    def on_connected(self, headers, body):
+    def on_connected(self, frame):
         if self.connect_callback:
-            self.connect_callback()
+            self.connect_callback(frame)
 
     def on_disconnected(self):
         if self.disconnect_callback:
             self.disconnect_callback()
-        elif self._auto_reconnect:
-            self._connect()
 
-    def on_message(self, headers, body):
+    def on_message(self, frame):
         """
         Called by the STOMP connection when a MESSAGE frame is received.
 
-        :param dict headers: a dictionary containing all headers sent by the server as key/value pairs.
-        :param body: the frame's payload - the message body.
+        :param Frame frame: the stomp frame. stomp.utils.Frame
+            A STOMP frame's attributes:
+                cmd: the protocol command
+                headers: a map of headers for the frame
+                body: the content of the frame.
         """
+        headers = frame.headers
+        body = frame.body
         try:
             response_type = headers.get('ret-type')
             if response_type == str(ResponseType.GET_SUB_SYMBOLS_END.value):
@@ -255,11 +260,17 @@ def on_message(self, headers, body):
         except Exception as e:
             logging.error(e, exc_info=True)
 
-    def on_error(self, headers, body):
+    def on_error(self, frame):
+        body = json.loads(frame.body)
+        if body.get('code') == 4001:
+            logging.error(body)
+            self.disconnect_callback = None
+            raise ApiException(4001, body.get('message'))
+
         if self.error_callback:
-            self.error_callback(body)
+            self.error_callback(frame)
         else:
-            logging.error(body)
+            logging.error(frame.body)
 
     def _update_subscribe_id(self, destination):
         self._destination_counter_map[destination] += 1