Merge pull request #173 from WeBankFinTech/develop

release eventmesh v1.1.1

README.md

@@ -36,6 +36,7 @@ Key components:
 | ----    | ----    |
 | v1.0.0  |Support DeFiBus as eventstore, support pub/sub, http api, java-sdk|
 | v1.1.0  |Support rocketmq as eventstore|
+| v1.1.1  |Support https|
 | v1.2.0  |Support transaction event|
 | v1.3.0  |Support Plug-in architectur|
 |         |Support Event Sourcing|

README.zh-CN.md

@@ -34,6 +34,7 @@ Event Mesh的一般功能:
 | ----    | ----    |
 | v1.0.0  |Support DeFiBus as eventstore, support pub/sub, http api, java-sdk|
 | v1.1.0  |Support rocketmq as eventstore|
+| v1.1.1  |Support https|
 | v1.2.0  |Support transaction event|
 | v1.3.0  |Support Plug-in architectur|
 |         |Support Event Sourcing|

docs/cn/features/https.zh-CN.md

@@ -0,0 +1,31 @@
+1.在eventmesh-emesher 中配置
+
+```
+proxy.properties(添加如下配置)
+proxy.server.useTls.enabled=true   //默认值 false
+
+
+config env varible
+-Dssl.server.protocol=TLSv1.1   //默认值 TLSv1.1 
+-Dssl.server.cer=sChat2.jks     //把文件放到启动脚本start.sh 指定的conPath目录下
+-Dssl.server.pass=sNetty
+
+```
+
+
+2.在eventmesh-sdk-java 中配置
+```
+//创建producer
+LiteClientConfig liteClientConfig = new liteClientConfig();
+...
+
+//设置开启TLS
+liteClientConfig.setUseTls(true);
+LiteProducer producer = new LiteProducer(liteClientConfig);
+
+
+//配置环境变量
+-Dssl.client.protocol=TLSv1.1   //默认值 TLSv1.1 
+-Dssl.client.cer=sChat2.jks     //把文件放到应用指定的conPath目录下
+-Dssl.client.pass=sNetty
+```

docs/en/features/https.md

@@ -0,0 +1,30 @@
+1.config in eventmesh-emesher
+
+```
+proxy.properties(add config as follows)
+proxy.server.useTls.enabled=true   //default value is false
+
+
+config env varible
+-Dssl.server.protocol=TLSv1.1   //default value is TLSv1.1
+-Dssl.server.cer=sChat2.jks     //put the file in confPath which is configured in start.sh 
+-Dssl.server.pass=sNetty
+
+```
+
+
+2.config in eventmesh-sdk-java 
+```
+// create producer
+LiteClientConfig liteClientConfig = new liteClientConfig();
+...
+// enable TLS
+liteClientConfig.setUseTls(true);
+LiteProducer producer = new LiteProducer(liteClientConfig);
+
+
+config env varible
+-Dssl.client.protocol=TLSv1.1   //default value is TLSv1.1
+-Dssl.client.cer=sChat2.jks     //put the file in confPath of your application
+-Dssl.client.pass=sNetty
+```

docs/eventmesh.md

