会话管理

dlink-admin/src/main/java/com/dlink/controller/StudioController.java

 package com.dlink.controller;
 
 import com.dlink.common.result.Result;
+import com.dlink.dto.SessionDTO;
 import com.dlink.dto.StudioCADTO;
 import com.dlink.dto.StudioDDLDTO;
 import com.dlink.dto.StudioExecuteDTO;
@@ -66,6 +67,14 @@ public class StudioController {
         }
     }
 
+    /**
+     * 创建session
+     */
+    @PostMapping("/createSession")
+    public Result createSession(@RequestBody SessionDTO sessionDTO)  {
+        return Result.succeed(studioService.createSession(sessionDTO,"admin"),"创建成功");
+    }
+
     /**
      * 清除指定session
      */
@@ -88,4 +97,12 @@ public class StudioController {
             return Result.failed("请选择要清除的记录");
         }
     }
+
+    /**
+     * 获取session列表
+     */
+    @GetMapping("/listSession")
+    public Result listSession()  {
+        return Result.succeed(studioService.listSession("admin"),"获取成功");
+    }
 }

dlink-admin/src/main/java/com/dlink/dto/SessionDTO.java

+package com.dlink.dto;
+
+import com.dlink.session.SessionConfig;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * SessionDTO
+ *
+ * @author wenmo
+ * @since 2021/7/6 22:10
+ */
+@Getter
+@Setter
+public class SessionDTO {
+    private String session;
+    private String type;
+    private boolean useRemote;
+    private Integer clusterId;
+
+}

dlink-admin/src/main/java/com/dlink/service/StudioService.java

 package com.dlink.service;
 
+import com.dlink.dto.SessionDTO;
 import com.dlink.dto.StudioDDLDTO;
 import com.dlink.dto.StudioExecuteDTO;
 import com.dlink.explainer.ca.TableCANode;
 import com.dlink.job.JobResult;
 import com.dlink.result.IResult;
 import com.dlink.result.SelectResult;
+import com.dlink.session.SessionInfo;
 
 import java.util.List;
 
@@ -23,8 +25,12 @@ public interface StudioService {
 
     SelectResult getJobData(String jobId);
 
+    boolean createSession(SessionDTO sessionDTO,String createUser);
+
     boolean clearSession(String session);
 
+    List<SessionInfo> listSession(String createUser);
+
     List<TableCANode> getOneTableCAByStatement(String statement);
 
     List<TableCANode> getOneTableColumnCAByStatement(String statement);

dlink-admin/src/main/java/com/dlink/service/impl/ClusterServiceImpl.java

@@ -43,7 +43,6 @@ public class ClusterServiceImpl extends SuperServiceImpl<ClusterMapper, Cluster>
 
     @Override
     public String buildEnvironmentAddress(boolean useRemote, Integer id) {
-        String address = FlinkConstant.LOCAL_HOST;
         if(useRemote) {
             return getJobManagerAddress(getById(id));
         }else{
@@ -56,7 +55,7 @@ public class ClusterServiceImpl extends SuperServiceImpl<ClusterMapper, Cluster>
                 e.printStackTrace();
             }
         }
-        return address;
+        return FlinkConstant.LOCAL_HOST;
     }
 
     @Override

dlink-admin/src/main/java/com/dlink/service/impl/StudioServiceImpl.java

@@ -3,6 +3,7 @@ package com.dlink.service.impl;
 import com.dlink.assertion.Assert;
 import com.dlink.cluster.FlinkCluster;
 import com.dlink.constant.FlinkConstant;
+import com.dlink.dto.SessionDTO;
 import com.dlink.dto.StudioDDLDTO;
 import com.dlink.dto.StudioExecuteDTO;
 import com.dlink.exception.BusException;
@@ -20,6 +21,8 @@ import com.dlink.result.RunResult;
 import com.dlink.result.SelectResult;
 import com.dlink.service.ClusterService;
 import com.dlink.service.StudioService;
+import com.dlink.session.SessionConfig;
+import com.dlink.session.SessionInfo;
 import com.dlink.session.SessionPool;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -61,6 +64,16 @@ public class StudioServiceImpl implements StudioService {
         return JobManager.getJobData(jobId);
     }
 
