feat: handle mq message

api/core/workflow/nodes/mq/mq_node.py

@@ -8,7 +8,7 @@ from core.workflow.nodes.mq.entities import MqNodeData
 from core.workflow.nodes.mq.rabbitmq_client import RabbitMQClient
 
 # 创建全局RabbitMQ客户端实例
-rabbitmq_client = RabbitMQClient("dify_node") 
+rabbitmq_client = RabbitMQClient("dify_node")
 
 class MqNode(BaseNode[MqNodeData]):
     _node_data_cls = MqNodeData
@@ -20,8 +20,17 @@ class MqNode(BaseNode[MqNodeData]):
         :return:
         """
         print("go go go execute execute execute")
+        print("ddd:", self.node_data)
         node_inputs: dict[str, list] = {"conditions": []}
 
+        newMessage = self.graph_runtime_state.variable_pool.convert_template(
+            self.node_data.message
+        ).text
+        newChannel = self.graph_runtime_state.variable_pool.convert_template(
+            self.node_data.channel
+        ).text
+        print("new message:", newMessage)
+        print("new newChannel:", newChannel)
         process_data: dict[str, list] = {"condition_results": []}
 
         input_conditions = []
@@ -30,7 +39,8 @@ class MqNode(BaseNode[MqNodeData]):
         try:
             rabbitmq_client.publish_json({
                 "action": "downloadImage",
-                "msg": '你好'
+                "msg": newMessage,
+                "channel": newChannel
             })
         except Exception:
             print("err")

web/app/components/workflow/constants.ts

@@ -361,7 +361,7 @@ export const NODES_INITIAL_DATA = {
     title: '',
     desc: '',
     variables: [],
-    ...HttpRequestDefault.defaultValue,
+    ...MqDefault.defaultValue,
   },
   [BlockEnum.ParameterExtractor]: {
     type: BlockEnum.ParameterExtractor,

web/app/components/workflow/hooks/use-checklist.ts

@@ -100,9 +100,9 @@ export const useChecklist = (nodes: Node[], edges: Edge[]) => {
 
       if (node.type === CUSTOM_NODE) {
         const checkData = getCheckData(node.data)
-        console.log('node', node)
-        console.log('checkData', checkData)
-        console.log('nodesExtraData', nodesExtraData)
+        // console.log('node', node)
+        // console.log('checkData', checkData)
+        // console.log('nodesExtraData', nodesExtraData)
         const { errorMessage } = nodesExtraData[node.data.type].checkValid(checkData, t, moreDataForCheckValid)
 
         if (errorMessage || !validNodes.find(n => n.id === node.id)) {

web/app/components/workflow/nodes/constants.ts

@@ -39,6 +39,7 @@ import ListFilterPanel from './list-operator/panel'
 import AgentNode from './agent/node'
 import AgentPanel from './agent/panel'
 import MqNode from './mq/node'
+import MqPanel from './mq/panel'
 import { TransferMethod } from '@/types/app'
 
 export const NodeComponentMap: Record<string, ComponentType<any>> = {
@@ -86,7 +87,7 @@ export const PanelComponentMap: Record<string, ComponentType<any>> = {
   [BlockEnum.DocExtractor]: DocExtractorPanel,
   [BlockEnum.ListFilter]: ListFilterPanel,
   [BlockEnum.Agent]: AgentPanel,
-  [BlockEnum.MqNode]: MqNode,
+  [BlockEnum.MqNode]: MqPanel,
 }
 
 export const CUSTOM_NODE_TYPE = 'custom'

web/app/components/workflow/nodes/index.tsx

@@ -14,7 +14,7 @@ import BasePanel from './_base/panel'
 
 const CustomNode = (props: NodeProps) => {
   const nodeData = props.data
-  console.log('zzq see nodeData', props)
+  // console.log('zzq see nodeData', props)
   const NodeComponent = NodeComponentMap[nodeData.type]
 
   return (

web/app/components/workflow/nodes/mq/node.tsx

@@ -37,35 +37,35 @@ const MqNode: FC<NodeProps<any>> = ({ data }) => {
           readOnly
         />
       </div>
-      <div>
-        <label style={{
-          display: 'block',
-          marginBottom: '4px',
-          fontSize: '14px',
-          color: '#495057',
-          fontWeight: '500',
-        }}>
-          Message
-        </label>
-        <textarea
-          style={{
-            width: '100%',
-            padding: '8px 12px',
-            border: '1px solid #ced4da',
-            borderRadius: '4px',
-            fontSize: '14px',
-            transition: 'border-color 0.15s ease-in-out',
-            boxSizing: 'border-box',
-            minHeight: '100px',
-            resize: 'vertical',
-            fontFamily: 'inherit',
-            lineHeight: '1.5',
-          }}
-          value={data.mqValue || ''}
-          placeholder="请输入要发送的消息"
-          readOnly
-        />
-      </div>
+      {/* <div> */}
+      {/*  <label style={{ */}
+      {/*    display: 'block', */}
+      {/*    marginBottom: '4px', */}
+      {/*    fontSize: '14px', */}
+      {/*    color: '#495057', */}
+      {/*    fontWeight: '500', */}
+      {/*  }}> */}
+      {/*    Message */}
+      {/*  </label> */}
+      {/*  <textarea */}
+      {/*    style={{ */}
+      {/*      width: '100%', */}
+      {/*      padding: '8px 12px', */}
+      {/*      border: '1px solid #ced4da', */}
+      {/*      borderRadius: '4px', */}
+      {/*      fontSize: '14px', */}
+      {/*      transition: 'border-color 0.15s ease-in-out', */}
+      {/*      boxSizing: 'border-box', */}
+      {/*      minHeight: '100px', */}
+      {/*      resize: 'vertical', */}
+      {/*      fontFamily: 'inherit', */}
+      {/*      lineHeight: '1.5', */}
+      {/*    }} */}
+      {/*    value={data.mqValue || ''} */}
+      {/*    placeholder="请输入要发送的消息" */}
+      {/*    readOnly */}
+      {/*  /> */}
+      {/* </div> */}
     </div>
   )
 }

web/app/components/workflow/nodes/mq/panel.tsx

@@ -0,0 +1,58 @@
+import type { FC } from 'react'
+import React from 'react'
+import { useTranslation } from 'react-i18next'
+import useConfig from './use-config'
+import type { MqNodeType } from './types'
+import Editor from '@/app/components/workflow/nodes/_base/components/prompt/editor'
+import type { NodePanelProps } from '@/app/components/workflow/types'
+import useAvailableVarList from '@/app/components/workflow/nodes/_base/hooks/use-available-var-list'
+import InputWithVar from '@/app/components/workflow/nodes/_base/components/prompt/editor'
+const i18nPrefix = 'workflow.nodes.answer'
+
+const Panel: FC<NodePanelProps<MqNodeType>> = ({
+  id,
+  data,
+}) => {
+  const { t } = useTranslation()
+
+  const {
+    readOnly,
+    inputs,
+    handleMessageChange,
+    handleChannelNameChange,
+    filterVar,
+  } = useConfig(id, data)
+
+  const { availableVars, availableNodesWithParent } = useAvailableVarList(id, {
+    onlyLeafNodeVar: false,
+    hideChatVar: false,
+    hideEnv: false,
+    filterVar,
+  })
+
+  return (
+    <div className='mb-2 mt-2 space-y-4 px-4'>
+      <Editor
+        readOnly={readOnly}
+        justVar
+        title={'Channel'}
+        value={inputs.channelName}
+        onChange={handleChannelNameChange}
+        nodesOutputVars={availableVars}
+        availableNodes={availableNodesWithParent}
+        isSupportFileVar
+      />
+      <div className='mt-1'></div>
+      <InputWithVar
+        title='数据'
+        value={inputs.message}
+        onChange={handleMessageChange}
+        justVar
+        nodesOutputVars={availableVars}
+        availableNodes={availableNodesWithParent}
+      />
+    </div>
+  )
+}
+
+export default React.memo(Panel)

web/app/components/workflow/nodes/mq/types.ts

@@ -0,0 +1,6 @@
+import type { CommonNodeType } from '@/app/components/workflow/types'
+
+export type MqNodeType = CommonNodeType & {
+    channelName: string
+    message: string
+}

web/app/components/workflow/nodes/mq/use-config.ts

@@ -0,0 +1,40 @@
+import { useCallback } from 'react'
+import produce from 'immer'
+import { VarType } from '../../types'
+import type { Var } from '../../types'
+import useNodeCrud from '@/app/components/workflow/nodes/_base/hooks/use-node-crud'
+import {
+    useNodesReadOnly,
+} from '@/app/components/workflow/hooks'
+import type { MqNodeType } from '@/app/components/workflow/nodes/mq/types'
+
+const useConfig = (id: string, payload: MqNodeType) => {
+    const { nodesReadOnly: readOnly } = useNodesReadOnly()
+    const { inputs, setInputs } = useNodeCrud<MqNodeType>(id, payload)
+
+    const handleChannelNameChange = useCallback((value: string) => {
+        const newInputs = produce(inputs, (draft) => {
+            draft.channelName = value
+        })
+        setInputs(newInputs)
+    }, [inputs, setInputs])
+    const handleMessageChange = useCallback((value: string) => {
+        const newInputs = produce(inputs, (draft) => {
+            draft.message = value
+        })
+        setInputs(newInputs)
+    }, [inputs, setInputs])
+
+    const filterVar = useCallback((varPayload: Var) => {
+        return varPayload.type !== VarType.arrayObject
+    }, [])
+    return {
+        readOnly,
+        inputs,
+        handleChannelNameChange,
+        handleMessageChange,
+        filterVar,
+    }
+}
+
+export default useConfig