Fix unsubscription on useData

lib/experimental/Collections/useData.spec.tsx

@@ -10,6 +10,7 @@ import type {
   BaseResponse,
   DataSource,
   PaginatedDataAdapter,
+  PaginatedResponse,
   RecordType,
 } from "./types"
 import { useData } from "./useData"
@@ -220,5 +221,168 @@ describe("useData", () => {
 
       expect(result.current.data).toEqual([mockData[0]])
     })
+
+    it("should apply filters to synchronous data", () => {
+      const filters: Partial<FiltersState<TestFilters>> = {
+        search: "Test 1",
+      }
+      const source = createMockDataSource(
+        ({ filters }: { filters: FiltersState<TestFilters> }) =>
+          mockData.filter((item) =>
+            filters.search ? item.name.includes(filters.search) : true
+          )
+      )
+
+      const { result } = renderHook(() => useData(source, { filters }))
+
+      expect(result.current.data).toEqual([mockData[0]])
+    })
+  })
+
+  describe("subscription cleanup", () => {
+    it("should clean up previous subscription when filters change", async () => {
+      // Create a mock implementation for Observable.subscribe that we can track
+      let subscriptionCount = 0
+      let unsubscribeCalls = 0
+
+      // Create a source with a fetchData function that returns an observable
+      const source = createMockDataSource(() => {
+        return new Observable<PromiseState<TestRecord[]>>((subscriber) => {
+          subscriptionCount++
+          subscriber.next({ loading: true, data: null, error: null })
+
+          // Simulate async data loading
+          setTimeout(() => {
+            subscriber.next({ loading: false, data: mockData, error: null })
+            subscriber.complete()
+          }, 10)
+
+          // Return unsubscribe function that tracks when it's called
+          return () => {
+            unsubscribeCalls++
+          }
+        })
+      })
+
+      // Render the hook with initial filters
+      const { rerender } = renderHook(
+        (props) => useData(props.source, { filters: props.filters }),
+        {
+          initialProps: {
+            source,
+            filters: { search: "initial" } as Partial<
+              FiltersState<TestFilters>
+            >,
+          },
+        }
+      )
+
+      // Wait for initial data to load
+      await act(async () => {
+        await new Promise((resolve) => setTimeout(resolve, 20))
+      })
+
+      // Record the current counts before changing filters
+      const initialSubscriptionCount = subscriptionCount
+      const initialUnsubscribeCalls = unsubscribeCalls
+
+      // Change filters to trigger a re-fetch
+      rerender({
+        source,
+        filters: { search: "new search" } as Partial<FiltersState<TestFilters>>,
+      })
+
+      // Wait for the new data to load
+      await act(async () => {
+        await new Promise((resolve) => setTimeout(resolve, 20))
+      })
+
+      // Verify that a new subscription was created and the old one was cleaned up
+      expect(subscriptionCount).toBe(initialSubscriptionCount + 1)
+      expect(unsubscribeCalls).toBe(initialUnsubscribeCalls + 1)
+    })
+
+    it("should clean up previous subscription when changing pages", async () => {
+      // Create a mock implementation for Observable.subscribe that we can track
+      let subscriptionCount = 0
+      let unsubscribeCalls = 0
+
+      // Create a paginated source with a fetchData function that returns an observable
+      const source = {
+        dataAdapter: {
+          fetchData: () => {
+            return new Observable<PromiseState<PaginatedResponse<TestRecord>>>(
+              (subscriber) => {
+                subscriptionCount++
+                subscriber.next({
+                  loading: true,
+                  data: null,
+                  error: null,
+                })
+
+                // Simulate async data loading
+                setTimeout(() => {
+                  subscriber.next({
+                    loading: false,
+                    data: {
+                      records: mockData,
+                      total: mockData.length,
+                      currentPage: 1,
+                      perPage: 10,
+                      pagesCount: 1,
+                    },
+                    error: null,
+                  })
+                  subscriber.complete()
+                }, 10)
+
+                // Return unsubscribe function that tracks when it's called
+                return () => {
+                  unsubscribeCalls++
+                }
+              }
+            )
+          },
+          paginationType: "pages" as const,
+          perPage: 10,
+        },
+        currentFilters: {},
+        setCurrentFilters: vi.fn(),
+      }
+
+      // Render the hook
+      const { result } = renderHook(() =>
+        useData(
+          source as DataSource<
+            TestRecord,
+            TestFilters,
+            ActionsDefinition<TestRecord>
+          >
+        )
+      )
+
+      // Wait for initial data to load
+      await act(async () => {
+        await new Promise((resolve) => setTimeout(resolve, 20))
+      })
+
+      // Record the current counts before changing page
+      const initialSubscriptionCount = subscriptionCount
+      const initialUnsubscribeCalls = unsubscribeCalls
+
+      // Change page to trigger a re-fetch
+      act(() => {
+        result.current.setPage(2)
+      })
+
+      // Wait for the new data to load
+      await act(async () => {
+        await new Promise((resolve) => setTimeout(resolve, 20))
+      })
+
+      // Verify that a new subscription was created and the old one was cleaned up
+      expect(subscriptionCount).toBe(initialSubscriptionCount + 1)
+      expect(unsubscribeCalls).toBe(initialUnsubscribeCalls + 1)
+    })
   })
 })

lib/experimental/Collections/useData.ts

@@ -168,6 +168,8 @@ export function useData<
       })
       setIsInitialLoading(false)
       setIsLoading(false)
+      // Clear the cleanup reference when an error occurs
+      cleanup.current = undefined
     },
     [setError, setIsInitialLoading, setIsLoading]
   )
@@ -177,6 +179,12 @@ export function useData<
   const fetchDataAndUpdate = useCallback(
     async (filters: FiltersState<Filters>, currentPage = 1) => {
       try {
+        // Clean up any existing subscription before creating a new one
+        if (cleanup.current) {
+          cleanup.current()
+          cleanup.current = undefined
+        }
+
         const fetcher = (): PromiseOrObservable<ResultType> =>
           dataAdapter.paginationType === "pages"
             ? dataAdapter.fetchData({