Real-time коммуникация клиент-сервер с SignalR и C#

public abstract class PersistentConnection {
    protected virtual Task OnReceived(IRequest request, string connectionId, string data);
    protected virtual Task OnConnected(IRequest request, string connectionId);
    protected virtual Task OnDisconnected(IRequest request, string connectionId, bool stopCalled);
    // Другие методы...
}

// Настройка стратегии переподключения для .NET клиента
var connection = new HubConnectionBuilder()
    .WithUrl("https://example.com/chathub")
    .WithAutomaticReconnect(new[] { TimeSpan.Zero, TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(30) })
    .Build();

public class ChatHub : Hub
{
    public async Task SendMessage(string user, string message)
    {
        await Clients.All.SendAsync("ReceiveMessage", user, message);
    }
}

// Отправка конкретному клиенту
await Clients.Client(connectionId).SendAsync("ReceiveMessage", message);
 
// Отправка всем, кроме вызывающего клиента
await Clients.Others.SendAsync("UserJoined", Context.ConnectionId);
 
// Отправка группе клиентов
await Clients.Group("Админы").SendAsync("AdminAlert", alert);

public class StockUpdate
{
    public string Symbol { get; set; }
    public decimal Price { get; set; }
    public decimal Change { get; set; }
}
 
public async Task BroadcastStockPrice(StockUpdate update)
{
    await Clients.All.SendAsync("UpdateStockPrice", update);
}

// Принудительное использование только Long Polling на стороне сервера
services.AddSignalR().AddHubOptions<ChatHub>(options =>
{
    options.AllowedTransports = HttpTransportType.LongPolling;
});
 
// То же самое на стороне JavaScript-клиента
const connection = new signalR.HubConnectionBuilder()
    .withUrl("/chatHub", { transport: signalR.HttpTransportType.LongPolling })
    .build();

// Добавление клиента в группу по имени пользователя
public override async Task OnConnectedAsync()
{
    var username = Context.User.Identity.Name;
    await Groups.AddToGroupAsync(Context.ConnectionId, username);
    await base.OnConnectedAsync();
}
 
// Отправка сообщения группе (только одна копия на пользователя)
public async Task SendPrivateMessage(string user, string message)
{
    await Clients.Group(user).SendAsync("ReceiveMessage", message);
}

// Защита хаба аутентификацией
[Authorize]
public class SecureHub : Hub
{
    public async Task AdminOperation()
    {
        // Проверка роли пользователя
        if (Context.User.IsInRole("Admin"))
        {
            await Clients.All.SendAsync("AdminActionCompleted");
        }
        else
        {
            throw new HubException("У вас нет прав администратора");
        }
    }
}

private static ConcurrentDictionary<string, Timer> _disconnectionTimers = new ConcurrentDictionary<string, Timer>();
 
public override async Task OnDisconnectedAsync(Exception exception)
{
    var username = Context.User.Identity.Name;
    
    // Запускаем таймер на 30 секунд
    _disconnectionTimers.TryAdd(username, new Timer(async state =>
    {
        // Если таймер сработал, пользователь действительно оффлайн
        await Clients.All.SendAsync("UserWentOffline", username);
        Timer t;
        _disconnectionTimers.TryRemove(username, out t);
    }, null, TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(-1)));
    
    await base.OnDisconnectedAsync(exception);
}
 
public override async Task OnConnectedAsync()
{
    var username = Context.User.Identity.Name;
    
    // Если пользователь переподключился, отменяем таймер
    Timer timer;
    if (_disconnectionTimers.TryRemove(username, out timer))
    {
        timer.Dispose();
    }
    else
    {
        // Если таймера нет, значит это новое подключение
        await Clients.All.SendAsync("UserCameOnline", username);
    }
    
    await base.OnConnectedAsync();
}

public override async Task OnConnectedAsync()
{
    var connectionId = Context.ConnectionId;
    _logger.LogInformation($"Клиент {connectionId} подключился. Время: {DateTime.Now}");
    
    // Сохраняем информацию о подключении
    _connections.Add(connectionId, new ConnectionInfo { ConnectedAt = DateTime.Now });
    
    await base.OnConnectedAsync();
}

services.AddSignalR(hubOptions =>
{
    hubOptions.KeepAliveInterval = TimeSpan.FromSeconds(10);  // Интервал пингов
    hubOptions.ClientTimeoutInterval = TimeSpan.FromSeconds(30);  // Таймаут клиента
});

public override async Task OnDisconnectedAsync(Exception exception)
{
    var connectionId = Context.ConnectionId;
    
    if (exception != null)
    {
        _logger.LogWarning($"Клиент {connectionId} отключился с ошибкой: {exception.Message}");
    }
    else
    {
        _logger.LogInformation($"Клиент {connectionId} корректно завершил соединение");
    }
    
    // Очистка ресурсов
    _connections.Remove(connectionId);
    
    await base.OnDisconnectedAsync(exception);
}

var connection = new HubConnectionBuilder()
    .WithUrl("/chatHub")
    .WithAutomaticReconnect()  // Магическая строчка!
    .Build();

.WithAutomaticReconnect(new[] { TimeSpan.Zero, TimeSpan.FromSeconds(3), TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(10) })

public abstract class LoggingHub : Hub
{
  private readonly ILogger _logger;
  
  protected LoggingHub(ILogger logger)
  {
      _logger = logger;
  }
  
  public override async Task OnConnectedAsync()
  {
      _logger.LogInformation($"Подключение: {Context.ConnectionId} в {DateTime.Now}");
      await base.OnConnectedAsync();
  }
  
