项目初始化

datax-common/datax-common-mybatis/src/main/java/cn/datax/common/mybatis/config/DataMetaObjectHandler.java

@@ -26,7 +26,7 @@ public class DataMetaObjectHandler implements MetaObjectHandler {
 		}
 		boolean bolFlowStatus = metaObject.hasSetter("flowStatus");
 		if (bolFlowStatus) {
-			this.strictInsertFill(metaObject, "flowStatus", String.class, DataConstant.AuditState.WAIT);
+			this.strictInsertFill(metaObject, "flowStatus", String.class, DataConstant.AuditState.WAIT.getKey());
 		}
 	}
 

datax-config/src/main/resources/config/datax-gateway-dev.yml

@@ -204,3 +204,15 @@ spring:
               args:
                 name: workflowHystrix
                 fallbackUri: forward:/fallback
+        # 即时通讯消息中心
+        - id: datax-service-websocket
+          uri: ws://localhost:9876
+          predicates:
+            - Path=/websocket/**
+          filters:
+            - SwaggerHeaderFilter
+            - StripPrefix=1
+            - name: Hystrix
+              args:
+                name: websocketHystrix
+                fallbackUri: forward:/fallback
\ No newline at end of file

datax-config/src/main/resources/config/datax-service-websocket-dev.yml

+# 数据源配置
+spring:
+  redis:
+    database: 1
+    host: localhost
+    port: 6379
+    password: 1234@abcd   # 密码（默认为空）
+    timeout: 6000ms       # 连接超时时长（毫秒）
+    lettuce:
+      pool:
+        max-active: 1000  # 连接池最大连接数（使用负值表示没有限制）
+        max-wait: -1ms    # 连接池最大阻塞等待时间（使用负值表示没有限制）
+        max-idle: 10      # 连接池中的最大空闲连接
+        min-idle: 5       # 连接池中的最小空闲连接
+
+# spring security 配置
+security:
+  oauth2:
+    client:
+      access-token-uri: http://localhost:8612/auth/oauth/token
+      user-authorization-uri: http://localhost:8612/auth/oauth/authorize
+      client-id: datax
+      client-secret: 123456
+      scope: all
+    resource:
+      loadBalanced: true
+      token-info-uri: http://localhost:8612/auth/oauth/check_token
+
+# tio 配置
+tio:
+  websocket:
+    server:
+      # websocket port default 9876
+      port: 9876
+      # 心跳时间
+      heartbeat-timeout: 60000
+      use-scanner: true
+    # 集群配置 默认关闭
+    # 集群参考 https://www.t-io.org/doc/tio/322
+    cluster:
+      enabled: false
+      # 集群是通过redis的Pub/Sub实现，所以需要配置Redis
+      redis:
+        ip: localhost
+        port: 6379
+        password: 1234@abcd
+      all: true
+      group: true
+      ip: true
+      user: true
+
+# Swagger界面内容配置
+swagger:
+  enable: true
+  title: API接口文档
+  description: Api Documentation
+  version: 1.0.0
+  basePackage: cn.datax.service.websocket.controller
+  termsOfServiceUrl: http://www.baidu.com
+  contact:
+    name: yuwei
+    url: http://www.baidu.com
+    email: 312075478@qq.com

datax-modules/data-masterdata-service-parent/data-masterdata-service/src/main/java/cn/datax/service/data/masterdata/service/impl/ModelServiceImpl.java

@@ -248,7 +248,7 @@ public class ModelServiceImpl extends BaseServiceImpl<ModelDao, ModelEntity> imp
             // 流程实例标题(动态拼接)
             String tempalte = businessEntity.getBusinessTempalte();
             String businessName = businessEntity.getBusinessName();
-            Map<String, String> parameters = new HashMap<>();
+            Map<String, String> parameters = new HashMap<>(4);
             parameters.put(MsgFormatUtil.TEMPALTE_NICKNAME, SecurityUtil.getNickname());
             parameters.put(MsgFormatUtil.TEMPALTE_DATETIME, DateUtil.formatLocalDateTime(LocalDateTime.now()));
             parameters.put(MsgFormatUtil.TEMPALTE_BUSINESS_NAME, businessName);

datax-modules/pom.xml

@@ -25,5 +25,6 @@
         <module>data-masterdata-service-parent</module>
         <module>data-quality-service-parent</module>
         <module>workflow-service-parent</module>
+        <module>websocket-service</module>
     </modules>
 </project>

datax-modules/system-service-parent/system-service/src/main/java/cn/datax/service/system/controller/MenuController.java

@@ -92,6 +92,7 @@ public class MenuController extends BaseController {
         QueryWrapper<MenuEntity> queryWrapper = new QueryWrapper<>();
         queryWrapper.eq("status", DataConstant.EnableState.ENABLE.getKey());
         queryWrapper.eq("menu_type", DataConstant.MenuType.MENU.getKey());
+        queryWrapper.eq("menu_hidden", DataConstant.EnableState.DISABLE.getKey());
         queryWrapper.isNotNull("menu_code");
         List<MenuEntity> list = menuService.list(queryWrapper);
         List<MenuVo> collect = list.stream().map(menuMapper::toVO).collect(Collectors.toList());

datax-modules/websocket-service/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datax-modules</artifactId>
+        <groupId>cn.datax</groupId>
+        <version>2.0.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <version>2.0.0</version>
+    <description>即时通讯</description>
+    <artifactId>websocket-service</artifactId>
+
+    <dependencies>
+        <!--web 模块-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+            <exclusions>
+                <exclusion>
+                    <artifactId>spring-boot-starter-tomcat</artifactId>
+                    <groupId>org.springframework.boot</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-undertow</artifactId>
+        </dependency>
+        <!--配置中心客户端 -->
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-config</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.github.xiaoymin</groupId>
+            <artifactId>knife4j-micro-spring-boot-starter</artifactId>
+            <version>${knife4j.version}</version>
+        </dependency>
+        <!--feign 依赖-->
+        <dependency>
+            <groupId>io.github.openfeign</groupId>
+            <artifactId>feign-okhttp</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-openfeign</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>cn.datax</groupId>
+            <artifactId>datax-common-core</artifactId>
+            <version>2.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>cn.datax</groupId>
+            <artifactId>datax-common-security</artifactId>
+            <version>2.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>cn.datax</groupId>
+            <artifactId>datax-common-redis</artifactId>
+            <version>2.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.t-io</groupId>
+            <artifactId>tio-websocket-spring-boot-starter</artifactId>
+            <version>${tio-websocket.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>
\ No newline at end of file

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/DataxWebsocketApplication.java

+package cn.datax.service.websocket;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.cloud.client.SpringCloudApplication;
+import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.tio.websocket.starter.EnableTioWebSocketServer;
+
+@EnableTioWebSocketServer
+@EnableFeignClients
+@SpringCloudApplication
+public class DataxWebsocketApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(DataxWebsocketApplication.class);
+    }
+
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/config/DataResourceServerConfig.java

+package cn.datax.service.websocket.config;
+
+import cn.datax.common.security.handler.DataAccessDeniedHandler;
+import cn.datax.common.security.handler.DataAuthExceptionEntryPoint;
+import cn.datax.common.security.utils.DataRedisTokenServices;
+import cn.datax.common.security.utils.RedisTokenStore;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.security.config.annotation.method.configuration.EnableGlobalMethodSecurity;
+import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.oauth2.config.annotation.web.configuration.EnableResourceServer;
+import org.springframework.security.oauth2.config.annotation.web.configuration.ResourceServerConfigurerAdapter;
+import org.springframework.security.oauth2.config.annotation.web.configurers.ResourceServerSecurityConfigurer;
+import org.springframework.security.oauth2.provider.token.TokenStore;
+
+@Configuration
+@EnableResourceServer
+@EnableGlobalMethodSecurity(prePostEnabled = true)
+public class DataResourceServerConfig extends ResourceServerConfigurerAdapter {
+
+    @Autowired
+    private DataAccessDeniedHandler accessDeniedHandler;
+
+    @Autowired
+    private DataAuthExceptionEntryPoint exceptionEntryPoint;
+
+    @Autowired
+    private RedisConnectionFactory redisConnectionFactory;
+
+    @Bean
+    public TokenStore redisTokenStore() {
+        return new RedisTokenStore(redisConnectionFactory);
+    }
+
+    @Override
+    public void configure(ResourceServerSecurityConfigurer resources) {
+        DataRedisTokenServices dataTokenServices = new DataRedisTokenServices();
+        dataTokenServices.setTokenStore(redisTokenStore());
+
+        resources
+                .tokenStore(redisTokenStore())
+                .tokenServices(dataTokenServices)
+                .authenticationEntryPoint(exceptionEntryPoint)
+                .accessDeniedHandler(accessDeniedHandler);
+    }
+
+    @Override
+    public void configure(HttpSecurity http) throws Exception {
+        //允许使用iframe 嵌套，避免swagger-ui 不被加载的问题
+        http.headers().frameOptions().disable();
+        http.authorizeRequests()
+                .antMatchers(
+                        "/actuator/**",
+                        "/v2/api-docs/**",
+                        "/swagger-ui.html",
+                        "/doc.html",
+                        "/swagger-resources/**",
+                        "/webjars/**",
+                        // feign 内部调用不用授权
+                        "/inner/**"
+                ).permitAll()
+                .anyRequest().authenticated()
+                .and().csrf().disable();
+    }
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/config/StartedUpRunner.java

+package cn.datax.service.websocket.config;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.env.Environment;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+
+@Component
+@RequiredArgsConstructor
+public class StartedUpRunner implements ApplicationRunner {
+
+    private final ConfigurableApplicationContext context;
+    private final Environment environment;
+
+    @Override
+    public void run(ApplicationArguments args) {
+        if (context.isActive()) {
+            String banner = "-----------------------------------------\n" +
+                    "服务启动成功，时间：" + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(LocalDateTime.now()) + "\n" +
+                    "服务名称：" + environment.getProperty("spring.application.name") + "\n" +
+                    "端口号：" + environment.getProperty("server.port") + "\n" +
+                    "-----------------------------------------";
+            System.out.println(banner);
+        }
+    }
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/config/SwaggerConfig.java

+package cn.datax.service.websocket.config;
+
+import com.github.xiaoymin.knife4j.spring.annotations.EnableKnife4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.web.bind.annotation.RequestMethod;
+import springfox.bean.validators.configuration.BeanValidatorPluginsConfiguration;
+import springfox.documentation.builders.*;
+import springfox.documentation.schema.ModelRef;
+import springfox.documentation.service.*;
+import springfox.documentation.spi.DocumentationType;
+import springfox.documentation.spring.web.plugins.Docket;
+import springfox.documentation.swagger2.annotations.EnableSwagger2;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Configuration
+@ConditionalOnProperty(prefix = "swagger", name = "enable", havingValue = "true")
+@EnableConfigurationProperties(SwaggerProperties.class)
+@EnableSwagger2
+@EnableKnife4j
+@Import(BeanValidatorPluginsConfiguration.class)
+public class SwaggerConfig {
+
+    @Autowired
+    private SwaggerProperties swaggerProperties;
+
+    /**
+     * 创建API应用
+     * apiInfo() 增加API相关信息
+     * 通过select()函数返回一个ApiSelectorBuilder实例,用来控制哪些接口暴露给Swagger来展现，
+     * 本例采用指定扫描的包路径来定义指定要建立API的目录。
+     *
+     * @return
+     */
+    @Bean
+    public Docket createRestApi(){
+        //版本类型是swagger2
+        return new Docket(DocumentationType.SWAGGER_2)
+                //通过调用自定义方法apiInfo，获得文档的主要信息
+                .apiInfo(apiInfo())
+                //设置全局参数
+                .globalOperationParameters(globalParamBuilder())
+                //设置全局响应参数
+                .globalResponseMessage(RequestMethod.GET,responseBuilder())
+                .globalResponseMessage(RequestMethod.POST,responseBuilder())
+                .globalResponseMessage(RequestMethod.PUT,responseBuilder())
+                .globalResponseMessage(RequestMethod.DELETE,responseBuilder())
+                .select()
+                //扫描该包下面的API注解
+                .apis(RequestHandlerSelectors.basePackage(swaggerProperties.getBasePackage()))
+                .paths(PathSelectors.any())
+                .build()
+                //设置安全认证
+                .securitySchemes(security());
+    }
+    /**
+     * 创建该API的基本信息（这些基本信息会展现在文档页面中）
+     * 访问地址：http://项目实际地址/swagger-ui.html
+     * @return
+     */
+    private ApiInfo apiInfo() {
+        return new ApiInfoBuilder()
+                .title(swaggerProperties.getTitle())
+                .description(swaggerProperties.getDescription())
+                .termsOfServiceUrl(swaggerProperties.getTermsOfServiceUrl())
+                .version(swaggerProperties.getVersion())
+                .contact(new Contact(swaggerProperties.getContact().getName(), swaggerProperties.getContact().getUrl(), swaggerProperties.getContact().getEmail()))
+                .build();
+    }
+
+    /**
+     * 安全认证参数
+     * @return
+     */
+    private List<ApiKey> security() {
+        List<ApiKey> apiKeys = new ArrayList<>();
+        apiKeys.add(new ApiKey("Authorization", "Authorization", "header"));
+        return apiKeys;
+    }
+
+    /**
+     * 构建全局参数列表
+     * @return
+     */
+    private List<Parameter> globalParamBuilder(){
+        List<Parameter> pars = new ArrayList<>();
+        pars.add(parameterBuilder("Authorization","令牌","string","header",false).build());
+        return pars;
+    }
+
+    /**
+     * 创建参数
+     * @return
+     */
+    private ParameterBuilder parameterBuilder(String name, String desc, String type, String parameterType, boolean required) {
+        ParameterBuilder tokenPar = new ParameterBuilder();
+        tokenPar.name(name).description(desc).modelRef(new ModelRef(type)).parameterType(parameterType).required(required).build();
+        return tokenPar;
+    }
+
+    /**
+     * 创建全局响应值
+     * @return
+     */
+    private List<ResponseMessage> responseBuilder() {
+        List<ResponseMessage> responseMessageList = new ArrayList<>();
+        responseMessageList.add(new ResponseMessageBuilder().code(200).message("响应成功").build());
+        responseMessageList.add(new ResponseMessageBuilder().code(500).message("服务器内部错误").build());
+        return responseMessageList;
+    }
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/config/SwaggerProperties.java