+    @Override
+    public boolean createSession(SessionDTO sessionDTO,String createUser) {
+        Cluster cluster = clusterService.getById(sessionDTO.getClusterId());
+        SessionConfig sessionConfig = SessionConfig.build(
+                sessionDTO.getType(), sessionDTO.isUseRemote(),
+                cluster.getId(),cluster.getAlias(),
+                clusterService.buildEnvironmentAddress(sessionDTO.isUseRemote(),sessionDTO.getClusterId()));
+        return JobManager.createSession(sessionDTO.getSession(),sessionConfig,createUser);
+    }
+
     @Override
     public boolean clearSession(String session) {
         if(SessionPool.remove(session)>0){
@@ -70,6 +83,11 @@ public class StudioServiceImpl implements StudioService {
         }
     }
 
+    @Override
+    public List<SessionInfo> listSession(String createUser) {
+        return JobManager.listSession(createUser);
+    }
+
     @Override
     public List<TableCANode> getOneTableCAByStatement(String statement) {
         return CABuilder.getOneTableCAByStatement(statement);

dlink-core/src/main/java/com/dlink/executor/ExecutorSetting.java

@@ -20,7 +20,7 @@ public class ExecutorSetting {
     private String savePointPath;
     private String jobName;
     private Map<String,String> config;
-    public static final ExecutorSetting DEFAULT = new ExecutorSetting(false);
+    public static final ExecutorSetting DEFAULT = new ExecutorSetting(true);
 
     public ExecutorSetting(boolean useSqlFragment) {
         this.useSqlFragment = useSqlFragment;

dlink-core/src/main/java/com/dlink/job/JobManager.java

@@ -9,6 +9,8 @@ import com.dlink.interceptor.FlinkInterceptor;
 import com.dlink.parser.SqlType;
 import com.dlink.result.*;
 import com.dlink.session.ExecutorEntity;
+import com.dlink.session.SessionConfig;
+import com.dlink.session.SessionInfo;
 import com.dlink.session.SessionPool;
 import com.dlink.trans.Operations;
 import org.apache.flink.api.common.JobID;
@@ -264,7 +266,21 @@ public class JobManager extends RunTime {
         return ResultPool.get(jobId);
     }
 
-    public static void createSession(String session){
+    public static boolean createSession(String session, SessionConfig sessionConfig,String createUser){
+        if(SessionPool.exist(session)){
+            return false;
+        }
+        Executor sessionExecutor = null;
+        if (sessionConfig.isUseRemote()) {
+            sessionExecutor = Executor.buildRemoteExecutor(EnvironmentSetting.build(sessionConfig.getAddress()), ExecutorSetting.DEFAULT);
+        } else {
+            sessionExecutor = Executor.buildLocalExecutor(sessionConfig.getExecutorSetting());
+        }
+        SessionPool.push(new ExecutorEntity(session,sessionConfig,createUser,LocalDateTime.now(), sessionExecutor));
+        return true;
+    }
 
+    public static List<SessionInfo> listSession(String createUser){
+        return SessionPool.filter(createUser);
     }
 }

dlink-core/src/main/java/com/dlink/session/ExecutorEntity.java

@@ -16,31 +16,21 @@ import java.time.LocalDateTime;
 @Getter
 public class ExecutorEntity {
     private String sessionId;
-    private SessionType type;
-    private boolean useRemote;
-    private String address;
+    private SessionConfig sessionConfig;
     private String createUser;
     private LocalDateTime createTime;
     private Executor executor;
 
-    public enum SessionType{
-        PUBLIC,
-        PRIVATE
-    }
-
     public ExecutorEntity(String sessionId, Executor executor) {
         this.sessionId = sessionId;
         this.executor = executor;
     }
 
-    public ExecutorEntity(String sessionId, SessionType type, boolean useRemote, String address, String createUser, LocalDateTime createTime, Executor executor) {
+    public ExecutorEntity(String sessionId, SessionConfig sessionConfig, String createUser, LocalDateTime createTime, Executor executor) {
         this.sessionId = sessionId;
-        this.type = type;
-        this.useRemote = useRemote;
-        this.address = address;
+        this.sessionConfig = sessionConfig;
         this.createUser = createUser;
         this.createTime = createTime;
         this.executor = executor;
     }
-
 }

dlink-core/src/main/java/com/dlink/session/SessionConfig.java

+package com.dlink.session;
+
+import com.dlink.executor.ExecutorSetting;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * SessionConfig
+ *
+ * @author wenmo
+ * @since 2021/7/6 21:59
+ */
+@Getter
+@Setter
+public class SessionConfig {
+    private SessionType type;
+    private boolean useRemote;
+    private Integer clusterId;
+    private String clusterName;
+    private String address;
+
+    public enum SessionType{
+        PUBLIC,
+        PRIVATE
+    }
+
+    public SessionConfig(SessionType type, boolean useRemote, Integer clusterId, String clusterName, String address) {
+        this.type = type;
+        this.useRemote = useRemote;
+        this.clusterId = clusterId;
+        this.clusterName = clusterName;
+        this.address = address;
+    }
+
+    public static SessionConfig build(String type, boolean useRemote, Integer clusterId, String clusterName, String address){
+        return new SessionConfig(SessionType.valueOf(type),useRemote,clusterId,clusterName,address);
+    }
+
+    public ExecutorSetting getExecutorSetting(){
+        return new ExecutorSetting(true);
+    }
+
+}

dlink-core/src/main/java/com/dlink/session/SessionInfo.java

+package com.dlink.session;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.time.LocalDateTime;
+
+/**
+ * SessionInfo
+ *
+ * @author wenmo
+ * @since 2021/7/6 22:22
+ */
+@Setter
+@Getter
+public class SessionInfo {
+    private String session;
+    private SessionConfig sessionConfig;
+    private String createUser;
+    private LocalDateTime createTime;
+
+    public SessionInfo(String session, SessionConfig sessionConfig, String createUser, LocalDateTime createTime) {
+        this.session = session;
+        this.sessionConfig = sessionConfig;
+        this.createUser = createUser;
+        this.createTime = createTime;
+    }
+
+    public static SessionInfo build(ExecutorEntity executorEntity){
+        return new SessionInfo(executorEntity.getSessionId(),executorEntity.getSessionConfig(),executorEntity.getCreateUser(),executorEntity.getCreateTime());
+    }
+
+}

dlink-core/src/main/java/com/dlink/session/SessionPool.java

@@ -2,6 +2,7 @@ package com.dlink.session;
 
 import com.dlink.constant.FlinkConstant;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Vector;
 
@@ -14,7 +15,16 @@ import java.util.Vector;
 public class SessionPool {
 
     private static volatile List<ExecutorEntity> executorList = new Vector<>(FlinkConstant.DEFAULT_SESSION_COUNT);
-    
+
+    public static boolean exist(String sessionId) {
+        for (ExecutorEntity executorEntity : executorList) {
+            if (executorEntity.getSessionId().equals(sessionId)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     public static Integer push(ExecutorEntity executorEntity){
         if (executorList.size() >= FlinkConstant.DEFAULT_SESSION_COUNT * FlinkConstant.DEFAULT_FACTOR) {
             executorList.remove(0);
@@ -48,4 +58,18 @@ public class SessionPool {
     public static List<ExecutorEntity> list(){
         return executorList;
     }
+
+    public static List<SessionInfo> filter(String createUser){
+        List<SessionInfo> sessionInfos = new ArrayList<>();
+        for (ExecutorEntity item : executorList) {
+            if(item.getSessionConfig().getType()== SessionConfig.SessionType.PUBLIC){
+                sessionInfos.add(SessionInfo.build(item));
+            }else{
+                if(createUser!=null&&createUser.equals(item.getCreateUser())){
+                    sessionInfos.add(SessionInfo.build(item));
+                }
+            }
+        }
+        return sessionInfos;
+    }
 }

dlink-web/src/components/Studio/StudioLeftTool/StudioConnector/components/SessionForm.tsx

+import React, {useEffect, useState} from 'react';
+import {Form, Button, Input, Modal} from 'antd';
+
+import type {SessionItem} from '../data.d';
+
+export type UpdateFormProps = {
+  onCancel: (flag?: boolean, formVals?: Partial<SessionItem>) => void;
+  onSubmit: (values: Partial<SessionItem>) => void;
+  updateModalVisible: boolean;
+  isCreate: boolean;
+  values: Partial<SessionItem>;
+};
+
+const FormItem = Form.Item;
+
+const formLayout = {
+  labelCol: {span: 7},
+  wrapperCol: {span: 13},
+};
+
+const SessionForm: React.FC<UpdateFormProps> = (props) => {
+  const [formVals, setFormVals] = useState<Partial<SessionItem>>({
+    session: props.values.session,
+    type: props.values.sessionConfig?.type,
+    useRemote: props.values.sessionConfig?.useRemote,
+    address: props.values.sessionConfig?.address,
+    createUser: props.values.createUser,
+    createTime: props.values.createTime,
+  });
+
+  const [form] = Form.useForm();
+
+  const {
+    onSubmit: handleUpdate,
+    onCancel: handleUpdateModalVisible,
+    updateModalVisible,
+    values,
+    isCreate,
+  } = props;
+
+  const submitForm = async () => {
+    const fieldsValue = await form.validateFields();
+    setFormVals({...formVals, ...fieldsValue});
+    handleUpdate({...formVals, ...fieldsValue});
+  };
+
+  const renderContent = () => {
+    return (
+      <>
+        <FormItem
+          name="session"
+          label="名称"
+          rules={[{required: true, message: '请输入唯一名称！'}]}>
+          <Input placeholder="请输入"/>
+        </FormItem>
+        <FormItem
+          name="alias"
+          label="别名"
+          rules={[{required: true, message: '请输入别名！'}]}>
+          <Input placeholder="请输入"/>
+        </FormItem>
+      </>
+    );
+  };
+
+  const renderFooter = () => {
+    return (
+      <>
+        <Button onClick={() => handleUpdateModalVisible(false, values)}>取消</Button>
+        <Button type="primary" onClick={() => submitForm()}>
+          完成
+        </Button>
+      </>
+    );
+  };
+
+  return (
+    <Modal
+      width={640}
+      bodyStyle={{padding: '32px 40px 48px'}}
+      destroyOnClose
+      title={isCreate ? '创建新作业' : ('重命名作业-' + formVals.name)}
+      visible={updateModalVisible}
+      footer={renderFooter()}
+      onCancel={() => handleUpdateModalVisible()}
+    >
+      <Form
+        {...formLayout}
+        form={form}
+        initialValues={{
+          id: formVals.id,
+          name: formVals.name,
+          alias: formVals.alias,
+          parentId: formVals.parentId,
+        }}
+      >
+        {renderContent()}
+      </Form>
+    </Modal>
+  );
+};
+
+export default SessionForm;

dlink-web/src/components/Studio/StudioLeftTool/StudioConnector/data.d.ts

+
+export type SessionItem = {
+  session: string,
+  sessionConfig: {
+    type:string,
+    useRemote:boolean,
+    clusterId:number,
+  },
+  createUser: string,
+  createTime: string,
+};

dlink-web/src/components/Studio/StudioLeftTool/StudioConnector/index.tsx

@@ -4,7 +4,7 @@ import {StateType} from "@/pages/FlinkSqlStudio/model";
 import {connect} from "umi";
 import {useState} from "react";
 import styles from "./index.less";
-import { SearchOutlined,DownOutlined,DeleteOutlined,CommentOutlined ,MessageOutlined} from '@ant-design/icons';
+import { SearchOutlined,DownOutlined,DeleteOutlined,CommentOutlined ,MessageOutlined,PlusOutlined} from '@ant-design/icons';
 import React from "react";
 import {removeTable, showTables,clearSession} from "@/components/Studio/StudioEvent/DDL";
 import {
@@ -203,6 +203,13 @@ const StudioConnector = (props:any) => {
             onClick={showSessions}
           />
         </Tooltip>
+        <Tooltip title="新建会话">
+          <Button
+            type="text"
+            icon={<PlusOutlined />}
+            onClick={showSessions}
+          />
+        </Tooltip>
         <Tooltip title="刷新连接器">
           <Button
             type="text"