alteropen/jellyfin
1
0
Fork 0
mirror of https://github.com/jellyfin/jellyfin.git synced 2026-03-23 02:27:17 +00:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

Merge pull request #3194 from OancaAndrei/syncplay-enhanced

Browse Source 
SyncPlay for TV series (and Music)
This commit is contained in:
Joshua M. Boniface
2020-12-04 20:22:30 -05:00
committed by GitHub
parent b96d4ef0e8 cc64ee483d
commit bba01bf7b9
91 changed files with 5555 additions and 1356 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
Emby.Server.Implementations/HttpServer/WebSocketManager.cs
View File

@@ -2,9 +2,9 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.WebSockets;
using System.Threading.Tasks;
using Jellyfin.Data.Events;
using MediaBrowser.Controller.Net;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
@@ -13,32 +13,23 @@ namespace Emby.Server.Implementations.HttpServer
{
public class WebSocketManager : IWebSocketManager
{
private readonly Lazy<IEnumerable<IWebSocketListener>> _webSocketListeners;
private readonly IWebSocketListener[] _webSocketListeners;
private readonly ILogger<WebSocketManager> _logger;
private readonly ILoggerFactory _loggerFactory;
private bool _disposed = false;
public WebSocketManager(
Lazy<IEnumerable<IWebSocketListener>> webSocketListeners,
IEnumerable<IWebSocketListener> webSocketListeners,
ILogger<WebSocketManager> logger,
ILoggerFactory loggerFactory)
{
_webSocketListeners = webSocketListeners;
_webSocketListeners = webSocketListeners.ToArray();
_logger = logger;
_loggerFactory = loggerFactory;
}
public event EventHandler<GenericEventArgs<IWebSocketConnection>> WebSocketConnected;
/// <inheritdoc />
public async Task WebSocketRequestHandler(HttpContext context)
{
if (_disposed)
{
return;
}
try
{
_logger.LogInformation("WS {IP} request", context.Connection.RemoteIpAddress);
@@ -54,7 +45,13 @@ namespace Emby.Server.Implementations.HttpServer
OnReceive = ProcessWebSocketMessageReceived
};
WebSocketConnected?.Invoke(this, new GenericEventArgs<IWebSocketConnection>(connection));
var tasks = new Task[_webSocketListeners.Length];
for (var i = 0; i < _webSocketListeners.Length; ++i)
{
tasks[i] = _webSocketListeners[i].ProcessWebSocketConnectedAsync(connection);
}
await Task.WhenAll(tasks).ConfigureAwait(false);
await connection.ProcessAsync().ConfigureAwait(false);
_logger.LogInformation("WS {IP} closed", context.Connection.RemoteIpAddress);
@@ -75,21 +72,13 @@ namespace Emby.Server.Implementations.HttpServer
/// <param name="result">The result.</param>
private Task ProcessWebSocketMessageReceived(WebSocketMessageInfo result)
{
if (_disposed)
var tasks = new Task[_webSocketListeners.Length];
for (var i = 0; i < _webSocketListeners.Length; ++i)
{
return Task.CompletedTask;
tasks[i] = _webSocketListeners[i].ProcessMessageAsync(result);
}
IEnumerable<Task> GetTasks()
{
var listeners = _webSocketListeners.Value;
foreach (var x in listeners)
{
yield return x.ProcessMessageAsync(result);
}
}
return Task.WhenAll(GetTasks());
return Task.WhenAll(tasks);
}
}
}

6
Emby.Server.Implementations/Session/SessionManager.cs
View File

@@ -1181,18 +1181,16 @@ namespace Emby.Server.Implementations.Session
}
/// <inheritdoc />
public async Task SendSyncPlayCommand(string sessionId, SendCommand command, CancellationToken cancellationToken)
public async Task SendSyncPlayCommand(SessionInfo session, SendCommand command, CancellationToken cancellationToken)
{
CheckDisposed();
var session = GetSessionToRemoteControl(sessionId);
await SendMessageToSession(session, SessionMessageType.SyncPlayCommand, command, cancellationToken).ConfigureAwait(false);
}
/// <inheritdoc />
public async Task SendSyncPlayGroupUpdate<T>(string sessionId, GroupUpdate<T> command, CancellationToken cancellationToken)
public async Task SendSyncPlayGroupUpdate<T>(SessionInfo session, GroupUpdate<T> command, CancellationToken cancellationToken)
{
CheckDisposed();
var session = GetSessionToRemoteControl(sessionId);
await SendMessageToSession(session, SessionMessageType.SyncPlayGroupUpdate, command, cancellationToken).ConfigureAwait(false);
}

90
Emby.Server.Implementations/Session/SessionWebSocketListener.cs
View File