  public override async Task OnDisconnectedAsync(Exception exception)
  {
      _logger.LogInformation($"Отключение: {Context.ConnectionId} в {DateTime.Now}. Причина: {exception?.Message ?? "Штатное отключение"}");
      await base.OnDisconnectedAsync(exception);
  }
  
  // Расширенный метод для отправки сообщений с логированием
  protected async Task SendWithLogging(string method, object arg)
  {
      _logger.LogDebug($"Отправка {method} для {Context.ConnectionId}");
      await Clients.Caller.SendAsync(method, arg);
  }
}

// Регистрация middleware для хаба
public void Configure(IApplicationBuilder app)
{
  app.UseSignalR(routes =>
  {
      routes.MapHub<ChatHub>("/chatHub", options =>
      {
          // Добавляем кастомное middleware
          options.Transports = HttpTransportType.WebSockets;
          
          // Перехват сообщений перед отправкой клиенту
          var pipeline = routes.HubPipeline;
          pipeline.AddMiddleware(async (context, next) =>
          {
              // До обработки сообщения
              var originalMessage = context.Message;
              
              // Вызов следующего middleware в цепочке
              await next();
              
              // После обработки сообщения
              context.Result = SanitizeResult(context.Result);
          });
      });
  });
}

pipeline.AddMiddleware(async (context, next) =>
{
  await next();
  
  // Применяем сжатие только к большим сообщениям
  if (context.Result is string textResult && textResult.Length > 1000)
  {
      context.Result = CompressString(textResult);
      // Добавляем флаг, чтобы клиент знал о необходимости распаковки
      context.Headers["X-Compressed"] = "true";
  }
});

[AttributeUsage(AttributeTargets.Class | AttributeTargets.Method)]
public class AuditLogAttribute : HubFilterAttribute
{
  public override async ValueTask<object> InvokeMethodAsync(HubInvocationContext context, Func<HubInvocationContext, ValueTask<object>> next)
  {
      var logger = context.ServiceProvider.GetService<ILogger>();
      logger.LogInformation($"Вызов метода {context.HubMethodName} пользователем {context.Context.User?.Identity?.Name}");
      
      return await next(context);
  }
}
 
// Применение атрибута
[AuditLog]
public class AdminHub : Hub
{
  // Методы хаба...
}

// В .NET 6+ (в Program.cs)
builder.Services.AddSignalR();
 
// ИЛИ в более старых версиях (в Startup.cs)
public void ConfigureServices(IServiceCollection services)
{
    services.AddSignalR();
}

// В .NET 6+ (в Program.cs)
app.MapHub<ChatHub>("/chathub");
 
// ИЛИ в более старых версиях (в Startup.cs)
app.UseEndpoints(endpoints =>
{
    endpoints.MapHub<ChatHub>("/chathub");
});

// Hubs/ChatHub.cs
using Microsoft.AspNetCore.SignalR;
 
public class ChatHub : Hub
{
    public async Task SendMessage(string user, string message)
    {
        await Clients.All.SendAsync("ReceiveMessage", user, message);
    }
}

services.AddSignalR(options => 
{
    options.MaximumReceiveMessageSize = 102400; // 100 КБ
    options.StreamBufferCapacity = 20;
    options.HandshakeTimeout = TimeSpan.FromSeconds(15);
})

services.AddSignalR(options => 
{
    options.KeepAliveInterval = TimeSpan.FromMinutes(2); // Пинг каждые 2 минуты
    options.ClientTimeoutInterval = TimeSpan.FromMinutes(4); // Таймаут через 4 минуты
})

public class NotificationHub : Hub
{
    private readonly INotificationService _notificationService;
    private readonly IUserRepository _userRepository;
    
    public NotificationHub(
        INotificationService notificationService,
        IUserRepository userRepository)
    {
        _notificationService = notificationService;
        _userRepository = userRepository;
    }
    
    // Методы хаба...
}

// В appsettings.json
{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft": "Warning",
      "Microsoft.AspNetCore.SignalR": "Debug",
      "Microsoft.AspNetCore.Http.Connections": "Debug"
    }
  }
}

services.AddResponseCompression(options =>
{
    options.MimeTypes = ResponseCompressionDefaults.MimeTypes.Concat(
        new[] { "application/octet-stream" });
});
 
// Важно добавить до настройки SignalR!
app.UseResponseCompression();

services.AddSignalR(options =>
{
    // Максимальный размер входящего сообщения (по умолчанию 32KB)
    options.MaximumReceiveMessageSize = 1024 * 1024; // 1MB
    
    // Таймаут рукопожатия при установке соединения
    options.HandshakeTimeout = TimeSpan.FromSeconds(15);
    
    // Интервал отправки пингов для проверки активности соединения
    options.KeepAliveInterval = TimeSpan.FromSeconds(10);
    
    // Таймаут на клиентскую активность
    options.ClientTimeoutInterval = TimeSpan.FromSeconds(30);
    
    // Включение детального логирования
    options.EnableDetailedErrors = true;
    
    // Ограничение размера буфера для стримов
    options.StreamBufferCapacity = 20;
});

services.AddSignalR()
    .AddMessagePackProtocol(options =>
    {
        options.FormatterResolvers = new List<MessagePack.IFormatterResolver>
        {
            MessagePack.Resolvers.StandardResolver.Instance
        };
    });

app.MapHub<ChatHub>("/chatHub");

