Browse Source

[ui-core] adds useQueueProcessor hook to manage concurrent async task (#3890)

* [ui-core] adds useQueueProcessor hook to manage concurrent async task

* refactor hook to make it minimal
Ram Prasad Agarwal 11 months ago
parent
commit
74c592e619

+ 126 - 0
desktop/core/src/desktop/js/utils/hooks/useQueueProcessor.test.tsx

@@ -0,0 +1,126 @@
+// Licensed to Cloudera, Inc. under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  Cloudera, Inc. licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import { renderHook, act, waitFor } from '@testing-library/react';
+import useQueueProcessor from './useQueueProcessor';
+
+describe('useQueueProcessor', () => {
+  const mockProcessItem = jest.fn(() => new Promise<void>(resolve => setTimeout(resolve, 100)));
+
+  it('should process items concurrently with a maximum of 1 item', async () => {
+    const { result } = renderHook(() =>
+      useQueueProcessor(mockProcessItem, { concurrentProcess: 1 })
+    );
+
+    expect(result.current.queue).toEqual([]);
+    expect(result.current.isLoading).toBe(false);
+
+    act(() => {
+      result.current.enqueue(['item1', 'item2', 'item3']);
+    });
+
+    expect(result.current.queue).toEqual(['item2', 'item3']);
+    expect(result.current.isLoading).toBe(true);
+
+    await waitFor(() => expect(result.current.queue).toEqual(['item3']));
+    expect(result.current.isLoading).toBe(true);
+
+    await waitFor(() => expect(result.current.queue).toEqual([]));
+    await waitFor(() => expect(result.current.isLoading).toBe(false));
+  });
+
+  it('should handle processing multiple items concurrently', async () => {
+    const CONCURRENT_PROCESS = 2;
+    const { result } = renderHook(() =>
+      useQueueProcessor(mockProcessItem, { concurrentProcess: CONCURRENT_PROCESS })
+    );
+
+    expect(result.current.queue).toEqual([]);
+    act(() => {
+      result.current.enqueue(['item1', 'item2', 'item3', 'item4']);
+    });
+
+    expect(result.current.queue).toEqual(['item3', 'item4']);
+    expect(result.current.isLoading).toBe(true);
+
+    await waitFor(() => expect(result.current.queue).toEqual([]));
+    await waitFor(() => expect(result.current.isLoading).toBe(false));
+  });
+
+  it('should add items to the queue when enqueued', () => {
+    const { result } = renderHook(() =>
+      useQueueProcessor(mockProcessItem, { concurrentProcess: 1 })
+    );
+
+    expect(result.current.queue).toEqual([]);
+
+    act(() => {
+      result.current.enqueue(['item1', 'item2']);
+    });
+
+    expect(result.current.queue).toEqual(['item2']);
+
+    act(() => {
+      result.current.enqueue(['item3']);
+    });
+
+    expect(result.current.queue).toEqual(['item2', 'item3']);
+  });
+
+  it('should remove items from the queue when dequeued', async () => {
+    const { result } = renderHook(() =>
+      useQueueProcessor(mockProcessItem, { concurrentProcess: 1 })
+    );
+
+    act(() => {
+      result.current.enqueue(['item1', 'item2', 'item3']);
+    });
+
+    expect(result.current.queue).toEqual(['item2', 'item3']);
+    expect(result.current.isLoading).toBe(true);
+
+    act(() => {
+      result.current.dequeue('item2');
+    });
+
+    expect(result.current.queue).toEqual(['item3']);
+  });
+
+  it('should update isLoading when items are being processed', async () => {
+    const { result } = renderHook(() =>
+      useQueueProcessor(mockProcessItem, { concurrentProcess: 1 })
+    );
+
+    expect(result.current.isLoading).toBe(false);
+
+    act(() => {
+      result.current.enqueue(['item1', 'item2']);
+    });
+
+    await waitFor(() => {
+      expect(result.current.isLoading).toBe(true);
+    });
+
+    await waitFor(() => {
+      expect(result.current.queue).toEqual(['item2']);
+    });
+
+    await waitFor(() => {
+      expect(result.current.queue).toEqual([]);
+      expect(result.current.isLoading).toBe(false);
+    });
+  });
+});

+ 79 - 0
desktop/core/src/desktop/js/utils/hooks/useQueueProcessor.ts

@@ -0,0 +1,79 @@
+// Licensed to Cloudera, Inc. under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  Cloudera, Inc. licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import { useState, useEffect } from 'react';
+
+interface UseQueueProcessorResult<T> {
+  queue: T[];
+  enqueue: (newItems: T[]) => void;
+  dequeue: (item: T) => void;
+  isLoading: boolean;
+}
+
+interface UseQueueProcessorOptions {
+  concurrentProcess: number;
+  onSuccess?: () => void;
+}
+
+const useQueueProcessor = <T>(
+  onItemProcess: (item: T) => Promise<void>,
+  options: UseQueueProcessorOptions
+): UseQueueProcessorResult<T> => {
+  const [isLoading, setIsLoading] = useState<boolean>(false);
+  const [queue, setQueue] = useState<T[]>([]);
+  const [processingQueue, setProcessingQueue] = useState<T[]>([]);
+
+  const enqueue = (newItems: T[]) => {
+    setQueue(prevQueue => [...prevQueue, ...newItems]);
+  };
+
+  const dequeue = (item: T) => {
+    setQueue(prev => prev.filter(i => i !== item));
+  };
+
+  const processQueueItem = async (item: T) => {
+    if (!isLoading) {
+      setIsLoading(true);
+    }
+
+    setProcessingQueue(prev => [...prev, item]);
+    await onItemProcess(item);
+    setProcessingQueue(prev => prev.filter(i => i !== item));
+  };
+
+  useEffect(() => {
+    if (processingQueue.length < options.concurrentProcess && queue.length) {
+      const item = queue[0];
+      setQueue(prev => prev.slice(1));
+      processQueueItem(item);
+    }
+
+    // if all items are processed then call the onSuccess callback
+    if (isLoading && processingQueue.length === 0 && queue.length === 0) {
+      setIsLoading(false);
+      options.onSuccess && options.onSuccess();
+    }
+  }, [queue, processingQueue, options.concurrentProcess, options.onSuccess, isLoading]);
+
+  return {
+    queue,
+    isLoading,
+    enqueue,
+    dequeue
+  };
+};
+
+export default useQueueProcessor;