CompositeCompletion.cs

    public class CompositeCompletion
    {
        private int _lockCount;
        private bool _isTaskCreated = false;
        private Action _completeAction;
        private readonly object _syncObject = new object();

        public IObservable<T> Add<T>(IObservable<T> observable)
        {
            lock (_syncObject)
            {
                if (_isTaskCreated)
                    throw new InvalidOperationException($"Can not add new completion after task creation. See {nameof(CreateTask)}.");

                _lockCount++;
            }

            return observable.Finally(() =>
                {
                    lock (_syncObject)
                    {
                        _lockCount--;

                        if (_lockCount == 0)
                            _completeAction?.Invoke();
                    }
                });
        }

        public Task CreateTask()
        {
            lock (_syncObject)
            {
                if (_lockCount == 0)
                    return Task.CompletedTask;

                var tcs = new TaskCompletionSource<int>();
                _completeAction = () => tcs.SetResult(1);

                _isTaskCreated = true;
                return tcs.Task;
            }
        }
    }