app.MapHub<ChatHub>("/chatHub", options =>
{
    // Ограничение транспортов только до WebSockets
    options.Transports = HttpTransportType.WebSockets;
    
    // Требование авторизации
    options.RequireAuthorization();
    
    // Настройка CORS для хаба
    options.RequireCors(builder => builder
        .WithOrigins("https://example.com")
        .AllowCredentials());
    
    // Настройка таймаутов на уровне HTTP-соединения
    options.WebSockets.CloseTimeout = TimeSpan.FromSeconds(5);
});

// Глобальное требование авторизации для всех хабов
services.AddSignalR().RequireAuthorization();
 
// ИЛИ с именованной политикой
services.AddSignalR().RequireAuthorization("SignalRPolicy");
 
// Определение политики
services.AddAuthorization(options =>
{
    options.AddPolicy("SignalRPolicy", policy =>
    {
        policy.RequireAuthenticatedUser()
              .RequireRole("SignalRUser");
    });
});

services.AddResponseCompression(options =>
{
    options.MimeTypes = ResponseCompressionDefaults.MimeTypes.Concat(
        new[] { "application/octet-stream" });
    
    // Выбор алгоритма сжатия
    options.Providers.Add<BrotliCompressionProvider>();
    options.Providers.Add<GzipCompressionProvider>();
});
 
// Важно добавить до вызова UseRouting() и UseEndpoints()
app.UseResponseCompression();

app.Use(async (context, next) =>
{
    if (context.Request.Path.StartsWithSegments("/chatHub"))
    {
        // Кастомная логика для запросов к хабу
        context.Response.Headers.Add("X-SignalR-Custom", "Value");
    }
    
    await next.Invoke();
});

services.AddHealthChecks()
    .AddCheck<SignalRHealthCheck>("SignalR");
 
// Реализация проверки
public class SignalRHealthCheck : IHealthCheck
{
    private readonly IHubContext<ChatHub> _hubContext;
    
    public SignalRHealthCheck(IHubContext<ChatHub> hubContext)
    {
        _hubContext = hubContext;
    }
    
    public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken token)
    {
        // Проверка работоспособности хаба
        return Task.FromResult(HealthCheckResult.Healthy());
    }
}

// C# клиент для .NET MAUI, Xamarin, WPF, и т.д.
private HubConnection _connection;
 
public async Task InitializeConnection()
{
    _connection = new HubConnectionBuilder()
        .WithUrl("https://myserver.com/chathub", options =>
        {
            // Настройка аутентификации
            options.AccessTokenProvider = () => Task.FromResult(_tokenService.GetToken());
            
            // Настройка таймаутов
            options.CloseTimeout = TimeSpan.FromSeconds(15);
        })
        .WithAutomaticReconnect()
        .Build();
        
    // Настройка обработчиков событий и запуск соединения
    // ...
}
 
// Обработка жизненного цикла приложения
protected override void OnSleep()
{
    // В зависимости от требований:
    // 1. Оставить соединение (для важных уведомлений)
    // 2. Закрыть соединение для экономии батареи
    _connection.StopAsync();
    
    base.OnSleep();
}
 
protected override void OnResume()
{
    _connection.StartAsync();
    base.OnResume();
}

_connection.On<string, string>("ReceiveMessage", (user, message) => {
    // Переключаемся на UI-поток
    Application.Current.Dispatcher.Invoke(() => {
        chatMessages.Add(new ChatMessage(user, message));
    });
});

public override Task OnConnectedAsync()
{
    try
    {
        // Инициализация для нового клиента
        return base.OnConnectedAsync();
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "Ошибка при подключении клиента {ConnectionId}", Context.ConnectionId);
        throw; // Важно пробросить исключение, чтобы клиент получил уведомление
    }
}

app.UseSignalR(routes =>
{
    routes.MapHub<ChatHub>("/chatHub");
    
    // Глобальный обработчик исключений
    var hubPipeline = routes.GetType().GetField("HubPipeline", 
        BindingFlags.NonPublic | BindingFlags.Instance).GetValue(routes);
        
    hubPipeline.GetType().GetMethod("AddOutgoingMiddleware").Invoke(
        hubPipeline, new object[] { 
            new Func<HubMessage, Func<HubMessage, Task>, Task>(
                async (message, next) => {
                    if (message is InvocationResultMessage resultMessage && 
                        resultMessage.Error != null)
                    {
                        _logger.LogError("Ошибка при выполнении метода хаба: {Error}", 
                            resultMessage.Error);
                    }
                    await next(message);
                })
        });
});

// На сервере
services.AddSignalR()
    .AddMessagePackProtocol();
 
// На JavaScript-клиенте
const connection = new signalR.HubConnectionBuilder()
    .withUrl("/myhub")
    .withHubProtocol(new signalR.protocols.msgpack.MessagePackHubProtocol())
    .build();

// До оптимизации
public class StockUpdate
{
    public string CompanyName { get; set; }
    public decimal CurrentPrice { get; set; }
    public decimal PriceChange { get; set; }
    public DateTime UpdateTime { get; set; }
    public string CurrencyCode { get; set; }
    // И т.д.
}
 
// После оптимизации
[MessagePackObject]
public class StockUpdate
{
    [Key(0)]
    public string c { get; set; } // CompanyName
 
    [Key(1)]
    public decimal p { get; set; } // CurrentPrice
 
    [Key(2)]
    public decimal d { get; set; } // PriceChange
 
    // UpdateTime и CurrencyCode отправляются отдельно или вообще исключаются
}

services.AddResponseCompression();
app.UseResponseCompression(); // Важно: до UseSignalR!

// Настройка детального логирования для SignalR
services.AddLogging(builder =>
{
  builder.AddFilter("Microsoft.AspNetCore.SignalR", LogLevel.Debug);
  builder.AddFilter("Microsoft.AspNetCore.Http.Connections", LogLevel.Debug);
});

