ref #GH-85

rref #5440 p90 #5440
zetaops · Aug 15, 2016 · 1c5c560 · 1c5c560
1 parent 677d597
commit 1c5c560
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 34 deletions.
diff --git a/zengine/messaging/lib.py b/zengine/messaging/lib.py
@@ -63,16 +63,18 @@ def set_password(self, raw_password):
                                               salt_size=10)
 
     def is_online(self, status=None):
-        if not self.key:
+        # if not self.key:
             # FIXME: This should not happen!
-            return
+            # return
         if status is None:
             return ConnectionStatus(self.key).get() or False
         else:
             mq_channel = self._connect_mq()
             for sbs in self.subscriptions.objects.filter():
                 if sbs.channel.typ == 10:
-                    mq_channel.basic_publish(exchange=sbs.channel.code_name,
+                    other_party = self.get_prv_exchange(
+                        sbs.channel.code_name.replace('self.key', '').replace('_', ''))
+                    mq_channel.basic_publish(exchange=other_party,
                                              routing_key='',
                                              body=json.dumps({
                                                  'cmd': 'user_status',
@@ -129,7 +131,11 @@ def full_name(self):
 
     @property
     def prv_exchange(self):
-        return 'prv_%s' % str(self.key).lower()
+        return self.get_prv_exchange(self.key)
+
+    @staticmethod
+    def get_prv_exchange(key):
+        return 'prv_%s' % str(key).lower()
 
     def bind_private_channel(self, sess_id):
         mq_channel = self._connect_mq()

diff --git a/zengine/tornado_server/server.py b/zengine/tornado_server/server.py
@@ -141,29 +141,6 @@ def write_message(self, output):
         self.finish()
         self.flush()
 
-        #     # output = blocking_connection.send_message(h_sess_id, input_data)
-        #     out_obj = json_decode(output)
-        #
-        #     if 'http_headers' in out_obj:
-        #         for k, v in out_obj['http_headers']:
-        #             self.set_header(k, v)
-        #     if 'response' in out_obj:
-        #         output = out_obj['response']
-        #
-        #     self.set_status(int(out_obj.get('code', 200)))
-        # except HTTPError as e:
-        #     log.exception("HTTPError for %s: %s" % (sess_id, input_data))
-        #     output = {'cmd': 'error', 'error': e.message, "code": e.code}
-        #     self.set_status(int(e.code))
-        # except:
-        #     log.exception("HTTPError for %s: %s" % (sess_id, input_data))
-        #     if DEBUG:
-        #         self.set_status(500)
-        #         output = json.dumps({'error': traceback.format_exc()})
-        #     else:
-        #         output = {'cmd': 'error', 'error': "Internal Error", "code": 500}
-        # self.write_message(output)
-
 
 
 

diff --git a/zengine/tornado_server/ws_to_queue.py b/zengine/tornado_server/ws_to_queue.py
@@ -186,7 +186,10 @@ def publish_incoming_message(self, message, sess_id):
     def on_message(self, channel, method, header, body):
         sess_id = method.consumer_tag
         log.debug("WS RPLY for %s: %s" % (sess_id, body))
-        if sess_id in self.websockets:
-            log.info("write msg to client")
-            self.websockets[sess_id].write_message(body)
-        channel.basic_ack(delivery_tag=method.delivery_tag)
+        try:
+            if sess_id in self.websockets:
+                log.info("write msg to client")
+                self.websockets[sess_id].write_message(body)
+            channel.basic_ack(delivery_tag=method.delivery_tag)
+        except RuntimeError:
+            log.exception("CANT WRITE TO HTTP OR WS: \n%s" % body)
diff --git a/zengine/views/auth.py b/zengine/views/auth.py
@@ -54,18 +54,18 @@ class Login(SimpleView):
     does the authentication at ``do`` stage.
     """
 
-    def _user_is_online(self):
-        self.current.user.is_online(True)
+    # def _user_is_online(self):
+    #     self.current.user.is_online(True)
 
     def _do_upgrade(self):
         """ open websocket connection """
         self.current.output['cmd'] = 'upgrade'
         self.current.output['user_id'] = self.current.user_id
         self.terminate_existing_login()
-        self.current.user.is_online(True)
         self.current.user.bind_private_channel(self.current.session.sess_id)
         user_sess = UserSessionID(self.current.user_id)
         user_sess.set(self.current.session.sess_id)
+        self.current.user.is_online(True)
 
     def terminate_existing_login(self):
         existing_sess_id = UserSessionID(self.current.user_id).get()