C# и сети: Сокеты, gRPC и SignalR

public async Task HandleClientAsync(TcpClient client)
{
    using NetworkStream stream = client.GetStream();
    byte[] buffer = new byte[4096];
    
    try
    {
        int bytesRead;
        while ((bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length)) > 0)
        {
            // Обработка полученных данных
            var message = Encoding.UTF8.GetString(buffer, 0, bytesRead);
            var response = ProcessMessage(message);
            
            // Асинхронная отправка ответа
            byte[] responseBytes = Encoding.UTF8.GetBytes(response);
            await stream.WriteAsync(responseBytes, 0, responseBytes.Length);
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Ошибка при обработке клиента: {ex.Message}");
    }
}

using System.Net.Security;
using System.Security.Cryptography.X509Certificates;
 
public async Task SecureConnectionAsync(TcpClient client, X509Certificate serverCertificate)
{
    SslStream sslStream = new SslStream(client.GetStream(), false);
    
    // Аутентифицируем соединение как сервер с использованием сертификата
    await sslStream.AuthenticateAsServerAsync(
        serverCertificate,
        clientCertificateRequired: false,
        enabledSslProtocols: System.Security.Authentication.SslProtocols.Tls12,
        checkCertificateRevocation: true);
    
    // Теперь все данные, передаваемые через этот поток, автоматически шифруются
    // ...
}

using System.Net;
using System.Net.Sockets;
using System.Text;
 
public class SimpleServer
{
    public async Task StartAsync(int port)
    {
        // Создаём конечную точку для прослушивания
        IPEndPoint endpoint = new IPEndPoint(IPAddress.Any, port);
        
        // Создаём TCP-сокет
        Socket listener = new Socket(AddressFamily.InterNetwork, 
                                   SocketType.Stream, 
                                   ProtocolType.Tcp);
        
        try
        {
            // Привязываем сокет к конечной точке
            listener.Bind(endpoint);
            
            // Начинаем прослушивание (очередь до 100 подключений)
            listener.Listen(100);
            
            Console.WriteLine($"Сервер запущен на порту {port}");
            
            while (true)
            {
                // Асинхронно принимаем новое подключение
                Socket handler = await Task.Factory.FromAsync(
                    listener.BeginAccept, 
                    listener.EndAccept, 
                    null);
                
                // Обработка подключения в отдельной задаче
                _ = Task.Run(() => HandleClientAsync(handler));
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Ошибка сервера: {ex.Message}");
        }
    }
    
    private async Task HandleClientAsync(Socket clientSocket)
    {
        try
        {
            byte[] buffer = new byte[1024];
            
            // Читаем данные от клиента
            int received = await Task.Factory.FromAsync(
                clientSocket.BeginReceive(buffer, 0, buffer.Length, SocketFlags.None, null, clientSocket),
                clientSocket.EndReceive);
                
            string message = Encoding.UTF8.GetString(buffer, 0, received);
            Console.WriteLine($"Получено: {message}");
            
            // Отправляем ответ
            string response = $"Эхо: {message}";
            byte[] responseData = Encoding.UTF8.GetBytes(response);
            
            await Task.Factory.FromAsync(
                clientSocket.BeginSend(responseData, 0, responseData.Length, SocketFlags.None, null, clientSocket),
                clientSocket.EndSend);
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Ошибка при обработке клиента: {ex.Message}");
        }
        finally
        {
            // Закрываем соединение
            clientSocket.Shutdown(SocketShutdown.Both);
            clientSocket.Close();
        }
    }
}

public class SimpleClient
{
    public async Task ConnectAndSendAsync(string server, int port, string message)
    {
        // Создаём сокет для подключения
        Socket client = new Socket(AddressFamily.InterNetwork, 
                                  SocketType.Stream, 
                                  ProtocolType.Tcp);
        
        try
        {
            // Подключаемся к серверу
            await client.ConnectAsync(server, port);
            
            // Отправляем сообщение
            byte[] data = Encoding.UTF8.GetBytes(message);
            await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
            
            // Получаем ответ
            byte[] buffer = new byte[1024];
            int received = await client.ReceiveAsync(new ArraySegment<byte>(buffer), SocketFlags.None);
            
            string response = Encoding.UTF8.GetString(buffer, 0, received);
            Console.WriteLine($"Ответ сервера: {response}");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Ошибка клиента: {ex.Message}");
        }
        finally
        {
            // Закрываем соединение
            client.Shutdown(SocketShutdown.Both);
            client.Close();
        }
    }
}

public class BufferManager
{
    private readonly int _bufferSize;
    private readonly ConcurrentBag<byte[]> _freeBuffers = new();
    