public async Task SendMessage(string user, string message)
{
  _logger.LogInformation("Сообщение от {User}: {@Message}", user, message);
  await Clients.All.SendAsync("ReceiveMessage", user, message);
}

public class SignalRStats
{
  private static readonly ConcurrentDictionary<string, DateTime> _connections = 
      new ConcurrentDictionary<string, DateTime>();
  
  public static void AddConnection(string connectionId)
  {
      _connections[connectionId] = DateTime.UtcNow;
  }
  
  public static void RemoveConnection(string connectionId)
  {
      _connections.TryRemove(connectionId, out _);
  }
  
  public static SignalRStatsDto GetStats()
  {
      return new SignalRStatsDto
      {
          ConnectionCount = _connections.Count,
          OldestConnection = _connections.Any() ? 
              _connections.Values.Min() : DateTime.MinValue
      };
  }
}

// В Program.cs или Startup.cs
services.AddSignalR().AddHubOptions<ChatHub>(options =>
{
    options.EnableDetailedErrors = true; // Критически важно для отладки!
});

public async Task SendComplexObject(MyDto data)
{
    try {
        _logger.LogDebug("Начало отправки объекта: {@Data}", data);
        await Clients.All.SendAsync("ReceiveData", data);
        _logger.LogDebug("Объект успешно отправлен");
    }
    catch (Exception ex) {
        _logger.LogError(ex, "Ошибка при отправке объекта: {@Data}", data);
        throw;
    }
}

// Инжектируем IDistributedCache
private readonly IDistributedCache _cache;
 
public async Task GetLatestAlerts()
{
  // Проверяем кэш перед генерацией данных
  string cachedData = await _cache.GetStringAsync("latest_alerts");
  if (cachedData != null)
  {
      // Отправляем кэшированные данные
      await Clients.Caller.SendAsync("ReceiveAlerts", cachedData);
      return;
  }
  
  // Если в кэше нет, генерируем и сохраняем
  var alerts = await _alertService.GetLatestAlertsAsync();
  string serialized = JsonSerializer.Serialize(alerts);
  
  // Сохраняем в кэш на 30 секунд
  await _cache.SetStringAsync("latest_alerts", serialized, 
      new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = TimeSpan.FromSeconds(30) });
  
  await Clients.Caller.SendAsync("ReceiveAlerts", serialized);
}

// На клиенте храним идентификатор последней версии
let lastVersionId = null;
 
// На сервере
public async Task GetUpdates(string lastVersionId)
{
  if (lastVersionId == _currentVersionId)
  {
      // Данные не изменились, ничего не отправляем
      return;
  }
  
  var updates = lastVersionId == null ? 
      _dataStore.GetFullData() : // Полные данные для новых клиентов
      _dataStore.GetChangesSince(lastVersionId); // Только изменения
      
  await Clients.Caller.SendAsync("ReceiveUpdates", _currentVersionId, updates);
}

public async Task SendLargeData(string clientId, byte[] data)
{
  // Сжимаем данные перед отправкой
  using var memoryStream = new MemoryStream();
  using (var gzipStream = new GZipStream(memoryStream, CompressionLevel.Fastest))
  {
      gzipStream.Write(data, 0, data.Length);
  }
  
  byte[] compressed = memoryStream.ToArray();
  // Отправляем сжатые данные с флагом
  await Clients.Client(clientId).SendAsync("ReceiveCompressedData", compressed, true);
}

private static SemaphoreSlim _throttler = new SemaphoreSlim(100); // Максимум 100 одновременных операций
 
public async Task SendBroadcast(string message)
{
    // Пытаемся получить доступ к семафору с таймаутом
    bool entered = await _throttler.WaitAsync(TimeSpan.FromSeconds(5));
    
    if (!entered)
    {
        // Не смогли получить разрешение - сообщаем клиенту
        throw new HubException("Сервер перегружен, попробуйте позже");
    }
    
    try
    {
        // Обрабатываем сообщение с задержкой, если необходимо
        await Task.Delay(CalculateDelay());
        await Clients.All.SendAsync("ReceiveMessage", message);
    }
    finally
    {
        // Важно! Всегда освобождаем семафор
        _throttler.Release();
    }
}

private class TokenBucket
{
    private int _tokens;
    private readonly int _capacity;
    private readonly Timer _timer;
    
    public TokenBucket(int capacity, TimeSpan refillInterval)
    {
        _capacity = capacity;
        _tokens = capacity;
        _timer = new Timer(_ => Refill(), null, refillInterval, refillInterval);
    }
    
    private void Refill()
    {
        Interlocked.Exchange(ref _tokens, _capacity);
    }
    
    public bool TryConsume()
    {
        return Interlocked.Decrement(ref _tokens) >= 0;
    }
}

private TimeSpan GetUpdateInterval()
{
    int connections = _connectionManager.ActiveConnections;
    
    // Динамически меняем интервал в зависимости от нагрузки
    if (connections > 1000) return TimeSpan.FromSeconds(10);
    if (connections > 500) return TimeSpan.FromSeconds(5);
    return TimeSpan.FromSeconds(1);
}

public class MarketDataHub : Hub
{
  private readonly IStockService _stockService;
  
  public MarketDataHub(IStockService stockService)
  {
      _stockService = stockService;
  }
  
  public async Task SubscribeToSymbol(string symbol)
  {
      // Добавляем клиента в группу для конкретного символа
      await Groups.AddToGroupAsync(Context.ConnectionId, symbol);
      
      // Отправляем текущие данные сразу после подписки
      var snapshot = await _stockService.GetSymbolSnapshotAsync(symbol);
      await Clients.Caller.SendAsync("ReceiveSnapshot", symbol, snapshot);
  }
  
