github/plexpy
1
0
Fork 0
mirror of https://github.com/Tautulli/Tautulli.git synced 2025-07-10 15:32:38 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions

Pass copy of notification data to prevent multithreading issues

Browse source
This commit is contained in:
JonnyWong16 2018-01-30 23:04:44 -08:00
parent 7672f1955e
commit 9100e25a21
4 changed files with 18 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

12
plexpy/activity_handler.py
View file

@ -97,7 +97,7 @@ class ActivityHandler(object):
% (str(session['session_key']), str(session['user_id']), session['username'],
str(session['rating_key']), session['full_title']))
plexpy.NOTIFY_QUEUE.put({'stream_data': session, 'notify_action': 'on_play'})
plexpy.NOTIFY_QUEUE.put({'stream_data': session.copy(), 'notify_action': 'on_play'})
# Write the new session to our temp session table
self.update_db_session(session=session)
@ -122,7 +122,7 @@ class ActivityHandler(object):
# Retrieve the session data from our temp table
db_session = ap.get_session_by_key(session_key=self.get_session_key())
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session, 'notify_action': 'on_stop'})
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session.copy(), 'notify_action': 'on_stop'})
# Write it to the history table
monitor_proc = activity_processor.ActivityProcessor()
@ -159,7 +159,7 @@ class ActivityHandler(object):
db_session = ap.get_session_by_key(session_key=self.get_session_key())
if not still_paused:
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session, 'notify_action': 'on_pause'})
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session.copy(), 'notify_action': 'on_pause'})
def on_resume(self):
if self.is_valid_session():
@ -178,7 +178,7 @@ class ActivityHandler(object):
# Retrieve the session data from our temp table
db_session = ap.get_session_by_key(session_key=self.get_session_key())
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session, 'notify_action': 'on_resume'})
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session.copy(), 'notify_action': 'on_resume'})
def on_buffer(self):
if self.is_valid_session():
@ -216,7 +216,7 @@ class ActivityHandler(object):
# Retrieve the session data from our temp table
db_session = ap.get_session_by_key(session_key=self.get_session_key())
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session, 'notify_action': 'on_buffer'})
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session.copy(), 'notify_action': 'on_buffer'})
# This function receives events from our websocket connection
def process(self):
@ -279,7 +279,7 @@ class ActivityHandler(object):
db_session['media_type'] == 'episode' and progress_percent >= plexpy.CONFIG.TV_WATCHED_PERCENT or
db_session['media_type'] == 'track' and progress_percent >= plexpy.CONFIG.MUSIC_WATCHED_PERCENT) \
and not any(d['notify_action'] == 'on_watched' for d in notify_states):
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session, 'notify_action': 'on_watched'})
plexpy.NOTIFY_QUEUE.put({'stream_data': db_session.copy(), 'notify_action': 'on_watched'})
else:
# We don't have this session in our table yet, start a new one.

18
plexpy/activity_pinger.py
View file