    public BufferManager(int bufferSize, int initialCount)
    {
        _bufferSize = bufferSize;
        for (int i = 0; i < initialCount; i++)
        {
            _freeBuffers.Add(new byte[bufferSize]);
        }
    }
    
    public byte[] GetBuffer()
    {
        if (_freeBuffers.TryTake(out byte[] buffer))
            return buffer;
        
        return new byte[_bufferSize]; // Создаём новый, если пул пуст
    }
    
    public void ReturnBuffer(byte[] buffer)
    {
        if (buffer.Length == _bufferSize)
            _freeBuffers.Add(buffer);
    }
}

private void StartAccept(SocketAsyncEventArgs acceptEventArg)
{
    if (acceptEventArg == null)
    {
        acceptEventArg = new SocketAsyncEventArgs();
        acceptEventArg.Completed += AcceptCompleted;
    }
    else
    {
        acceptEventArg.AcceptSocket = null;
    }
    
    bool willRaiseEvent = _listenSocket.AcceptAsync(acceptEventArg);
    if (!willRaiseEvent)
    {
        ProcessAccept(acceptEventArg);
    }
}
 
private void AcceptCompleted(object sender, SocketAsyncEventArgs e)
{
    ProcessAccept(e);
}
 
private void ProcessAccept(SocketAsyncEventArgs e)
{
    // Обработка нового подключения...
    
    // Продолжаем принимать следующие соединения
    StartAccept(e);
}

socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true);
socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, true); // Отключаем алгоритм Нагла

private async Task SendWithLengthPrefixAsync(Socket socket, byte[] message)
{
    // Создаём буфер для длины и сообщения
    byte[] lengthBytes = BitConverter.GetBytes(message.Length);
    byte[] fullPacket = new byte[4 + message.Length];
    
    // Копируем длину и сообщение в общий буфер
    Buffer.BlockCopy(lengthBytes, 0, fullPacket, 0, 4);
    Buffer.BlockCopy(message, 0, fullPacket, 4, message.Length);
    
    // Отправляем весь пакет
    await socket.SendAsync(new ArraySegment<byte>(fullPacket), SocketFlags.None);
}
 
private async Task<byte[]> ReceiveWithLengthPrefixAsync(Socket socket)
{
    // Сначала получаем 4 байта с длиной сообщения
    byte[] lengthBytes = new byte[4];
    int bytesReceived = 0;
    
    while (bytesReceived < 4)
    {
        int currentBytes = await socket.ReceiveAsync(
            new ArraySegment<byte>(lengthBytes, bytesReceived, 4 - bytesReceived), 
            SocketFlags.None);
            
        if (currentBytes == 0) // Соединение закрыто
            return new byte[0];
            
        bytesReceived += currentBytes;
    }
    
    // Преобразуем байты в длину сообщения
    int messageLength = BitConverter.ToInt32(lengthBytes, 0);
    
    // Получаем само сообщение
    byte[] message = new byte[messageLength];
    bytesReceived = 0;
    
    while (bytesReceived < messageLength)
    {
        int currentBytes = await socket.ReceiveAsync(
            new ArraySegment<byte>(message, bytesReceived, messageLength - bytesReceived), 
            SocketFlags.None);
            
        if (currentBytes == 0) // Соединение закрыто
            return new byte[0];
            
        bytesReceived += currentBytes;
    }
    
    return message;
}

syntax = "proto3";
 
option csharp_namespace = "MessagingService";
 
package messaging;
 
// Определяем сервис
service Messenger {
  // Простой унарный вызов
  rpc SendMessage (MessageRequest) returns (MessageResponse);
  
  // Потоковая передача от сервера к клиенту
  rpc SubscribeToMessages (SubscriptionRequest) returns (stream MessageResponse);
}
 
// Запрос на отправку сообщения
message MessageRequest {
  string sender = 1;
  string recipient = 2;
  string content = 3;
  int64 timestamp = 4;
}
 
// Ответ на отправку сообщения
message MessageResponse {
  bool success = 1;
  string message_id = 2;
  int64 timestamp = 3;
}
 
// Запрос на подписку
message SubscriptionRequest {
  string subscriber_id = 1;
  repeated string channels = 2;
}

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;
using Grpc.Core;
using MessagingService;
 