  // Метод вызывается из фоновой службы при изменении цены
  public async Task BroadcastPriceUpdate(string symbol, decimal price, decimal change)
  {
      // Отправляем обновление только подписчикам конкретного символа
      await Clients.Group(symbol).SendAsync("PriceUpdate", symbol, price, change);
  }
}

public async Task MakeMove(string gameId, int x, int y)
{
  // Проверка валидности хода
  var moveResult = await _gameService.ProcessMoveAsync(gameId, Context.ConnectionId, x, y);
  
  if (moveResult.IsValid)
  {
      // Отправка хода всем игрокам в этой игровой сессии
      await Clients.Group(gameId).SendAsync("MoveMade", Context.ConnectionId, x, y, moveResult.GameState);
      
      // Если ход завершил игру
      if (moveResult.IsGameOver)
      {
          await Clients.Group(gameId).SendAsync("GameOver", moveResult.Winner);
      }
  }
  else
  {
      // Сообщаем только текущему игроку об ошибке
      await Clients.Caller.SendAsync("InvalidMove", moveResult.ErrorMessage);
  }
}

public class LocationHub : Hub
{
    private readonly IVehicleRepository _repository;
    
    public LocationHub(IVehicleRepository repository)
    {
        _repository = repository;
    }
    
    public async Task UpdateLocation(string vehicleId, double latitude, double longitude, double speed)
    {
        // Сохраняем позицию в базе данных
        await _repository.UpdateVehiclePositionAsync(vehicleId, latitude, longitude, speed);
        
        // Определяем, кому отправлять обновление
        var supervisorId = await _repository.GetSupervisorForVehicleAsync(vehicleId);
        if (!string.IsNullOrEmpty(supervisorId))
        {
            // Отправляем обновление только ответственному диспетчеру
            await Clients.User(supervisorId).SendAsync("VehicleMoved", vehicleId, latitude, longitude, speed);
        }
        
        // Проверяем геозоны
        var zones = await _repository.CheckGeofencesAsync(vehicleId, latitude, longitude);
        if (zones.Any())
        {
            foreach (var zone in zones)
            {
                await Clients.User(supervisorId).SendAsync("GeofenceAlert", vehicleId, zone.Name, zone.Type);
            }
        }
    }
}

public async Task PlaceBid(int auctionId, decimal amount)
{
    // Проверка валидности ставки
    var bidResult = await _auctionService.ProcessBidAsync(auctionId, Context.ConnectionId, amount);
    
    if (bidResult.IsValid)
    {
        // Транслируем новую ставку всем участникам аукциона
        await Clients.Group($"auction_{auctionId}").SendAsync("NewBid", bidResult.BidderName, amount, bidResult.ServerTimestamp);
        
        // Если ставка сделана в последнюю минуту, продлеваем аукцион
        if (bidResult.IsExtended)
        {
            await Clients.Group($"auction_{auctionId}").SendAsync("AuctionExtended", bidResult.NewEndTime);
        }
    }
    else
    {
        // Сообщаем только текущему пользователю об ошибке
        await Clients.Caller.SendAsync("BidRejected", bidResult.ErrorMessage);
    }
}

public async Task StartConsultation(string patientId)
{
    // Проверяем, доступен ли пациент
    if (await _patientService.IsPatientOnlineAsync(patientId))
    {
        // Создаем защищенную комнату для консультации
        var roomId = await _roomService.CreateSecureRoomAsync(Context.ConnectionId, patientId);
        
        // Отправляем приглашение пациенту
        await Clients.User(patientId).SendAsync("ConsultationInvite", roomId, Context.User.Identity.Name);
        
        // Уведомляем врача
        await Clients.Caller.SendAsync("PatientNotified", patientId);
    }
    else
    {
        await Clients.Caller.SendAsync("PatientUnavailable", patientId);
    }
}

// Подключение Redis Backplane в ASP.NET Core
services.AddSignalR()
  .AddStackExchangeRedis("redis_connection_string", options =>
  {
      options.Configuration.ChannelPrefix = "MyApp";
      options.Configuration.DefaultDatabase = 5;
      
      // Настройка обработки ошибок
      options.ConnectionFactory = async writer =>
      {
          var connection = await ConnectionMultiplexer
              .ConnectAsync(options.Configuration.GetConfigurationOptions());
          connection.ErrorMessage += (sender, e) =>
          {
              writer.WriteLine($"Ошибка Redis: {e.Message}");
          };
          return connection;
      };
  });

// Расширенная конфигурация для high-load систем
services.AddSignalR()
  .AddStackExchangeRedis(options =>
  {
      // Настраиваем пул соединений
      options.ConnectionMultiplexerFactory = () => 
          ConnectionMultiplexer.Connect(new ConfigurationOptions
          {
              EndPoints = { { "redis1:6379" }, { "redis2:6379" } },
              AbortOnConnectFail = false,
              ConnectTimeout = 5000
          });
      
      // Ограничиваем размер сообщений
      options.MaximumMessageSize = 1024 * 64; // 64KB
  });

