Skip to content

Commit

Permalink
Merge ef3a584 into 7f3fc8a
Browse files Browse the repository at this point in the history
  • Loading branch information
@pandaapo
pandaapo committed May 3, 2023
2 parents 7f3fc8a + ef3a584 commit a543043
Show file tree
Hide file tree
Showing 6 changed files with 64 additions and 157 deletions.
15 changes: 11 additions & 4 deletions ...entmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/RegistryService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.eventmesh.spi.EventMeshExtensionType;
import org.apache.eventmesh.spi.EventMeshSPI;

import java.util.Collections;
import java.util.List;
import java.util.Map;

Expand All @@ -45,16 +46,22 @@ public interface RegistryService {

List<EventMeshDataInfo> findAllEventMeshInfo() throws RegistryException;

Map<String/*eventMeshName*/, Map<String/*purpose*/, Integer/*num*/>> findEventMeshClientDistributionData(
String clusterName, String group, String purpose) throws RegistryException;
default Map<String/*eventMeshName*/, Map<String/*purpose*/, Integer/*num*/>> findEventMeshClientDistributionData(
String clusterName, String group, String purpose) throws RegistryException {
return Collections.emptyMap();
}

void registerMetadata(Map<String, String> metadataMap);

boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws RegistryException;

boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throws RegistryException;

EventMeshAppSubTopicInfo findEventMeshAppSubTopicInfoByGroup(String group) throws RegistryException;
default EventMeshAppSubTopicInfo findEventMeshAppSubTopicInfoByGroup(String group) throws RegistryException {
return null;
}

List<EventMeshServicePubTopicInfo> findEventMeshServicePubTopicInfos() throws RegistryException;
default List<EventMeshServicePubTopicInfo> findEventMeshServicePubTopicInfos() throws RegistryException {
return Collections.emptyList();
}
}
28 changes: 5 additions & 23 deletions ...sul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,6 @@

import org.apache.eventmesh.api.exception.RegistryException;
import org.apache.eventmesh.api.registry.RegistryService;
import org.apache.eventmesh.api.registry.bo.EventMeshAppSubTopicInfo;
import org.apache.eventmesh.api.registry.bo.EventMeshServicePubTopicInfo;
import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
Expand All @@ -30,7 +28,6 @@
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
Expand All @@ -42,6 +39,7 @@
import com.ecwid.consul.v1.health.HealthServicesRequest;
import com.ecwid.consul.v1.health.model.HealthService;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
Expand All @@ -57,6 +55,7 @@ public class ConsulRegistryService implements RegistryService {

private String consulPort;

@Getter
private ConsulClient consulClient;

private String token;
Expand Down Expand Up @@ -107,6 +106,9 @@ public void shutdown() throws RegistryException {
public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws RegistryException {
try {
String[] ipPort = eventMeshRegisterInfo.getEndPoint().split(IP_PORT_SEPARATOR);
if (ipPort == null || ipPort.length < 2) {
return false;

Check warning on line 110 in eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java#L110 

Added line #L110 was not covered by tests
}
NewService service = new NewService();
service.setPort(Integer.parseInt(ipPort[1]));
service.setAddress(ipPort[0]);
Expand All @@ -132,16 +134,6 @@ public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throw
return true;
}

@Override
public List<EventMeshServicePubTopicInfo> findEventMeshServicePubTopicInfos() throws RegistryException {
return null;
}

@Override
public EventMeshAppSubTopicInfo findEventMeshAppSubTopicInfoByGroup(String group) throws RegistryException {
return null;
}

@Override
public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) throws RegistryException {
HealthServicesRequest request = HealthServicesRequest.newBuilder().setPassing(true).setToken(token).build();
Expand All @@ -166,18 +158,8 @@ public List<EventMeshDataInfo> findAllEventMeshInfo() throws RegistryException {
return eventMeshDataInfos;
}

@Override
public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(String clusterName, String group, String purpose)
throws RegistryException {
return Collections.emptyMap();
}

@Override
public void registerMetadata(Map<String, String> metadataMap) {

}

public ConsulClient getConsulClient() {
return consulClient;
}
}
14 changes: 7 additions & 7 deletions ...stry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdCustomService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,24 +27,24 @@
import org.apache.commons.collections4.CollectionUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import javax.annotation.Nullable;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.options.GetOption;

import lombok.extern.slf4j.Slf4j;

@Slf4j

Check warning on line 42 in eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdCustomService.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdCustomService.java#L42 

Added line #L42 was not covered by tests
public class EtcdCustomService extends EtcdRegistryService {

private static final String KEY_PREFIX = "eventMesh" + EtcdConstant.KEY_SEPARATOR;
private static final String KEY_APP = "app";
private static final String KEY_SERVICE = "service";
private static final Logger logger = LoggerFactory.getLogger(EtcdCustomService.class);

@Nullable
public List<EventMeshServicePubTopicInfo> findEventMeshServicePubTopicInfos() throws RegistryException {
Expand All @@ -68,11 +68,11 @@ public List<EventMeshServicePubTopicInfo> findEventMeshServicePubTopicInfos() th
return eventMeshServicePubTopicInfoList;
}
} catch (Exception e) {
logger.error("[EtcdRegistryService][findEventMeshServicePubTopicInfos] error", e);
log.error("[EtcdRegistryService][findEventMeshServicePubTopicInfos] error", e);

Check warning on line 71 in eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdCustomService.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdCustomService.java#L71 

Added line #L71 was not covered by tests
throw new RegistryException(e.getMessage());
}

return null;
return Collections.emptyList();

Check warning on line 75 in eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdCustomService.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdCustomService.java#L75 

Added line #L75 was not covered by tests
}

@Nullable
Expand All @@ -93,7 +93,7 @@ public EventMeshAppSubTopicInfo findEventMeshAppSubTopicInfoByGroup(String group
return eventMeshAppSubTopicInfo;
}
} catch (Exception e) {
logger.error("[EtcdRegistryService][findEventMeshAppSubTopicInfoByGroup] error, group: {}", group, e);
log.error("[EtcdRegistryService][findEventMeshAppSubTopicInfoByGroup] error, group: {}", group, e);

Check warning on line 96 in eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdCustomService.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdCustomService.java#L96 

Added line #L96 was not covered by tests
throw new RegistryException(e.getMessage());
}
return null;
Expand Down
20 changes: 6 additions & 14 deletions ...ry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@