public class MessengerService : Messenger.MessengerBase
{
    // Хранилище сообщений (в реальном приложении было бы использовано что-то более постоянное)
    private readonly ConcurrentDictionary<string, ConcurrentBag<MessageResponse>> _messagesByUser = 
        new ConcurrentDictionary<string, ConcurrentBag<MessageResponse>>();
    
    public override Task<MessageResponse> SendMessage(MessageRequest request, ServerCallContext context)
    {
        Console.WriteLine($"Получено сообщение от {request.Sender} для {request.Recipient}: {request.Content}");
        
        // Создаём ответ
        var response = new MessageResponse
        {
            Success = true,
            MessageId = Guid.NewGuid().ToString(),
            Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
        };
        
        // Сохраняем сообщение для получателя
        _messagesByUser.AddOrUpdate(
            request.Recipient,
            new ConcurrentBag<MessageResponse>(new[] { response }),
            (_, bag) => { bag.Add(response); return bag; }
        );
        
        return Task.FromResult(response);
    }
    
    public override async Task SubscribeToMessages(
        SubscriptionRequest request, 
        IServerStreamWriter<MessageResponse> responseStream, 
        ServerCallContext context)
    {
        Console.WriteLine($"Пользователь {request.SubscriberId} подписался на сообщения");
        
        // Получаем или создаём очередь сообщений для пользователя
        var userMessages = _messagesByUser.GetOrAdd(
            request.SubscriberId,
            new ConcurrentBag<MessageResponse>()
        );
        
        // Отправляем существующие сообщения
        foreach (var message in userMessages)
        {
            await responseStream.WriteAsync(message);
        }
        
        // В реальном приложении здесь был бы код для отправки новых сообщений
        // по мере их поступления, например, с использованием каналов или событий
    }
}

using System;
using Grpc.Core;
using MessagingService;
 
class Program
{
    const int Port = 50051;
    
    static void Main(string[] args)
    {
        Server server = new Server
        {
            Services = { Messenger.BindService(new MessengerService()) },
            Ports = { new ServerPort("localhost", Port, ServerCredentials.Insecure) }
        };
        
        server.Start();
        
        Console.WriteLine($"gRPC сервер запущен на порту {Port}");
        Console.WriteLine("Нажмите любую клавишу для завершения...");
        Console.ReadKey();
        
        server.ShutdownAsync().Wait();
    }
}

using System;
using System.Threading.Tasks;
using Grpc.Core;
using MessagingService;
 
class Program
{
    static async Task Main(string[] args)
    {
        // Создаём канал для подключения к серверу
        Channel channel = new Channel("localhost:50051", ChannelCredentials.Insecure);
        
        // Создаём клиент
        var client = new Messenger.MessengerClient(channel);
        
        try
        {
            // Отправляем сообщение
            var response = await client.SendMessageAsync(new MessageRequest
            {
                Sender = "Алиса",
                Recipient = "Боб",
                Content = "Привет! Как дела?",
                Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
            });
            
            Console.WriteLine($"Сообщение отправлено, ID: {response.MessageId}");
            
            // Подписываемся на сообщения
            using (var call = client.SubscribeToMessages(new SubscriptionRequest
            {
                SubscriberId = "Алиса"
            }))
            {
                // Читаем сообщения по мере поступления
                while (await call.ResponseStream.MoveNext())
                {
                    var message = call.ResponseStream.Current;
                    Console.WriteLine($"Новое сообщение, ID: {message.MessageId}");
                }
            }
        }
        finally
        {
            await channel.ShutdownAsync();
        }
    }
}

service Chat {
  rpc ChatSession (stream ChatMessage) returns (stream ChatMessage);
}
 
message ChatMessage {
  string user_id = 1;
  string content = 2;
  int64 timestamp = 3;
}

public override async Task ChatSession(
    IAsyncStreamReader<ChatMessage> requestStream,
    IServerStreamWriter<ChatMessage> responseStream,
    ServerCallContext context)
{
    // Словарь для хранения активных клиентов
    var clients = new ConcurrentDictionary<string, IServerStreamWriter<ChatMessage>>();
    
    // Регистрируем текущего клиента
    string clientId = context.RequestHeaders.GetValue("client-id");
    clients.TryAdd(clientId, responseStream);
    
    // Пытаемся обработать входящие сообщения
    try
    {
        // Читаем сообщения от клиента
        await foreach (var message in requestStream.ReadAllAsync())
        {
            // Рассылаем сообщение всем клиентам
            foreach (var client in clients.Values)
            {
                try
                {
                    await client.WriteAsync(message);
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"Ошибка при отправке сообщения: {ex.Message}");
                    // Здесь можно добавить логику удаления недоступных клиентов
                }
            }
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Ошибка в сессии чата: {ex.Message}");
    }
    finally
    {
        // Удаляем клиента при завершении сессии
        clients.TryRemove(clientId, out _);
    }
}