// Настройка отслеживания изменений в БД
public async Task SetupChangeTracking()
{
  using var connection = new SqlConnection(_connectionString);
  await connection.OpenAsync();
  
  // Включаем Service Broker для базы данных
  await connection.ExecuteAsync("ALTER DATABASE [MyDatabase] SET ENABLE_BROKER");
  
  // Создаем триггер на таблице
  await connection.ExecuteAsync(@"
    CREATE TRIGGER [dbo].[Notify_ProductChanged] ON [dbo].[Products]
    AFTER INSERT, UPDATE, DELETE AS
    BEGIN
      DECLARE @message NVARCHAR(MAX) = N'Products';
      DECLARE @conversationHandle UNIQUEIDENTIFIER;
      
      BEGIN DIALOG CONVERSATION @conversationHandle
      FROM SERVICE [NotificationService]
      TO SERVICE 'NotificationService'
      ON CONTRACT [DEFAULT]
      WITH ENCRYPTION = OFF;
      
      SEND ON CONVERSATION @conversationHandle MESSAGE TYPE [DEFAULT] (@message);
      END CONVERSATION @conversationHandle;
    END
  ");
}

public class DatabaseChangeListener : BackgroundService
{
  private readonly IHubContext<ProductHub> _hubContext;
  private readonly string _connectionString;
  
  protected override async Task ExecuteAsync(CancellationToken stoppingToken)
  {
    using var connection = new SqlConnection(_connectionString);
    await connection.OpenAsync(stoppingToken);
    
    while (!stoppingToken.IsCancellationRequested)
    {
      // Слушаем сообщения от Service Broker
      var message = await connection.QueryFirstOrDefaultAsync<string>(@"
        DECLARE @conversation_handle UNIQUEIDENTIFIER;
        DECLARE @message_body NVARCHAR(MAX);
        
        WAITFOR (
          RECEIVE TOP(1)
            @conversation_handle = conversation_handle,
            @message_body = message_body
          FROM [NotificationQueue]
        ), TIMEOUT 30000;
        
        SELECT @message_body;
      ");
      
      if (message == "Products")
      {
        // Получаем обновленные данные
        var products = await GetUpdatedProductsAsync(connection);
        
        // Отправляем клиентам через SignalR
        await _hubContext.Clients.All.SendAsync("ProductsUpdated", products, stoppingToken);
      }
    }
  }
}

public async Task MonitorMongoChanges()
{
  var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(change => change.OperationType == ChangeStreamOperationType.Insert || 
                     change.OperationType == ChangeStreamOperationType.Update);
  
  var options = new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup };
  var cursor = await _collection.WatchAsync(pipeline, options);
  
  await cursor.ForEachAsync(async change => 
  {
    await _hubContext.Clients.All.SendAsync("DataChanged", change.FullDocument);
  });
}

[Authorize]
public class SecureHub : Hub
{
    public async Task SendPrivateMessage(string user, string message)
    {
        // Только авторизованные пользователи попадут сюда
        await Clients.User(user).SendAsync("ReceiveMessage", Context.User.Identity.Name, message);
    }
}

[Authorize(Policy = "PremiumOnly")]
public async Task SendPriorityAlert(string message)
{
    // Только пользователи с премиум-подпиской
    await Clients.All.SendAsync("PriorityAlert", message);
}

// На стороне сервера
services.AddAuthentication(JwtBearerDefaults.AuthenticationScheme)
  .AddJwtBearer(options =>
  {
      options.TokenValidationParameters = new TokenValidationParameters
      {
          // Настройки валидации токена
      };
      
      // Критически важно для WebSocket!
      options.Events = new JwtBearerEvents
      {
          OnMessageReceived = context =>
          {
              var accessToken = context.Request.Query["access_token"];
              var path = context.HttpContext.Request.Path;
              
              if (!string.IsNullOrEmpty(accessToken) && path.StartsWithSegments("/hubs"))
              {
                  context.Token = accessToken;
              }
              
              return Task.CompletedTask;
          }
      };
  });

// Так делать опасно!
public async Task DeleteMessage(int messageId)
{
    await _messageService.DeleteAsync(messageId);
    await Clients.All.SendAsync("MessageDeleted", messageId);
}
 
// Правильный подход - всегда проверять права на объект
public async Task DeleteMessage(int messageId)
{
    var message = await _messageService.GetByIdAsync(messageId);
    if (message.AuthorId != Context.UserIdentifier)
    {
        throw new HubException("У вас нет прав на удаление этого сообщения");
    }
    
    await _messageService.DeleteAsync(messageId);
    await Clients.All.SendAsync("MessageDeleted", messageId);
}

// Для .NET MAUI/Xamarin клиента
var connection = new HubConnectionBuilder()
    .WithUrl("https://myserver.com/hub")
    .WithAutomaticReconnect(new[] { 
        TimeSpan.Zero,  // Мгновенная первая попытка
        TimeSpan.FromSeconds(1),
        TimeSpan.FromSeconds(3),
        TimeSpan.FromSeconds(5),
        // Далее экспоненциальное увеличение до 60 секунд
    })
    .Build();

// Xamarin.iOS
public override void DidEnterBackground(UIApplication application)
{
    base.DidEnterBackground(application);
    
    // Сохраняем состояние и закрываем соединение корректно
    var backgroundTask = application.BeginBackgroundTask("CloseConnection", () => {});
    
    Task.Run(async () => {
        try {
            // Пометить клиента как отошедшего
            await connection.InvokeAsync("UpdatePresence", "away");
            await connection.StopAsync();
        }
        finally {
            application.EndBackgroundTask(backgroundTask);
        }
    });
}

private Queue<Message> _messageBuffer = new Queue<Message>();
 
public async Task SendMessage(string content)
{
    var message = new Message { Content = content, Timestamp = DateTime.UtcNow };
    
    if (connection.State == HubConnectionState.Connected)
    {
        try {
            await connection.InvokeAsync("SendMessage", message);
        }
        catch {
            // При сбое добавляем в буфер
            _messageBuffer.Enqueue(message);
            SaveBufferToStorage();
        }
    }
    else
    {
        _messageBuffer.Enqueue(message);
        SaveBufferToStorage();
    }
}

connection.Reconnected += async (connectionId) => {
    while (_messageBuffer.Count > 0)
    {
        var message = _messageBuffer.Peek();
        try {
            await connection.InvokeAsync("SendMessage", message);
            _messageBuffer.Dequeue();
        }
        catch {
            // Если снова ошибка, прекращаем попытки — соединение нестабильно
            break;
        }
    }
    SaveBufferToStorage(); // Сохраняем оставшиеся сообщения
};

public class RabbitToSignalRBridge : BackgroundService
{
    private readonly IHubContext<NotificationHub> _hubContext;
    private readonly IModel _channel;
 
    public RabbitToSignalRBridge(IHubContext<NotificationHub> hubContext, IConnection rabbitConnection)
    {
        _hubContext = hubContext;
        _channel = rabbitConnection.CreateModel();
        _channel.QueueDeclare("transaction_events", durable: true, exclusive: false, autoDelete: false);
    }
 
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var consumer = new EventingBasicConsumer(_channel);
        
        consumer.Received += async (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            var transaction = JsonSerializer.Deserialize<TransactionEvent>(message);
            
            // Отправляем уведомление конкретному пользователю
            await _hubContext.Clients.User(transaction.UserId)
                .SendAsync("TransactionProcessed", transaction, stoppingToken);
        };
        
        _channel.BasicConsume("transaction_events", autoAck: true, consumer: consumer);
        
        await Task.Delay(Timeout.Infinite, stoppingToken);
    }
}