+package cn.datax.service.websocket.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(ignoreUnknownFields = false, prefix = "swagger")
+public class SwaggerProperties {
+
+    private Boolean enable;
+    private String title;
+    private String description;
+    private String version;
+    private String termsOfServiceUrl;
+    private String basePackage;
+    private Contact contact;
+
+    public Boolean getEnable() {
+        return enable;
+    }
+
+    public void setEnable(Boolean enable) {
+        this.enable = enable;
+    }
+
+    public String getTitle() {
+        return title;
+    }
+
+    public void setTitle(String title) {
+        this.title = title;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    public void setDescription(String description) {
+        this.description = description;
+    }
+
+    public String getVersion() {
+        return version;
+    }
+
+    public void setVersion(String version) {
+        this.version = version;
+    }
+
+    public String getTermsOfServiceUrl() {
+        return termsOfServiceUrl;
+    }
+
+    public void setTermsOfServiceUrl(String termsOfServiceUrl) {
+        this.termsOfServiceUrl = termsOfServiceUrl;
+    }
+
+    public String getBasePackage() {
+        return basePackage;
+    }
+
+    public void setBasePackage(String basePackage) {
+        this.basePackage = basePackage;
+    }
+
+    public Contact getContact() {
+        return contact;
+    }
+
+    public void setContact(Contact contact) {
+        this.contact = contact;
+    }
+
+    public static class Contact {
+        private String name;
+        private String url;
+        private String email;
+
+        public String getName() {
+            return name;
+        }
+
+        public void setName(String name) {
+            this.name = name;
+        }
+
+        public String getUrl() {
+            return url;
+        }
+
+        public void setUrl(String url) {
+            this.url = url;
+        }
+
+        public String getEmail() {
+            return email;
+        }
+
+        public void setEmail(String email) {
+            this.email = email;
+        }
+    }
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/controller/InnerController.java

+package cn.datax.service.websocket.controller;
+
+import cn.datax.common.base.BaseController;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/inner")
+public class InnerController extends BaseController {
+
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/controller/PushController.java

+package cn.datax.service.websocket.controller;
+
+import cn.datax.common.base.BaseController;
+import cn.hutool.core.util.StrUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.tio.core.Tio;
+import org.tio.websocket.common.WsResponse;
+import org.tio.websocket.starter.TioWebSocketServerBootstrap;
+
+@RestController
+@RequestMapping("/push")
+public class PushController extends BaseController {
+
+    @Autowired
+    private TioWebSocketServerBootstrap bootstrap;
+
+    @GetMapping("/msg")
+    public void pushMessage(String msg){
+        if (StrUtil.isEmpty(msg)){
+            msg = "hello tio websocket";
+        }
+        Tio.sendToAll(bootstrap.getServerTioConfig(), WsResponse.fromText(msg,"utf-8"));
+    }
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/socket/MyWsMsgHandler.java

+package cn.datax.service.websocket.socket;
+
+import org.tio.core.ChannelContext;
+import org.tio.core.Tio;
+import org.tio.http.common.HttpRequest;
+import org.tio.http.common.HttpResponse;
+import org.tio.websocket.common.WsRequest;
+import org.tio.websocket.common.WsResponse;
+import org.tio.websocket.server.handler.IWsMsgHandler;
+
+/**
+ * 和 Tio WebSocket 用法一致，需要实现 IWsMsgHandler 接口，
+ * 可以添加 @Service 注解，不加的话会自动扫描该类(需要配置 tio.websocket.server.use-scanner: true)
+ */
+public class MyWsMsgHandler implements IWsMsgHandler {
+
+    /**
+     * 握手
+     * @param httpRequest
+     * @param httpResponse
+     * @param channelContext
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public HttpResponse handshake(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
+        System.out.println("握手成功");
+        return httpResponse;
+    }
+
+    /**
+     * 握手完毕
+     * @param httpRequest
+     * @param httpResponse
+     * @param channelContext
+     * @throws Exception
+     */
+    @Override
+    public void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
+        System.out.println("握手完毕");
+    }
+
+    /**
+     * binaryType = arraybuffer
+     * @param wsRequest
+     * @param bytes
+     * @param channelContext
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public Object onBytes(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
+        System.out.println("接收到bytes消息:" + new String(bytes));
+        return null;
+    }
+
+    /**
+     * binaryType = blob
+     * @param wsRequest
+     * @param s
+     * @param channelContext
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public Object onText(WsRequest wsRequest, String s, ChannelContext channelContext) throws Exception {
+        System.out.println("接收到blob消息:" + s);
+        Tio.sendToAll(channelContext.getTioConfig(), WsResponse.fromText("服务端收到了消息：" + s, "utf-8"));
+        return null;
+    }
+
+    /**
+     * 关闭
+     * @param wsRequest
+     * @param bytes
+     * @param channelContext
+     * @return
+     * @throws Exception
+     */
+    @Override
+    public Object onClose(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
+        return null;
+    }
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/socket/listener/MyAioListener.java

+package cn.datax.service.websocket.socket.listener;
+
+import org.tio.core.ChannelContext;
+import org.tio.core.intf.Packet;
+import org.tio.websocket.server.WsServerAioListener;
+
+public class MyAioListener extends WsServerAioListener {
+
+    @Override
+    public void onAfterConnected(ChannelContext channelContext, boolean isConnected, boolean isReconnect) throws Exception {
+        super.onAfterConnected(channelContext, isConnected, isReconnect);
+    }
+
+    @Override
+    public void onBeforeClose(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove) throws Exception {
+        super.onBeforeClose(channelContext, throwable, remark, isRemove);
+    }
+
+    @Override
+    public void onAfterSent(ChannelContext channelContext, Packet packet, boolean isSentSuccess) throws Exception {
+        super.onAfterSent(channelContext, packet, isSentSuccess);
+    }
+
+    @Override
+    public void onAfterDecoded(ChannelContext channelContext, Packet packet, int packetSize) throws Exception {
+        super.onAfterDecoded(channelContext, packet, packetSize);
+    }
+
+    @Override
+    public void onAfterHandled(ChannelContext channelContext, Packet packet, long cost) throws Exception {
+        super.onAfterHandled(channelContext, packet, cost);
+    }
+
+    @Override
+    public void onAfterReceivedBytes(ChannelContext channelContext, int receivedBytes) throws Exception {
+        super.onAfterReceivedBytes(channelContext, receivedBytes);
+    }
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/socket/listener/MyGroupListener.java

+package cn.datax.service.websocket.socket.listener;
+
+import org.springframework.stereotype.Service;
+import org.tio.core.ChannelContext;
+import org.tio.core.intf.GroupListener;
+
+@Service
+public class MyGroupListener implements GroupListener {
+
+    @Override
+    public void onAfterBind(ChannelContext channelContext, String s) throws Exception {
+
+    }
+
+    @Override
+    public void onAfterUnbind(ChannelContext channelContext, String s) throws Exception {
+
+    }
+}

datax-modules/websocket-service/src/main/java/cn/datax/service/websocket/socket/listener/MyIpStatListener.java

+package cn.datax.service.websocket.socket.listener;
+
+import org.tio.core.ChannelContext;
+import org.tio.core.TioConfig;
+import org.tio.core.intf.Packet;
+import org.tio.core.stat.IpStat;
+import org.tio.core.stat.IpStatListener;
+
+public class MyIpStatListener implements IpStatListener {
+
+    @Override
+    public void onExpired(TioConfig tioConfig, IpStat ipStat) {
+
+    }
+
+    @Override
+    public void onAfterConnected(ChannelContext channelContext, boolean b, boolean b1, IpStat ipStat) throws Exception {
+
+    }
+
+    @Override
+    public void onDecodeError(ChannelContext channelContext, IpStat ipStat) {
+
+    }
+
+    @Override
+    public void onAfterSent(ChannelContext channelContext, Packet packet, boolean b, IpStat ipStat) throws Exception {
+
+    }
+
+    @Override
+    public void onAfterDecoded(ChannelContext channelContext, Packet packet, int i, IpStat ipStat) throws Exception {
+
+    }
+
+    @Override
+    public void onAfterReceivedBytes(ChannelContext channelContext, int i, IpStat ipStat) throws Exception {
+
+    }
+
+    @Override
+    public void onAfterHandled(ChannelContext channelContext, Packet packet, IpStat ipStat, long l) throws Exception {
+
+    }
+}

datax-modules/websocket-service/src/main/resources/bootstrap.yml

+server:
+  port: 8815
+
+spring:
+  application:
+    name: datax-service-websocket
+  profiles:
+    active: dev
+  cloud:
+    config:
+      fail-fast: true
+      name: ${spring.application.name}
+      profile: ${spring.profiles.active}
+      discovery:
+        enabled: true
+        service-id: datax-config
+
+# 注册中心配置
+eureka:
+  instance:
+    lease-renewal-interval-in-seconds: 20
+  client:
+    register-with-eureka: true
+    fetch-registry: true
+    instance-info-replication-interval-seconds: 30
+    registry-fetch-interval-seconds: 3
+    service-url:
+      defaultZone: http://localhost:8610/eureka
\ No newline at end of file

datax-modules/websocket-service/src/main/resources/logback-spring.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<configuration scan="true" scanPeriod="60 seconds" debug="false">
+    <springProperty scope="context" name="springAppName" source="spring.application.name"/>
+    <property name="log.path" value="logs/datax-service-websocket"/>
+    <property name="log.maxHistory" value="15"/>
+    <property name="log.totalSizeCap" value="500MB"/>
+    <property name="log.maxFileSize" value="10MB"/>
+    <property name="log.colorPattern"
+              value="%magenta(%d{yyyy-MM-dd HH:mm:ss}) %highlight(%-5level) %boldCyan(${springAppName:-}) %yellow(%thread) %green(%logger) %msg%n"/>
+    <property name="log.pattern" value="%d{yyyy-MM-dd HH:mm:ss} %-5level ${springAppName:-} %thread %logger %msg%n"/>
+
+    <!--输出到控制台-->
+    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>${log.colorPattern}</pattern>
+        </encoder>
+    </appender>
+
+    <!--输出到文件-->
+    <!-- RollingFileAppender：滚动记录文件，先将日志记录到指定文件，当符合某个条件时，将日志记录到其他文件 -->
+    <!-- 以下的大概意思是：1.先按日期存日志，日期变了，将前一天的日志文件名重命名为XXX%日期%索引，新的日志仍然是project_info.log -->
+    <!--             2.如果日期没有发生变化，但是当前日志的文件大小超过10MB时，对当前日志进行分割 重命名-->
+    <appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <!--日志文件路径和名称-->
+        <File>${log.path}/info/info.log</File>
+        <!--是否追加到文件末尾,默认为true-->
+        <append>true</append>
+        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+            <!-- 日志文件的名字会根据fileNamePattern的值，每隔一段时间改变一次 -->
+            <!-- 文件名：logs/project_info.2017-12-05.0.log -->
+            <!-- 注意：SizeAndTimeBasedRollingPolicy中 ％i和％d令牌都是强制性的，必须存在，要不会报错 -->
+            <fileNamePattern>${log.path}/info/info.%d.%i.log</fileNamePattern>
+            <!-- 每产生一个日志文件，该日志文件的保存期限为30天, ps:maxHistory的单位是根据fileNamePattern中的翻转策略自动推算出来的,例如上面选用了yyyy-MM-dd,则单位为天
+            如果上面选用了yyyy-MM,则单位为月,另外上面的单位默认为yyyy-MM-dd-->
+            <MaxHistory>${log.maxHistory}</MaxHistory>
+            <!-- 每个日志文件到2mb的时候开始切分，最多保留30天，但最大到500MB，哪怕没到30天也要删除多余的日志 -->
+            <totalSizeCap>${log.totalSizeCap}</totalSizeCap>
+            <!-- maxFileSize:这是活动文件的大小，默认值是10MB，测试时可改成5KB看效果 -->
+            <maxFileSize>${log.maxFileSize}</maxFileSize>
+        </rollingPolicy>
+        <encoder>
+            <pattern>${log.pattern}</pattern>
+        </encoder>
+        <filter class="ch.qos.logback.classic.filter.LevelFilter">
+            <level>INFO</level>
+            <onMatch>ACCEPT</onMatch>
+            <onMismatch>DENY</onMismatch>
+        </filter>
+    </appender>
+
+    <appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <File>${log.path}/error/error.log</File>
+        <append>true</append>
+        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+            <fileNamePattern>${log.path}/error/error.%d.%i.log</fileNamePattern>
+            <MaxHistory>${log.maxHistory}</MaxHistory>
+            <totalSizeCap>${log.totalSizeCap}</totalSizeCap>
+            <maxFileSize>${log.maxFileSize}</maxFileSize>
+        </rollingPolicy>
+        <encoder>
+            <pattern>${log.pattern}</pattern>
+        </encoder>
+        <filter class="ch.qos.logback.classic.filter.LevelFilter">
+            <level>ERROR</level>
+            <onMatch>ACCEPT</onMatch>
+            <onMismatch>DENY</onMismatch>
+        </filter>
+    </appender>
+
+    <root level="debug">
+        <appender-ref ref="console"/>
+    </root>
+
+    <root level="info">
+        <appender-ref ref="file_info"/>
+        <appender-ref ref="file_error"/>
+    </root>
+</configuration>
\ No newline at end of file

datax-ui/src/utils/websocket.js

+const socket = {
+  ws: null,
+  url: 'ws://localhost:8612/websocket',
+  // 'blob' or 'arraybuffer'
+  binaryType: 'blob',
+  ping: 'heartbeat',
+  timeout: 15 * 1000,
+  // 心跳时间 单位：毫秒
+  pingTimeout: 10 * 1000,
+  // 重连间隔时间 单位：毫秒
+  reconnInterval: 2000,
+  timeoutObj: null,
+  pingTimeoutObj: null,
+  // 重连标识
+  lockReconnect: false,
+  // 重连限制
+  repeatLimit: 5,
+  // 重连次数
+  repeat: 0,
+  // 手动关闭连接，不再重连
+  closed: false,
+  // 初始化
+  init() {
+    this.ws = new WebSocket(this.url)
+    this.ws.binaryType = this.binaryType
+    this.ws.onopen = function(event) {
+      console.log('onopen:' + JSON.stringify(event))
+      socket.heartCheckStart()
+    }
+    this.ws.onmessage = function(event) {
+      console.log('onmessage:' + JSON.stringify(event.data))
+      socket.heartCheckReset()
+      socket.heartCheckStart()
+    }
+    this.ws.onclose = function(event) {
+      console.log('onclose:' + JSON.stringify(event))
+      socket.heartCheckReconn()
+    }
+    this.ws.onerror = function(event) {
+      console.log('onerror:' + JSON.stringify(event))
+      socket.heartCheckReconn()
+    }
+  },
+  // 发送消息
+  send(data) {
+    console.log('send:' + JSON.stringify(data))
+    this.ws.send(JSON.stringify(data))
+  },
+  // 手动关闭
+  close() {
+    this.closed = true
+    socket.heartCheckReset()
+    this.ws.close()
+  },
+  // 心跳检测重置
+  heartCheckReset() {
+    const that = this
+    clearTimeout(that.timeoutObj)
+    clearTimeout(that.pingTimeoutObj)
+  },
+  // 心跳检测开始
+  heartCheckStart() {
+    const that = this
+    // 不再重连就不再执行心跳
+    if (that.closed) return
+    that.timeoutObj = setTimeout(function() {
+      // 发送一个心跳，后端收到后，返回一个心跳消息，onmessage拿到返回的心跳就说明连接正常
+      that.send({ 'command': that.ping })
+      // 如果超过一定时间还没重置，说明后端主动断开了
+      that.pingTimeoutObj = setTimeout(function() {
+        that.ws.close()
+      }, that.pingTimeout)
+    }, that.timeout)
+  },
+  // 心跳检测重连
+  heartCheckReconn() {
+    const that = this
+    if (that.repeatLimit > 0 && that.repeatLimit <= that.repeat) return
+    if (that.lockReconnect || that.closed) return
+    that.lockReconnect = true
+    that.repeat++
+    // 没连接上会一直重连，设置延迟避免请求过多
+    setTimeout(() => {
+      that.init()
+      that.lockReconnect = false
+    }, that.reconnInterval)
+  }
+}
+
+export default socket
+

datax-ui/src/views/dashboard/index.vue

@@ -6,6 +6,7 @@
 
 <script>
 import { mapGetters } from 'vuex'
+import websocket from '@/utils/websocket'
 
 export default {
   name: 'Dashboard',
@@ -13,6 +14,12 @@ export default {
     ...mapGetters([
       'user'
     ])
+  },
+  mounted() {
+    websocket.init()
+  },
+  beforeDestroy() {
+    websocket.close()
   }
 }
 </script>

datax-ui/src/views/masterdata/datamodel/DataModelDetail.vue

@@ -3,6 +3,7 @@
     <div slot="header" class="clearfix">
       <span>{{ title }}</span>
       <el-button-group style="float: right;">
+        <el-button size="mini" icon="el-icon-s-data" round @click="submit">提交</el-button>
         <el-button size="mini" icon="el-icon-s-data" round @click="createTable">建模</el-button>
         <el-button size="mini" icon="el-icon-back" round @click="showCard">返回</el-button>
       </el-button-group>
@@ -229,7 +230,7 @@
 </template>
 
 <script>
-import { getDataModel, createTable } from '@/api/masterdata/datamodel'
+import { getDataModel, submitDataModel, createTable } from '@/api/masterdata/datamodel'
 import { listDataDictType } from '@/api/standard/datadict'
 
 export default {
@@ -307,10 +308,20 @@ export default {
     getDataModel: function(id) {
       getDataModel(id).then(response => {
         if (response.success) {
+          console.log(response.data)
           this.form = response.data
         }
       })
     },
+    submit() {
+      submitDataModel(this.data.id).then(response => {
+        if (response.success) {
+          this.$message.success('提交成功')
+        } else {
+          this.$message.warning('提交失败')
+        }
+      })
+    },
     createTable() {
       createTable(this.data.id).then(response => {
         if (response.success) {

datax-ui/src/views/workflow/task/done/TaskDoneList.vue

@@ -89,6 +89,7 @@ export default {
       // 表格头
       tableColumns: [
         { prop: 'name', label: '节点名称', show: true },
+        { prop: 'businessName', label: '业务名称', show: true },
         { prop: 'createTime', label: '开始时间', show: true },
         { prop: 'endTime', label: '结束时间', show: true },
         { prop: 'durationInMillis', label: '耗时', show: true }

datax-ui/src/views/workflow/task/todo/TaskTodoList.vue

@@ -140,6 +140,7 @@ export default {
       // 表格头
       tableColumns: [
         { prop: 'name', label: '节点名称', show: true },
+        { prop: 'businessName', label: '业务名称', show: true },
         { prop: 'createTime', label: '创建时间', show: true }
       ],
       // 表格数据