// Создаём двунаправленный вызов
using var call = client.ChatSession();
 
// Запускаем задачу для чтения входящих сообщений
var readTask = Task.Run(async () =>
{
    await foreach (var message in call.ResponseStream.ReadAllAsync())
    {
        Console.WriteLine($"{message.UserId}: {message.Content}");
    }
});
 
// Отправляем сообщения
while (true)
{
    string input = Console.ReadLine();
    if (string.IsNullOrEmpty(input)) break;
    
    await call.RequestStream.WriteAsync(new ChatMessage
    {
        UserId = "CurrentUser",
        Content = input,
        Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
    });
}
 
// Завершаем стрим запросов
await call.RequestStream.CompleteAsync();
// Дожидаемся завершения чтения ответов
await readTask;

public class ServiceDiscoveryGrpcClientFactory
{
    private readonly ConsulClient _consulClient;
    
    public ServiceDiscoveryGrpcClientFactory(string consulAddress)
    {
        _consulClient = new ConsulClient(config => 
        {
            config.Address = new Uri(consulAddress);
        });
    }
    
    public async Task<T> CreateClientAsync<T>(string serviceName) where T : ClientBase
    {
        // Получаем адреса всех экземпляров сервиса
        var services = await _consulClient.Health.Service(serviceName, "", true);
        
        if (!services.Response.Any())
            throw new Exception($"Сервис {serviceName} не найден или недоступен");
        
        // Выбираем случайный экземпляр
        var service = services.Response[new Random().Next(services.Response.Count)];
        var address = $"{service.Service.Address}:{service.Service.Port}";
        
        // Создаём канал и клиент
        var channel = new Channel(address, ChannelCredentials.Insecure);
        return (T)Activator.CreateInstance(typeof(T), channel);
    }
}

// На стороне клиента
var callOptions = new CallOptions(compressionRequestAlgorithm: "gzip");
var response = await client.SomeMethodAsync(request, callOptions);
 
// На стороне сервера
services.AddGrpc(options =>
{
    options.ResponseCompressionAlgorithm = "gzip";
    options.ResponseCompressionLevel = CompressionLevel.Optimal;
});

// Создаём канал один раз
var channel = GrpcChannel.ForAddress("https://example.com");
 
// Используем для множества разных клиентов
var userClient = new UserService.UserServiceClient(channel);
var orderClient = new OrderService.OrderServiceClient(channel);

// Вместо этого
message User {
  string id = 1;
  string firstName = 2;
  string lastName = 3;
  string email = 4;
  string phone = 5;
}
 
// Лучше использовать 
message User {
  string id = 1;
  string first_name = 2;
  string last_name = 3;
  
  oneof contact {
    string email = 4;
    string phone = 5;
  }
}

public class ChatHub : Hub
{
    public async Task SendMessage(string user, string message)
    {
        // Вызываем метод на всех подключенных клиентах
        await Clients.All.SendAsync("ReceiveMessage", user, message);
    }
    
    public override async Task OnConnectedAsync()
    {
        await Clients.All.SendAsync("UserConnected", Context.ConnectionId);
        await base.OnConnectedAsync();
    }
    
    public override async Task OnDisconnectedAsync(Exception exception)
    {
        await Clients.All.SendAsync("UserDisconnected", Context.ConnectionId);
        await base.OnDisconnectedAsync(exception);
    }
}

// В Startup.cs или Program.cs (для .NET 6+)
 
// Добавляем службы
public void ConfigureServices(IServiceCollection services)
{
    services.AddSignalR();
}
 
// Настраиваем конечные точки
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
    // ... другие настройки middleware
    
    app.UseRouting();
    
    app.UseEndpoints(endpoints =>
    {
        endpoints.MapHub<ChatHub>("/chathub");
        // Можно добавить дополнительные хабы
    });
}

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddSignalR();
 
var app = builder.Build();
app.MapHub<ChatHub>("/chathub");
app.Run();

using Microsoft.AspNetCore.SignalR.Client;
 
