Checkpoint: working subscribe/unsubscribe and message transit, through the debug page

api.py

@@ -51,7 +51,7 @@ def SendMessage(google_user, client, args):
   subject = args['subject']
   message = args['message']
 
-  models.Subject.FindOrCreate(subject).SendMessage(message)
+  models.Subject.FindOrCreate(subject).SendMessage(message, client.parent_key())
 
   return {}
 
@@ -96,6 +96,13 @@ def Subscribe(google_user, client, args):
   }
 
 
+def Unsubscribe(google_user, client, args):
+  subject = models.Subject.FindOrCreate(args['subject'])
+  models.Subscription.Remove(subject, client)
+
+  return {}
+
+
 class APIWrapper(webapp2.RequestHandler):
 
   _COMMANDS = {
@@ -103,6 +110,7 @@ class APIWrapper(webapp2.RequestHandler):
       'sendMessage': SendMessage,
       'setValue': SetValue,
       'subscribe': Subscribe,
+      'unsubscribe': Unsubscribe,
   }
 
   @utils.chaos_monkey

lib/models.py

@@ -142,8 +142,8 @@ class Subject(db.Model):
     return query.run(limit=num_messages)
 
   @db.transactional()
-  def SendMessage(self, message):
+  def SendMessage(self, message, sender):
-    obj = Message(parent=self, message=message)
+    obj = Message(parent=self, message=message, sender=sender)
     obj.put()
 
     event = obj.ToEvent()
@@ -166,22 +166,34 @@ class Subscription(db.Model):
         .filter('client =', client)
         .fetch(1))
     if not subscriptions:
-      logging.info('no subscriptions found')
       cls(parent=subject, client=client).put()
     if messages == 0:
       return []
     return [m.ToEvent() for m in subject.RecentMessages(messages)]
 
+  @classmethod
+  @db.transactional()
+  def Remove(cls, subject, client):
+    subscriptions = (
+        cls.all()
+        .ancestor(subject)
+        .filter('client =', client))
+    for subscription in subscriptions:
+      subscription.delete()
+
 
 class Message(db.Model):
   # parent=Subject
 
   created = db.DateTimeProperty(required=True, auto_now_add=True)
   message = db.TextProperty(required=True)
