add crons job and controller

client/src/App.tsx

@@ -2,14 +2,17 @@ import Router from './router/Router';
 import { RootProvider } from './context/Root';
 import { NavProvider } from './context/Navigation';
 import { AuthProvider } from './context/Auth';
+import { WebSocketProvider } from './context/WebSocket';
 
 function App() {
   return (
     <RootProvider>
       <AuthProvider>
-        <NavProvider>
-          <Router></Router>
-        </NavProvider>
+        <WebSocketProvider>
+          <NavProvider>
+            <Router></Router>
+          </NavProvider>
+        </WebSocketProvider>
       </AuthProvider>
     </RootProvider>
   );

client/src/context/WebSocket.tsx

@@ -0,0 +1,29 @@
+import { createContext, useEffect, useState } from 'react';
+import { io } from 'socket.io-client';
+import { CollectionHttp } from '../http/Collection';
+
+export const navContext = createContext<{}>({});
+
+const { Provider } = navContext;
+
+export const WebSocketProvider = (props: { children: React.ReactNode }) => {
+  // test code for socket
+  useEffect(() => {
+    console.log('----in websocket-----');
+    const socket = io('http://localhost:3000');
+    socket.on('connect', function () {
+      console.log('Connected');
+
+      socket.emit('identity', 0, (res: any) => console.log(res));
+
+      socket.emit('events', { test: 'events' });
+
+      socket.emit('senddata', { test: 'senddata' });
+    });
+    socket.on('COLLECTION', (data: any) => {
+      const collections = data.map((v: any) => new CollectionHttp(v));
+      console.log('event', collections);
+    });
+  }, []);
+  return <Provider value={{}}>{props.children}</Provider>;
+};

client/src/pages/connect/Connect.tsx

@@ -14,7 +14,6 @@ import { MilvusHttp } from '../../http/Milvus';
 import { rootContext } from '../../context/Root';
 import { MILVUS_ADDRESS } from '../../consts/Localstorage';
 import { formatAddress } from '../../utils/Format';
-// import { io } from "socket.io-client";
 
 const useStyles = makeStyles((theme: Theme) => ({
   wrapper: {
@@ -100,33 +99,6 @@ const Connect = () => {
     defaultValue: form.address,
   };
 
-  // test code for socket
-  // useEffect(() => {
-  //   const socket = io('http://localhost:3002');
-  //   socket.on('connect', function () {
-  //     console.log('Connected');
-
-  //     socket.emit('identity', 0, (res: any) =>
-  //       console.log(res));
-
-  //     socket.emit('events', { test: 'events' });
-
-  //     socket.emit('senddata', { test: 'senddata' });
-  //   });
-  //   socket.on('events', (data: any) => {
-  //     console.log('event', data);
-  //   });
-  //   socket.on('senddata', (data: any) => {
-  //     console.log('senddata', data);
-  //   });
-  //   socket.on('exception', (data: any) => {
-  //     console.log('event', data);
-  //   });
-  //   socket.on('disconnect', () => {
-  //     console.log('Disconnected');
-  //   });
-  // }, []);
-
   return (
     <ConnectContainer>
       <section className={classes.wrapper}>

server/package.json

@@ -27,6 +27,7 @@
     "@nestjs/passport": "^8.0.0",
     "@nestjs/platform-express": "^8.0.4",
     "@nestjs/platform-socket.io": "^8.0.4",
+    "@nestjs/schedule": "^1.0.1",
     "@nestjs/serve-static": "^2.2.2",
     "@nestjs/swagger": "^5.0.8",
     "@nestjs/websockets": "^8.0.4",

server/src/app.module.ts

@@ -11,6 +11,8 @@ import { PartitionsModule } from './partitions/partitions.module';
 import { SchemaModule } from './schema/schema.module';
 import { EventsModule } from './events/events.module';
 import { LoggingInterceptor } from './interceptors/index';
+import { CronsModule } from './crons/crons.module';
+import { ScheduleModule } from '@nestjs/schedule';
 
 @Module({
   imports: [
@@ -30,6 +32,8 @@ import { LoggingInterceptor } from './interceptors/index';
     SchemaModule,
     // used for events communication
     EventsModule,
+    CronsModule,
+    ScheduleModule.forRoot(),
   ],
   controllers: [AppController],
   providers: [

server/src/collections/collections.controller.ts

@@ -11,8 +11,8 @@ import {
   ValidationPipe,
   // CACHE_MANAGER,
   // Inject,
-  UseInterceptors,
-  CacheInterceptor,
+  // UseInterceptors,
+  // CacheInterceptor,
 } from '@nestjs/common';
 // import { Cache } from 'cache-manager';
 import { ApiTags } from '@nestjs/swagger';
@@ -64,7 +64,7 @@ export class CollectionsController {
 
   // use interceptor to control cache automatically
   @Get('statistics')
-  @UseInterceptors(CacheInterceptor)
+  // @UseInterceptors(CacheInterceptor)
   async getStatistics() {
     return await this.collectionsService.getStatistics();
   }

server/src/collections/collections.module.ts

@@ -13,5 +13,6 @@ import { ttl } from '../cache/config';
   ],
   providers: [CollectionsService],
   controllers: [CollectionsController],
+  exports: [CollectionsService],
 })
-export class CollectionsModule { }
+export class CollectionsModule {}

server/src/crons/crons.controller.spec.ts

@@ -0,0 +1,18 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { CronsController } from './crons.controller';
+
+describe('CronsController', () => {
+  let controller: CronsController;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      controllers: [CronsController],
+    }).compile();
+
+    controller = module.get<CronsController>(CronsController);
+  });
+
+  it('should be defined', () => {
+    expect(controller).toBeDefined();
+  });
+});

server/src/crons/crons.controller.ts

@@ -0,0 +1,14 @@
+import { Body, Controller, Put } from '@nestjs/common';
+import { CronsService } from './crons.service';
+import { ToggleCron } from './dto';
+
+@Controller('crons')
+export class CronsController {
+  constructor(private cronsService: CronsService) {}
+
+  @Put()
+  async toggleCron(@Body() data: ToggleCron) {
+    console.log(data);
+    return await this.cronsService.toggleCronJobByName(data);
+  }
+}

server/src/crons/crons.module.ts

@@ -0,0 +1,12 @@
+import { Module } from '@nestjs/common';
+import { CollectionsModule } from 'src/collections/collections.module';
+import { EventsModule } from 'src/events/events.module';
+import { CronsService } from './crons.service';
+import { CronsController } from './crons.controller';
+
+@Module({
+  imports: [EventsModule, CollectionsModule],
+  providers: [CronsService],
+  controllers: [CronsController],
+})
+export class CronsModule {}

server/src/crons/crons.service.spec.ts

@@ -0,0 +1,18 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { CronsService } from './crons.service';
+
+describe('CronsService', () => {
+  let service: CronsService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [CronsService],
+    }).compile();
+
+    service = module.get<CronsService>(CronsService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+});

server/src/crons/crons.service.ts

@@ -0,0 +1,28 @@
+import { Injectable } from '@nestjs/common';
+import { EventsGateway } from '../events/events.gateway';
+import { Cron, CronExpression, SchedulerRegistry } from '@nestjs/schedule';
+import { CollectionsService } from '../collections/collections.service';
+import { WS_EVENTS, WS_EVENTS_TYPE } from 'src/utils/Const';
+@Injectable()
+export class CronsService {
+  constructor(
+    private eventService: EventsGateway,
+    private collectionService: CollectionsService,
+    private schedulerRegistry: SchedulerRegistry,
+  ) {}
+
+  async toggleCronJobByName(data: { name: string; type: WS_EVENTS_TYPE }) {
+    const { name, type } = data;
+    const cronJob = this.schedulerRegistry.getCronJob(name);
+    return Number(type) === WS_EVENTS_TYPE.STOP
+      ? cronJob.stop()
+      : cronJob.start();
+  }
+
+  @Cron(CronExpression.EVERY_30_SECONDS, { name: WS_EVENTS.COLLECTION })
+  async getCollections() {
+    const res = await this.collectionService.getAllCollections();
+    this.eventService.server.emit(WS_EVENTS.COLLECTION, res);
+    return res;
+  }
+}

server/src/crons/dto.ts

@@ -0,0 +1,24 @@
+import { ApiProperty } from '@nestjs/swagger';
+import { IsEnum, IsNotEmpty, IsString } from 'class-validator';
+import { WS_EVENTS_TYPE } from '../utils/Const';
+
+export class ToggleCron {
+  @ApiProperty({
+    description: 'Cron job  name',
+  })
+  @IsString()
+  @IsNotEmpty({
+    message: 'Cron job name is empty',
+  })
+  readonly name: string;
+
+  @ApiProperty({
+    description: 'Type allow start->0 stop->1',
+    enum: WS_EVENTS_TYPE,
+  })
+  @IsEnum(WS_EVENTS_TYPE, { message: 'start -> 0, stop -> 1' })
+  @IsNotEmpty({
+    message: 'Toggle type is empty',
+  })
+  readonly type: WS_EVENTS_TYPE;
+}

server/src/events/events.gateway.ts

@@ -8,21 +8,21 @@ import {
 import { from, Observable } from 'rxjs';
 import { map } from 'rxjs/operators';
 import { Server } from 'socket.io';
+import { WS_EVENTS } from 'src/utils/Const';
 
-@WebSocketGateway(3002, {
+@WebSocketGateway({
   cors: {
-    origin: "*",
-    methods: ["GET", "POST"],
-  }
+    origin: '*',
+    methods: ['GET', 'POST'],
+  },
 })
 export class EventsGateway {
   @WebSocketServer()
   server: Server;
 
-  @SubscribeMessage('senddata')
+  @SubscribeMessage(WS_EVENTS.COLLECTION)
   data(@MessageBody() data: unknown): WsResponse<unknown> {
-    const event = 'senddata';
-    return { event, data };
+    return { event: WS_EVENTS.COLLECTION + '', data };
   }
 
   @SubscribeMessage('events')
@@ -30,9 +30,7 @@ export class EventsGateway {
     const event = 'events';
     const response = [1, 2, 3];
 
-    return from(response).pipe(
-      map(data => ({ event, data })),
-    );
+    return from(response).pipe(map((data) => ({ event, data })));
   }
 
   @SubscribeMessage('identity')

server/src/events/events.module.ts

@@ -3,5 +3,6 @@ import { EventsGateway } from './events.gateway';
 
 @Module({
   providers: [EventsGateway],
+  exports: [EventsGateway],
 })
 export class EventsModule {}

server/src/utils/Const.ts

@@ -5,3 +5,12 @@ export enum LOADING_STATE {
   LOADING,
   UNLOADED,
 }
+
+export enum WS_EVENTS {
+  COLLECTION = 'COLLECTION',
+}
+
+export enum WS_EVENTS_TYPE {
+  START,
+  STOP,
+}

server/yarn.lock

@@ -741,6 +741,14 @@
     socket.io "4.1.3"
     tslib "2.3.0"
 
+"@nestjs/schedule@^1.0.1":
+  version "1.0.1"
+  resolved "https://registry.yarnpkg.com/@nestjs/schedule/-/schedule-1.0.1.tgz#4ab3f753c2a0f606cf969b0ec7aa4e655ea444d3"
+  integrity sha512-EU2tB4rxuEgum8JlorAFvXkU982EYZm/IBa7n6kgkyps5BbxQSFf7iR1CLkP9zODO9ApZTWk5z3q9L3O7vrkoQ==
+  dependencies:
+    cron "1.7.2"
+    uuid "8.3.2"
+
 "@nestjs/schematics@^8.0.0", "@nestjs/schematics@^8.0.2":
   version "8.0.2"
   resolved "https://registry.yarnpkg.com/@nestjs/schematics/-/schematics-8.0.2.tgz#ff0d92fcdfd885e66f69132d4ebbd97e4ab56ef5"
@@ -2262,6 +2270,13 @@ create-require@^1.1.0:
   resolved "https://registry.yarnpkg.com/create-require/-/create-require-1.1.1.tgz#c1d7e8f1e5f6cfc9ff65f9cd352d37348756c333"
   integrity sha512-dcKFX3jn0MpIaXjisoRvexIJVEKzaq7z2rZKxf+MSr9TkdmHmsU4m2lcLojrj/FHl8mk5VxMmYA+ftRkP/3oKQ==
 
+cron@1.7.2:
+  version "1.7.2"
+  resolved "https://registry.yarnpkg.com/cron/-/cron-1.7.2.tgz#2ea1f35c138a07edac2ac5af5084ed6fee5723db"
+  integrity sha512-+SaJ2OfeRvfQqwXQ2kgr0Y5pzBR/lijf5OpnnaruwWnmI799JfWr2jN2ItOV9s3A/+TFOt6mxvKzQq5F0Jp6VQ==
+  dependencies:
+    moment-timezone "^0.5.x"
+
 cross-spawn@^6.0.0:
   version "6.0.5"
   resolved "https://registry.yarnpkg.com/cross-spawn/-/cross-spawn-6.0.5.tgz#4a5ec7c64dfae22c3a14124dbacdee846d80cbc4"
@@ -4571,6 +4586,18 @@ mkdirp@^0.5.1:
   dependencies:
     minimist "^1.2.5"
 
+moment-timezone@^0.5.x:
+  version "0.5.33"
+  resolved "https://registry.yarnpkg.com/moment-timezone/-/moment-timezone-0.5.33.tgz#b252fd6bb57f341c9b59a5ab61a8e51a73bbd22c"
+  integrity sha512-PTc2vcT8K9J5/9rDEPe5czSIKgLoGsH8UNpA4qZTVw0Vd/Uz19geE9abbIOQKaAQFcnQ3v5YEXrbSc5BpshH+w==
+  dependencies:
+    moment ">= 2.9.0"
+
+"moment@>= 2.9.0":
+  version "2.29.1"
+  resolved "https://registry.yarnpkg.com/moment/-/moment-2.29.1.tgz#b2be769fa31940be9eeea6469c075e35006fa3d3"
+  integrity sha512-kHmoybcPV8Sqy59DwNDY3Jefr64lK/by/da0ViFcuA4DH0vQg5Q6Ze5VimxkfQNSC+Mls/Kx53s7TjP1RhFEDQ==
+
 ms@2.0.0:
   version "2.0.0"
   resolved "https://registry.yarnpkg.com/ms/-/ms-2.0.0.tgz#5608aeadfc00be6c2901df5f9861788de0d597c8"