add : eventmesh-retry/eventmesh-retry-kafka

docs/en/instructions/eventmesh-sdk-java-quickstart.md

@@ -0,0 +1,120 @@
+## How to run eventmesh-sdk-java demo
+
+> Eventmesh-sdk-java , as the client, communicated with eventmesh-runtime, used to complete the sending and receiving of message.  
+>
+> Eventmesh-sdk-java supports sync msg, async msg and broadcast msg. Sync msg means the producer sends msg which need the consumer supplies the response msg, Async msg means the producer just sends msg and does not care reply msg.Broadcast msg means the producer send msg once and all the consumer subscribed the broadcast topic will receive the msg. 
+>
+> Eventmesh-sdk-java supports  the protocol  of HTTP and TCP.  
+
+
+
+###  1. TCP DEMO
+
+#### Sync msg 
+
+- create topic
+
+```
+sh runadmin.sh updateTopic  -c ${ClusterName} -t ${topic} -n ${namesrvAddr}
+```
+
+
+
+* start consumer ,subscribe topic in previous step. 
+
+```
+Run the main method of cn.webank.eventmesh.client.tcp.demo.SyncResponse
+```
+
+
+
+* start producer, send message
+
+```
+Run the main method of cn.webank.eventmesh.client.tcp.demo.SyncRequest
+```
+
+
+
+#### Async msg 
+
+- create topic
+
+```
+sh runadmin.sh updateTopic  -c ${ClusterName} -t ${topic} -n ${namesrvAddr}
+```
+
+
+
+- start consumer ,subscribe topic in previous step. 
+
+```
+Run the main method of cn.webank.eventmesh.client.tcp.demo.AsyncSubscribe
+```
+
+
+
+start producer, send  message
+
+```
+Run the main method of cn.webank.eventmesh.client.tcp.demo.AsyncPublish
+```
+
+
+
+#### Broadcast msg 
+
+- create topic
+
+```
+sh runadmin.sh updateTopic  -c ${ClusterName} -t ${topic} -n ${namesrvAddr}
+```
+
+
+
+- start consumer ,subscribe topic in previous step. 
+
+```
+Run the main method of cn.webank.eventmesh.client.tcp.demo.AsyncSubscribeBroadcast
+```
+
+
+
+* start producer, send broadcast message
+
+```
+Run the main method of cn.webank.eventmesh.client.tcp.demo.AsyncPublishBroadcast
+```
+
+### 2. HTTP DEMO
+
+> As to http, eventmesh-sdk-java just implements  the sending of msg. And it already  supports sync msg and async msg.
+>
+> In the demo ,the field of `content` of the java class `LiteMessage` represents a special protocal, so if you want to use http-client of eventmesh-sdk-java, you just need to design the content of protocal and supply the consumer appliacation at the same time.
+
+
+
+#### Sync msg
+
+> send msg ,producer need waiting until receive the response msg of consumer
+
+```
+Run the main method of cn.webank.eventmesh.client.http.demo.SyncRequestInstance
+```
+
+
+
+> send msg,producer handles the reponse msg in callback
+
+```
+Run the main method ofcn.webank.eventmesh.client.http.demo.AsyncSyncRequestInstance
+```
+
+
+
+#### Async msg
+
+```
+Run the main method of cn.webank.eventmesh.client.http.demo.AsyncPublishInstance
+```
+

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

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cn.webank.eventmesh.common;
+
+public class Constants {
+
+    public static final String DEFAULT_CHARSET = "UTF-8";
+
+    public static final String TARGET_PROXY_REGION = "TARGET_PROXY_REGION";
+
+    public static final String CONSTANTS_DEFAULT_REGION_KEY = "default";
+
+    public static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss.SSS";
+
+    public static final String LANGUAGE_JAVA = "JAVA";
+
+    public static final String HTTP_PROTOCOL_PREFIX = "http://";
+
+    public static final int DEFAULT_HTTP_TIME_OUT = 3000;
+
+    public static final String PROXY_MESSAGE_CONST_TTL = "ttl";
+
+    public static final Integer DEFAULT_CLIENT_UNACK = 12;
+
+    public static final String CONSTANTS_SERVICE_DESC_ENV = "env";
+
+    public static final String CONSTANTS_SERVICE_DESC_VERSION = "version";
+
+    public static final String CONSTANTS_INSTANCE_DESC_ENV = "env";
+
+    public static final String CONSTANTS_INSTANCE_DESC_IDC = "idc";
+
+    public static final String CONSTANTS_INSTANCE_DESC_DCN = "dcn";
+
+    public static final String CONSTANTS_INSTANCE_DESC_SYSID = "sysId";
+
+    public static final String CONSTANTS_INSTANCE_DESC_IP = "ip";
+
+    public static final String CONSTANTS_INSTANCE_DESC_PORT = "port";
+
+    public static final String KEY_CONSTANTS_INSTANCE_DESC_PID = "pid";
+
+    public static final String RMB_UNIQ_ID = "RMB_UNIQ_ID";
+
+    public static final String IDC_SEPERATER = "-";
+
+}

