Revert "e2e-removal-refactor finished. instant messaging working"

This reverts commit 82de442.

zengine/messaging/lib.py

@@ -74,8 +74,7 @@ def encrypt_password(self):
         if self.password and not self.password.startswith('$pbkdf2'):
             self.set_password(self.password)
 
-    def prepare_user_channel(self):
-        """should be called from User.post_creation hook"""
+    def prepare_channels(self):
         from zengine.messaging.model import Channel, Subscriber
         # create private channel of user
         ch, new = Channel.objects.get_or_create(owner=self, typ=5)
@@ -118,13 +117,12 @@ def full_name(self):
     def prv_exchange(self):
         return 'prv_%s' % str(self.key).lower()
 
-    def bind_channels_to_session_queue(self, sess_id):
+    def bind_private_channel(self, sess_id):
         mq_channel = pika.BlockingConnection(BLOCKING_MQ_PARAMS).channel()
         mq_channel.queue_declare(queue=sess_id, arguments={'x-expires': 40000})
         log.debug("Binding private exchange to client queue: Q:%s --> E:%s" % (sess_id,
                                                                                self.prv_exchange))
-        for sbs in self.subscriptions.objects.filter():
-            mq_channel.queue_bind(exchange=sbs.channel.code_name, queue=sess_id)
+        mq_channel.queue_bind(exchange=self.prv_exchange, queue=sess_id)
 
     def send_notification(self, title, message, typ=1, url=None):
         """

zengine/messaging/model.py

@@ -17,9 +17,8 @@
 from pyoko.fields import DATE_TIME_FORMAT
 from pyoko.lib.utils import get_object_from_path
 from zengine.client_queue import BLOCKING_MQ_PARAMS
-from zengine.lib.cache import UserSessionID
 from zengine.lib.utils import to_safe_str
-from zengine.log import log
+
 UserModel = get_object_from_path(settings.USER_MODEL)
 
 
@@ -106,7 +105,7 @@ def add_message(cls, channel_key, body, title=None, sender=None, url=None, typ=2
         msg_object = Message(sender=sender, body=body, msg_title=title, url=url,
                              typ=typ, channel_id=channel_key, receiver=receiver, key=uuid4().hex)
         mq_channel.basic_publish(exchange=channel_key,
-                                 routing_key='#',
+                                 routing_key='',
                                  body=json.dumps(msg_object.serialize()))
         return msg_object.save()
 
@@ -206,29 +205,35 @@ def unread_count(self):
         else:
             self.channel.message_set.objects.filter().count()
 
+    def create_exchange(self):
+        """
+        Creates user's private exchange
+
+        Actually user's private channel needed to be defined only once,
+        and this should be happened when user first created.
+        But since this has a little performance cost,
+        to be safe we always call it before binding to the channel we currently subscribe
+        """
+        channel = self._connect_mq()
+        channel.exchange_declare(exchange='prv_%s' % self.user.key.lower(),
+                                 exchange_type='fanout',
+                                 durable=True)
+
     @classmethod
     def mark_seen(cls, key, datetime_str):
         cls.objects.filter(key=key).update(last_seen=datetime_str)
 
     def bind_to_channel(self):
         """
-        Binds (subscribes) users session queue to channel exchange
+        Binds (subscribes) users private exchange to channel exchange
         Automatically called at creation of subscription record.
-        Same operation done at user login
-        (zengine.messaging.lib.BaseUser#bind_channels_to_session_queue)
         """
-        try:
-            sess_id = UserSessionID(self.user_id).get()
-            mq_channel = self._connect_mq()
-            mq_channel.queue_bind(exchange=self.channel.code_name,
-                                  queue=sess_id)
-        except:
-            log.exception("Cant create subscription  binding for %s : %s" % (self.name,
-                                                                             self.user.full_name))
-
+        if self.channel.code_name != self.user.prv_exchange:
+            channel = self._connect_mq()
+            channel.exchange_bind(source=self.channel.code_name, destination=self.user.prv_exchange)
 
     def post_creation(self):
-        # self.create_exchange()
+        self.create_exchange()
         self.bind_to_channel()
 
         if not self.name:
@@ -324,9 +329,9 @@ def _republish(self):
         mq_channel.basic_publish(exchange=self.channel.key, routing_key='',
                                  body=json.dumps(self.serialize()))
 
-    # def pre_save(self):
-    #     if self.exist:
-    #         self._republish()
+    def pre_save(self):
+        if self.exist:
+            self._republish()
 
 
 ATTACHMENT_TYPES = (

zengine/models/auth.py

@@ -89,7 +89,7 @@ def pre_save(self):
         self.encrypt_password()
 
     def post_creation(self):
-        self.prepare_user_channel()
+        self.prepare_channels()
 
     def get_permissions(self):
         """

zengine/tornado_server/ws_to_queue.py

@@ -118,7 +118,6 @@ def send_message(self, sess_id, input_data):
             {'code': 503, 'error': 'Retry'})
 
 
-
 class QueueManager(object):
     """
     Async RabbitMQ & Tornado websocket connector
@@ -135,7 +134,7 @@ def __init__(self, io_loop=None):
         self.out_channels = {}
         self.out_channel = None
         self.websockets = {}
-
+        # self.connect()
 
     def connect(self):
         """
@@ -188,6 +187,12 @@ def on_input_queue_declare(self, queue):
                                    exchange='input_exc',
                                    queue=self.INPUT_QUEUE_NAME,
                                    routing_key="#")