// Создаём подключение к хабу
var connection = new HubConnectionBuilder()
    .WithUrl("https://example.com/chathub")
    .WithAutomaticReconnect() // Автоматическое переподключение при обрыве связи
    .Build();
 
// Обработчик для получения сообщений
connection.On<string, string>("ReceiveMessage", (user, message) =>
{
    Console.WriteLine($"{user}: {message}");
});
 
// Обработчики событий подключения/отключения
connection.On<string>("UserConnected", (connectionId) =>
{
    Console.WriteLine($"Пользователь подключился: {connectionId}");
});
 
connection.On<string>("UserDisconnected", (connectionId) =>
{
    Console.WriteLine($"Пользователь отключился: {connectionId}");
});
 
// Обработка событий состояния подключения
connection.Closed += async (error) =>
{
    Console.WriteLine($"Соединение закрыто: {error?.Message}");
    // Здесь можно добавить логику при закрытии соединения
};
 
// Запускаем подключение
await connection.StartAsync();
 
// Отправляем сообщение
await connection.InvokeAsync("SendMessage", "АлексейDev", "Привет, SignalR!");

public class GroupChatHub : Hub
{
    public async Task JoinGroup(string groupName)
    {
        await Groups.AddToGroupAsync(Context.ConnectionId, groupName);
        await Clients.Group(groupName).SendAsync("UserJoined", Context.ConnectionId, groupName);
    }
    
    public async Task LeaveGroup(string groupName)
    {
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, groupName);
        await Clients.Group(groupName).SendAsync("UserLeft", Context.ConnectionId, groupName);
    }
    
    public async Task SendToGroup(string groupName, string message)
    {
        await Clients.Group(groupName).SendAsync("ReceiveGroupMessage", Context.ConnectionId, message);
    }
}

// Добавление SignalR с поддержкой Redis backplane
public void ConfigureServices(IServiceCollection services)
{
    services.AddSignalR()
        .AddStackExchangeRedis("redis-server:6379", options =>
        {
            options.Configuration.ChannelPrefix = "MyApp";
            options.Configuration.DefaultDatabase = 5;
            options.Configuration.AbortOnConnectFail = false;
        });
}

// Настройка Azure SignalR Service
services.AddSignalR()
    .AddAzureSignalR(options =>
    {
        options.ServerStickyMode = ServerStickyMode.Required;
        options.ConnectionCount = 5; // Количество соединений между приложением и сервисом
    });

[Authorize]
public class SecureHub : Hub
{
    public async Task SendMessage(string message)
    {
        // Только авторизованные пользователи могут вызывать этот метод
        var username = Context.User.Identity.Name;
        await Clients.All.SendAsync("ReceiveMessage", username, message);
    }
    
    [Authorize(Roles = "Admin")]
    public async Task AnnounceToAll(string announcement)
    {
        // Только администраторы могут делать объявления
        await Clients.All.SendAsync("ReceiveAnnouncement", announcement);
    }
}

var connection = new HubConnectionBuilder()
    .WithUrl("https://example.com/securehub", options =>
    {
        options.AccessTokenProvider = () => Task.FromResult(myAccessToken);
    })
    .Build();

public class ProjectHub : Hub
{
    private readonly IProjectService _projectService;
    
    public ProjectHub(IProjectService projectService)
    {
        _projectService = projectService;
    }
    
    public async Task JoinProject(int projectId)
    {
        // Проверяем права пользователя на доступ к проекту
        var userId = Context.User.FindFirst(ClaimTypes.NameIdentifier)?.Value;
        if (!await _projectService.CanAccessProject(userId, projectId))
        {
            throw new HubException("У вас нет доступа к этому проекту");
        }
        
        // Добавляем пользователя в группу проекта
        await Groups.AddToGroupAsync(Context.ConnectionId, $"project-{projectId}");
    }
    
    public async Task UpdateProject(int projectId, ProjectUpdateDto update)
    {
        var userId = Context.User.FindFirst(ClaimTypes.NameIdentifier)?.Value;
        if (!await _projectService.CanModifyProject(userId, projectId))
        {
            throw new HubException("У вас нет прав на изменение этого проекта");
        }
        
        // Обновляем проект и уведомляем всех участников
        await _projectService.UpdateProjectAsync(projectId, update);
        await Clients.Group($"project-{projectId}").SendAsync("ProjectUpdated", projectId, update);
    }
}