eventmesh-common/src/main/java/cn/webank/eventmesh/common/IPUtil.java

@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cn.webank.eventmesh.common;
+
+import io.netty.channel.Channel;
+
+import java.net.Inet6Address;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Enumeration;
+
+public class IPUtil {
+
+    public static String getLocalAddress() {
+        //获取docker宿主机ip
+        String dockerHostIp = System.getenv("webank_docker_host_ip");
+        if (dockerHostIp != null && !"".equals(dockerHostIp))
+            return dockerHostIp;
+        //priority of networkInterface when generating client ip
+        String priority = System.getProperty("networkInterface.priority", "eth0<eth1<bond1");
+//        LOGGER.info("networkInterface.priority: {}", priority);
+        ArrayList<String> preferList = new ArrayList<String>();
+        for (String eth : priority.split("<")) {
+            preferList.add(eth);
+        }
+        NetworkInterface preferNetworkInterface = null;
+
+        try {
+            Enumeration<NetworkInterface> enumeration1 = NetworkInterface.getNetworkInterfaces();
+            while (enumeration1.hasMoreElements()) {
+                final NetworkInterface networkInterface = enumeration1.nextElement();
+//                LOGGER.debug("networkInterface:{}", networkInterface);
+
+                if (!preferList.contains(networkInterface.getName())) {
+                    continue;
+                } else if (preferNetworkInterface == null) {
+                    preferNetworkInterface = networkInterface;
+                }
+                //get the networkInterface that has higher priority
+                else if (preferList.indexOf(networkInterface.getName())
+                        > preferList.indexOf(preferNetworkInterface.getName())) {
+                    preferNetworkInterface = networkInterface;
+                }
+            }
+
+            // Traversal Network interface to get the first non-loopback and non-private address
+            ArrayList<String> ipv4Result = new ArrayList<String>();
+            ArrayList<String> ipv6Result = new ArrayList<String>();
+
+            if (preferNetworkInterface != null) {
+//                LOGGER.info("use preferNetworkInterface:{}", preferNetworkInterface);
+                final Enumeration<InetAddress> en = preferNetworkInterface.getInetAddresses();
+                getIpResult(ipv4Result, ipv6Result, en);
+            } else {
+//                LOGGER.info("no preferNetworkInterface");
+                Enumeration<NetworkInterface> enumeration = NetworkInterface.getNetworkInterfaces();
+                while (enumeration.hasMoreElements()) {
+                    final NetworkInterface networkInterface = enumeration.nextElement();
+                    final Enumeration<InetAddress> en = networkInterface.getInetAddresses();
+                    getIpResult(ipv4Result, ipv6Result, en);
+                }
+            }
+
+            // prefer ipv4
+            if (!ipv4Result.isEmpty()) {
+                for (String ip : ipv4Result) {
+                    if (ip.startsWith("127.0") || ip.startsWith("192.168")) {
+                        continue;
+                    }
+
+                    return ip;
+                }
+
+                return ipv4Result.get(ipv4Result.size() - 1);
+            } else if (!ipv6Result.isEmpty()) {
+                return ipv6Result.get(0);
+            }
+            //If failed to find,fall back to localhost
+            final InetAddress localHost = InetAddress.getLocalHost();
+            return normalizeHostAddress(localHost);
+        } catch (SocketException e) {
+            e.printStackTrace();
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+        }
+
+        return null;
+    }
+
+    private static void getIpResult(ArrayList<String> ipv4Result, ArrayList<String> ipv6Result,
+                                    Enumeration<InetAddress> en) {
+        while (en.hasMoreElements()) {
+            final InetAddress address = en.nextElement();
+            if (!address.isLoopbackAddress()) {
+                if (address instanceof Inet6Address) {
+                    ipv6Result.add(normalizeHostAddress(address));
+                } else {
+                    ipv4Result.add(normalizeHostAddress(address));
+                }
+            }
+        }
+    }
+
+    private static String normalizeHostAddress(final InetAddress localHost) {
+        if (localHost instanceof Inet6Address) {
+            return "[" + localHost.getHostAddress() + "]";
+        } else {
+            return localHost.getHostAddress();
+        }
+    }
+
+
+    public static String parseChannelRemoteAddr(final Channel channel) {
+        if (null == channel) {
+            return "";
+        }
+        SocketAddress remote = channel.remoteAddress();
+        final String addr = remote != null ? remote.toString() : "";
+
+        if (addr.length() > 0) {
+            int index = addr.lastIndexOf("/");
+            if (index >= 0) {
+                return addr.substring(index + 1);
+            }
+
+            return addr;
+        }
+
+        return "";
+    }
+}