using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MediaBrowser.Controller.Session;
using MediaBrowser.Controller.SyncPlay;
using MediaBrowser.Model.Session;
using MediaBrowser.Model.SyncPlay;
namespace Emby.Server.Implementations.SyncPlay
{
///
/// Class SyncPlayController.
///
///
/// Class is not thread-safe, external locking is required when accessing methods.
///
public class SyncPlayController : ISyncPlayController
{
///
/// Used to filter the sessions of a group.
///
private enum BroadcastType
{
///
/// All sessions will receive the message.
///
AllGroup = 0,
///
/// Only the specified session will receive the message.
///
CurrentSession = 1,
///
/// All sessions, except the current one, will receive the message.
///
AllExceptCurrentSession = 2,
///
/// Only sessions that are not buffering will receive the message.
///
AllReady = 3
}
///
/// The session manager.
///
private readonly ISessionManager _sessionManager;
///
/// The SyncPlay manager.
///
private readonly ISyncPlayManager _syncPlayManager;
///
/// The group to manage.
///
private readonly GroupInfo _group = new GroupInfo();
///
public Guid GetGroupId() => _group.GroupId;
///
public Guid GetPlayingItemId() => _group.PlayingItem.Id;
///
public bool IsGroupEmpty() => _group.IsEmpty();
///
/// Initializes a new instance of the class.
///
/// The session manager.
/// The SyncPlay manager.
public SyncPlayController(
ISessionManager sessionManager,
ISyncPlayManager syncPlayManager)
{
_sessionManager = sessionManager;
_syncPlayManager = syncPlayManager;
}
///
/// Converts DateTime to UTC string.
///
/// The date to convert.
/// The UTC string.
private string DateToUTCString(DateTime date)
{
return date.ToUniversalTime().ToString("o");
}
///
/// Filters sessions of this group.
///
/// The current session.
/// The filtering type.
/// The array of sessions matching the filter.
private SessionInfo[] FilterSessions(SessionInfo from, BroadcastType type)
{
switch (type)
{
case BroadcastType.CurrentSession:
return new SessionInfo[] { from };
case BroadcastType.AllGroup:
return _group.Participants.Values.Select(
session => session.Session
).ToArray();
case BroadcastType.AllExceptCurrentSession:
return _group.Participants.Values.Select(
session => session.Session
).Where(
session => !session.Id.Equals(from.Id)
).ToArray();
case BroadcastType.AllReady:
return _group.Participants.Values.Where(
session => !session.IsBuffering
).Select(
session => session.Session
).ToArray();
default:
return Array.Empty();
}
}
///
/// Sends a GroupUpdate message to the interested sessions.
///
/// The current session.
/// The filtering type.
/// The message to send.
/// The cancellation token.
/// The task.
private Task SendGroupUpdate(SessionInfo from, BroadcastType type, GroupUpdate message, CancellationToken cancellationToken)
{
IEnumerable GetTasks()
{
SessionInfo[] sessions = FilterSessions(from, type);
foreach (var session in sessions)
{
yield return _sessionManager.SendSyncPlayGroupUpdate(session.Id.ToString(), message, cancellationToken);
}
}
return Task.WhenAll(GetTasks());
}
///
/// Sends a playback command to the interested sessions.
///
/// The current session.
/// The filtering type.
/// The message to send.
/// The cancellation token.
/// The task.
private Task SendCommand(SessionInfo from, BroadcastType type, SendCommand message, CancellationToken cancellationToken)
{
IEnumerable GetTasks()
{
SessionInfo[] sessions = FilterSessions(from, type);
foreach (var session in sessions)
{
yield return _sessionManager.SendSyncPlayCommand(session.Id.ToString(), message, cancellationToken);
}
}
return Task.WhenAll(GetTasks());
}
///
/// Builds a new playback command with some default values.
///
/// The command type.
/// The SendCommand.
private SendCommand NewSyncPlayCommand(SendCommandType type)
{
return new SendCommand()
{
GroupId = _group.GroupId.ToString(),
Command = type,
PositionTicks = _group.PositionTicks,
When = DateToUTCString(_group.LastActivity),
EmittedAt = DateToUTCString(DateTime.UtcNow)
};
}
///
/// Builds a new group update message.
///
/// The update type.
/// The data to send.
/// The GroupUpdate.
private GroupUpdate NewSyncPlayGroupUpdate(GroupUpdateType type, T data)
{
return new GroupUpdate()
{
GroupId = _group.GroupId.ToString(),
Type = type,
Data = data
};
}
///
public void InitGroup(SessionInfo session, CancellationToken cancellationToken)
{
_group.AddSession(session);
_syncPlayManager.AddSessionToGroup(session, this);
_group.PlayingItem = session.FullNowPlayingItem;
_group.IsPaused = true;
_group.PositionTicks = session.PlayState.PositionTicks ?? 0;
_group.LastActivity = DateTime.UtcNow;
var updateSession = NewSyncPlayGroupUpdate(GroupUpdateType.GroupJoined, DateToUTCString(DateTime.UtcNow));
SendGroupUpdate(session, BroadcastType.CurrentSession, updateSession, cancellationToken);
var pauseCommand = NewSyncPlayCommand(SendCommandType.Pause);
SendCommand(session, BroadcastType.CurrentSession, pauseCommand, cancellationToken);
}
///
public void SessionJoin(SessionInfo session, JoinGroupRequest request, CancellationToken cancellationToken)
{
if (session.NowPlayingItem?.Id == _group.PlayingItem.Id && request.PlayingItemId == _group.PlayingItem.Id)
{
_group.AddSession(session);
_syncPlayManager.AddSessionToGroup(session, this);
var updateSession = NewSyncPlayGroupUpdate(GroupUpdateType.GroupJoined, DateToUTCString(DateTime.UtcNow));
SendGroupUpdate(session, BroadcastType.CurrentSession, updateSession, cancellationToken);
var updateOthers = NewSyncPlayGroupUpdate(GroupUpdateType.UserJoined, session.UserName);
SendGroupUpdate(session, BroadcastType.AllExceptCurrentSession, updateOthers, cancellationToken);
// Client join and play, syncing will happen client side
if (!_group.IsPaused)
{
var playCommand = NewSyncPlayCommand(SendCommandType.Play);
SendCommand(session, BroadcastType.CurrentSession, playCommand, cancellationToken);
}
else
{
var pauseCommand = NewSyncPlayCommand(SendCommandType.Pause);
SendCommand(session, BroadcastType.CurrentSession, pauseCommand, cancellationToken);
}
}
else
{
var playRequest = new PlayRequest();
playRequest.ItemIds = new Guid[] { _group.PlayingItem.Id };
playRequest.StartPositionTicks = _group.PositionTicks;
var update = NewSyncPlayGroupUpdate(GroupUpdateType.PrepareSession, playRequest);
SendGroupUpdate(session, BroadcastType.CurrentSession, update, cancellationToken);
}
}
///
public void SessionLeave(SessionInfo session, CancellationToken cancellationToken)
{
_group.RemoveSession(session);
_syncPlayManager.RemoveSessionFromGroup(session, this);
var updateSession = NewSyncPlayGroupUpdate(GroupUpdateType.GroupLeft, _group.PositionTicks);
SendGroupUpdate(session, BroadcastType.CurrentSession, updateSession, cancellationToken);
var updateOthers = NewSyncPlayGroupUpdate(GroupUpdateType.UserLeft, session.UserName);
SendGroupUpdate(session, BroadcastType.AllExceptCurrentSession, updateOthers, cancellationToken);
}
///
public void HandleRequest(SessionInfo session, PlaybackRequest request, CancellationToken cancellationToken)
{
// The server's job is to mantain a consistent state to which clients refer to,
// as also to notify clients of state changes.
// The actual syncing of media playback happens client side.
// Clients are aware of the server's time and use it to sync.
switch (request.Type)
{
case PlaybackRequestType.Play:
HandlePlayRequest(session, request, cancellationToken);
break;
case PlaybackRequestType.Pause:
HandlePauseRequest(session, request, cancellationToken);
break;
case PlaybackRequestType.Seek:
HandleSeekRequest(session, request, cancellationToken);
break;
case PlaybackRequestType.Buffering:
HandleBufferingRequest(session, request, cancellationToken);
break;
case PlaybackRequestType.BufferingDone:
HandleBufferingDoneRequest(session, request, cancellationToken);
break;
case PlaybackRequestType.UpdatePing:
HandlePingUpdateRequest(session, request);
break;
}
}
///
/// Handles a play action requested by a session.
///
/// The session.
/// The play action.
/// The cancellation token.
private void HandlePlayRequest(SessionInfo session, PlaybackRequest request, CancellationToken cancellationToken)
{
if (_group.IsPaused)
{
// Pick a suitable time that accounts for latency
var delay = _group.GetHighestPing() * 2;
delay = delay < _group.DefaulPing ? _group.DefaulPing : delay;
// Unpause group and set starting point in future
// Clients will start playback at LastActivity (datetime) from PositionTicks (playback position)
// The added delay does not guarantee, of course, that the command will be received in time
// Playback synchronization will mainly happen client side
_group.IsPaused = false;
_group.LastActivity = DateTime.UtcNow.AddMilliseconds(
delay
);
var command = NewSyncPlayCommand(SendCommandType.Play);
SendCommand(session, BroadcastType.AllGroup, command, cancellationToken);
}
else
{
// Client got lost, sending current state
var command = NewSyncPlayCommand(SendCommandType.Play);
SendCommand(session, BroadcastType.CurrentSession, command, cancellationToken);
}
}
///
/// Handles a pause action requested by a session.
///
/// The session.
/// The pause action.
/// The cancellation token.
private void HandlePauseRequest(SessionInfo session, PlaybackRequest request, CancellationToken cancellationToken)
{
if (!_group.IsPaused)
{
// Pause group and compute the media playback position
_group.IsPaused = true;
var currentTime = DateTime.UtcNow;
var elapsedTime = currentTime - _group.LastActivity;
_group.LastActivity = currentTime;
// Seek only if playback actually started
// (a pause request may be issued during the delay added to account for latency)
_group.PositionTicks += elapsedTime.Ticks > 0 ? elapsedTime.Ticks : 0;
var command = NewSyncPlayCommand(SendCommandType.Pause);
SendCommand(session, BroadcastType.AllGroup, command, cancellationToken);
}
else
{
// Client got lost, sending current state
var command = NewSyncPlayCommand(SendCommandType.Pause);
SendCommand(session, BroadcastType.CurrentSession, command, cancellationToken);
}
}
///
/// Handles a seek action requested by a session.
///
/// The session.
/// The seek action.
/// The cancellation token.
private void HandleSeekRequest(SessionInfo session, PlaybackRequest request, CancellationToken cancellationToken)
{
// Sanitize PositionTicks
var ticks = SanitizePositionTicks(request.PositionTicks);
// Pause and seek
_group.IsPaused = true;
_group.PositionTicks = ticks;
_group.LastActivity = DateTime.UtcNow;
var command = NewSyncPlayCommand(SendCommandType.Seek);
SendCommand(session, BroadcastType.AllGroup, command, cancellationToken);
}
///
/// Handles a buffering action requested by a session.
///
/// The session.
/// The buffering action.
/// The cancellation token.
private void HandleBufferingRequest(SessionInfo session, PlaybackRequest request, CancellationToken cancellationToken)
{
if (!_group.IsPaused)
{
// Pause group and compute the media playback position
_group.IsPaused = true;
var currentTime = DateTime.UtcNow;
var elapsedTime = currentTime - _group.LastActivity;
_group.LastActivity = currentTime;
_group.PositionTicks += elapsedTime.Ticks > 0 ? elapsedTime.Ticks : 0;
_group.SetBuffering(session, true);
// Send pause command to all non-buffering sessions
var command = NewSyncPlayCommand(SendCommandType.Pause);
SendCommand(session, BroadcastType.AllReady, command, cancellationToken);
var updateOthers = NewSyncPlayGroupUpdate(GroupUpdateType.GroupWait, session.UserName);
SendGroupUpdate(session, BroadcastType.AllExceptCurrentSession, updateOthers, cancellationToken);
}
else
{
// Client got lost, sending current state
var command = NewSyncPlayCommand(SendCommandType.Pause);
SendCommand(session, BroadcastType.CurrentSession, command, cancellationToken);
}
}
///
/// Handles a buffering-done action requested by a session.
///
/// The session.
/// The buffering-done action.
/// The cancellation token.
private void HandleBufferingDoneRequest(SessionInfo session, PlaybackRequest request, CancellationToken cancellationToken)
{
if (_group.IsPaused)
{
_group.SetBuffering(session, false);
var requestTicks = SanitizePositionTicks(request.PositionTicks);
var when = request.When ?? DateTime.UtcNow;
var currentTime = DateTime.UtcNow;
var elapsedTime = currentTime - when;
var clientPosition = TimeSpan.FromTicks(requestTicks) + elapsedTime;
var delay = _group.PositionTicks - clientPosition.Ticks;
if (_group.IsBuffering())
{
// Others are still buffering, tell this client to pause when ready
var command = NewSyncPlayCommand(SendCommandType.Pause);
var pauseAtTime = currentTime.AddMilliseconds(delay);
command.When = DateToUTCString(pauseAtTime);
SendCommand(session, BroadcastType.CurrentSession, command, cancellationToken);
}
else
{
// Let other clients resume as soon as the buffering client catches up
_group.IsPaused = false;
if (delay > _group.GetHighestPing() * 2)
{
// Client that was buffering is recovering, notifying others to resume
_group.LastActivity = currentTime.AddMilliseconds(
delay
);
var command = NewSyncPlayCommand(SendCommandType.Play);
SendCommand(session, BroadcastType.AllExceptCurrentSession, command, cancellationToken);
}
else
{
// Client, that was buffering, resumed playback but did not update others in time
delay = _group.GetHighestPing() * 2;
delay = delay < _group.DefaulPing ? _group.DefaulPing : delay;
_group.LastActivity = currentTime.AddMilliseconds(
delay
);
var command = NewSyncPlayCommand(SendCommandType.Play);
SendCommand(session, BroadcastType.AllGroup, command, cancellationToken);
}
}
}
else
{
// Group was not waiting, make sure client has latest state
var command = NewSyncPlayCommand(SendCommandType.Play);
SendCommand(session, BroadcastType.CurrentSession, command, cancellationToken);
}
}
///
/// Sanitizes the PositionTicks, considers the current playing item when available.
///
/// The PositionTicks.
/// The sanitized PositionTicks.
private long SanitizePositionTicks(long? positionTicks)
{
var ticks = positionTicks ?? 0;
ticks = ticks >= 0 ? ticks : 0;
if (_group.PlayingItem != null)
{
var runTimeTicks = _group.PlayingItem.RunTimeTicks ?? 0;
ticks = ticks > runTimeTicks ? runTimeTicks : ticks;
}
return ticks;
}
///
/// Updates ping of a session.
///
/// The session.
/// The update.
private void HandlePingUpdateRequest(SessionInfo session, PlaybackRequest request)
{
// Collected pings are used to account for network latency when unpausing playback
_group.UpdatePing(session, request.Ping ?? _group.DefaulPing);
}
///
public GroupInfoView GetInfo()
{
return new GroupInfoView()
{
GroupId = GetGroupId().ToString(),
PlayingItemName = _group.PlayingItem.Name,
PlayingItemId = _group.PlayingItem.Id.ToString(),
PositionTicks = _group.PositionTicks,
Participants = _group.Participants.Values.Select(session => session.Session.UserName).Distinct().ToList()
};
}
}
}