fixed

DalgoT4D · Dec 15, 2024 · bbb12b7 · bbb12b7
2 parents 8dd28d0 + f70335e
commit bbb12b7
Show file tree

Hide file tree

Showing 9 changed files with 445 additions and 300 deletions.
diff --git a/package.json b/package.json
@@ -33,8 +33,9 @@
     "d3": "^7.8.5",
     "eslint": "8.57.1",
     "eslint-config-next": "^14.2.18",
+    "mock-socket": "^9.3.1",
     "moment": "^2.30.1",
-    "next": "^14.2.15",
+    "next": "^14.2.18",
     "next-auth": "^4.24.10",
     "react": "^18.3.1",
     "react-arborist": "^3.4.0",
@@ -49,7 +50,7 @@
     "remark-gfm": "^4.0.0",
     "superset-ui": "^0.0.0-1",
     "swr": "^2.2.4",
-    "typescript": "5.6.2",
+    "typescript": "5.6.3",
     "use-resize-observer": "^9.1.0"
   },
   "resolutions": {

diff --git a/src/components/Connections/CreateConnectionForm.tsx b/src/components/Connections/CreateConnectionForm.tsx
@@ -10,6 +10,8 @@ import { GlobalContext } from '@/contexts/ContextProvider';
 import { useSession } from 'next-auth/react';
 import { demoAccDestSchema } from '@/config/constant';
 import Input from '../UI/Input/Input';
+import { generateWebsocketUrl } from '@/helpers/websocket';
+import useWebSocket from 'react-use-websocket';
 
 interface CreateConnectionFormProps {
   connectionId: string;
@@ -59,7 +61,7 @@ const CreateConnectionForm = ({
   const isAnyCursorAbsent = useMemo(() => {
     return filteredSourceStreams.some((stream) => !stream.cursorField);
   }, [filteredSourceStreams]);
-
+  console.log(filteredSourceStreams, 'filtered source stream');
   const [loading, setLoading] = useState<boolean>(false);
   const [someStreamSelected, setSomeStreamSelected] = useState<boolean>(false);
   const [normalize, setNormalize] = useState<boolean>(false);
@@ -77,7 +79,7 @@ const CreateConnectionForm = ({
   const setupInitialStreamsState = (catalog: any, connectionId: string | undefined | null) => {
     const action = connectionId ? 'edit' : 'create';
 
-    const streams = catalog.streams.map((el: any) => {
+    const streams = catalog?.streams.map((el: any) => {
       const stream = {
         name: el.stream.name,
         supportsIncremental: el.stream.supportedSyncModes.indexOf('incremental') > -1,
@@ -168,33 +170,51 @@ const CreateConnectionForm = ({
   }, [sourcesData]);
 
   // source selection changes
+  const [socketUrl, setSocketUrl] = useState<string | null>(null);
+  const { sendJsonMessage, lastJsonMessage }: any = useWebSocket(socketUrl, {
+    share: false,
+    onError(event) {
+      console.error('Socket error:', event);
+      setLoading(false);
+    },
+  });
+
+  useEffect(() => {
+    if (session) {
+      setSocketUrl(generateWebsocketUrl('airbyte/connection/schema_catalog', session));
+    }
+  }, [session]);
   useEffect(() => {
     if (watchSourceSelection?.id && !connectionId) {
-      (async () => {
-        setLoading(true);
-        try {
-          const message = await httpGet(
-            session,
-            `airbyte/sources/${watchSourceSelection.id}/schema_catalog`
-          );
-          const streams: SourceStream[] = setupInitialStreamsState(
-            message['catalog'],
-            connectionId
-          );
-          setSourceStreams(streams);
-          setFilteredSourceStreams(streams);
-        } catch (err: any) {
-          if (err.cause) {
-            errorToast(err.cause.detail, [], globalContext);
-          } else {
-            errorToast(err.message, [], globalContext);
-          }
-        }
-        setLoading(false);
-      })();
+      setLoading(true);
+      sendJsonMessage({
+        sourceId: watchSourceSelection.id,
+      });
+      setLoading(true);
+      sendJsonMessage({
+        sourceId: watchSourceSelection.id,
+      });
     }
   }, [watchSourceSelection]);
 
+  useEffect(() => {
+    if (!lastJsonMessage) return;
+
+    const { data, message, status } = lastJsonMessage;
+    const source_schema_catalog = data?.result?.catalog;
+
+    if (status == 'success' && source_schema_catalog) {
+      const streams: SourceStream[] = setupInitialStreamsState(source_schema_catalog, connectionId);
+      setSourceStreams(streams);
+      setFilteredSourceStreams(streams);
+    } else if (status == 'error') {
+      setSourceStreams([]);
+      setFilteredSourceStreams([]);
+      errorToast(message, [], globalContext);
+    }
+    setLoading(false);
+  }, [lastJsonMessage]);
+
   const handleClose = () => {
     reset();
     setConnectionId('');

diff --git a/src/components/Connections/__tests__/ClearConnection.test.tsx b/src/components/Connections/__tests__/ClearConnection.test.tsx
@@ -3,6 +3,8 @@ import { SessionProvider } from 'next-auth/react';
 import { SWRConfig } from 'swr';
 import '@testing-library/jest-dom';
 import { Connections } from '../Connections';
+import { generateWebsocketUrl } from '@/helpers/websocket';
+import { Server } from 'mock-socket';
 
 jest.mock('next/router', () => ({
   useRouter() {
@@ -12,6 +14,10 @@ jest.mock('next/router', () => ({
   },
 }));
 
+jest.mock('@/helpers/websocket', () => ({
+  generateWebsocketUrl: jest.fn(),
+}));
+
 describe('Clear connection suite', () => {
   const mockSession = {
     expires: '1',
@@ -34,6 +40,13 @@ describe('Clear connection suite', () => {
   });
 
   it('clears the connection successfully and handles failure', async () => {
+    const mockServer = new Server('wss://mock-websocket-url');
+    const mockGenerateWebsocketUrl = generateWebsocketUrl;
+
+    // Mock the generateWebsocketUrl function
+    mockGenerateWebsocketUrl.mockImplementation((path, session) => {
+      return 'wss://mock-websocket-url';
+    });
     // Mock fetch for initial connections data
     (global as any).fetch = jest.fn().mockResolvedValueOnce({
       ok: true,