public class KafkaConsumerService : BackgroundService
{
    private readonly IHubContext<TelemetryHub> _hubContext;
    private readonly ConsumerConfig _config;
    
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using var consumer = new ConsumerBuilder<string, string>(_config).Build();
        consumer.Subscribe("device-telemetry");
        
        while (!stoppingToken.IsCancellationRequested)
        {
            var result = consumer.Consume(TimeSpan.FromMilliseconds(100));
            if (result != null)
            {
                var telemetry = JsonSerializer.Deserialize<DeviceTelemetry>(result.Message.Value);
                
                // Группируем данные по идентификатору устройства
                await _hubContext.Clients.Group($"device_{telemetry.DeviceId}")
                    .SendAsync("TelemetryUpdate", telemetry, stoppingToken);
            }
        }
    }
}

public class DeviceCommandHub : Hub
{
    private readonly IProducer<string, string> _producer;
    
    public async Task SendCommand(string deviceId, DeviceCommand command)
    {
        // Отправляем команду в Kafka
        var message = new Message<string, string>
        {
            Key = deviceId,
            Value = JsonSerializer.Serialize(command)
        };
        
        await _producer.ProduceAsync("device-commands", message);
        
        // Подтверждаем принятие команды клиенту
        await Clients.Caller.SendAsync("CommandQueued");
    }
}

public class RegionSyncService : BackgroundService
{
  private readonly IHubContext<ChatHub> _hubContext;
  private readonly IRegionEventBus _eventBus;
  
  protected override async Task ExecuteAsync(CancellationToken stoppingToken)
  {
      await _eventBus.SubscribeAsync("global-events", async message => {
          // Сообщения из других регионов
          if (message.OriginRegion != _currentRegion)
          {
              await _hubContext.Clients.Group(message.GroupName)
                  .SendAsync(message.Method, message.Payload);
          }
      });
  }
}

services.AddSignalR()
  .AddAzureSignalR(options => {
      options.ServerStickyMode = ServerStickyMode.Required;
      options.GracefulShutdown.Mode = GracefulShutdownMode.WaitForClientsClose;
      
      // Несколько региональных эндпоинтов
      options.Endpoints = new[]
      {
          new ServiceEndpoint("<conn-string-eu>", EndpointType.Primary, "Europe"),
          new ServiceEndpoint("<conn-string-us>", EndpointType.Primary, "America"),
          new ServiceEndpoint("<conn-string-asia>", EndpointType.Primary, "Asia")
      };
  });

[Fact]
public async Task SendMessage_ShouldBroadcastToAllClients()
{
    // Arrange
    var mockClients = new Mock<IHubCallerClients>();
    var mockClientProxy = new Mock<IClientProxy>();
    mockClients.Setup(clients => clients.All).Returns(mockClientProxy.Object);
    
    var hub = new ChatHub
    {
        Clients = mockClients.Object,
        Context = new DefaultHubCallerContext()
    };
    
    // Act
    await hub.SendMessage("testUser", "Hello, world!");
    
    // Assert
    mockClientProxy.Verify(
        clientProxy => clientProxy.SendCoreAsync(
            "ReceiveMessage",
            It.Is<object[]>(o => o != null && o[0].ToString() == "testUser" && o[1].ToString() == "Hello, world!"),
            default(CancellationToken)),
        Times.Once);
}

[Fact]
public async Task ClientShouldReceiveMessages()
{
    // Запускаем тестовый сервер
    using var host = await CreateHostBuilder().StartAsync();
    var connection = new HubConnectionBuilder()
        .WithUrl("http://localhost:5000/chathub")
        .Build();
    
    string receivedUser = null;
    string receivedMessage = null;
    
    connection.On<string, string>("ReceiveMessage", (user, message) => {
        receivedUser = user;
        receivedMessage = message;
    });
    
    await connection.StartAsync();
    
    // Отправляем тестовое сообщение через другое соединение
    var sender = new HubConnectionBuilder()
        .WithUrl("http://localhost:5000/chathub")
        .Build();
    await sender.StartAsync();
    await sender.InvokeAsync("SendMessage", "testUser", "Test message");
    
    // Ждем получения сообщения
    await Task.Delay(500);
    
    Assert.Equal("testUser", receivedUser);
    Assert.Equal("Test message", receivedMessage);
}

