upgrade stomp.py

Author: tigersk

requirements.txt

@@ -1,9 +1,9 @@
-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

tigeropen/__init__.py

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

tigeropen/push/__init__.py

@@ -3,4 +3,26 @@
 Created on 2018/10/29
 
 @author: gaoan
-"""
+"""
+import ssl
+import time
+
+
+def _patch_ssl(wait=0.006):
+    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

@@ -26,6 +26,7 @@
 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',
@@ -73,7 +74,7 @@ def __init__(self, host, port, use_ssl=True, connection_timeout=120, auto_reconn
         :param host:
         :param port:
         :param use_ssl:
-        :param connection_timeout: second
+        param connection_timeout: unit: second. The timeout value should be greater the heartbeats interval
         :param auto_reconnect:
         :param heartbeats: tuple of millisecond
         """
@@ -99,6 +100,7 @@ def __init__(self, host, port, use_ssl=True, connection_timeout=120, auto_reconn
         self._connection_timeout = connection_timeout
         self._auto_reconnect = auto_reconnect
         self._heartbeats = heartbeats
+        _patch_ssl()
 
     def _connect(self):
         try:
@@ -108,12 +110,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,7 +131,7 @@ 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()
 
@@ -138,13 +141,18 @@ def on_disconnected(self):
         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 +263,11 @@ 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):
         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