Observable.FromAsync memory leaks [MelbourneDeveloper / Device.Net]

Question

I am Using a slightly modified version of the HID Example from melbournedeveloper/Device.NET library to poll an HID device every 100ms with a TransferResult(byte[] data, uint bytesRead) callback, using DotMemory the returned TransferResult seems to be leaking on every call

_hidObserver = Observable
          .Interval(TimeSpan.FromMilliseconds(1000))
          .SelectMany(_ => Observable.FromAsync(() => _hidIDevice.ReadAsync()))
          .DefaultIfEmpty()
          .Subscribe(onNext: tuple =>
              {
                  Console.WriteLine("QUEUE | bytes transferred: " + tuple.BytesTransferred);
                  Console.WriteLine("QUEUE | bytes: " + tuple.Data);
                  return;
              },
              onCompleted: () => Console.WriteLine("HID Button Observer | Completed."),
              onError: exception => Console.WriteLine($"HID Button Observer | Error | {exception.Message}.")
          );

App normally starts with 70MB of memory, leaving it running for 17 hours memory grew up to 1.8 GB

commenting out the observable part memory stays stable.

---

Update 1:

Applying @theodor-zoulias answer fixed the memory leak from continuous calls, however not all, after extensive analysis, it tends out to be a problem with the HID Library itself leaking from the inside MelbourneDeveloper/Device.Net#219

related to these lines of codes:

private static extern bool HidD_FreePreparsedData(ref IntPtr pointerToPreparsedData); 
isSuccess = HidD_FreePreparsedData(ref pointerToPreParsedData); 

should be without the ref keyword in regard of MS doc here: https://learn.microsoft.com/en-us/windows-hardware/drivers/ddi/hidsdi/nf-hidsdi-hidd_freepreparseddata

private static extern bool HidD_FreePreparsedData(IntPtr pointerToPreparsedData); 
isSuccess = HidD_FreePreparsedData(pointerToPreParsedData); 

Fixed by forking the Device.Net library and updating these 2 lines.

Answer 1

_hidObserver = Observable
    .Interval(TimeSpan.FromMilliseconds(1000))
    .SelectMany(_ => Observable.FromAsync(() => _hidIDevice.ReadAsync()))

The Observable.Interval sequence produces a value every second, each value is projected to an asynchronous operation, and each operation is started immediately. There is no provision for avoiding overlapping. In case the _hidIDevice.ReadAsync() takes more than 1 second, a second _hidIDevice.ReadAsync() operation will start before the pervious has completed. Obviously this is not going to scale well. My guess is that the _hidIDevice.ReadAsync() has some internal serialization mechanism that queues incoming requests and executes them one at a time. There are also other possible scenarios, like ThreadPool starvation.

My suggestion is to prevent the overlapping from happening, by not starting a new operation in case the previous has not completed yet. You can find in this question a custom ExhaustMap operator that could be used like this:

_hidObserver = Observable
    .Interval(TimeSpan.FromMilliseconds(1000))
    .ExhaustMap(_ => _hidIDevice.ReadAsync())