@@ -4,7 +4,6 @@ using System.Linq;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Data.Events;
using MediaBrowser.Controller.Net;
using MediaBrowser.Controller.Session;
using MediaBrowser.Model.Net;
@@ -22,35 +21,17 @@ namespace Emby.Server.Implementations.Session
/// <summary>
/// The timeout in seconds after which a WebSocket is considered to be lost.
/// </summary>
public const int WebSocketLostTimeout = 60;
private const int WebSocketLostTimeout = 60;
/// <summary>
/// The keep-alive interval factor; controls how often the watcher will check on the status of the WebSockets.
/// </summary>
public const float IntervalFactor = 0.2f;
private const float IntervalFactor = 0.2f;
/// <summary>
/// The ForceKeepAlive factor; controls when a ForceKeepAlive is sent.
/// </summary>
public const float ForceKeepAliveFactor = 0.75f;
/// <summary>
/// The _session manager.
/// </summary>
private readonly ISessionManager _sessionManager;
/// <summary>
/// The _logger.
/// </summary>
private readonly ILogger<SessionWebSocketListener> _logger;
private readonly ILoggerFactory _loggerFactory;
private readonly IWebSocketManager _webSocketManager;
/// <summary>
/// The KeepAlive cancellation token.
/// </summary>
private CancellationTokenSource _keepAliveCancellationToken;
private const float ForceKeepAliveFactor = 0.75f;
/// <summary>
/// Lock used for accesing the KeepAlive cancellation token.
@@ -63,42 +44,68 @@ namespace Emby.Server.Implementations.Session
private readonly HashSet<IWebSocketConnection> _webSockets = new HashSet<IWebSocketConnection>();
/// <summary>
/// Lock used for accesing the WebSockets watchlist.
/// Lock used for accessing the WebSockets watchlist.
/// </summary>
private readonly object _webSocketsLock = new object();
/// <summary>
/// The _session manager.
/// </summary>
private readonly ISessionManager _sessionManager;
/// <summary>
/// The _logger.
/// </summary>
private readonly ILogger<SessionWebSocketListener> _logger;
private readonly ILoggerFactory _loggerFactory;
/// <summary>
/// The KeepAlive cancellation token.
/// </summary>
private CancellationTokenSource _keepAliveCancellationToken;
/// <summary>
/// Initializes a new instance of the <see cref="SessionWebSocketListener" /> class.
/// </summary>
/// <param name="logger">The logger.</param>
/// <param name="sessionManager">The session manager.</param>
/// <param name="loggerFactory">The logger factory.</param>
/// <param name="webSocketManager">The HTTP server.</param>
public SessionWebSocketListener(
ILogger<SessionWebSocketListener> logger,
ISessionManager sessionManager,
ILoggerFactory loggerFactory,
IWebSocketManager webSocketManager)
ILoggerFactory loggerFactory)
{
_logger = logger;
_sessionManager = sessionManager;
_loggerFactory = loggerFactory;
_webSocketManager = webSocketManager;
webSocketManager.WebSocketConnected += OnServerManagerWebSocketConnected;
}
private async void OnServerManagerWebSocketConnected(object sender, GenericEventArgs<IWebSocketConnection> e)
/// <inheritdoc />
public void Dispose()
{
var session = GetSession(e.Argument.QueryString, e.Argument.RemoteEndPoint.ToString());
StopKeepAlive();
}
/// <summary>
/// Processes the message.
/// </summary>
/// <param name="message">The message.</param>
/// <returns>Task.</returns>
public Task ProcessMessageAsync(WebSocketMessageInfo message)
=> Task.CompletedTask;
/// <inheritdoc />
public async Task ProcessWebSocketConnectedAsync(IWebSocketConnection connection)
{
var session = GetSession(connection.QueryString, connection.RemoteEndPoint.ToString());
if (session != null)
{
EnsureController(session, e.Argument);
await KeepAliveWebSocket(e.Argument).ConfigureAwait(false);
EnsureController(session, connection);
await KeepAliveWebSocket(connection).ConfigureAwait(false);
}
else
{
_logger.LogWarning("Unable to determine session based on query string: {0}", e.Argument.QueryString);
_logger.LogWarning("Unable to determine session based on query string: {0}", connection.QueryString);
}
}
@@ -119,21 +126,6 @@ namespace Emby.Server.Implementations.Session
return _sessionManager.GetSessionByAuthenticationToken(token, deviceId, remoteEndpoint);
}
/// <inheritdoc />
public void Dispose()
{
_webSocketManager.WebSocketConnected -= OnServerManagerWebSocketConnected;
StopKeepAlive();
}
/// <summary>
/// Processes the message.
/// </summary>
/// <param name="message">The message.</param>
/// <returns>Task.</returns>
public Task ProcessMessageAsync(WebSocketMessageInfo message)
=> Task.CompletedTask;
private void EnsureController(SessionInfo session, IWebSocketConnection connection)
{
var controllerInfo = session.EnsureController<WebSocketController>(

674
Emby.Server.Implementations/SyncPlay/Group.cs Normal file
View File

@@ -0,0 +1,674 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Data.Entities;
using MediaBrowser.Controller.Library;
using MediaBrowser.Controller.Session;
using MediaBrowser.Controller.SyncPlay;
using MediaBrowser.Controller.SyncPlay.GroupStates;
using MediaBrowser.Controller.SyncPlay.Queue;
using MediaBrowser.Controller.SyncPlay.Requests;
using MediaBrowser.Model.SyncPlay;
using Microsoft.Extensions.Logging;
namespace Emby.Server.Implementations.SyncPlay
{
/// <summary>
/// Class Group.
/// </summary>
/// <remarks>
/// Class is not thread-safe, external locking is required when accessing methods.
/// </remarks>
public class Group : IGroupStateContext
{
/// <summary>
/// The logger.
/// </summary>
private readonly ILogger<Group> _logger;
/// <summary>
/// The logger factory.
/// </summary>
private readonly ILoggerFactory _loggerFactory;
/// <summary>
/// The user manager.
/// </summary>
private readonly IUserManager _userManager;
/// <summary>
/// The session manager.
/// </summary>
private readonly ISessionManager _sessionManager;
/// <summary>
/// The library manager.
/// </summary>
private readonly ILibraryManager _libraryManager;
/// <summary>
/// The participants, or members of the group.
/// </summary>
private readonly Dictionary<string, GroupMember> _participants =
new Dictionary<string, GroupMember>(StringComparer.OrdinalIgnoreCase);
/// <summary>
/// The internal group state.
/// </summary>
private IGroupState _state;
/// <summary>
/// Initializes a new instance of the <see cref="Group" /> class.
/// </summary>
/// <param name="loggerFactory">The logger factory.</param>
/// <param name="userManager">The user manager.</param>
/// <param name="sessionManager">The session manager.</param>
/// <param name="libraryManager">The library manager.</param>
public Group(
ILoggerFactory loggerFactory,
IUserManager userManager,
ISessionManager sessionManager,
ILibraryManager libraryManager)
{
_loggerFactory = loggerFactory;
_userManager = userManager;
_sessionManager = sessionManager;
_libraryManager = libraryManager;
_logger = loggerFactory.CreateLogger<Group>();
_state = new IdleGroupState(loggerFactory);
}
/// <summary>
/// Gets the default ping value used for sessions.
/// </summary>
/// <value>The default ping.</value>
public long DefaultPing { get; } = 500;
/// <summary>
/// Gets the maximum time offset error accepted for dates reported by clients, in milliseconds.
/// </summary>
/// <value>The maximum time offset error.</value>
public long TimeSyncOffset { get; } = 2000;
/// <summary>
/// Gets the maximum offset error accepted for position reported by clients, in milliseconds.
/// </summary>
/// <value>The maximum offset error.</value>
public long MaxPlaybackOffset { get; } = 500;
/// <summary>
/// Gets the group identifier.
/// </summary>
/// <value>The group identifier.</value>
public Guid GroupId { get; } = Guid.NewGuid();
/// <summary>
/// Gets the group name.
/// </summary>
/// <value>The group name.</value>
public string GroupName { get; private set; }
/// <summary>
/// Gets the group identifier.
/// </summary>
/// <value>The group identifier.</value>
public PlayQueueManager PlayQueue { get; } = new PlayQueueManager();
/// <summary>
/// Gets the runtime ticks of current playing item.
/// </summary>
/// <value>The runtime ticks of current playing item.</value>
public long RunTimeTicks { get; private set; }
/// <summary>
/// Gets or sets the position ticks.
/// </summary>
/// <value>The position ticks.</value>
public long PositionTicks { get; set; }
/// <summary>
/// Gets or sets the last activity.
/// </summary>
/// <value>The last activity.</value>
public DateTime LastActivity { get; set; }
/// <summary>
/// Adds the session to the group.
/// </summary>
/// <param name="session">The session.</param>
private void AddSession(SessionInfo session)
{
_participants.TryAdd(
session.Id,
new GroupMember(session)
{
Ping = DefaultPing,
IsBuffering = false
});
}
/// <summary>
/// Removes the session from the group.
/// </summary>
/// <param name="session">The session.</param>
private void RemoveSession(SessionInfo session)
{
_participants.Remove(session.Id);
}
/// <summary>
/// Filters sessions of this group.
/// </summary>
/// <param name="from">The current session.</param>
/// <param name="type">The filtering type.</param>
/// <returns>The list of sessions matching the filter.</returns>
private IEnumerable<SessionInfo> FilterSessions(SessionInfo from, SyncPlayBroadcastType type)
{
return type switch
{
SyncPlayBroadcastType.CurrentSession => new SessionInfo[] { from },
SyncPlayBroadcastType.AllGroup => _participants
.Values
.Select(session => session.Session),
SyncPlayBroadcastType.AllExceptCurrentSession => _participants
.Values
.Select(session => session.Session)
.Where(session => !session.Id.Equals(from.Id, StringComparison.OrdinalIgnoreCase)),
SyncPlayBroadcastType.AllReady => _participants
.Values
.Where(session => !session.IsBuffering)
.Select(session => session.Session),
_ => Enumerable.Empty<SessionInfo>()
};
}
/// <summary>
/// Checks if a given user can access all items of a given queue, that is,
/// the user has the required minimum parental access and has access to all required folders.
/// </summary>
/// <param name="user">The user.</param>
/// <param name="queue">The queue.</param>
/// <returns><c>true</c> if the user can access all the items in the queue, <c>false</c> otherwise.</returns>
private bool HasAccessToQueue(User user, IReadOnlyList<Guid> queue)
{
// Check if queue is empty.
if (queue == null || queue.Count == 0)
{
return true;
}
foreach (var itemId in queue)
{
var item = _libraryManager.GetItemById(itemId);
if (!item.IsVisibleStandalone(user))
{
return false;
}
}
return true;
}
private bool AllUsersHaveAccessToQueue(IReadOnlyList<Guid> queue)
{
// Check if queue is empty.
if (queue == null || queue.Count == 0)
{
return true;
}
// Get list of users.
var users = _participants
.Values
.Select(participant => _userManager.GetUserById(participant.Session.UserId));
// Find problematic users.
var usersWithNoAccess = users.Where(user => !HasAccessToQueue(user, queue));
// All users must be able to access the queue.
return !usersWithNoAccess.Any();
}
/// <summary>
/// Checks if the group is empty.
/// </summary>
/// <returns><c>true</c> if the group is empty, <c>false</c> otherwise.</returns>
public bool IsGroupEmpty() => _participants.Count == 0;
/// <summary>
/// Initializes the group with the session's info.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The request.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public void CreateGroup(SessionInfo session, NewGroupRequest request, CancellationToken cancellationToken)
{
GroupName = request.GroupName;
AddSession(session);
var sessionIsPlayingAnItem = session.FullNowPlayingItem != null;
RestartCurrentItem();
if (sessionIsPlayingAnItem)
{
var playlist = session.NowPlayingQueue.Select(item => item.Id).ToList();
PlayQueue.Reset();
PlayQueue.SetPlaylist(playlist);
PlayQueue.SetPlayingItemById(session.FullNowPlayingItem.Id);
RunTimeTicks = session.FullNowPlayingItem.RunTimeTicks ?? 0;
PositionTicks = session.PlayState.PositionTicks ?? 0;
// Maintain playstate.
var waitingState = new WaitingGroupState(_loggerFactory)
{
ResumePlaying = !session.PlayState.IsPaused
};
SetState(waitingState);
}
var updateSession = NewSyncPlayGroupUpdate(GroupUpdateType.GroupJoined, GetInfo());
SendGroupUpdate(session, SyncPlayBroadcastType.CurrentSession, updateSession, cancellationToken);
_state.SessionJoined(this, _state.Type, session, cancellationToken);
_logger.LogInformation("Session {SessionId} created group {GroupId}.", session.Id, GroupId.ToString());
}
/// <summary>
/// Adds the session to the group.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The request.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public void SessionJoin(SessionInfo session, JoinGroupRequest request, CancellationToken cancellationToken)
{
AddSession(session);
var updateSession = NewSyncPlayGroupUpdate(GroupUpdateType.GroupJoined, GetInfo());
SendGroupUpdate(session, SyncPlayBroadcastType.CurrentSession, updateSession, cancellationToken);
var updateOthers = NewSyncPlayGroupUpdate(GroupUpdateType.UserJoined, session.UserName);
SendGroupUpdate(session, SyncPlayBroadcastType.AllExceptCurrentSession, updateOthers, cancellationToken);
_state.SessionJoined(this, _state.Type, session, cancellationToken);
_logger.LogInformation("Session {SessionId} joined group {GroupId}.", session.Id, GroupId.ToString());
}
/// <summary>
/// Removes the session from the group.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The request.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public void SessionLeave(SessionInfo session, LeaveGroupRequest request, CancellationToken cancellationToken)
{
_state.SessionLeaving(this, _state.Type, session, cancellationToken);
RemoveSession(session);
var updateSession = NewSyncPlayGroupUpdate(GroupUpdateType.GroupLeft, GroupId.ToString());
SendGroupUpdate(session, SyncPlayBroadcastType.CurrentSession, updateSession, cancellationToken);
var updateOthers = NewSyncPlayGroupUpdate(GroupUpdateType.UserLeft, session.UserName);
SendGroupUpdate(session, SyncPlayBroadcastType.AllExceptCurrentSession, updateOthers, cancellationToken);
_logger.LogInformation("Session {SessionId} left group {GroupId}.", session.Id, GroupId.ToString());
}
/// <summary>
/// Handles the requested action by the session.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The requested action.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public void HandleRequest(SessionInfo session, IGroupPlaybackRequest 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.
_logger.LogInformation("Session {SessionId} requested {RequestType} in group {GroupId} that is {StateType}.", session.Id, request.Action, GroupId.ToString(), _state.Type);
// Apply requested changes to this group given its current state.
// Every request has a slightly different outcome depending on the group's state.
// There are currently four different group states that accomplish different goals:
// - Idle: in this state no media is playing and clients should be idle (playback is stopped).
// - Waiting: in this state the group is waiting for all the clients to be ready to start the playback,
// that is, they've either finished loading the media for the first time or they've finished buffering.
// Once all clients report to be ready the group's state can change to Playing or Paused.
// - Playing: clients have some media loaded and playback is unpaused.
// - Paused: clients have some media loaded but playback is currently paused.
request.Apply(this, _state, session, cancellationToken);
}
/// <summary>
/// Gets the info about the group for the clients.
/// </summary>
/// <returns>The group info for the clients.</returns>
public GroupInfoDto GetInfo()
{
var participants = _participants.Values.Select(session => session.Session.UserName).Distinct().ToList();
return new GroupInfoDto(GroupId, GroupName, _state.Type, participants, DateTime.UtcNow);
}
/// <summary>
/// Checks if a user has access to all content in the play queue.
/// </summary>
/// <param name="user">The user.</param>
/// <returns><c>true</c> if the user can access the play queue; <c>false</c> otherwise.</returns>
public bool HasAccessToPlayQueue(User user)
{
var items = PlayQueue.GetPlaylist().Select(item => item.ItemId).ToList();
return HasAccessToQueue(user, items);
}
/// <inheritdoc />
public void SetIgnoreGroupWait(SessionInfo session, bool ignoreGroupWait)
{
if (_participants.TryGetValue(session.Id, out GroupMember value))
{
value.IgnoreGroupWait = ignoreGroupWait;
}
}
/// <inheritdoc />
public void SetState(IGroupState state)
{
_logger.LogInformation("Group {GroupId} switching from {FromStateType} to {ToStateType}.", GroupId.ToString(), _state.Type, state.Type);
this._state = state;
}
/// <inheritdoc />
public Task SendGroupUpdate<T>(SessionInfo from, SyncPlayBroadcastType type, GroupUpdate<T> message, CancellationToken cancellationToken)
{
IEnumerable<Task> GetTasks()
{
foreach (var session in FilterSessions(from, type))
{
yield return _sessionManager.SendSyncPlayGroupUpdate(session, message, cancellationToken);
}
}
return Task.WhenAll(GetTasks());
}
/// <inheritdoc />
public Task SendCommand(SessionInfo from, SyncPlayBroadcastType type, SendCommand message, CancellationToken cancellationToken)
{
IEnumerable<Task> GetTasks()
{
foreach (var session in FilterSessions(from, type))
{
yield return _sessionManager.SendSyncPlayCommand(session, message, cancellationToken);
}
}
return Task.WhenAll(GetTasks());
}
/// <inheritdoc />
public SendCommand NewSyncPlayCommand(SendCommandType type)
{
return new SendCommand(
GroupId,
PlayQueue.GetPlayingItemPlaylistId(),
LastActivity,
type,
PositionTicks,
DateTime.UtcNow);
}
/// <inheritdoc />
public GroupUpdate<T> NewSyncPlayGroupUpdate<T>(GroupUpdateType type, T data)
{
return new GroupUpdate<T>(GroupId, type, data);
}
/// <inheritdoc />
public long SanitizePositionTicks(long? positionTicks)
{
var ticks = positionTicks ?? 0;
return Math.Clamp(ticks, 0, RunTimeTicks);
}
/// <inheritdoc />
public void UpdatePing(SessionInfo session, long ping)
{
if (_participants.TryGetValue(session.Id, out GroupMember value))
{
value.Ping = ping;
}
}
/// <inheritdoc />
public long GetHighestPing()
{
long max = long.MinValue;
foreach (var session in _participants.Values)
{
max = Math.Max(max, session.Ping);
}
return max;
}
/// <inheritdoc />
public void SetBuffering(SessionInfo session, bool isBuffering)
{
if (_participants.TryGetValue(session.Id, out GroupMember value))
{
value.IsBuffering = isBuffering;
}
}
/// <inheritdoc />
public void SetAllBuffering(bool isBuffering)
{
foreach (var session in _participants.Values)
{
session.IsBuffering = isBuffering;
}
}
/// <inheritdoc />
public bool IsBuffering()
{
foreach (var session in _participants.Values)
{
if (session.IsBuffering && !session.IgnoreGroupWait)
{
return true;
}
}
return false;
}
/// <inheritdoc />
public bool SetPlayQueue(IReadOnlyList<Guid> playQueue, int playingItemPosition, long startPositionTicks)
{
// Ignore on empty queue or invalid item position.
if (playQueue.Count == 0 || playingItemPosition >= playQueue.Count || playingItemPosition < 0)
{
return false;
}
// Check if participants can access the new playing queue.
if (!AllUsersHaveAccessToQueue(playQueue))
{
return false;
}
PlayQueue.Reset();
PlayQueue.SetPlaylist(playQueue);
PlayQueue.SetPlayingItemByIndex(playingItemPosition);
var item = _libraryManager.GetItemById(PlayQueue.GetPlayingItemId());
RunTimeTicks = item.RunTimeTicks ?? 0;
PositionTicks = startPositionTicks;
LastActivity = DateTime.UtcNow;
return true;
}
/// <inheritdoc />
public bool SetPlayingItem(Guid playlistItemId)
{
var itemFound = PlayQueue.SetPlayingItemByPlaylistId(playlistItemId);
if (itemFound)
{
var item = _libraryManager.GetItemById(PlayQueue.GetPlayingItemId());
RunTimeTicks = item.RunTimeTicks ?? 0;
}
else
{
RunTimeTicks = 0;
}
RestartCurrentItem();
return itemFound;
}
/// <inheritdoc />
public bool RemoveFromPlayQueue(IReadOnlyList<Guid> playlistItemIds)
{
var playingItemRemoved = PlayQueue.RemoveFromPlaylist(playlistItemIds);
if (playingItemRemoved)
{
var itemId = PlayQueue.GetPlayingItemId();
if (!itemId.Equals(Guid.Empty))
{
var item = _libraryManager.GetItemById(itemId);
RunTimeTicks = item.RunTimeTicks ?? 0;
}
else
{
RunTimeTicks = 0;
}
RestartCurrentItem();
}
return playingItemRemoved;
}
/// <inheritdoc />
public bool MoveItemInPlayQueue(Guid playlistItemId, int newIndex)
{
return PlayQueue.MovePlaylistItem(playlistItemId, newIndex);
}
/// <inheritdoc />
public bool AddToPlayQueue(IReadOnlyList<Guid> newItems, GroupQueueMode mode)
{
// Ignore on empty list.
if (newItems.Count == 0)
{
return false;
}
// Check if participants can access the new playing queue.
if (!AllUsersHaveAccessToQueue(newItems))
{
return false;
}
if (mode.Equals(GroupQueueMode.QueueNext))
{
PlayQueue.QueueNext(newItems);
}
else
{
PlayQueue.Queue(newItems);
}
return true;
}
/// <inheritdoc />
public void RestartCurrentItem()
{
PositionTicks = 0;
LastActivity = DateTime.UtcNow;
}
/// <inheritdoc />
public bool NextItemInQueue()
{
var update = PlayQueue.Next();
if (update)
{
var item = _libraryManager.GetItemById(PlayQueue.GetPlayingItemId());
RunTimeTicks = item.RunTimeTicks ?? 0;
RestartCurrentItem();
return true;
}
else
{
return false;
}
}
/// <inheritdoc />
public bool PreviousItemInQueue()
{
var update = PlayQueue.Previous();
if (update)
{
var item = _libraryManager.GetItemById(PlayQueue.GetPlayingItemId());
RunTimeTicks = item.RunTimeTicks ?? 0;
RestartCurrentItem();
return true;
}
else
{
return false;
}
}
/// <inheritdoc />
public void SetRepeatMode(GroupRepeatMode mode)
{
PlayQueue.SetRepeatMode(mode);
}
/// <inheritdoc />
public void SetShuffleMode(GroupShuffleMode mode)
{
PlayQueue.SetShuffleMode(mode);
}
/// <inheritdoc />
public PlayQueueUpdate GetPlayQueueUpdate(PlayQueueUpdateReason reason)
{
var startPositionTicks = PositionTicks;
if (_state.Type.Equals(GroupStateType.Playing))
{
var currentTime = DateTime.UtcNow;
var elapsedTime = currentTime - LastActivity;
// Elapsed time is negative if event happens
// during the delay added to account for latency.
// In this phase clients haven't started the playback yet.
// In other words, LastActivity is in the future,
// when playback unpause is supposed to happen.
// Adjust ticks only if playback actually started.
startPositionTicks += Math.Max(elapsedTime.Ticks, 0);
}
return new PlayQueueUpdate(
reason,
PlayQueue.LastChange,
PlayQueue.GetPlaylist(),
PlayQueue.PlayingItemIndex,
startPositionTicks,
PlayQueue.ShuffleMode,
PlayQueue.RepeatMode);
}
}
}

514
Emby.Server.Implementations/SyncPlay/SyncPlayController.cs
View File

@@ -1,514 +0,0 @@
using System;
using System.Collections.Generic;
using System.Globalization;
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
{
/// <summary>
/// Class SyncPlayController.
/// </summary>
/// <remarks>
/// Class is not thread-safe, external locking is required when accessing methods.
/// </remarks>
public class SyncPlayController : ISyncPlayController
{
/// <summary>
/// Used to filter the sessions of a group.
/// </summary>
private enum BroadcastType
{
/// <summary>
/// All sessions will receive the message.
/// </summary>
AllGroup = 0,
/// <summary>
/// Only the specified session will receive the message.
/// </summary>
CurrentSession = 1,
/// <summary>
/// All sessions, except the current one, will receive the message.
/// </summary>
AllExceptCurrentSession = 2,
/// <summary>
/// Only sessions that are not buffering will receive the message.
/// </summary>
AllReady = 3
}
/// <summary>
/// The session manager.
/// </summary>
private readonly ISessionManager _sessionManager;
/// <summary>
/// The SyncPlay manager.
/// </summary>
private readonly ISyncPlayManager _syncPlayManager;
/// <summary>
/// The group to manage.
/// </summary>
private readonly GroupInfo _group = new GroupInfo();
/// <summary>
/// Initializes a new instance of the <see cref="SyncPlayController" /> class.
/// </summary>
/// <param name="sessionManager">The session manager.</param>
/// <param name="syncPlayManager">The SyncPlay manager.</param>
public SyncPlayController(
ISessionManager sessionManager,
ISyncPlayManager syncPlayManager)
{
_sessionManager = sessionManager;
_syncPlayManager = syncPlayManager;
}
/// <inheritdoc />
public Guid GetGroupId() => _group.GroupId;
/// <inheritdoc />
public Guid GetPlayingItemId() => _group.PlayingItem.Id;
/// <inheritdoc />
public bool IsGroupEmpty() => _group.IsEmpty();
/// <summary>
/// Converts DateTime to UTC string.
/// </summary>
/// <param name="date">The date to convert.</param>
/// <value>The UTC string.</value>
private string DateToUTCString(DateTime date)
{
return date.ToUniversalTime().ToString("o", CultureInfo.InvariantCulture);
}
/// <summary>
/// Filters sessions of this group.
/// </summary>
/// <param name="from">The current session.</param>
/// <param name="type">The filtering type.</param>
/// <value>The array of sessions matching the filter.</value>
private IEnumerable<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);
case BroadcastType.AllExceptCurrentSession:
return _group.Participants.Values
.Select(session => session.Session)
.Where(session => !session.Id.Equals(from.Id, StringComparison.Ordinal));
case BroadcastType.AllReady:
return _group.Participants.Values
.Where(session => !session.IsBuffering)
.Select(session => session.Session);
default:
return Array.Empty<SessionInfo>();
}
}
/// <summary>
/// Sends a GroupUpdate message to the interested sessions.
/// </summary>
/// <param name="from">The current session.</param>
/// <param name="type">The filtering type.</param>
/// <param name="message">The message to send.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <value>The task.</value>
private Task SendGroupUpdate<T>(SessionInfo from, BroadcastType type, GroupUpdate<T> message, CancellationToken cancellationToken)
{
IEnumerable<Task> GetTasks()
{
foreach (var session in FilterSessions(from, type))
{
yield return _sessionManager.SendSyncPlayGroupUpdate(session.Id, message, cancellationToken);
}
}
return Task.WhenAll(GetTasks());
}
/// <summary>
/// Sends a playback command to the interested sessions.
/// </summary>
/// <param name="from">The current session.</param>
/// <param name="type">The filtering type.</param>
/// <param name="message">The message to send.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <value>The task.</value>
private Task SendCommand(SessionInfo from, BroadcastType type, SendCommand message, CancellationToken cancellationToken)
{
IEnumerable<Task> GetTasks()
{
foreach (var session in FilterSessions(from, type))
{
yield return _sessionManager.SendSyncPlayCommand(session.Id, message, cancellationToken);
}
}
return Task.WhenAll(GetTasks());
}
/// <summary>
/// Builds a new playback command with some default values.
/// </summary>
/// <param name="type">The command type.</param>
/// <value>The SendCommand.</value>
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)
};
}
/// <summary>
/// Builds a new group update message.
/// </summary>
/// <param name="type">The update type.</param>
/// <param name="data">The data to send.</param>
/// <value>The GroupUpdate.</value>
private GroupUpdate<T> NewSyncPlayGroupUpdate<T>(GroupUpdateType type, T data)
{
return new GroupUpdate<T>()
{
GroupId = _group.GroupId.ToString(),
Type = type,
Data = data
};
}
/// <inheritdoc />
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);
}
/// <inheritdoc />
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
{
ItemIds = new Guid[] { _group.PlayingItem.Id },
StartPositionTicks = _group.PositionTicks
};
var update = NewSyncPlayGroupUpdate(GroupUpdateType.PrepareSession, playRequest);
SendGroupUpdate(session, BroadcastType.CurrentSession, update, cancellationToken);
}
}
/// <inheritdoc />
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);
}
/// <inheritdoc />
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;
}
}
/// <summary>
/// Handles a play action requested by a session.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The play action.</param>
/// <param name="cancellationToken">The cancellation token.</param>
private void HandlePlayRequest(SessionInfo session, PlaybackRequest request, CancellationToken cancellationToken)
{
if (_group.IsPaused)
{
// Pick a suitable time that accounts for latency
var delay = Math.Max(_group.GetHighestPing() * 2, GroupInfo.DefaultPing);
// 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);
}
}
/// <summary>
/// Handles a pause action requested by a session.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The pause action.</param>
/// <param name="cancellationToken">The cancellation token.</param>
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);
}
}
/// <summary>
/// Handles a seek action requested by a session.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The seek action.</param>
/// <param name="cancellationToken">The cancellation token.</param>
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);
}
/// <summary>
/// Handles a buffering action requested by a session.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The buffering action.</param>
/// <param name="cancellationToken">The cancellation token.</param>
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);
}
}
/// <summary>
/// Handles a buffering-done action requested by a session.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The buffering-done action.</param>
/// <param name="cancellationToken">The cancellation token.</param>
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 = Math.Max(_group.GetHighestPing() * 2, GroupInfo.DefaultPing);
_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);
}
}
/// <summary>
/// Sanitizes the PositionTicks, considers the current playing item when available.
/// </summary>
/// <param name="positionTicks">The PositionTicks.</param>
/// <value>The sanitized PositionTicks.</value>
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;
}
/// <summary>
/// Updates ping of a session.
/// </summary>
/// <param name="session">The session.</param>
/// <param name="request">The update.</param>
private void HandlePingUpdateRequest(SessionInfo session, PlaybackRequest request)
{
// Collected pings are used to account for network latency when unpausing playback
_group.UpdatePing(session, request.Ping ?? GroupInfo.DefaultPing);
}
/// <inheritdoc />
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()
};
}
}
}

