rudder-csharp-sdk/Services/RealtimeService.cs

using System;
using System.Threading;
using System.Threading.Tasks;
using RudderSdk.Core.Abstractions;

namespace RudderSdk.Core;

/// <summary>
/// Realtime websocket channel. Requires
/// <see cref="RudderClientOptions.RealtimeUrl"/> and
/// <see cref="RudderClientOptions.RealtimeTransportFactory"/>.
/// </summary>
public sealed class RealtimeService
{
    private static readonly TimeSpan DefaultTimeout = TimeSpan.FromSeconds(10);

    private readonly RudderClient _client;
    private RealtimeSession? _session;

    internal RealtimeService(RudderClient client) => _client = client;

    /// <summary>The current session, or null when not connected.</summary>
    public RealtimeSession? Session => _session;

    /// <summary>True while a session is connected.</summary>
    public bool IsConnected => _session?.IsConnected == true;

    /// <summary>Connects to the configured realtime URL.</summary>
    public Task<RealtimeSession> ConnectAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default)
    {
        if (string.IsNullOrEmpty(_client.RealtimeUrl))
            throw new InvalidOperationException("RealtimeUrl is not configured.");

        return ConnectAsync(new Uri(_client.RealtimeUrl), timeout, cancellationToken);
    }

    /// <summary>Connects to an explicit realtime URL.</summary>
    public async Task<RealtimeSession> ConnectAsync(Uri uri, TimeSpan? timeout = null, CancellationToken cancellationToken = default)
    {
        var factory = _client.Options.RealtimeTransportFactory
            ?? throw new InvalidOperationException("RealtimeTransportFactory is not configured.");

        var token = _client.TokenStore.GetAccessToken();
        if (string.IsNullOrEmpty(token))
            throw new InvalidOperationException("LiveOps access token is required for realtime authorization.");

        var transport = factory.Create();
        await transport.ConnectAsync(uri, timeout ?? DefaultTimeout, cancellationToken).ConfigureAwait(false);
        _session = new RealtimeSession(transport, token);
        return _session;
    }

    /// <summary>Closes the current session, if any.</summary>
    public Task DisconnectAsync(CancellationToken cancellationToken = default)
        => _session?.DisconnectAsync(cancellationToken) ?? Task.CompletedTask;

    /// <summary>Pumps the underlying transport; call every frame.</summary>
    public void Update(float deltaTime)
    {
        _session?.Update(deltaTime);
    }
}

/// <summary>An open realtime connection.</summary>
public sealed class RealtimeSession
{
    private readonly IRealtimeTransport _transport;

    internal RealtimeSession(IRealtimeTransport transport, string accessToken)
    {
        _transport = transport;
        AccessToken = accessToken;
        _transport.Closed += () => Closed?.Invoke();
        _transport.Error += ex => Error?.Invoke(ex);
        _transport.Received += data => MessageReceived?.Invoke(data);
    }

    /// <summary>Access token the connection was authorized with.</summary>
    public string AccessToken { get; }

    /// <summary>True while the connection is open.</summary>
    public bool IsConnected => _transport.IsConnected;

    /// <summary>Raised when the connection closes.</summary>
    public event Action? Closed;

    /// <summary>Raised on transport errors.</summary>
    public event Action<Exception>? Error;

    /// <summary>Raised for every incoming message.</summary>
    public event Action<ArraySegment<byte>>? MessageReceived;

    /// <summary>Sends one message.</summary>
    public Task SendAsync(byte[] payload, CancellationToken cancellationToken = default)
        => _transport.SendAsync(new ArraySegment<byte>(payload ?? Array.Empty<byte>()), cancellationToken);

    /// <summary>Closes the connection.</summary>
    public Task DisconnectAsync(CancellationToken cancellationToken = default)
        => _transport.CloseAsync(cancellationToken);

    internal void Update(float deltaTime)
    {
        _transport.Update(deltaTime);
    }
}