// Базовая настройка тестового окружения для хаба
public class NotificationHubTests
{
  private Mock<IHubCallerClients> _mockClients;
  private Mock<IClientProxy> _mockClientProxy;
  private Mock<HubCallerContext> _mockContext;
  private Mock<IGroupManager> _mockGroups;
  private string _connectionId = "test-connection-id";
  
  public NotificationHubTests()
  {
      // Создаем моки для всех зависимостей хаба
      _mockClients = new Mock<IHubCallerClients>();
      _mockClientProxy = new Mock<IClientProxy>();
      _mockContext = new Mock<HubCallerContext>();
      _mockGroups = new Mock<IGroupManager>();
      
      // Настраиваем поведение моков
      _mockContext.Setup(c => c.ConnectionId).Returns(_connectionId);
      _mockClients.Setup(c => c.Caller).Returns(_mockClientProxy.Object);
      _mockClients.Setup(c => c.All).Returns(_mockClientProxy.Object);
  }
  
  [Fact]
  public async Task SendNotification_ShouldInvokeClientMethod()
  {
      // Arrange
      var hub = new NotificationHub
      {
          Clients = _mockClients.Object,
          Context = _mockContext.Object,
          Groups = _mockGroups.Object
      };
      
      // Act
      await hub.SendNotification("Тестовое уведомление");
      
      // Assert
      _mockClientProxy.Verify(
          x => x.SendCoreAsync(
              "ReceiveNotification",
              It.Is<object[]>(o => o.Length == 1 && (string)o[0] == "Тестовое уведомление"),
              default),
          Times.Once);
  }
}

// Настройка мока для работы с группами
_mockGroups
  .Setup(g => g.AddToGroupAsync(_connectionId, "TestGroup", default))
  .Returns(Task.CompletedTask);
 
// Проверка что клиент был добавлен в группу
_mockGroups.Verify(
  g => g.AddToGroupAsync(_connectionId, "TestGroup", default),
  Times.Once);

// Тестирование отключения клиента
[Fact]
public async Task OnDisconnected_ShouldCleanupResources()
{
  // Arrange
  var hub = new ChatHub() {
      Context = _mockContext.Object,
      Clients = _mockClients.Object
  };
  
  // Act - имитируем подключение, а затем отключение
  await hub.OnConnectedAsync();
  await hub.OnDisconnectedAsync(new Exception("Тест отключения"));
  
  // Assert - проверяем, что было выполнено нужное действие
  _mockRepository.Verify(r => r.RemoveUserConnection(_connectionId), Times.Once);
}

public async Task SendFile(string fileName, string base64Content)
{
  // Проверка размера файла (не более 4MB для Base64)
  if (base64Content.Length > 4 * 1024 * 1024)
  {
      throw new HubException("Файл слишком большой для прямой передачи");
  }
  
  await Clients.All.SendAsync("ReceiveFile", fileName, base64Content);
}

public async Task InitiateFileUpload(string fileName, long fileSize)
{
  // Генерируем уникальный идентификатор для загрузки
  var uploadId = Guid.NewGuid().ToString("N");
  
  // Сохраняем информацию о файле
  _uploadTracker.RegisterUpload(uploadId, Context.ConnectionId, fileName);
  
  // Отправляем клиенту URL для загрузки
  var uploadUrl = $"/api/fileupload/{uploadId}";
  await Clients.Caller.SendAsync("FileUploadReady", uploadUrl, uploadId);
}

// В контроллере обработки загрузки
[HttpPost("/api/fileupload/{uploadId}")]
public async Task<IActionResult> UploadFile(string uploadId)
{
  // Сохраняем файл...
  
  // Уведомляем всех через SignalR
  await _hubContext.Clients.All.SendAsync(
      "FileShared", 
      _uploadTracker.GetFileName(uploadId),
      _uploadTracker.GetFileUrl(uploadId));
  
  return Ok();
}

public async IAsyncEnumerable<FileChunk> UploadFileStream(
    string fileName, 
    [EnumeratorCancellation] CancellationToken cancellationToken)
{
    var totalChunks = 0;
    var processedChunks = 0;
    
    // Получаем стрим от клиента через параметр Stream в методе вызова
    await foreach (var chunk in Clients.Caller.StreamAsync<byte[]>("UploadChunk", cancellationToken))
    {
        if (totalChunks == 0) 
        {
            // Первый чанк содержит метаданные о количестве чанков
            totalChunks = BitConverter.ToInt32(chunk, 0);
            continue;
        }
        
        // Сохраняем чанк в файл
        await _fileService.AppendChunkAsync(fileName, chunk);
        
        processedChunks++;
        var progress = (float)processedChunks / totalChunks;
        
        // Отправляем информацию о прогрессе
        yield return new FileChunk 
        { 
            ProgressPercentage = progress * 100,
            ChunkIndex = processedChunks,
            TotalChunks = totalChunks
        };
    }
    
    // Завершаем сборку файла
    var fileUrl = await _fileService.FinalizeFileAsync(fileName);
    
    yield return new FileChunk 
    { 
        ProgressPercentage = 100,
        ChunkIndex = totalChunks,
        TotalChunks = totalChunks,
        CompletedFileUrl = fileUrl
    };
}