0x2620/openmedialibrary
1
0
Fork 0
Code Issues 131 Pull requests Releases Wiki Activity

better online/offline status handling

Browse source
This commit is contained in:
j 2015-12-02 22:05:23 +01:00
commit 8a26285c88
11 changed files with 92 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

12
oml/downloads.py
View file

@ -74,12 +74,12 @@ class ScrapeThread(Thread):
return True
logger.debug('scrape %s', s.item)
try:
s.item.scrape()
for f in s.item.files:
f.move()
s.item.update_icons()
s.remove()
trigger_event('change', {})
if s.item.scrape():
for f in s.item.files:
f.move()
s.item.update_icons()
s.remove()
trigger_event('change', {})
scraped = True
except:
logger.debug('scrape failed %s', s.item, exc_info=1)

5
oml/item/models.py
View file

@ -353,7 +353,10 @@ class Item(db.Model):
m['primaryid'] = primaryid
self.meta = m
self.modified = datetime.utcnow()
self.update()
self.update()
return True
return False
return True
def queue_download(self):
u = state.user()

1
oml/localnodes.py
View file

@ -148,6 +148,7 @@ class LocalNodesBase(Thread):
state.nodes.queue('add', u.id)
self.send()
def get_ip(self):
pass

40
oml/nodes.py
View file

@ -38,7 +38,7 @@ class Node(Thread):
_pulling = False
host = None
local = None
online = False
_online = None
download_speed = 0
TIMEOUT = 5
@ -78,11 +78,12 @@ class Node(Thread):
#return Thread.join(self)
def pull(self):
if not self._pulling:
if state.online and not self._pulling:
self._q.put('pull')
def ping(self):
self._q.put('ping')
if state.online:
self._q.put('ping')
def go_online(self):
self._q.put('go_online')
@ -99,6 +100,18 @@ class Node(Thread):
url = 'https://%s.onion:9851' % self.user_id
return url
@property
def online(self):
return self._online
@online.setter
def online(self, online):
if self._online != online:
self._online = online
self.trigger_status()
else:
self._online = online
def resolve(self):
#logger.debug('resolve node %s', self.user_id)
r = self.get_local()
@ -263,13 +276,13 @@ class Node(Thread):
self.online = False
else:
self.online = False
self.trigger_status()
def trigger_status(self):
trigger_event('status', {
'id': self.user_id,
'online': self.online
})
if self.online is not None:
trigger_event('status', {
'id': self.user_id,
'online': self.online
})
def pullChanges(self):
if not self.online or not self.user.peered:
@ -280,7 +293,6 @@ class Node(Thread):
changes = self.request('pullChanges', from_revision)
except:
self.online = False
self.trigger_status()
logger.debug('%s went offline', self.user.name)
return False
if not changes:
@ -296,7 +308,6 @@ class Node(Thread):
r = self.request('pushChanges', changes)
except:
self.online = False
self.trigger_status()
r = False
logger.debug('pushedChanges %s %s', r, self.user_id)
@ -407,6 +418,15 @@ class Nodes(Thread):
def __init__(self):
self._q = Queue()
self._running = True
with db.session():
for u in user.models.User.query.filter_by(peered=True):
if 'local' in u.info:
del u.info['local']
u.save()
self.queue('add', 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)
self._local = LocalNodes()
self._cleanup = PeriodicCallback(lambda: self.queue('cleanup'), 120000)
self._cleanup.start()

17
oml/server.py
View file

@ -53,6 +53,7 @@ def log_request(handler):
request_time = 1000.0 * handler.request.request_time()
log_method("%d %s %.2fms", handler.get_status(),
handler._request_summary(), request_time)
def run():
setup.create_db()
PID = sys.argv[2] if len(sys.argv) > 2 else None
@ -106,7 +107,6 @@ def run():
state.tasks = tasks.Tasks()
def start_node():
import user
import downloads
import nodes
import tor
@ -117,21 +117,12 @@ def run():
state.downloads = downloads.Downloads()
state.scraping = downloads.ScrapeThread()
state.nodes = nodes.Nodes()
def add_users():
def publish():
if not state.tor.is_online():
state.main.add_callback(add_users)
state.main.call_later(1, publish)
else:
with db.session():
for u in user.models.User.query.filter_by(peered=True):
if 'local' in u.info:
del u.info['local']
u.save()
state.nodes.queue('add', u.id)
for u in user.models.User.query.filter_by(queued=True):
logger.debug('adding queued node... %s', u.id)
state.nodes.queue('add', u.id)
nodes.publish_node()
state.main.add_callback(add_users)
state.main.add_callback(publish)
state.main.add_callback(start_node)
if ':' in settings.server['address']:
host = '[%s]' % settings.server['address']

5
oml/tor.py
View file

@ -8,6 +8,7 @@ from stem.control import Controller
import settings
import state
import utils
import logging
logger = logging.getLogger(__name__)
@ -127,7 +128,7 @@ class Tor(object):
self.connected = True
self.socks_port = int(self.controller.get_conf('SocksPort').split(' ')[0])
self.publish()
state.online = True
state.online = self.is_online()
return True
def reconnect(self):
@ -193,4 +194,4 @@ class Tor(object):
state.online = False
def is_online(self):
return self.connected and self.controller.is_alive()
return self.connected and self.controller.is_alive() and utils.can_connect_dns()

13
oml/utils.py
View file

@ -346,3 +346,16 @@ def open_folder(folder=None, path=None):
else:
logger.debug('unsupported platform %s', sys.platform)
subprocess.Popen(cmd, close_fds=True)
def can_connect_dns(host="8.8.8.8", port=53):
"""
host: 8.8.8.8 (google-public-dns-a.google.com)
port: 53/tcp
"""
try:
socket.setdefaulttimeout(1)
socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect((host, port))
return True
except:
pass
return False

8
oml/websocket.py
View file

@ -9,6 +9,7 @@ import json
from oxtornado import json_dumps
import state
import settings
import logging
logger = logging.getLogger(__name__)
@ -33,6 +34,13 @@ class Handler(WebSocketHandler):
self.close()
if self not in state.websockets:
state.websockets.append(self)
trigger_event('status', {
'id': settings.USER_ID,
'online': state.online
})
if state.nodes:
for node in state.nodes._nodes.values():
node.trigger_status()
#websocket calls