@ -61,12 +61,12 @@ def check_active_sessions(ws_request=False):
if session['state'] == 'paused':
logger.debug(u"Tautulli Monitor :: Session %s paused." % stream['session_key'])
plexpy.NOTIFY_QUEUE.put({'stream_data': stream, 'notify_action': 'on_pause'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream.copy(), 'notify_action': 'on_pause'})
if session['state'] == 'playing' and stream['state'] == 'paused':
logger.debug(u"Tautulli Monitor :: Session %s resumed." % stream['session_key'])
plexpy.NOTIFY_QUEUE.put({'stream_data': stream, 'notify_action': 'on_resume'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream.copy(), 'notify_action': 'on_resume'})
if stream['state'] == 'paused' and not ws_request:
# The stream is still paused so we need to increment the paused_counter
@ -104,7 +104,7 @@ def check_active_sessions(ws_request=False):
'WHERE session_key = ? AND rating_key = ?',
[stream['session_key'], stream['rating_key']])
plexpy.NOTIFY_QUEUE.put({'stream_data': stream, 'notify_action': 'on_buffer'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream.copy(), 'notify_action': 'on_buffer'})
else:
# Subsequent buffer notifications after wait time
@ -118,7 +118,7 @@ def check_active_sessions(ws_request=False):
'WHERE session_key = ? AND rating_key = ?',
[stream['session_key'], stream['rating_key']])
plexpy.NOTIFY_QUEUE.put({'stream_data': stream, 'notify_action': 'on_buffer'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream.copy(), 'notify_action': 'on_buffer'})
logger.debug(u"Tautulli Monitor :: Session %s is buffering. Count is now %s. Last triggered %s."
% (stream['session_key'],
@ -135,7 +135,7 @@ def check_active_sessions(ws_request=False):
session['media_type'] == 'episode' and progress_percent >= plexpy.CONFIG.TV_WATCHED_PERCENT or
session['media_type'] == 'track' and progress_percent >= plexpy.CONFIG.MUSIC_WATCHED_PERCENT) \
and not any(d['notify_action'] == 'on_watched' for d in notify_states):
plexpy.NOTIFY_QUEUE.put({'stream_data': stream, 'notify_action': 'on_watched'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream.copy(), 'notify_action': 'on_watched'})
else:
# The user has stopped playing a stream
@ -155,9 +155,9 @@ def check_active_sessions(ws_request=False):
stream['media_type'] == 'episode' and progress_percent >= plexpy.CONFIG.TV_WATCHED_PERCENT or
stream['media_type'] == 'track' and progress_percent >= plexpy.CONFIG.MUSIC_WATCHED_PERCENT) \
and not any(d['notify_action'] == 'on_watched' for d in notify_states):
plexpy.NOTIFY_QUEUE.put({'stream_data': stream, 'notify_action': 'on_watched'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream.copy(), 'notify_action': 'on_watched'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream, 'notify_action': 'on_stop'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream.copy(), 'notify_action': 'on_stop'})
# Write the item history on playback stop
row_id = monitor_process.write_session_history(session=stream)
@ -243,7 +243,7 @@ def check_recently_added():
if 0 < time_threshold - int(item['added_at']) <= time_interval:
logger.debug(u"Tautulli Monitor :: Library item %s added to Plex." % str(item['rating_key']))
plexpy.NOTIFY_QUEUE.put({'timeline_data': item, 'notify_action': 'on_created'})
plexpy.NOTIFY_QUEUE.put({'timeline_data': item.copy(), 'notify_action': 'on_created'})
else:
item = max(metadata, key=lambda x:x['added_at'])
@ -261,7 +261,7 @@ def check_recently_added():
logger.debug(u"Tautulli Monitor :: Library item %s added to Plex." % str(item['rating_key']))
# Check if any notification agents have notifications enabled
plexpy.NOTIFY_QUEUE.put({'timeline_data': item, 'notify_action': 'on_created'})
plexpy.NOTIFY_QUEUE.put({'timeline_data': item.copy(), 'notify_action': 'on_created'})
def check_server_response():

2
plexpy/activity_processor.py
View file

@ -127,7 +127,7 @@ class ActivityProcessor(object):
if result == 'insert':
# Check if any notification agents have notifications enabled
if notify:
plexpy.NOTIFY_QUEUE.put({'stream_data': values, 'notify_action': 'on_play'})
plexpy.NOTIFY_QUEUE.put({'stream_data': values.copy(), 'notify_action': 'on_play'})
# If it's our first write then time stamp it.
started = int(time.time())

4
plexpy/notification_handler.py
View file

@ -122,8 +122,8 @@ def add_notifier_each(notifier_id=None, notify_action=None, stream_data=None, ti
# Add on_concurrent and on_newdevice to queue if action is on_play
if notify_action == 'on_play':
plexpy.NOTIFY_QUEUE.put({'stream_data': stream_data, 'notify_action': 'on_concurrent'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream_data, 'notify_action': 'on_newdevice'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream_data.copy(), 'notify_action': 'on_concurrent'})
plexpy.NOTIFY_QUEUE.put({'stream_data': stream_data.copy(), 'notify_action': 'on_newdevice'})
def notify_conditions(notify_action=None, stream_data=None, timeline_data=None):