import org.apache.eventmesh.api.exception.RegistryException;
import org.apache.eventmesh.api.registry.RegistryService;
import org.apache.eventmesh.api.registry.bo.EventMeshAppSubTopicInfo;
import org.apache.eventmesh.api.registry.bo.EventMeshServicePubTopicInfo;
import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
Expand All @@ -36,10 +35,11 @@
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
Expand All @@ -51,7 +51,7 @@
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;


import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
Expand All @@ -70,9 +70,10 @@ public class EtcdRegistryService implements RegistryService {

private String password;

@Getter

Check warning on line 73 in eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java

View check run for this annotation

Codecov / codecov/patch

eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java#L73 

Added line #L73 was not covered by tests
private Client etcdClient;

private Map<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap;
private ConcurrentMap<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap;

private ScheduledExecutorService etcdRegistryMonitorExecutorService;

Expand All @@ -82,7 +83,7 @@ public void init() throws RegistryException {
if (!initStatus.compareAndSet(false, true)) {
return;
}
eventMeshRegisterInfoMap = new HashMap<>(ConfigurationContextUtil.KEYS.size());
eventMeshRegisterInfoMap = new ConcurrentHashMap<>(ConfigurationContextUtil.KEYS.size());
for (String key : ConfigurationContextUtil.KEYS) {
CommonConfiguration commonConfiguration = ConfigurationContextUtil.get(key);
if (null == commonConfiguration) {
Expand Down Expand Up @@ -237,20 +238,11 @@ public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throw
}
}

@Override
public EventMeshAppSubTopicInfo findEventMeshAppSubTopicInfoByGroup(String group) throws RegistryException {
return null;
}

@Override
public List<EventMeshServicePubTopicInfo> findEventMeshServicePubTopicInfos() throws RegistryException {
return null;
}

public Client getEtcdClient() {
return etcdClient;
}

public long getLeaseId() {
return EtcdClientFactory.getLeaseId(serverAddr);
}
Expand Down
Loading

0 comments on commit a543043

Please sign in to comment.