public class SignalRMetricsMiddleware
{
    private readonly RequestDelegate _next;
    private readonly Counter _connectionCounter;
    private readonly Counter _messageCounter;
    private readonly Counter _errorCounter;
    
    public SignalRMetricsMiddleware(RequestDelegate next, IMetricsFactory metricsFactory)
    {
        _next = next;
        _connectionCounter = metricsFactory.CreateCounter("signalr_connections_total", "Total number of SignalR connections");
        _messageCounter = metricsFactory.CreateCounter("signalr_messages_total", "Total number of SignalR messages");
        _errorCounter = metricsFactory.CreateCounter("signalr_errors_total", "Total number of SignalR errors");
    }
    
    public async Task InvokeAsync(HttpContext context)
    {
        // Реализация промежуточного ПО для сбора метрик
        await _next(context);
    }
}

services.AddSignalR()
    .AddHubOptions<ChatHub>(options =>
    {
        options.EnableDetailedErrors = true; // Передает детали ошибок клиентам (только для разработки!)
    });

public class ErrorHandlingFilter : IHubFilter
{
    private readonly ILogger<ErrorHandlingFilter> _logger;
    
    public ErrorHandlingFilter(ILogger<ErrorHandlingFilter> logger)
    {
        _logger = logger;
    }
    
    public async ValueTask<object> InvokeMethodAsync(
        HubInvocationContext invocationContext,
        Func<HubInvocationContext, ValueTask<object>> next)
    {
        try
        {
            return await next(invocationContext);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка при вызове метода {Method} хаба {Hub}",
                invocationContext.HubMethodName, invocationContext.Hub.GetType().Name);
            
            // Преобразуем исключение в понятное клиенту сообщение
            throw new HubException($"Не удалось выполнить действие: {ex.Message}");
        }
    }
}
 
// Регистрация фильтра
services.AddSignalR(options =>
{
    options.AddFilter<ErrorHandlingFilter>();
});

// 1. Низкоуровневый сервис сбора биржевых данных (сокеты)
public class MarketDataReceiver
{
    private Socket _socket;
    
    public async Task StartReceivingAsync()
    {
        // Высокопроизводительный обработчик потока рыночных данных
        byte[] buffer = new byte[8192];
        while (true)
        {
            int received = await _socket.ReceiveAsync(new ArraySegment<byte>(buffer), SocketFlags.None);
            if (received > 0)
            {
                ProcessMarketData(buffer, received);
            }
        }
    }
    
    private void ProcessMarketData(byte[] data, int length)
    {
        // Обработка и публикация данных через брокер сообщений
        _messageBroker.Publish("market-updates", data);
    }
}
 
// 2. Сервис обработки и аналитики (gRPC)
public class AnalyticsService : Analytics.AnalyticsBase
{
    public override async Task GetLiveAnalytics(
        AnalyticsRequest request, 
        IServerStreamWriter<AnalyticsResponse> responseStream, 
        ServerCallContext context)
    {
        // Стриминг аналитических данных
        // ...
    }
}
 
// 3. Слой для веб-клиентов (SignalR)
public class MarketHub : Hub
{
    private readonly IMarketDataService _marketService;
    
    public MarketHub(IMarketDataService marketService)
    {
        _marketService = marketService;
    }
    
    public async Task SubscribeToSymbol(string symbol)
    {
        await Groups.AddToGroupAsync(Context.ConnectionId, symbol);
    }
}

public class LargeFileTransferService : FileTransfer.FileTransferBase
{
    public override async Task<FileTransferInfo> InitiateTransfer(FileMetadata request, ServerCallContext context)
    {
        // Создаём временное хранилище для файла
        var transferId = Guid.NewGuid().ToString();
        var port = GetAvailablePort();
        
        // Запускаем специальный файловый сервер для передачи
        _ = Task.Run(() => StartFileReceiverServer(transferId, port, request.FileSize));
        
        // Возвращаем клиенту информацию для прямого подключения
        return new FileTransferInfo 
        { 
            TransferId = transferId,
            ServerPort = port
        };
    }
    
    private async Task StartFileReceiverServer(string transferId, int port, long fileSize)
    {
        var listener = new TcpListener(IPAddress.Any, port);
        listener.Start();
        
        using (var client = await listener.AcceptTcpClientAsync())
        using (var stream = client.GetStream())
        using (var fileStream = File.Create($"uploads/{transferId}"))
        {
            // Быстрая передача файла напрямую
            await stream.CopyToAsync(fileStream);
        }
        
        listener.Stop();
    }
}