Merge branch 'ww' into 'main_dev'

fix(socket): socket消息未能成功取消订阅

See merge request yunteng/thingskit-view!39

@@ -8,6 +8,7 @@ import { setOption } from '@/packages/public/chart'
 import { useChartDataSocket } from './useChartDataSocket'
 import { customRequest } from '@/api/external/customRequest'
 import { useFilterFn } from './useFilterFn'
+import { RequestContentTypeEnum } from '@/enums/external/httpEnum'
 
 // 获取类型
 type ChartEditStoreType = typeof useChartEditStore
@@ -26,9 +27,6 @@ export const useChartDataFetch = (
   const vChartRef = ref<typeof VChart | null>(null)
   let fetchInterval: any = 0
 
-  // 数据池
-  const { addGlobalDataInterface } = useChartDataPondFetch()
-
   // 组件类型
   const { chartFrame } = targetComponent.chartConfig
 
@@ -43,6 +41,7 @@ export const useChartDataFetch = (
 
   const requestIntervalFn = () => {
     const chartEditStore = useChartEditStore()
+    if ((targetComponent.request.requestContentType as RequestContentTypeEnum) === RequestContentTypeEnum.WEB_SOCKET) return
 
     // 全局数据
     const {
@@ -107,9 +106,9 @@ export const useChartDataFetch = (
     } catch (error) {
       console.log(error)
     }
-  }
+  } 
 
-  if (isPreview()) {
+  if (isPreview()) { 
     requestIntervalFn()
     const { initial } = useChartDataSocket()
     initial(targetComponent, useChartEditStore, updateCallback)

@@ -35,17 +35,19 @@ const getSocketInstance = (request: ExtraRequestConfigType) => {
   const { requestUrl, requestOriginUrl } = request
   const socketStore = useSocketStore()
   const index = socketConnectionPool.findIndex(item => item.url === requestUrl)
+
   if (~index) {
     return socketConnectionPool[index].ws
   }
   const token = getJwtToken()
   const socketUrl = `${getOriginUrl(requestOriginUrl || '')}${requestUrl}?token=${token}`
 
-  const instance = useWebSocket(socketUrl.replace('undefined', ''), {
+  const instance = useWebSocket(socketUrl, {
     onMessage() {
       const { data: originData } = instance
       const value = parse(unref(originData)) as SocketReceiveMessageType
       socketStore.updateComponentDataset(value)
+      socketStore.unsubscribe(value, instance.send)
     },
     onDisconnected(ws, event) {
       console.log('连接断开')
@@ -64,25 +66,28 @@ export const useChartDataSocket = () => {
   const socketStore = useSocketStore()
 
   const initial = (targetComponent: CreateComponentType, useChartEditStore: ChartEditStoreType, updateCallback?: (...args: any) => any) => {
-    const { request } = targetComponent
 
+    const { request } = targetComponent
     const { requestUrl, requestContentType } = request
 
     if ((requestContentType as RequestContentTypeEnum) !== RequestContentTypeEnum.WEB_SOCKET) return
-
     const { send } = getSocketInstance(request!)
 
     onMounted(() => {
       const message = socketStore.subscribe(targetComponent)
-      send(JSON.stringify(message))
+      if (!message) return
+      const { subscribeMessage } = message
+      send(JSON.stringify(subscribeMessage))
     })
   }
 
   const sendMessage = async (targetComponent: CreateComponentType) => {
     const { request } = unref(targetComponent)
     const message = socketStore.subscribe(unref(targetComponent))
+    if (!message) return
+    const { subscribeMessage } = message
     const { send, data } = getSocketInstance(request)
-    message && send(JSON.stringify(message))
+    send(JSON.stringify(subscribeMessage))
     return socketStore.getComponentValueByKeys(targetComponent, parse(unref(data)))
   }
 

@@ -3,7 +3,8 @@ export enum SocketStoreEnum {
   CONNECTION_POOL = 'connectionPool',
   SUBSCRIBE_POOL = 'subscribePool',
   CACHE_MESSAGE = 'cacheMessage',
-  CURRENT_SUBSCRIBE_ID = 'currentSubscribeId'
+  CURRENT_SUBSCRIBE_ID = 'currentSubscribeId',
+  UNSUBSCRIBE_POOL = 'unsubscribePool'
 }
 
 export interface KeyBoundComponentList {
@@ -55,10 +56,15 @@ export interface SocketComponentRecord {
   keys: string[]
 }
 
+export interface UnsubscribePoolType {
+  subscribeId: number,
+  message: SocketSendMessageType 
+}
+
 export interface SocketStoreType {
   [SocketStoreEnum.CONNECTION_POOL]: SocketConnectionPoolType,
   [SocketStoreEnum.SUBSCRIBE_POOL]: SubscribePoolType[],
   [SocketStoreEnum.CACHE_MESSAGE]: CacheMessageType,
-  [SocketStoreEnum.CURRENT_SUBSCRIBE_ID]: number
-  
+  [SocketStoreEnum.CURRENT_SUBSCRIBE_ID]: number,
+  [SocketStoreEnum.UNSUBSCRIBE_POOL]: UnsubscribePoolType[]
 } 

 import { defineStore } from "pinia";
-import { KeyBoundComponentList, SocketComponentRecord, SocketReceiveMessageType, SocketSendMessageItemType, SocketSendMessageType, SocketStoreType } from '@/store/external/modules/socketStore.d'
+import { KeyBoundComponentList, SocketComponentRecord, SocketReceiveMessageType, SocketSendMessageItemType, SocketSendMessageType, SocketStoreType, UnsubscribePoolType } from '@/store/external/modules/socketStore.d'
 import { CreateComponentType } from "@/packages/index.d";
 import { RequestContentTypeEnum } from "@/enums/external/httpEnum";
 import { useChartEditStore } from "@/store/modules/chartEditStore/chartEditStore";
@@ -16,7 +16,8 @@ export const useSocketStore = defineStore({
     connectionPool: {},
     subscribePool: [],
     cacheMessage: {},
-    currentSubscribeId: 0
+    currentSubscribeId: 0,
+    unsubscribePool: []
   }),
   getters: {
     /**
@@ -39,7 +40,16 @@ export const useSocketStore = defineStore({
   },
   actions: {
     getSubscribeId() {
-      return this.currentSubscribeId += this.currentSubscribeId
+      return this.currentSubscribeId++
+    },
+
+    setUnsubscribePool(message: UnsubscribePoolType) {
+      this.unsubscribePool.push(message)
+    },
+
+    removeUnsubscribePool(message: SocketReceiveMessageType) {
+      const index = this.unsubscribePool.findIndex(item => item.subscribeId === message.subscriptionId)
+      this.unsubscribePool.splice(index, 1)
     },
 
     /**
@@ -76,12 +86,27 @@ export const useSocketStore = defineStore({
      */
     refreshSubscribedMessage(entityId: string) {
       const isExist = this.subscribePool.findIndex(item => item.entityId === entityId)
+      const needUnsubscribe = !!~isExist
+
+      const newSubscribeId = this.getSubscribeId()
+      let oldSubscribeId: number
+      // 订阅设备不存在时
       if (!~isExist) {
-        const subscribeId = this.getSubscribeId()
-        this.subscribePool.push({ subscribeId, entityId })
+
+        this.subscribePool.push({ subscribeId: newSubscribeId, entityId })
+      } else {
+        oldSubscribeId = this.subscribePool.findIndex(item => item.entityId === entityId)
+        this.subscribePool.splice(oldSubscribeId, 1)
+        this.subscribePool.push({ subscribeId: newSubscribeId, entityId })
+      }
+
+      const unSubscribeMessage = needUnsubscribe ? this.createUnSubscribeMessage(oldSubscribeId!, entityId) : null
+      const subscribeMessage = this.createMessage(newSubscribeId, entityId)
+
+      return {
+        unSubscribeMessage,
+        subscribeMessage
       }
-      const subscribeId = this.subscribePool.find(item => item.entityId === entityId)!.subscribeId!
-      return this.createMessage(subscribeId, entityId, !!~isExist)
     },
 
     /**
@@ -90,18 +115,32 @@ export const useSocketStore = defineStore({
      * @param entityId 
      * @returns 
      */
-    createMessage(subscribeId: number, entityId: string, needUnsubscribe: boolean): SocketSendMessageType {
+    createMessage(subscribeId: number, entityId: string): SocketSendMessageType {
       const keys = Object.keys(Reflect.get(this.connectionPool, entityId)).join(',')
       const messageInfo = { entityType: 'DEVICE', entityId: entityId, scope: "LATEST_TELEMETRY", cmdId: subscribeId, keys }
       return {
         tsSubCmds: [
-          ...(needUnsubscribe ? [{ ...messageInfo, unsubscribe: true }] as SocketSendMessageItemType[] : []),
-          messageInfo,
+          messageInfo
         ]
       }
     },
 
     /**
+     * @description 创建取消订阅的消息
+     * @param subscribeId 
+     * @param entityId 
+     */
+    createUnSubscribeMessage(subscribeId: number, entityId: string): SocketSendMessageType {
+      const keys = Object.keys(Reflect.get(this.connectionPool, entityId)).join(',')
+      const messageInfo = { entityType: 'DEVICE', entityId: entityId, scope: "LATEST_TELEMETRY", cmdId: subscribeId, keys, unsubscribe: true } as SocketSendMessageItemType
+      const message = {
+        tsSubCmds: [messageInfo]
+      }
+      this.setUnsubscribePool({ subscribeId, message })
+      return message
+    },
+
+    /**
      * @description 订阅
      * @param targetComponent 
      */
@@ -138,6 +177,7 @@ export const useSocketStore = defineStore({
      */
     getNeedUpdateComponentsIdBySubscribeId(subscribeId: number, keys: string[]) {
       const entityId = this.subscribePool.find(item => item.subscribeId === subscribeId)?.entityId
+
       if (entityId) {
         const keysRecord = Reflect.get(this.connectionPool, entityId)
         const needUpdateComponents = keys.map(key => keysRecord[key])
@@ -160,14 +200,14 @@ export const useSocketStore = defineStore({
       const { keys = '' } = Params
       const targetComponentBindKeys = keys.split(KEYS_SEPARATOR)
 
-      const _value = cloneDeep(value)
+      const _value = cloneDeep(value) || { data: {}, latestValues: {} }
       _value.data = targetComponentBindKeys.reduce((prev, next) => {
-        return { ...prev, [next]: value.data[next] }
+        return { ...prev, [next]: _value.data[next] }
       }, {})
       _value.latestValues = targetComponentBindKeys.reduce((prev, next) => {
-        return { ...prev, [next]: value.latestValues[next] }
+        return { ...prev, [next]: _value.latestValues[next] }
       }, {})
-      
+
       return _value
     },
 
@@ -198,10 +238,23 @@ export const useSocketStore = defineStore({
       const { subscriptionId, data } = value
       const keys = Object.keys(data)
       const componentIds = this.getNeedUpdateComponentsIdBySubscribeId(subscriptionId, keys)
+      console.log(componentIds)
       componentIds?.forEach((targetComponentId) => {
-        this.updateComponentById(targetComponentId as string , value)
+        this.updateComponentById(targetComponentId as string, value)
       })
+    },
 
+    /**
+     * @description socket接受到消息后,从需要取消的订阅池中取消订阅消息
+     */
+    unsubscribe(message: SocketReceiveMessageType, unsubscribeFn: Fn) {
+      const { subscriptionId } = message
+      if (subscriptionId === undefined) return
+      const index = this.unsubscribePool.findIndex(item => item.subscribeId === subscriptionId)
+      if (!~index) return
+      const sendMessage = this.unsubscribePool[index].message
+      unsubscribeFn(JSON.stringify(sendMessage))
+      this.removeUnsubscribePool(message)
     }
   }
 })

@@ -114,6 +114,7 @@ const getConfigurationData = () => {
   record.requestParams[RequestParamsTypeEnum.BODY] = bodyValue
   record.requestInterval = unref(requestIntervalValueRef)
   record.requestIntervalUnit = unref(requestIntervalUnitRef)
+  record.requestContentType = unref(requestContentTypeRef) as unknown as any
   return record
 }
 

@@ -22,8 +22,7 @@ const updateValue = (value: string) => {
       <NTag type="success" style="width: fit-content;">sql</NTag>
     </SettingItemBox>
     <SettingItemBox name="键值">
-      <MonacoEditor v-model:modelValue="value" @update:modelValue="updateValue" width="600px" height="200px"
-        language="sql" />
+      <MonacoEditor :modelValue="value" @update:modelValue="updateValue" width="600px" height="200px" language="sql" />
     </SettingItemBox>
   </section>
 </template>

@@ -52,7 +52,7 @@ const title = ref('')
 const comTitle = computed(() => {
   // eslint-disable-next-line vue/no-side-effects-in-computed-properties  
   const newTitle = projectInfoStore.getProjectInfo.dataViewName
-  setTitle(`工作空间-${newTitle}`)
+  setTitle(`工作空间-${newTitle || ''}`)
   chartEditStore.setEditCanvasConfig(EditCanvasConfigEnum.PROJECT_NAME, newTitle)
   projectInfoStore.setProjectInfoByKey(ProjectInfoEnum.DATA_VIEW_NAME, newTitle)
   return newTitle