@@ -0,0 +1,34 @@
+## 什么是Event Mesh？
+该图显示了Event Mesh相对于应用程序框架中其他类似技术(例如Service Mesh)的定位.
+![architecture1](docs/images/eventmesh-define.png)
+
+Event Mesh是一个动态的插件式云原生基础服务层，用于分离应用程序和中间件层。它提供了灵活，可靠和快速的事件分发，并且可以进行管理：
+![architecture1](docs/images/eventmesher.png)
+
+云原生Event Mesh：
+![architecture2](docs/images/bus.png)
+
+Event Mesh允许将来自一个应用程序的事件动态路由到任何其他应用程序.
+Event Mesh的一般功能:
+* 事件驱动;
+* 事件治理;
+* 动态路由;
+* 云原生;
+
+依赖部件：
+* 可选1：DeFiBus：具有低延迟，高性能和可靠性，和灵活可伸缩性的分布式消息传递平台 [DeFiBus](https://github.com/WeBankFinTech/DeFiBus)
+* 可选2：RocketMQ
+
+关键部件：
+* event mesher：一种中间件，用于在事件产生者和使用者之间传输事件，支持云原生应用程序和微服务
+* sdk：当前支持HTTP和TCP协议，未来会支持gRPC等
+* registry：自动在连接到事件网格的应用程序和服务之间路由事件, 管理event mesher
+
+
+## 开源地址
+* https://github.com/WeBankFinTech/DeFiBus
+* https://github.com/WeBankFinTech/EventMesh
+* https://gitee.com/WeBank/DeFiBus
+* https://gitee.com/WeBank/EventMesh
+
+欢迎加入社区进行交流

eventmesh-common/gradle.properties

@@ -16,7 +16,7 @@
 #
 
 group=com.webank.eventmesh
-version=1.1.0
+version=1.1.1
 defibus_version=1.0.1
 jdk=1.7
 snapshot=false

eventmesh-common/src/main/java/com/webank/eventmesh/common/Constants.java

@@ -31,6 +31,8 @@ public class Constants {
 
     public static final String HTTP_PROTOCOL_PREFIX = "http://";
 
+    public static final String HTTPS_PROTOCOL_PREFIX = "https://";
+
     public static final int DEFAULT_HTTP_TIME_OUT = 3000;
 
     public static final String PROXY_MESSAGE_CONST_TTL = "ttl";

eventmesh-emesher/conf/server.env

@@ -1 +1 @@
-APP_START_JVM_OPTION:::-server -Xms128M -Xmx256M -Xmn128m -XX:SurvivorRatio=4 -Duser.language=zh -Dclient.executor.corePoolSize=8 -Dclient.schedule.corePoolSize=4 -Dclient.instance.corePoolSize=4 -Dclient.checkExecutor.corePoolSize=4 -Dclient.rebalanceImpl.corePoolSize=4 -Dclient.rebalanceService.corePoolSize=2 -Dclient.pullMessage.corePoolSize=8
+APP_START_JVM_OPTION:::-server -Xms128M -Xmx256M -Xmn128m -XX:SurvivorRatio=4 -Duser.language=zh

eventmesh-emesher/gradle.properties

@@ -16,7 +16,7 @@
 #
 
 group=com.webank.eventmesh
-version=1.1.0
+version=1.1.1
 defibus_version=1.0.1
 mavenUserName=
 mavenPassword=

eventmesh-emesher/src/main/java/com/webank/emesher/boot/AbrstractHTTPServer.java

@@ -38,6 +38,7 @@
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.codec.http.*;
 import io.netty.handler.codec.http.multipart.*;
+import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
 import org.apache.commons.collections4.MapUtils;
@@ -47,6 +48,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -67,6 +70,8 @@ public abstract class AbrstractHTTPServer extends AbstractRemotingServer {
 
     private AtomicBoolean started = new AtomicBoolean(false);
 
+    private boolean useTLS;
+
     public ThreadPoolExecutor asyncContextCompleteHandler =
             ThreadPoolFactory.createThreadPoolExecutor(10, 10, "proxy-http-asyncContext-");
 
@@ -77,8 +82,9 @@ public abstract class AbrstractHTTPServer extends AbstractRemotingServer {
     protected HashMap<Integer/* request code */, Pair<HttpRequestProcessor, ThreadPoolExecutor>> processorTable =
             new HashMap<Integer, Pair<HttpRequestProcessor, ThreadPoolExecutor>>(64);
 
-    public AbrstractHTTPServer(int port) {
+    public AbrstractHTTPServer(int port, boolean useTLS) {
         this.port = port;
+        this.useTLS = useTLS;
     }
 
     public Map<String, Object> parseHTTPHeader(HttpRequest fullReq) {
@@ -126,18 +132,7 @@ public void start() throws Exception {
             ServerBootstrap b = new ServerBootstrap();
             b.group(this.bossGroup, this.workerGroup)
                     .channel(NioServerSocketChannel.class)
-                    .childHandler(new ChannelInitializer<SocketChannel>() {
-                        @Override
-                        public void initChannel(SocketChannel ch)
-                                throws Exception {
-                            ch.pipeline()
-                                    .addLast(new HttpRequestDecoder(),
-                                            new HttpResponseEncoder(),
-                                            new HttpConnectionHandler(),
-                                            new HttpObjectAggregator(Integer.MAX_VALUE),
-                                            new HTTPHandler());
-                        }
-                    }).childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
+                    .childHandler(new HttpsServerInitializer(SSLContextFactory.getSslContext())).childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
             try {
                 logger.info("HTTPServer[port={}] started......", this.port);
                 ChannelFuture future = b.bind(this.port).sync();
@@ -389,4 +384,30 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
             ctx.fireUserEventTriggered(evt);
         }
     }
+
+    class HttpsServerInitializer extends ChannelInitializer<SocketChannel> {
+
+        private SSLContext sslContext;
+
+        public HttpsServerInitializer(SSLContext sslContext) {
+            this.sslContext = sslContext;
+        }
+
+        @Override
+        protected void initChannel(SocketChannel channel) throws Exception {
+            ChannelPipeline pipeline = channel.pipeline();
+
+
+            if(sslContext != null && useTLS){
+                SSLEngine sslEngine = sslContext.createSSLEngine();
+                sslEngine.setUseClientMode(false);
+                pipeline.addFirst("ssl", new SslHandler(sslEngine));
+            }
+            pipeline.addLast(new HttpRequestDecoder(),
+                    new HttpResponseEncoder(),
+                    new HttpConnectionHandler(),
+                    new HttpObjectAggregator(Integer.MAX_VALUE),
+                    new HTTPHandler());
+        }
+    }
 }

eventmesh-emesher/src/main/java/com/webank/emesher/boot/ProxyHTTPServer.java

@@ -43,7 +43,7 @@ public class ProxyHTTPServer extends AbrstractHTTPServer {
 
     public ProxyHTTPServer(ProxyServer proxyServer,
                            ProxyConfiguration proxyConfiguration) {
-        super(proxyConfiguration.httpServerPort);
+        super(proxyConfiguration.httpServerPort, proxyConfiguration.proxyServerUseTls);
         this.proxyServer = proxyServer;
         this.proxyConfiguration = proxyConfiguration;
     }

eventmesh-emesher/src/main/java/com/webank/emesher/boot/SSLContextFactory.java

@@ -0,0 +1,51 @@
+package com.webank.emesher.boot;
+
+import com.webank.emesher.constants.ProxyConstants;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.security.KeyStore;
+
+public class SSLContextFactory {
+    private static Logger httpLogger = LoggerFactory.getLogger("http");
+
+    private static String protocol = "TLSv1.1";
+
+    private static String fileName;
+
+    private static String pass;
+
+
+    public static SSLContext getSslContext(){
+        SSLContext sslContext = null;
+        try{
+            protocol = System.getProperty("ssl.server.protocol", "TLSv1.1");
+
+            fileName = System.getProperty("ssl.server.cer","sChat2.jks");
+
+            char[] filePass = null;
+            pass = System.getProperty("ssl.server.pass","sNetty");
+            if(StringUtils.isNotBlank(pass)){
+                filePass = pass.toCharArray();
+            }
+            sslContext = SSLContext.getInstance(protocol);
+            KeyStore keyStore = KeyStore.getInstance("JKS");
+            keyStore.load(Files.newInputStream(Paths.get(ProxyConstants.PROXY_CONF_HOME
+                    + File.separator
+                    + fileName), StandardOpenOption.READ), filePass);
+            KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+            kmf.init(keyStore, filePass);
+            sslContext.init(kmf.getKeyManagers(), null, null);
+        }catch (Exception e){
+            httpLogger.warn("sslContext init failed", e);
+            sslContext = null;
+        }
+        return sslContext;
+    }
+}

eventmesh-emesher/src/main/java/com/webank/emesher/configuration/ProxyConfiguration.java

@@ -63,6 +63,8 @@ public class ProxyConfiguration extends CommonConfiguration {
 
     public boolean proxyServerConsumerEnabled = false;
 
+    public boolean proxyServerUseTls = false;
+
     public ProxyConfiguration(ConfigurationWraper configurationWraper){
         super(configurationWraper);
     }
@@ -168,6 +170,11 @@ public void init(){
         if (StringUtils.isNotEmpty(proxyServerRetryThreadNumStr) && StringUtils.isNumeric(proxyServerRetryThreadNumStr)) {
             proxyServerRetryThreadNum = Integer.valueOf(StringUtils.deleteWhitespace(proxyServerRetryThreadNumStr));
         }
+
+        String proxyServerUseTlsStr = configurationWraper.getProp(ConfKeys.KEY_PROXY_HTTPS_ENABLED);
+        if (StringUtils.isNotEmpty(proxyServerUseTlsStr)) {
+            proxyServerUseTls = Boolean.valueOf(StringUtils.deleteWhitespace(proxyServerUseTlsStr));
+        }
     }
 
     static class ConfKeys{
@@ -212,5 +219,6 @@ static class ConfKeys{
 
         public static String KEY_PROXY_CONSUMER_ENABLED = "proxy.server.consumer.enabled";
 
+        public static String KEY_PROXY_HTTPS_ENABLED = "proxy.server.useTls.enabled";
     }
 }

eventmesh-sdk-java/gradle.properties

@@ -16,7 +16,6 @@
 #
 
 group=com.webank.eventmesh
-version=1.1.0
+version=1.1.1
 jdk=1.8
 snapshot=false
-proxy_common_version=1.1.0

eventmesh-sdk-java/src/main/java/com/webank/eventmesh/client/http/conf/LiteClientConfig.java

@@ -43,6 +43,8 @@ public class LiteClientConfig {
 
     private String password = "password";
 
+    private boolean useTls = false;
+
     public int getConsumeThreadMax() {
         return consumeThreadMax;
     }
@@ -151,6 +153,14 @@ public LiteClientConfig setLiteProxyAddr(String liteProxyAddr) {
         return this;
     }
 
+    public boolean isUseTls() {
+        return useTls;
+    }
+
+    public void setUseTls(boolean useTls) {
+        this.useTls = useTls;
+    }
+
     @Override
     public String toString() {
         StringBuilder sb = new StringBuilder();
@@ -166,7 +176,8 @@ public String toString() {
                 .append("pid=").append(pid).append(",")
                 .append("sys=").append(sys).append(",")
                 .append("userName=").append(userName).append(",")
-                .append("password=").append(password).append("}");
+                .append("password=").append(password).append(",")
+                .append("useTls=").append(useTls).append("}");
         return sb.toString();
     }
 }