+  sender = db.ReferenceProperty(required=True, reference_class=Profile)
 
   def ToEvent(self):
     return {
       'event_type':   'message',
+      'id':           self.key().id(),
+      'sender':       str(Message.sender.get_value_for_datastore(self)),
       'subject':      self.parent_key().name(),
       'created':      self.created,
       'message':      self.message,

static/cosmopolite.js

@@ -22,6 +22,7 @@ cosmopolite.Client = function(opt_callbacks, opt_urlPrefix, opt_namespace) {
   this.namespace_ = opt_namespace || 'cosmopolite';
 
   this.stateCache_ = {};
+  this.subscriptions_ = {};
 
   var scriptUrls = [
     'https://ajax.googleapis.com/ajax/libs/jquery/2.1.0/jquery.min.js',
@@ -36,6 +37,56 @@ cosmopolite.Client = function(opt_callbacks, opt_urlPrefix, opt_namespace) {
   }, this);
 };
 
+cosmopolite.Client.prototype.setValue = function(key, value, is_public) {
+  this.sendRPC_('setValue', {
+    'key': key,
+    'value': value,
+    'public': is_public,
+  });
+  // Provide immediate feedback without waiting for a round trip.
+  // We'll also get a response from the server, so this should be eventually
+  // consistent.
+  if ('onStateChange' in this.callbacks_) {
+    this.callbacks_['onStateChange'](key, value);
+  }
+};
+
+cosmopolite.Client.prototype.getValue = function(key) {
+  return this.stateCache_[key];
+};
+
+cosmopolite.Client.prototype.subscribe = function(subject, messages) {
+  if (subject in this.subscriptions_) {
+    console.log('Not sending duplication subscription request for subject:', subject);
+    return;
+  }
+  this.subscriptions_[subject] = {
+    'messages': [],
+  };
+  this.sendRPC_('subscribe', {
+    'subject': subject,
+    'messages': messages,
+  });
+};
+
+cosmopolite.Client.prototype.unsubscribe = function(subject) {
+  delete this.subscriptions_[subject];
+  this.sendRPC_('unsubscribe', {
+    'subject': subject,
+  });
+};
+
+cosmopolite.Client.prototype.sendMessage = function(subject, message) {
+  this.sendRPC_('sendMessage', {
+    'subject': subject,
+    'message': message,
+  });
+};
+
+cosmopolite.Client.prototype.getMessages = function(subject) {
+  return this.subscriptions_[subject].messages;
+};
+
 cosmopolite.Client.prototype.onLoad_ = function() {
   if (--this.numScriptsToLoad_ > 0) {
     return;
@@ -151,46 +202,24 @@ cosmopolite.Client.prototype.sendRPCs_ = function(commands, delay) {
     });
 };
 
-cosmopolite.Client.prototype.setValue = function(key, value, is_public) {
-  this.sendRPC_('setValue', {
-    'key': key,
-    'value': value,
-    'public': is_public,
-  })
-  // Provide immediate feedback without waiting for a round trip.
-  // We'll also get a response from the server, so this should be eventually
-  // consistent.
-  if ('onStateChange' in this.callbacks_) {
-    this.callbacks_['onStateChange'](key, value);
-  }
-};
-
-cosmopolite.Client.prototype.getValue = function(key) {
-  return this.stateCache_[key];
-};
-
 cosmopolite.Client.prototype.createChannel_ = function() {
-  this.sendRPCs_([
+  var rpcs = [
     {
       'command':   'createChannel',
       'onSuccess': this.onCreateChannel_,
     },
-      // TODO(flamingcow): Remove debugging below.
+  ];
-      {
+  // TODO(flamingcow): Need to restart from the latest message.
+  for (var subject in this.subscriptions_) {
+    rpcs.push({
       'command': 'subscribe',
       'arguments': {
-          'subject': 'books',
+        'subject':  subject,
-          'messages': 5,
+        'messages': 0,
       }
-      },
+    });
-      {
-        'command': 'sendMessage',
-        'arguments': {
-          'subject': 'books',
-          'message': 'foobar',
   }
-      },
+  this.sendRPCs_(rpcs);
-  ]);
 };
 
 cosmopolite.Client.prototype.onCreateChannel_ = function(data) {
@@ -244,7 +273,7 @@ cosmopolite.Client.prototype.onServerEvent_ = function(e) {
     case 'login':
       if ('onLogin' in this.callbacks_) {
         this.callbacks_['onLogin'](
-            e.google_user,
+            e['google_user'],
             this.urlPrefix_ + '/auth/logout');
       }
       break;
@@ -254,6 +283,24 @@ cosmopolite.Client.prototype.onServerEvent_ = function(e) {
           this.urlPrefix_ + '/auth/login');
       }
       break;
+    case 'message':
+      if ('onMessage' in this.callbacks_) {
+        if (!(e['subject'] in this.subscriptions_)) {
+          console.log('Message from unrecognized subject:', e);
+          break;
+        }
+        var subscription = this.subscriptions_[e['subject']];
+        var duplicate = subscription.messages.some(function(message) {
+          return message['id'] == e.id;
+        });
+        if (duplicate) {
+          console.log('Duplicate message:', e);
+          break;
+        }
+        subscription.messages.push(e);
+        this.callbacks_['onMessage'](e);
+      }
+      break;
     default:
       // Client out of date? Force refresh?
       console.log('Unknown channel event:', e);

static/debug.html

@@ -28,12 +28,33 @@ a {
         <textarea id="value" rows="10" cols="40"></textarea>
       </div>
     </div>
+    <hr>
+    <div>
+      <div>
+        Subject:
+        <input type="text" id="subject">
+        <input type="button" id="subscribe" value="Subscribe">
+      </div>
+      <div>
+        <select id="subscriptions"></select>
+        <input type="button" id="unsubscribe" value="Unsubscribe">
+      </div>
+      <div>
+        <input type="text" id="message">
+        <input type="button" id="send" value="Send">
+      </div>
+      <div id="messages"></div>
     <script>
 var keys = document.getElementById('keys');
 var value = document.getElementById('value');
 var last_set = document.getElementById('last_set');
 var is_public = document.getElementById('public');
 
+var subject = document.getElementById('subject');
+var subscriptions = document.getElementById('subscriptions');
+var messages = document.getElementById('messages');
+var message = document.getElementById('message');
+
 var selectKey = function(new_key) {
   keys.value = new_key;
   keys.dispatchEvent(new CustomEvent('change'));
@@ -48,6 +69,16 @@ var addKey = function(new_key, index) {
   }
 };
 
+var addMessage = function(message) {
+  var messageDiv = document.createElement('div');
+  messageDiv.innerHTML = (
+    (new Date(message['created'] * 1000)).toString() +
+    ' &lt;' + message['sender'] + '&gt; ' +
+    message['message']
+  );
+  messages.appendChild(messageDiv);
+};
+
 window.addEventListener('load', function() {
   var googleUser = document.getElementById('google_user');
 
@@ -74,6 +105,12 @@ window.addEventListener('load', function() {
       // Sorts at the end
       addKey(new_key, keys.options.length);
     },
+    onMessage: function(message) {
+      if (subscriptions.value != message['subject']) {
+        return;
+      }
+      addMessage(message);
+    },
   };
 
   var debug = new cosmopolite.Client(callbacks);
@@ -97,6 +134,33 @@ window.addEventListener('load', function() {
     last_set.innerHTML = (new Date(value_obj.last_set * 1000)).toString();
     is_public.checked = value_obj['public'];
   });
+
+  document.getElementById('subscribe').addEventListener('click', function() {
+    debug.subscribe(subject.value, -1);
+    var option = document.createElement('option');
+    option.text = subject.value;
+    subscriptions.options.add(option);
+    subscriptions.dispatchEvent(new CustomEvent('change'));
+  });
+
+  document.getElementById('unsubscribe').addEventListener('click', function() {
+    debug.unsubscribe(subscriptions.value);
+    subscriptions.options.remove(subscriptions.options.selectedIndex);
+    subscriptions.dispatchEvent(new CustomEvent('change'));
+  });
+
+  document.getElementById('send').addEventListener('click', function() {
+    debug.sendMessage(subscriptions.value, message.value);
+  });
+
+  document.getElementById('subscriptions').addEventListener('change', function() {
+    messages.innerHTML = '';
+    if (!subscriptions.value) {
+      return;
+    }
+    debug.getMessages(subscriptions.value).forEach(addMessage);
+  });
+
 });
     </script>
   </body>