+    def ask_for_user_id(self, sess_id):
+        log.debug(sess_id)
+        # TODO: add remote ip
+        self.publish_incoming_message(dict(_zops_remote_ip='',
+                                           data={'view': 'sessid_to_userid'}), sess_id)
+
 
     def register_websocket(self, sess_id, ws):
         """
@@ -196,8 +201,15 @@ def register_websocket(self, sess_id, ws):
             sess_id:
             ws:
         """
-        self.websockets[sess_id] = ws
-        self.create_out_channel(sess_id)
+        log.debug("GET SESSUSERS: %s" % sys.sessid_to_userid)
+        try:
+            user_id = sys.sessid_to_userid[sess_id]
+            self.websockets[user_id] = ws
+        except KeyError:
+            self.ask_for_user_id(sess_id)
+            self.websockets[sess_id] = ws
+            user_id = sess_id
+        self.create_out_channel(sess_id, user_id)
 
     def inform_disconnection(self, sess_id):
         self.in_channel.basic_publish(exchange='input_exc',
@@ -208,22 +220,25 @@ def inform_disconnection(self, sess_id):
                                           _zops_remote_ip='')))
 
     def unregister_websocket(self, sess_id):
+        user_id = sys.sessid_to_userid.get(sess_id, None)
         try:
             self.inform_disconnection(sess_id)
-            del self.websockets[sess_id]
-        except:
-            log.exception("WS already deleted")
+            del self.websockets[user_id]
+        except KeyError:
+            log.exception("Non-existent websocket for %s" % user_id)
         if sess_id in self.out_channels:
             try:
                 self.out_channels[sess_id].close()
             except ChannelClosed:
                 log.exception("Pika client (out) channel already closed")
 
-    def create_out_channel(self, sess_id):
+    def create_out_channel(self, sess_id, user_id):
         def _on_output_channel_creation(channel):
             def _on_output_queue_decleration(queue):
-                channel.basic_consume(self.on_message, queue=sess_id, consumer_tag=sess_id)
-                log.debug("BIND QUEUE TO WS Q.%s " % sess_id)
+                channel.basic_consume(self.on_message, queue=sess_id)
+                log.debug("BIND QUEUE TO WS Q.%s on Ch.%s WS.%s" % (sess_id,
+                                                                    channel.consumer_tags[0],
+                                                                    user_id))
             self.out_channels[sess_id] = channel
 
             channel.queue_declare(callback=_on_output_queue_decleration,
@@ -248,9 +263,18 @@ def publish_incoming_message(self, message, sess_id):
                                       body=json_encode(message))
 
     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 %s" % sess_id)
-            self.websockets[sess_id].write_message(body)
+        user_id = method.exchange[4:]
+        log.debug("WS RPLY for %s: %s" % (user_id, body))
+        if user_id in self.websockets:
+            log.info("write msg to client")
+            self.websockets[user_id].write_message(body)
+            # channel.basic_ack(delivery_tag=method.delivery_tag)
+        elif 'sessid_to_userid' in body:
+            reply = json_decode(body)
+            sys.sessid_to_userid[reply['sess_id']] = reply['user_id']
+            self.websockets[reply['user_id']] = self.websockets[reply['sess_id']]
+            del self.websockets[reply['sess_id']]
         channel.basic_ack(delivery_tag=method.delivery_tag)
+
+            # else:
+            #     channel.basic_reject(delivery_tag=method.delivery_tag)

zengine/views/auth.py

@@ -61,8 +61,9 @@ def _do_upgrade(self):
         self.current.output['cmd'] = 'upgrade'
         self.current.output['user_id'] = self.current.user_id
         self.current.user.is_online(True)
-        self.current.user.bind_channels_to_session_queue(self.current.session.sess_id)
-        UserSessionID(self.current.user_id).set(self.current.session.sess_id)
+        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)
 
     def do_view(self):
         """
@@ -81,6 +82,12 @@ def do_view(self):
                 self.current.task_data['login_successful'] = auth_result
                 if auth_result:
                     self._do_upgrade()
+
+                    # old_sess_id = user_sess.get()
+                    # notify = Notify(self.current.user_id)
+                    # notify.cache_to_queue()
+                    # if old_sess_id:
+                    #     notify.old_to_new_queue(old_sess_id)
             except:
                 raise
                 self.current.log.exception("Wrong username or another error occurred")

zengine/views/system.py

@@ -11,7 +11,7 @@
 def sessid_to_userid(current):
     current.output['user_id'] = current.user_id.lower()
     current.output['sess_id'] = current.session.sess_id
-    current.user.bind_channels_to_session_queue(current.session.sess_id)
+    current.user.bind_private_channel(current.session.sess_id)
     current.output['sessid_to_userid'] = True
 
 def mark_offline_user(current):

zengine/wf_daemon.py

@@ -176,10 +176,10 @@ def handle_message(self, ch, method, properties, body):
     def send_output(self, output):
         # TODO: This is ugly, we should separate login process
         log.debug("SEND_OUTPUT: %s" % output)
-        # if self.current.user_id is None or 'login_process' in output:
-        self.client_queue.send_to_default_exchange(self.sessid, output)
-        # else:
-        #     self.client_queue.send_to_prv_exchange(self.current.user_id, output)
+        if self.current.user_id is None or 'login_process' in output:
+            self.client_queue.send_to_default_exchange(self.sessid, output)
+        else:
+            self.client_queue.send_to_prv_exchange(self.current.user_id, output)