githublaohu commented on code in PR #1029: URL: https://github.com/apache/incubator-eventmesh/pull/1029#discussion_r923645172
########## eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java: ########## @@ -0,0 +1,217 @@ +/* + * 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 org.apache.eventmesh.registry.etcd.service; + +import io.etcd.jetcd.ByteSequence; +import io.etcd.jetcd.Client; +import io.etcd.jetcd.KeyValue; +import io.etcd.jetcd.options.GetOption; +import io.etcd.jetcd.options.PutOption; +import org.apache.commons.collections4.CollectionUtils; +import org.apache.eventmesh.api.exception.RegistryException; +import org.apache.eventmesh.api.registry.RegistryService; +import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo; +import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo; +import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo; +import org.apache.eventmesh.common.config.CommonConfiguration; +import org.apache.eventmesh.common.utils.ConfigurationContextUtil; + +import org.apache.commons.lang3.StringUtils; + +import java.util.*; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.eventmesh.common.utils.JsonUtils; +import org.apache.eventmesh.registry.etcd.constant.EtcdConstant; +import org.apache.eventmesh.registry.etcd.factory.EtcdClientFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class EtcdRegistryService implements RegistryService { + + private static final Logger logger = LoggerFactory.getLogger(EtcdRegistryService.class); + + private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false); + + private static final AtomicBoolean START_STATUS = new AtomicBoolean(false); + + private static final String KEY_PREFIX = EtcdConstant.KEY_SEPARATOR + "eventMesh" + EtcdConstant.KEY_SEPARATOR + "registry" + EtcdConstant.KEY_SEPARATOR; + + private String serverAddr; + + private String username; + + private String password; + + private EtcdClientFactory etcdClientFactory; + + private Client etcdClient; + + private Long leaseId; + + private Map<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap; + + @Override + public void init() throws RegistryException { + boolean update = INIT_STATUS.compareAndSet(false, true); + if (!update) { + return; + } + etcdClientFactory = new EtcdClientFactory(); + eventMeshRegisterInfoMap = new HashMap<>(ConfigurationContextUtil.KEYS.size()); + for (String key : ConfigurationContextUtil.KEYS) { + CommonConfiguration commonConfiguration = ConfigurationContextUtil.get(key); + if (null == commonConfiguration) { + continue; + } + if (StringUtils.isBlank(commonConfiguration.namesrvAddr)) { + throw new RegistryException("namesrvAddr cannot be null"); + } + this.serverAddr = commonConfiguration.namesrvAddr; + this.username = commonConfiguration.eventMeshRegistryPluginUsername; + this.password = commonConfiguration.eventMeshRegistryPluginPassword; + break; + } + } + + @Override + public void start() throws RegistryException { + boolean update = START_STATUS.compareAndSet(false, true); + if (!update) { + return; + } + try { + Properties properties = new Properties(); + properties.setProperty(EtcdConstant.SERVER_ADDR, serverAddr); + properties.setProperty(EtcdConstant.USERNAME, username); + properties.setProperty(EtcdConstant.PASSWORD, password); + this.etcdClient = etcdClientFactory.createClient(properties); + this.leaseId = etcdClientFactory.getLeaseId(serverAddr); + } catch (Exception e) { + logger.error("[EtcdRegistryService][start] error", e); + throw new RegistryException(e.getMessage()); + } + } + + @Override + public void shutdown() throws RegistryException { + INIT_STATUS.compareAndSet(true, false); + START_STATUS.compareAndSet(true, false); + logger.info("EtcdRegistryService closed"); + } + + @Override + public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) throws RegistryException { + List<EventMeshDataInfo> eventMeshDataInfoList = new ArrayList<>(); + + try { + String keyPrefix = clusterName == null? KEY_PREFIX : KEY_PREFIX + EtcdConstant.KEY_SEPARATOR + clusterName; + GetOption getOption = GetOption.newBuilder().withPrefix(ByteSequence.from(keyPrefix.getBytes())).build(); + List<KeyValue> keyValues = getEtcdClient().getKVClient().get(ByteSequence.from(keyPrefix.getBytes()), getOption).get().getKvs(); + + if (CollectionUtils.isNotEmpty(keyValues)) { + for (KeyValue kv : keyValues) { + EventMeshDataInfo eventMeshDataInfo = JsonUtils.deserialize(new String(kv.getValue().getBytes()), EventMeshDataInfo.class); + eventMeshDataInfoList.add(eventMeshDataInfo); + } + } + } catch (Exception e) { + logger.error("[EtcdRegistryService][findEventMeshInfoByCluster] error", e); + throw new RegistryException(e.getMessage()); + } + return eventMeshDataInfoList; + } + + @Override + public List<EventMeshDataInfo> findAllEventMeshInfo() throws RegistryException { + try { + return findEventMeshInfoByCluster(null); + } catch (Exception e) { + logger.error("[EtcdRegistryService][findEventMeshInfoByCluster] error", e); + throw new RegistryException(e.getMessage()); + } + } + + @Override + public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(String clusterName, String group, String purpose) + throws RegistryException { + // todo find metadata + return null; + } + + @Override + public void registerMetadata(Map<String, String> metadataMap) { + for (Map.Entry<String, EventMeshRegisterInfo> eventMeshRegisterInfo : eventMeshRegisterInfoMap.entrySet()) { + EventMeshRegisterInfo registerInfo = eventMeshRegisterInfo.getValue(); + registerInfo.setMetadata(metadataMap); + this.register(registerInfo); + } + } + + @Override + public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws RegistryException { + try { + String eventMeshClusterName = eventMeshRegisterInfo.getEventMeshClusterName(); + String eventMeshName = eventMeshRegisterInfo.getEventMeshName(); + ByteSequence etcdKey = ByteSequence.from((KEY_PREFIX + eventMeshClusterName + EtcdConstant.KEY_SEPARATOR + eventMeshName + EtcdConstant.KEY_SEPARATOR + eventMeshRegisterInfo.getEndPoint()).getBytes()); + EventMeshDataInfo eventMeshDataInfo = + new EventMeshDataInfo(eventMeshClusterName, eventMeshName, + eventMeshRegisterInfo.getEndPoint(), new Date().getTime(), eventMeshRegisterInfo.getMetadata()); + ByteSequence etcdValue = ByteSequence.from(JsonUtils.serialize(eventMeshDataInfo).getBytes()); + etcdClient.getKVClient().put(etcdKey, etcdValue, PutOption.newBuilder().withLeaseId(leaseId).build()); + eventMeshRegisterInfoMap.put(eventMeshName, eventMeshRegisterInfo); + } catch (Exception e) { + logger.error("[EtcdRegistryService][register] error", e); + throw new RegistryException(e.getMessage()); + } + logger.info("EventMesh successfully registered to etcd"); + return true; + } + + @Override + public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throws RegistryException { + try { + String eventMeshClusterName = eventMeshUnRegisterInfo.getEventMeshClusterName(); + String eventMeshName = eventMeshUnRegisterInfo.getEventMeshName(); + ByteSequence etcdKey = ByteSequence.from((KEY_PREFIX + eventMeshClusterName + EtcdConstant.KEY_SEPARATOR + eventMeshName + EtcdConstant.KEY_SEPARATOR + eventMeshUnRegisterInfo.getEndPoint()).getBytes()); + etcdClient.getKVClient().delete(etcdKey); + eventMeshRegisterInfoMap.remove(eventMeshName); + } catch (Exception e) { + logger.error("[EtcdRegistryService][unRegister] error", e); + throw new RegistryException(e.getMessage()); + } + logger.info("EventMesh successfully logout to etcd"); + return true; + } + + public String getServerAddr() { Review Comment: 这个与下面三个方法有存在的意义吗？ ########## eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java: ########## @@ -0,0 +1,217 @@ +/* + * 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 org.apache.eventmesh.registry.etcd.service; + +import io.etcd.jetcd.ByteSequence; +import io.etcd.jetcd.Client; +import io.etcd.jetcd.KeyValue; +import io.etcd.jetcd.options.GetOption; +import io.etcd.jetcd.options.PutOption; +import org.apache.commons.collections4.CollectionUtils; +import org.apache.eventmesh.api.exception.RegistryException; +import org.apache.eventmesh.api.registry.RegistryService; +import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo; +import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo; +import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo; +import org.apache.eventmesh.common.config.CommonConfiguration; +import org.apache.eventmesh.common.utils.ConfigurationContextUtil; + +import org.apache.commons.lang3.StringUtils; + +import java.util.*; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.eventmesh.common.utils.JsonUtils; +import org.apache.eventmesh.registry.etcd.constant.EtcdConstant; +import org.apache.eventmesh.registry.etcd.factory.EtcdClientFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class EtcdRegistryService implements RegistryService { + + private static final Logger logger = LoggerFactory.getLogger(EtcdRegistryService.class); + + private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false); + + private static final AtomicBoolean START_STATUS = new AtomicBoolean(false); + + private static final String KEY_PREFIX = EtcdConstant.KEY_SEPARATOR + "eventMesh" + EtcdConstant.KEY_SEPARATOR + "registry" + EtcdConstant.KEY_SEPARATOR; + + private String serverAddr; + + private String username; + + private String password; + + private EtcdClientFactory etcdClientFactory; + + private Client etcdClient; + + private Long leaseId; + + private Map<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap; + + @Override + public void init() throws RegistryException { + boolean update = INIT_STATUS.compareAndSet(false, true); + if (!update) { + return; + } + etcdClientFactory = new EtcdClientFactory(); + eventMeshRegisterInfoMap = new HashMap<>(ConfigurationContextUtil.KEYS.size()); + for (String key : ConfigurationContextUtil.KEYS) { + CommonConfiguration commonConfiguration = ConfigurationContextUtil.get(key); + if (null == commonConfiguration) { + continue; + } + if (StringUtils.isBlank(commonConfiguration.namesrvAddr)) { + throw new RegistryException("namesrvAddr cannot be null"); + } + this.serverAddr = commonConfiguration.namesrvAddr; + this.username = commonConfiguration.eventMeshRegistryPluginUsername; + this.password = commonConfiguration.eventMeshRegistryPluginPassword; + break; + } + } + + @Override + public void start() throws RegistryException { + boolean update = START_STATUS.compareAndSet(false, true); + if (!update) { + return; + } + try { + Properties properties = new Properties(); + properties.setProperty(EtcdConstant.SERVER_ADDR, serverAddr); + properties.setProperty(EtcdConstant.USERNAME, username); + properties.setProperty(EtcdConstant.PASSWORD, password); + this.etcdClient = etcdClientFactory.createClient(properties); + this.leaseId = etcdClientFactory.getLeaseId(serverAddr); + } catch (Exception e) { + logger.error("[EtcdRegistryService][start] error", e); + throw new RegistryException(e.getMessage()); + } + } + + @Override + public void shutdown() throws RegistryException { Review Comment: 应该调用etcdclient.close(); ########## eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/factory/EtcdClientFactory.java: ########## @@ -0,0 +1,243 @@ +/* + * 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 org.apache.eventmesh.registry.etcd.factory; + +import io.etcd.jetcd.*; +import io.etcd.jetcd.lease.LeaseKeepAliveResponse; +import io.etcd.jetcd.options.LeaseOption; +import io.grpc.stub.StreamObserver; +import org.apache.commons.lang3.StringUtils; +import org.apache.eventmesh.registry.etcd.constant.EtcdConstant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.Objects; +import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; + +public class EtcdClientFactory { Review Comment: 是否应该是单列模式，否则etcdregistrySercie每一个都创建一次。如果每次都创建一次etcdLeaseIdMap，是否有存在的必要 ########## eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java: ########## @@ -0,0 +1,217 @@ +/* + * 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 org.apache.eventmesh.registry.etcd.service; + +import io.etcd.jetcd.ByteSequence; +import io.etcd.jetcd.Client; +import io.etcd.jetcd.KeyValue; +import io.etcd.jetcd.options.GetOption; +import io.etcd.jetcd.options.PutOption; +import org.apache.commons.collections4.CollectionUtils; +import org.apache.eventmesh.api.exception.RegistryException; +import org.apache.eventmesh.api.registry.RegistryService; +import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo; +import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo; +import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo; +import org.apache.eventmesh.common.config.CommonConfiguration; +import org.apache.eventmesh.common.utils.ConfigurationContextUtil; + +import org.apache.commons.lang3.StringUtils; + +import java.util.*; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.eventmesh.common.utils.JsonUtils; +import org.apache.eventmesh.registry.etcd.constant.EtcdConstant; +import org.apache.eventmesh.registry.etcd.factory.EtcdClientFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class EtcdRegistryService implements RegistryService { + + private static final Logger logger = LoggerFactory.getLogger(EtcdRegistryService.class); + + private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false); + + private static final AtomicBoolean START_STATUS = new AtomicBoolean(false); + + private static final String KEY_PREFIX = EtcdConstant.KEY_SEPARATOR + "eventMesh" + EtcdConstant.KEY_SEPARATOR + "registry" + EtcdConstant.KEY_SEPARATOR; + + private String serverAddr; + + private String username; + + private String password; + + private EtcdClientFactory etcdClientFactory; + + private Client etcdClient; + + private Long leaseId; + + private Map<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap; + + @Override + public void init() throws RegistryException { + boolean update = INIT_STATUS.compareAndSet(false, true); + if (!update) { + return; + } + etcdClientFactory = new EtcdClientFactory(); + eventMeshRegisterInfoMap = new HashMap<>(ConfigurationContextUtil.KEYS.size()); + for (String key : ConfigurationContextUtil.KEYS) { + CommonConfiguration commonConfiguration = ConfigurationContextUtil.get(key); + if (null == commonConfiguration) { + continue; + } + if (StringUtils.isBlank(commonConfiguration.namesrvAddr)) { + throw new RegistryException("namesrvAddr cannot be null"); + } + this.serverAddr = commonConfiguration.namesrvAddr; + this.username = commonConfiguration.eventMeshRegistryPluginUsername; + this.password = commonConfiguration.eventMeshRegistryPluginPassword; + break; + } + } + + @Override + public void start() throws RegistryException { + boolean update = START_STATUS.compareAndSet(false, true); + if (!update) { + return; + } + try { + Properties properties = new Properties(); + properties.setProperty(EtcdConstant.SERVER_ADDR, serverAddr); + properties.setProperty(EtcdConstant.USERNAME, username); + properties.setProperty(EtcdConstant.PASSWORD, password); + this.etcdClient = etcdClientFactory.createClient(properties); + this.leaseId = etcdClientFactory.getLeaseId(serverAddr); + } catch (Exception e) { + logger.error("[EtcdRegistryService][start] error", e); + throw new RegistryException(e.getMessage()); + } + } + + @Override + public void shutdown() throws RegistryException { + INIT_STATUS.compareAndSet(true, false); + START_STATUS.compareAndSet(true, false); + logger.info("EtcdRegistryService closed"); + } + + @Override + public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) throws RegistryException { + List<EventMeshDataInfo> eventMeshDataInfoList = new ArrayList<>(); + + try { + String keyPrefix = clusterName == null? KEY_PREFIX : KEY_PREFIX + EtcdConstant.KEY_SEPARATOR + clusterName; + GetOption getOption = GetOption.newBuilder().withPrefix(ByteSequence.from(keyPrefix.getBytes())).build(); + List<KeyValue> keyValues = getEtcdClient().getKVClient().get(ByteSequence.from(keyPrefix.getBytes()), getOption).get().getKvs(); + + if (CollectionUtils.isNotEmpty(keyValues)) { + for (KeyValue kv : keyValues) { + EventMeshDataInfo eventMeshDataInfo = JsonUtils.deserialize(new String(kv.getValue().getBytes()), EventMeshDataInfo.class); + eventMeshDataInfoList.add(eventMeshDataInfo); + } + } + } catch (Exception e) { + logger.error("[EtcdRegistryService][findEventMeshInfoByCluster] error", e); + throw new RegistryException(e.getMessage()); + } + return eventMeshDataInfoList; + } + + @Override + public List<EventMeshDataInfo> findAllEventMeshInfo() throws RegistryException { + try { + return findEventMeshInfoByCluster(null); + } catch (Exception e) { + logger.error("[EtcdRegistryService][findEventMeshInfoByCluster] error", e); + throw new RegistryException(e.getMessage()); + } + } + + @Override + public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(String clusterName, String group, String purpose) + throws RegistryException { + // todo find metadata + return null; + } + + @Override + public void registerMetadata(Map<String, String> metadataMap) { + for (Map.Entry<String, EventMeshRegisterInfo> eventMeshRegisterInfo : eventMeshRegisterInfoMap.entrySet()) { + EventMeshRegisterInfo registerInfo = eventMeshRegisterInfo.getValue(); + registerInfo.setMetadata(metadataMap); + this.register(registerInfo); + } + } + + @Override + public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws RegistryException { + try { + String eventMeshClusterName = eventMeshRegisterInfo.getEventMeshClusterName(); + String eventMeshName = eventMeshRegisterInfo.getEventMeshName(); + ByteSequence etcdKey = ByteSequence.from((KEY_PREFIX + eventMeshClusterName + EtcdConstant.KEY_SEPARATOR + eventMeshName + EtcdConstant.KEY_SEPARATOR + eventMeshRegisterInfo.getEndPoint()).getBytes()); Review Comment: etcdkey的创建有多处，麻烦提取出来 ########## eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/constant/EtcdConstant.java: ########## @@ -0,0 +1,35 @@ +/* + * 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 org.apache.eventmesh.registry.etcd.constant; + +/** + * EtcdConstant. + */ +public class EtcdConstant { Review Comment: common包里面是否有这些变量？ ########## eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/factory/EtcdClientFactory.java: ########## @@ -0,0 +1,243 @@ +/* + * 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 org.apache.eventmesh.registry.etcd.factory; + +import io.etcd.jetcd.*; +import io.etcd.jetcd.lease.LeaseKeepAliveResponse; +import io.etcd.jetcd.options.LeaseOption; +import io.grpc.stub.StreamObserver; +import org.apache.commons.lang3.StringUtils; +import org.apache.eventmesh.registry.etcd.constant.EtcdConstant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.Objects; +import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; + +public class EtcdClientFactory { + + private static final Logger logger = LoggerFactory.getLogger(EtcdClientFactory.class); + + private Map<String, EtcdLeaseId> etcdLeaseIdMap = new ConcurrentHashMap<>(); + + + public Client createClient(Properties properties) { + String serverAddr = properties.getProperty(EtcdConstant.SERVER_ADDR); + String username = properties.getProperty(EtcdConstant.USERNAME); + String password = properties.getProperty(EtcdConstant.PASSWORD); + + EtcdLeaseId etcdLeaseId = etcdLeaseIdMap.get(serverAddr); + if (Objects.nonNull(etcdLeaseId)) { + return etcdLeaseId.getClientWrapper(); + } + ClientBuilder clientBuilder = Client.builder(); + String[] addresss = serverAddr.split(","); + String[] httpAddress = new String[addresss.length]; + for (int i = 0; i < addresss.length; i++) { + if (!addresss[i].startsWith("http://";)) { + httpAddress[i] = "http://"; + addresss[i]; + } + } + etcdLeaseId = new EtcdLeaseId(); + try { + etcdLeaseId.setUrl(serverAddr); + etcdLeaseId.setClientBuilder(clientBuilder.endpoints(httpAddress)); + if (StringUtils.isNoneBlank(username)) { + etcdLeaseId.getClientBuilder().user(ByteSequence.from(username.getBytes())); + } + if (StringUtils.isNoneBlank(password)) { + etcdLeaseId.getClientBuilder().password(ByteSequence.from(password.getBytes())); + } + etcdLeaseId.setClientWrapper(new ClientWrapper(etcdLeaseId.getClientBuilder().build())); + ClientWrapper client = etcdLeaseId.getClientWrapper(); + long leaseId = client.getLeaseClient().grant(10L).get().getID(); + etcdLeaseId.setLeaseId(leaseId); + EtcdStreamObserver etcdStreamObserver = new EtcdStreamObserver(); + etcdStreamObserver.etcdLeaseId = etcdLeaseId; + etcdLeaseId.setEtcdStreamObserver(etcdStreamObserver); + client.getLeaseClient().keepAlive(leaseId, etcdStreamObserver); + + etcdLeaseIdMap.put(serverAddr, etcdLeaseId); + } catch (Throwable e) { + logger.error(e.getMessage(), e); Review Comment: 不抛出异常吗? ########## eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java: ########## @@ -0,0 +1,217 @@ +/* + * 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 org.apache.eventmesh.registry.etcd.service; + +import io.etcd.jetcd.ByteSequence; +import io.etcd.jetcd.Client; +import io.etcd.jetcd.KeyValue; +import io.etcd.jetcd.options.GetOption; +import io.etcd.jetcd.options.PutOption; +import org.apache.commons.collections4.CollectionUtils; +import org.apache.eventmesh.api.exception.RegistryException; +import org.apache.eventmesh.api.registry.RegistryService; +import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo; +import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo; +import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo; +import org.apache.eventmesh.common.config.CommonConfiguration; +import org.apache.eventmesh.common.utils.ConfigurationContextUtil; + +import org.apache.commons.lang3.StringUtils; + +import java.util.*; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.eventmesh.common.utils.JsonUtils; +import org.apache.eventmesh.registry.etcd.constant.EtcdConstant; +import org.apache.eventmesh.registry.etcd.factory.EtcdClientFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class EtcdRegistryService implements RegistryService { + + private static final Logger logger = LoggerFactory.getLogger(EtcdRegistryService.class); + + private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false); + + private static final AtomicBoolean START_STATUS = new AtomicBoolean(false); + + private static final String KEY_PREFIX = EtcdConstant.KEY_SEPARATOR + "eventMesh" + EtcdConstant.KEY_SEPARATOR + "registry" + EtcdConstant.KEY_SEPARATOR; + + private String serverAddr; + + private String username; + + private String password; + + private EtcdClientFactory etcdClientFactory; + + private Client etcdClient; + + private Long leaseId; + + private Map<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap; + + @Override + public void init() throws RegistryException { + boolean update = INIT_STATUS.compareAndSet(false, true); + if (!update) { + return; + } + etcdClientFactory = new EtcdClientFactory(); Review Comment: 是否可以放到 start() 方法创建 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
