How to properly cancel notification subscriptions when listening to object changes?

[johanneszellinger]

I have a minio wrapper class for my application utilizing the minio-dotnet sdk. One of the functions is as follows:

    public void AddObjectEventListener(
        string userID,
        string datasetID,
        string name,
        CancellationToken ct,
        Func<CustomMinioNotification, Task> callback)
    {
        _logger.LogDebug($"Listening for Object creation: {name}");
        var events = new List<EventType> { EventType.ObjectCreatedAll };
        var prefix = GetObjectName(datasetID, name);

        ListenBucketNotificationsArgs args = new ListenBucketNotificationsArgs()
                                                 .WithBucket(userID)
                                                 .WithEvents(events)
                                                 .WithPrefix(prefix);
        IObservable<MinioNotificationRaw> observable = _minio.ListenBucketNotificationsAsync(args, ct);
        IDisposable subscription = observable.Subscribe(
            notification =>
            {
                var notificationEvent = JsonSerializer.Deserialize<CustomMinioNotification>(notification.json);
                callback(notificationEvent!);
            },
            ex => _logger.LogError($"Minio Error while listening to bucket event: {ex}"),
            () => _logger.LogDebug($"Stopped listening for bucket notifications"));
    }

The idea is the following:

1. Request PresignedPutURL and send it back to a client
2. Subscribe to object created events with the method above
3. Start upload
4. When notified execute the callback (this is database stuff to keep track of the uploaded objects)
5. Unsubscribe using the CancellationToken (I keep track of a CancellationTokenSource in the initial caller of the method above).

The process above works fine for my purpose, except for the unsubscribing part. As mentioned, I keep track of a CancellationTokenSource which is the parent of the passed CancellationToken and call its Cancel() function in the callback. However () => _logger.LogDebug($"Stopped listening for bucket notifications") is never executed. I guess this means I have some event subscription floating around, which never get disposed/cleaned up. How should one handle this properly?