537
Emby.Server.Implementations/SyncPlay/SyncPlayManager.cs
View File

@@ -1,13 +1,11 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Threading;
using Jellyfin.Data.Entities;
using Jellyfin.Data.Enums;
using MediaBrowser.Controller.Library;
using MediaBrowser.Controller.Session;
using MediaBrowser.Controller.SyncPlay;
using MediaBrowser.Controller.SyncPlay.Requests;
using MediaBrowser.Model.SyncPlay;
using Microsoft.Extensions.Logging;
@@ -23,6 +21,11 @@ namespace Emby.Server.Implementations.SyncPlay
/// </summary>
private readonly ILogger<SyncPlayManager> _logger;
/// <summary>
/// The logger factory.
/// </summary>
private readonly ILoggerFactory _loggerFactory;
/// <summary>
/// The user manager.
/// </summary>
@@ -41,18 +44,21 @@ namespace Emby.Server.Implementations.SyncPlay
/// <summary>
/// The map between sessions and groups.
/// </summary>
private readonly Dictionary<string, ISyncPlayController> _sessionToGroupMap =
new Dictionary<string, ISyncPlayController>(StringComparer.OrdinalIgnoreCase);
private readonly ConcurrentDictionary<string, Group> _sessionToGroupMap =
new ConcurrentDictionary<string, Group>(StringComparer.OrdinalIgnoreCase);
/// <summary>
/// The groups.
/// </summary>
private readonly Dictionary<Guid, ISyncPlayController> _groups =
new Dictionary<Guid, ISyncPlayController>();
private readonly ConcurrentDictionary<Guid, Group> _groups =
new ConcurrentDictionary<Guid, Group>();
/// <summary>
/// Lock used for accessing any group.
/// Lock used for accessing multiple groups at once.
/// </summary>
/// <remarks>
/// This lock has priority on locks made on <see cref="Group"/>.
/// </remarks>
private readonly object _groupsLock = new object();
private bool _disposed = false;
@@ -60,31 +66,24 @@ namespace Emby.Server.Implementations.SyncPlay
/// <summary>
/// Initializes a new instance of the <see cref="SyncPlayManager" /> class.
/// </summary>
/// <param name="logger">The logger.</param>
/// <param name="loggerFactory">The logger factory.</param>
/// <param name="userManager">The user manager.</param>
/// <param name="sessionManager">The session manager.</param>
/// <param name="libraryManager">The library manager.</param>
public SyncPlayManager(
ILogger<SyncPlayManager> logger,
ILoggerFactory loggerFactory,
IUserManager userManager,
ISessionManager sessionManager,
ILibraryManager libraryManager)
{
_logger = logger;
_loggerFactory = loggerFactory;
_userManager = userManager;
_sessionManager = sessionManager;
_libraryManager = libraryManager;
_sessionManager.SessionEnded += OnSessionManagerSessionEnded;
_sessionManager.PlaybackStopped += OnSessionManagerPlaybackStopped;
_logger = loggerFactory.CreateLogger<SyncPlayManager>();
_sessionManager.SessionStarted += OnSessionManagerSessionStarted;
}
/// <summary>
/// Gets all groups.
/// </summary>
/// <value>All groups.</value>
public IEnumerable<ISyncPlayController> Groups => _groups.Values;
/// <inheritdoc />
public void Dispose()
{
@@ -92,6 +91,233 @@ namespace Emby.Server.Implementations.SyncPlay
GC.SuppressFinalize(this);
}
/// <inheritdoc />
public void NewGroup(SessionInfo session, NewGroupRequest request, CancellationToken cancellationToken)
{
if (session == null)
{
throw new InvalidOperationException("Session is null!");
}
if (request == null)
{
throw new InvalidOperationException("Request is null!");
}
// Locking required to access list of groups.
lock (_groupsLock)
{
// Make sure that session has not joined another group.
if (_sessionToGroupMap.ContainsKey(session.Id))
{
var leaveGroupRequest = new LeaveGroupRequest();
LeaveGroup(session, leaveGroupRequest, cancellationToken);
}
var group = new Group(_loggerFactory, _userManager, _sessionManager, _libraryManager);
_groups[group.GroupId] = group;
if (!_sessionToGroupMap.TryAdd(session.Id, group))
{
throw new InvalidOperationException("Could not add session to group!");
}
group.CreateGroup(session, request, cancellationToken);
}
}
/// <inheritdoc />
public void JoinGroup(SessionInfo session, JoinGroupRequest request, CancellationToken cancellationToken)
{
if (session == null)
{
throw new InvalidOperationException("Session is null!");
}
if (request == null)
{
throw new InvalidOperationException("Request is null!");
}
var user = _userManager.GetUserById(session.UserId);
// Locking required to access list of groups.
lock (_groupsLock)
{
_groups.TryGetValue(request.GroupId, out Group group);
if (group == null)
{
_logger.LogWarning("Session {SessionId} tried to join group {GroupId} that does not exist.", session.Id, request.GroupId);
var error = new GroupUpdate<string>(Guid.Empty, GroupUpdateType.GroupDoesNotExist, string.Empty);
_sessionManager.SendSyncPlayGroupUpdate(session, error, CancellationToken.None);
return;
}
// Group lock required to let other requests end first.
lock (group)
{
if (!group.HasAccessToPlayQueue(user))
{
_logger.LogWarning("Session {SessionId} tried to join group {GroupId} but does not have access to some content of the playing queue.", session.Id, group.GroupId.ToString());
var error = new GroupUpdate<string>(group.GroupId, GroupUpdateType.LibraryAccessDenied, string.Empty);
_sessionManager.SendSyncPlayGroupUpdate(session, error, CancellationToken.None);
return;
}
if (_sessionToGroupMap.TryGetValue(session.Id, out var existingGroup))
{
if (existingGroup.GroupId.Equals(request.GroupId))
{
// Restore session.
group.SessionJoin(session, request, cancellationToken);
return;
}
var leaveGroupRequest = new LeaveGroupRequest();
LeaveGroup(session, leaveGroupRequest, cancellationToken);
}
if (!_sessionToGroupMap.TryAdd(session.Id, group))
{
throw new InvalidOperationException("Could not add session to group!");
}
group.SessionJoin(session, request, cancellationToken);
}
}
}
/// <inheritdoc />
public void LeaveGroup(SessionInfo session, LeaveGroupRequest request, CancellationToken cancellationToken)
{
if (session == null)
{
throw new InvalidOperationException("Session is null!");
}
if (request == null)
{
throw new InvalidOperationException("Request is null!");
}
// Locking required to access list of groups.
lock (_groupsLock)
{
if (_sessionToGroupMap.TryGetValue(session.Id, out var group))
{
// Group lock required to let other requests end first.
lock (group)
{
if (_sessionToGroupMap.TryRemove(session.Id, out var tempGroup))
{
if (!tempGroup.GroupId.Equals(group.GroupId))
{
throw new InvalidOperationException("Session was in wrong group!");
}
}
else
{
throw new InvalidOperationException("Could not remove session from group!");
}
group.SessionLeave(session, request, cancellationToken);
if (group.IsGroupEmpty())
{
_logger.LogInformation("Group {GroupId} is empty, removing it.", group.GroupId);
_groups.Remove(group.GroupId, out _);
}
}
}
else
{
_logger.LogWarning("Session {SessionId} does not belong to any group.", session.Id);
var error = new GroupUpdate<string>(Guid.Empty, GroupUpdateType.NotInGroup, string.Empty);
_sessionManager.SendSyncPlayGroupUpdate(session, error, CancellationToken.None);
return;
}
}
}
/// <inheritdoc />
public List<GroupInfoDto> ListGroups(SessionInfo session, ListGroupsRequest request)
{
if (session == null)
{
throw new InvalidOperationException("Session is null!");
}
if (request == null)
{
throw new InvalidOperationException("Request is null!");
}
var user = _userManager.GetUserById(session.UserId);
List<GroupInfoDto> list = new List<GroupInfoDto>();
foreach (var group in _groups.Values)
{
// Locking required as group is not thread-safe.
lock (group)
{
if (group.HasAccessToPlayQueue(user))
{
list.Add(group.GetInfo());
}
}
}
return list;
}
/// <inheritdoc />
public void HandleRequest(SessionInfo session, IGroupPlaybackRequest request, CancellationToken cancellationToken)
{
if (session == null)
{
throw new InvalidOperationException("Session is null!");
}
if (request == null)
{
throw new InvalidOperationException("Request is null!");
}
if (_sessionToGroupMap.TryGetValue(session.Id, out var group))
{
// Group lock required as Group is not thread-safe.
lock (group)
{
// Make sure that session still belongs to this group.
if (_sessionToGroupMap.TryGetValue(session.Id, out var checkGroup) && !checkGroup.GroupId.Equals(group.GroupId))
{
// Drop request.
return;
}
// Drop request if group is empty.
if (group.IsGroupEmpty())
{
return;
}
// Apply requested changes to group.
group.HandleRequest(session, request, cancellationToken);
}
}
else
{
_logger.LogWarning("Session {SessionId} does not belong to any group.", session.Id);
var error = new GroupUpdate<string>(Guid.Empty, GroupUpdateType.NotInGroup, string.Empty);
_sessionManager.SendSyncPlayGroupUpdate(session, error, CancellationToken.None);
}
}
/// <summary>
/// Releases unmanaged and optionally managed resources.
/// </summary>
@@ -103,275 +329,18 @@ namespace Emby.Server.Implementations.SyncPlay
return;
}
_sessionManager.SessionEnded -= OnSessionManagerSessionEnded;
_sessionManager.PlaybackStopped -= OnSessionManagerPlaybackStopped;
_sessionManager.SessionStarted -= OnSessionManagerSessionStarted;
_disposed = true;
}
private void OnSessionManagerSessionEnded(object sender, SessionEventArgs e)
private void OnSessionManagerSessionStarted(object sender, SessionEventArgs e)
{
var session = e.SessionInfo;
if (!IsSessionInGroup(session))
if (_sessionToGroupMap.TryGetValue(session.Id, out var group))
{
return;
}
LeaveGroup(session, CancellationToken.None);
}
private void OnSessionManagerPlaybackStopped(object sender, PlaybackStopEventArgs e)
{
var session = e.Session;
if (!IsSessionInGroup(session))
{
return;
}
LeaveGroup(session, CancellationToken.None);
}
private bool IsSessionInGroup(SessionInfo session)
{
return _sessionToGroupMap.ContainsKey(session.Id);
}
private bool HasAccessToItem(User user, Guid itemId)
{
var item = _libraryManager.GetItemById(itemId);
// Check ParentalRating access
var hasParentalRatingAccess = !user.MaxParentalAgeRating.HasValue
|| item.InheritedParentalRatingValue <= user.MaxParentalAgeRating;
if (!user.HasPermission(PermissionKind.EnableAllFolders) && hasParentalRatingAccess)
{
var collections = _libraryManager.GetCollectionFolders(item).Select(
folder => folder.Id.ToString("N", CultureInfo.InvariantCulture));
return collections.Intersect(user.GetPreference(PreferenceKind.EnabledFolders)).Any();
}
return hasParentalRatingAccess;
}
private Guid? GetSessionGroup(SessionInfo session)
{
_sessionToGroupMap.TryGetValue(session.Id, out var group);
return group?.GetGroupId();
}
/// <inheritdoc />
public void NewGroup(SessionInfo session, CancellationToken cancellationToken)
{
var user = _userManager.GetUserById(session.UserId);
if (user.SyncPlayAccess != SyncPlayAccess.CreateAndJoinGroups)
{
_logger.LogWarning("NewGroup: {0} does not have permission to create groups.", session.Id);
var error = new GroupUpdate<string>
{
Type = GroupUpdateType.CreateGroupDenied
};
_sessionManager.SendSyncPlayGroupUpdate(session.Id, error, CancellationToken.None);
return;
}
lock (_groupsLock)
{
if (IsSessionInGroup(session))
{
LeaveGroup(session, cancellationToken);
}
var group = new SyncPlayController(_sessionManager, this);
_groups[group.GetGroupId()] = group;
group.CreateGroup(session, cancellationToken);
}
}
/// <inheritdoc />
public void JoinGroup(SessionInfo session, Guid groupId, JoinGroupRequest request, CancellationToken cancellationToken)
{
var user = _userManager.GetUserById(session.UserId);
if (user.SyncPlayAccess == SyncPlayAccess.None)
{
_logger.LogWarning("JoinGroup: {0} does not have access to SyncPlay.", session.Id);
var error = new GroupUpdate<string>()
{
Type = GroupUpdateType.JoinGroupDenied
};
_sessionManager.SendSyncPlayGroupUpdate(session.Id, error, CancellationToken.None);
return;
}
lock (_groupsLock)
{
ISyncPlayController group;
_groups.TryGetValue(groupId, out group);
if (group == null)
{
_logger.LogWarning("JoinGroup: {0} tried to join group {0} that does not exist.", session.Id, groupId);
var error = new GroupUpdate<string>()
{
Type = GroupUpdateType.GroupDoesNotExist
};
_sessionManager.SendSyncPlayGroupUpdate(session.Id, error, CancellationToken.None);
return;
}
if (!HasAccessToItem(user, group.GetPlayingItemId()))
{
_logger.LogWarning("JoinGroup: {0} does not have access to {1}.", session.Id, group.GetPlayingItemId());
var error = new GroupUpdate<string>()
{
GroupId = group.GetGroupId().ToString(),
Type = GroupUpdateType.LibraryAccessDenied
};
_sessionManager.SendSyncPlayGroupUpdate(session.Id, error, CancellationToken.None);
return;
}
if (IsSessionInGroup(session))
{
if (GetSessionGroup(session).Equals(groupId))
{
return;
}
LeaveGroup(session, cancellationToken);
}
group.SessionJoin(session, request, cancellationToken);
}
}
/// <inheritdoc />
public void LeaveGroup(SessionInfo session, CancellationToken cancellationToken)
{
// TODO: determine what happens to users that are in a group and get their permissions revoked
lock (_groupsLock)
{
_sessionToGroupMap.TryGetValue(session.Id, out var group);
if (group == null)
{
_logger.LogWarning("LeaveGroup: {0} does not belong to any group.", session.Id);
var error = new GroupUpdate<string>()
{
Type = GroupUpdateType.NotInGroup
};
_sessionManager.SendSyncPlayGroupUpdate(session.Id, error, CancellationToken.None);
return;
}
group.SessionLeave(session, cancellationToken);
if (group.IsGroupEmpty())
{
_logger.LogInformation("LeaveGroup: removing empty group {0}.", group.GetGroupId());
_groups.Remove(group.GetGroupId(), out _);
}
}
}
/// <inheritdoc />
public List<GroupInfoView> ListGroups(SessionInfo session, Guid filterItemId)
{
var user = _userManager.GetUserById(session.UserId);
if (user.SyncPlayAccess == SyncPlayAccess.None)
{
return new List<GroupInfoView>();
}
// Filter by item if requested
if (!filterItemId.Equals(Guid.Empty))
{
return _groups.Values.Where(
group => group.GetPlayingItemId().Equals(filterItemId) && HasAccessToItem(user, group.GetPlayingItemId())).Select(
group => group.GetInfo()).ToList();
}
else
{
// Otherwise show all available groups
return _groups.Values.Where(
group => HasAccessToItem(user, group.GetPlayingItemId())).Select(
group => group.GetInfo()).ToList();
}
}
/// <inheritdoc />
public void HandleRequest(SessionInfo session, PlaybackRequest request, CancellationToken cancellationToken)
{
var user = _userManager.GetUserById(session.UserId);
if (user.SyncPlayAccess == SyncPlayAccess.None)
{
_logger.LogWarning("HandleRequest: {0} does not have access to SyncPlay.", session.Id);
var error = new GroupUpdate<string>()
{
Type = GroupUpdateType.JoinGroupDenied
};
_sessionManager.SendSyncPlayGroupUpdate(session.Id, error, CancellationToken.None);
return;
}
lock (_groupsLock)
{
_sessionToGroupMap.TryGetValue(session.Id, out var group);
if (group == null)
{
_logger.LogWarning("HandleRequest: {0} does not belong to any group.", session.Id);
var error = new GroupUpdate<string>()
{
Type = GroupUpdateType.NotInGroup
};
_sessionManager.SendSyncPlayGroupUpdate(session.Id, error, CancellationToken.None);
return;
}
group.HandleRequest(session, request, cancellationToken);
}
}
/// <inheritdoc />
public void AddSessionToGroup(SessionInfo session, ISyncPlayController group)
{
if (IsSessionInGroup(session))
{
throw new InvalidOperationException("Session in other group already!");
}
_sessionToGroupMap[session.Id] = group;
}
/// <inheritdoc />
public void RemoveSessionFromGroup(SessionInfo session, ISyncPlayController group)
{
if (!IsSessionInGroup(session))
{
throw new InvalidOperationException("Session not in any group!");
}
_sessionToGroupMap.Remove(session.Id, out var tempGroup);
if (!tempGroup.GetGroupId().Equals(group.GetGroupId()))
{
throw new InvalidOperationException("Session was in wrong group!");
var request = new JoinGroupRequest(group.GroupId);
JoinGroup(session, request, CancellationToken.None);
}
}
}