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 CreateGroup(SessionInfo session, CancellationToken cancellationToken) { _group.AddSession(session); _syncPlayManager.AddSessionToGroup(session, this); _group.PlayingItem = session.FullNowPlayingItem; _group.IsPaused = session.PlayState.IsPaused; _group.PositionTicks = session.PlayState.PositionTicks ?? 0; _group.LastActivity = DateTime.UtcNow; var updateSession = NewSyncPlayGroupUpdate(GroupUpdateType.GroupJoined, DateToUTCString(DateTime.UtcNow)); SendGroupUpdate(session, BroadcastType.CurrentSession, updateSession, cancellationToken); } /// public void SessionJoin(SessionInfo session, JoinGroupRequest request, CancellationToken cancellationToken) { if (session.NowPlayingItem?.Id == _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); // 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 maintain a consistent state for clients to reference // and 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.Buffer: HandleBufferingRequest(session, request, cancellationToken); break; case PlaybackRequestType.Ready: HandleBufferingDoneRequest(session, request, cancellationToken); break; case PlaybackRequestType.Ping: 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.DefaultPing ? _group.DefaultPing : 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 // 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.DefaultPing ? _group.DefaultPing : 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.DefaultPing); } /// 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() }; } } }