user per peer library (sqlitedict)

oml/nodes.py

@@ -22,9 +22,10 @@ from changelog import Changelog
 from websocket import trigger_event
 from localnodes import LocalNodes
 from tor_request import get_opener
-from utils import user_sort_key
+from utils import user_sort_key, get_peer
 import state
 import db
+import library
 
 import logging
 logger = logging.getLogger(__name__)
@@ -247,24 +248,21 @@ class Node(Thread):
     def pullChanges(self):
         with db.session():
             u = user.models.User.get_or_create(self.user_id)
-            if not self.online or not u.peered:
+            if not u or not self.online or not u.peered:
                 return True
-            last = Changelog.query.filter_by(user_id=self.user_id).order_by('-revision').first()
-            from_revision = last.revision + 1 if last else 0
-            try:
-                changes = self.request('pullChanges', from_revision)
-            except:
-                self.online = False
-                logger.debug('%s went offline', u.name)
-                return False
-            if not changes:
-                return False
-            try:
-                r = Changelog.apply_changes(u, changes, first=from_revision == 0)
-            except:
-                logger.debug('apply_changes failed', exc_info=True)
-                r = False
-            return r
+        peer = get_peer(self.user_id)
+        from_revision = peer.info.get('revision', -1) + 1
+        try:
+            changes = self.request('pullChanges', from_revision)
+        except:
+            self.online = False
+            logger.debug('%s went offline', u.name, ext_info=True)
+            return False
+        if not changes:
+            return False
+        #with open('/tmp/changelog_%s_%s.json' % (self.user_id, from_revision), 'w') as f:
+        #    json.dump(changes, f, ensure_ascii=False, indent=4)
+        return peer.apply_changes(changes)
 
     def peering(self, action):
         with db.session():
@@ -411,13 +409,14 @@ class Nodes(Thread):
                     del u.info['local']
                     u.save()
                 self.queue('add', u.id)
+                state.peers[u.id] = library.Peer(u.id)
             for u in user.models.User.query.filter_by(queued=True):
                 logger.debug('adding queued node... %s', u.id)
                 self.queue('add', u.id, True)
         self._local = LocalNodes()
         self._cleanup = PeriodicCallback(lambda: self.queue('cleanup'), 120000)
         self._cleanup.start()
-        self._pullcb = PeriodicCallback(self.pull, 60000)
+        self._pullcb = PeriodicCallback(self.pull, settings.server['pull_interval'])
         self._pullcb.start()
         Thread.__init__(self)
         self.daemon = True
@@ -475,6 +474,7 @@ class Nodes(Thread):
         if state.activity and state.activity.get('activity') == 'import':
             return
         self._pulling = True
+        library.sync_db()
         users = []
         with db.session():
             from user.models import User
@@ -493,6 +493,7 @@ class Nodes(Thread):
         self._pulling = False
 
     def run(self):
+        self.queue('pull')
         while not state.shutdown:
             args = self._q.get()
             if args:
@@ -519,7 +520,6 @@ def publish_node():
     state.check_nodes.start()
     state._online = PeriodicCallback(update_online, 60000)
     state._online.start()
-    state.nodes.pull()
 
 def update_online():
     online = state.tor and state.tor.is_online()