diff --git a/apollo-adminservice/src/main/java/com/ctrip/apollo/adminservice/AdminServiceAutoConfiguration.java b/apollo-adminservice/src/main/java/com/ctrip/apollo/adminservice/AdminServiceAutoConfiguration.java deleted file mode 100644 index 07a2a73addb45b471c92bbd58adf8127c45e640a..0000000000000000000000000000000000000000 --- a/apollo-adminservice/src/main/java/com/ctrip/apollo/adminservice/AdminServiceAutoConfiguration.java +++ /dev/null @@ -1,56 +0,0 @@ -package com.ctrip.apollo.adminservice; - -import com.ctrip.apollo.biz.message.DummyMessageSender; -import com.ctrip.apollo.biz.message.MessageSender; -import com.ctrip.apollo.biz.message.RedisMessageSender; - -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.data.redis.core.StringRedisTemplate; - -/** - * @author Jason Song(song_s@ctrip.com) - */ -@Configuration -public class AdminServiceAutoConfiguration { - @ConditionalOnProperty(value = "apollo.redis.enabled", havingValue = "true", matchIfMissing = false) - public static class AdminRedisConfiguration { - @Value("${apollo.redis.host}") - private String host; - @Value("${apollo.redis.port}") - private int port; - @Bean - public JedisConnectionFactory redisConnectionFactory() { - JedisConnectionFactory factory = new JedisConnectionFactory(); - factory.setHostName(host); - factory.setPort(port); - return factory; - } - - @Bean - public RedisTemplate redisTemplate(RedisConnectionFactory factory) { - StringRedisTemplate template = new StringRedisTemplate(factory); - return template; - } - - @Bean - public MessageSender redisMessageSender(RedisTemplate redisTemplate) { - return new RedisMessageSender(redisTemplate); - } - - } - - @Configuration - @ConditionalOnProperty(value = "apollo.redis.enabled", havingValue = "false", matchIfMissing = true) - public static class ConfigDefaultConfiguration { - @Bean - public MessageSender defaultMessageSender() { - return new DummyMessageSender(); - } - } -} diff --git a/apollo-adminservice/src/main/java/com/ctrip/apollo/adminservice/controller/ReleaseController.java b/apollo-adminservice/src/main/java/com/ctrip/apollo/adminservice/controller/ReleaseController.java index 714d20bbaaa98c4dcafcbc8c4389947697598cfe..1fbdaf24607a8a466a2a6c031501176fe5f7a312 100644 --- a/apollo-adminservice/src/main/java/com/ctrip/apollo/adminservice/controller/ReleaseController.java +++ b/apollo-adminservice/src/main/java/com/ctrip/apollo/adminservice/controller/ReleaseController.java @@ -1,14 +1,6 @@ package com.ctrip.apollo.adminservice.controller; -import java.util.List; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.security.core.userdetails.UserDetails; -import org.springframework.web.bind.annotation.PathVariable; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RequestMethod; -import org.springframework.web.bind.annotation.RequestParam; -import org.springframework.web.bind.annotation.RestController; +import com.google.common.base.Joiner; import com.ctrip.apollo.biz.entity.Namespace; import com.ctrip.apollo.biz.entity.Release; @@ -19,9 +11,20 @@ import com.ctrip.apollo.biz.service.NamespaceService; import com.ctrip.apollo.biz.service.ReleaseService; import com.ctrip.apollo.common.auth.ActiveUser; import com.ctrip.apollo.common.utils.BeanUtils; +import com.ctrip.apollo.core.ConfigConsts; import com.ctrip.apollo.core.dto.ReleaseDTO; import com.ctrip.apollo.core.exception.NotFoundException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.security.core.userdetails.UserDetails; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; + +import java.util.List; + @RestController public class ReleaseController { @@ -37,26 +40,29 @@ public class ReleaseController { @Autowired private MessageSender messageSender; + private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR); + @RequestMapping("/release/{releaseId}") public ReleaseDTO get(@PathVariable("releaseId") long releaseId) { Release release = releaseService.findOne(releaseId); - if (release == null) + if (release == null) { throw new NotFoundException(String.format("release not found for %s", releaseId)); + } return BeanUtils.transfrom(ReleaseDTO.class, release); } @RequestMapping("/apps/{appId}/clusters/{clusterName}/namespaces/{namespaceName}/releases") public List find(@PathVariable("appId") String appId, - @PathVariable("clusterName") String clusterName, - @PathVariable("namespaceName") String namespaceName) { + @PathVariable("clusterName") String clusterName, + @PathVariable("namespaceName") String namespaceName) { List releases = releaseService.findReleases(appId, clusterName, namespaceName); return BeanUtils.batchTransform(ReleaseDTO.class, releases); } @RequestMapping("/apps/{appId}/clusters/{clusterName}/namespaces/{namespaceName}/releases/latest") public ReleaseDTO getLatest(@PathVariable("appId") String appId, - @PathVariable("clusterName") String clusterName, - @PathVariable("namespaceName") String namespaceName) { + @PathVariable("clusterName") String clusterName, + @PathVariable("namespaceName") String namespaceName) { Release release = configService.findRelease(appId, clusterName, namespaceName); if (release == null) { throw new NotFoundException(String.format("latest release not found for %s %s %s", appId, @@ -68,10 +74,11 @@ public class ReleaseController { @RequestMapping(path = "/apps/{appId}/clusters/{clusterName}/namespaces/{namespaceName}/releases", method = RequestMethod.POST) public ReleaseDTO buildRelease(@PathVariable("appId") String appId, - @PathVariable("clusterName") String clusterName, - @PathVariable("namespaceName") String namespaceName, @RequestParam("name") String name, - @RequestParam(name = "comment", required = false) String comment, - @ActiveUser UserDetails user) { + @PathVariable("clusterName") String clusterName, + @PathVariable("namespaceName") String namespaceName, + @RequestParam("name") String name, + @RequestParam(name = "comment", required = false) String comment, + @ActiveUser UserDetails user) { Namespace namespace = namespaceService.findOne(appId, clusterName, namespaceName); if (namespace == null) { throw new NotFoundException(String.format("Could not find namespace for %s %s %s", appId, @@ -84,6 +91,6 @@ public class ReleaseController { } private String assembleKey(String appId, String cluster, String namespace) { - return String.format("%s-%s-%s", appId, cluster, namespace); + return STRING_JOINER.join(appId, cluster, namespace); } } diff --git a/apollo-adminservice/src/test/java/com/ctrip/apollo/AdminServiceTestConfiguration.java b/apollo-adminservice/src/test/java/com/ctrip/apollo/AdminServiceTestConfiguration.java index 88619fb0c21572b55048d9c4587f34b329ce26da..9df291f2b50f6af02f5792da21d32301b26c1d7b 100644 --- a/apollo-adminservice/src/test/java/com/ctrip/apollo/AdminServiceTestConfiguration.java +++ b/apollo-adminservice/src/test/java/com/ctrip/apollo/AdminServiceTestConfiguration.java @@ -5,6 +5,8 @@ import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.ComponentScan.Filter; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.FilterType; +import org.springframework.security.config.annotation.web.builders.HttpSecurity; +import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter; @Configuration @ComponentScan(excludeFilters = {@Filter(type = FilterType.ASSIGNABLE_TYPE, value = { diff --git a/apollo-adminservice/src/test/java/com/ctrip/apollo/adminservice/controller/ReleaseControllerTest.java b/apollo-adminservice/src/test/java/com/ctrip/apollo/adminservice/controller/ReleaseControllerTest.java index 1b87fd5df8f5915a1a8cd55520484e02107c59cd..b2b4f150e2013d6d359fd30957d08c5bcce986cc 100644 --- a/apollo-adminservice/src/test/java/com/ctrip/apollo/adminservice/controller/ReleaseControllerTest.java +++ b/apollo-adminservice/src/test/java/com/ctrip/apollo/adminservice/controller/ReleaseControllerTest.java @@ -1,7 +1,20 @@ package com.ctrip.apollo.adminservice.controller; -import java.util.HashMap; -import java.util.Map; +import com.google.common.base.Joiner; +import com.google.gson.Gson; + +import com.ctrip.apollo.biz.entity.Namespace; +import com.ctrip.apollo.biz.message.MessageSender; +import com.ctrip.apollo.biz.message.Topics; +import com.ctrip.apollo.biz.repository.ReleaseRepository; +import com.ctrip.apollo.biz.service.NamespaceService; +import com.ctrip.apollo.biz.service.ReleaseService; +import com.ctrip.apollo.core.ConfigConsts; +import com.ctrip.apollo.core.dto.AppDTO; +import com.ctrip.apollo.core.dto.ClusterDTO; +import com.ctrip.apollo.core.dto.ItemDTO; +import com.ctrip.apollo.core.dto.NamespaceDTO; +import com.ctrip.apollo.core.dto.ReleaseDTO; import org.junit.Assert; import org.junit.Test; @@ -16,18 +29,8 @@ import org.springframework.test.util.ReflectionTestUtils; import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; -import com.ctrip.apollo.biz.entity.Namespace; -import com.ctrip.apollo.biz.message.MessageSender; -import com.ctrip.apollo.biz.message.Topics; -import com.ctrip.apollo.biz.repository.ReleaseRepository; -import com.ctrip.apollo.biz.service.NamespaceService; -import com.ctrip.apollo.biz.service.ReleaseService; -import com.ctrip.apollo.core.dto.AppDTO; -import com.ctrip.apollo.core.dto.ClusterDTO; -import com.ctrip.apollo.core.dto.ItemDTO; -import com.ctrip.apollo.core.dto.NamespaceDTO; -import com.ctrip.apollo.core.dto.ReleaseDTO; -import com.google.gson.Gson; +import java.util.HashMap; +import java.util.Map; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -119,7 +122,8 @@ public class ReleaseControllerTest extends AbstractControllerTest { .buildRelease(someAppId, someCluster, someNamespaceName, someName, someComment, someUser); verify(someMessageSender, times(1)) - .sendMessage(String.format("%s-%s-%s", someAppId, someCluster, someNamespaceName), + .sendMessage(Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(someAppId, someCluster, someNamespaceName), Topics.APOLLO_RELEASE_TOPIC); } diff --git a/apollo-adminservice/src/test/java/com/ctrip/apollo/adminservice/controller/TestWebSecurityConfig.java b/apollo-adminservice/src/test/java/com/ctrip/apollo/adminservice/controller/TestWebSecurityConfig.java index 7654901cc3ea07d55c86580393940e3cc8390673..48743f0be4efe1de9d5cca40126d56efcb93e64d 100644 --- a/apollo-adminservice/src/test/java/com/ctrip/apollo/adminservice/controller/TestWebSecurityConfig.java +++ b/apollo-adminservice/src/test/java/com/ctrip/apollo/adminservice/controller/TestWebSecurityConfig.java @@ -15,6 +15,10 @@ public class TestWebSecurityConfig extends WebSecurityConfigurerAdapter { protected void configure(HttpSecurity http) throws Exception { http.httpBasic(); http.csrf().disable(); + http.authorizeRequests().antMatchers("/").permitAll().and() + .authorizeRequests().antMatchers("/console/**").permitAll(); + + http.headers().frameOptions().disable(); } @Autowired diff --git a/apollo-adminservice/src/test/resources/data.sql b/apollo-adminservice/src/test/resources/data.sql index 58cfbee298332ce8538d32f96c2b3c0a14148f7d..18c8103b5784a824348201b0ca9bd4ba504b5f31 100644 --- a/apollo-adminservice/src/test/resources/data.sql +++ b/apollo-adminservice/src/test/resources/data.sql @@ -11,24 +11,23 @@ INSERT INTO Cluster (AppId, Name) VALUES ('100003173', 'default'); INSERT INTO Cluster (AppId, Name) VALUES ('100003173', 'cluster3'); INSERT INTO Cluster (AppId, Name) VALUES ('fxhermesproducer', 'default'); -INSERT INTO AppNamespace (AppId, Name) VALUES ('100003171', '100003171'); +INSERT INTO AppNamespace (AppId, Name) VALUES ('100003171', 'application'); INSERT INTO AppNamespace (AppId, Name) VALUES ('100003171', 'fx.apollo.config'); -INSERT INTO AppNamespace (AppId, Name) VALUES ('100003172', '100003172'); +INSERT INTO AppNamespace (AppId, Name) VALUES ('100003172', 'application'); INSERT INTO AppNamespace (AppId, Name) VALUES ('100003172', 'fx.apollo.admin'); -INSERT INTO AppNamespace (AppId, Name) VALUES ('100003173', '100003173'); +INSERT INTO AppNamespace (AppId, Name) VALUES ('100003173', 'application'); INSERT INTO AppNamespace (AppId, Name) VALUES ('100003173', 'fx.apollo.portal'); INSERT INTO AppNamespace (AppID, Name) VALUES ('fxhermesproducer', 'fx.hermes.producer'); -INSERT INTO Namespace (Id, AppId, ClusterName, NamespaceName) VALUES (1, '100003171', 'default', '100003171'); +INSERT INTO Namespace (Id, AppId, ClusterName, NamespaceName) VALUES (1, '100003171', 'default', 'application'); INSERT INTO Namespace (Id, AppId, ClusterName, NamespaceName) VALUES (2, 'fxhermesproducer', 'default', 'fx.hermes.producer'); -INSERT INTO Namespace (Id, AppId, ClusterName, NamespaceName) VALUES (3, '100003172', 'default', '100003172'); -INSERT INTO Namespace (Id, AppId, ClusterName, NamespaceName) VALUES (4, '100003173', 'default', '100003173'); -INSERT INTO Namespace (Id, AppId, ClusterName, NamespaceName) VALUES (5, '100003171', 'default', '100003171'); +INSERT INTO Namespace (Id, AppId, ClusterName, NamespaceName) VALUES (3, '100003172', 'default', 'application'); +INSERT INTO Namespace (Id, AppId, ClusterName, NamespaceName) VALUES (4, '100003173', 'default', 'application'); INSERT INTO Item (NamespaceId, `Key`, Value, Comment) VALUES (1, 'k1', 'v1', 'comment1'); INSERT INTO Item (NamespaceId, `Key`, Value, Comment) VALUES (1, 'k2', 'v2', 'comment2'); INSERT INTO Item (NamespaceId, `Key`, Value, Comment) VALUES (2, 'k3', 'v3', 'comment3'); INSERT INTO Item (NamespaceId, `Key`, Value, Comment) VALUES (5, 'k3', 'v4', 'comment4'); -INSERT INTO RELEASE (Name, Comment, AppId, ClusterName, NamespaceName, Configurations) VALUES ('REV1','First Release','100003171', 'default', '100003171', '{"k1":"v1"}'); +INSERT INTO RELEASE (Name, Comment, AppId, ClusterName, NamespaceName, Configurations) VALUES ('REV1','First Release','100003171', 'default', 'application', '{"k1":"v1"}'); diff --git a/apollo-biz/pom.xml b/apollo-biz/pom.xml index 1b08a15d467479f6f75066795dec74be8effc22e..ac5085867fb69f378d04b83006c4e18024ef5cb2 100644 --- a/apollo-biz/pom.xml +++ b/apollo-biz/pom.xml @@ -22,10 +22,6 @@ org.springframework.boot spring-boot-starter-data-jpa - - org.springframework.boot - spring-boot-starter-redis - mysql mysql-connector-java diff --git a/apollo-biz/src/main/java/com/ctrip/apollo/biz/entity/ReleaseMessage.java b/apollo-biz/src/main/java/com/ctrip/apollo/biz/entity/ReleaseMessage.java new file mode 100644 index 0000000000000000000000000000000000000000..2e05f6af6ee4c663e09375b9bca1d546f477a96d --- /dev/null +++ b/apollo-biz/src/main/java/com/ctrip/apollo/biz/entity/ReleaseMessage.java @@ -0,0 +1,58 @@ +package com.ctrip.apollo.biz.entity; + +import java.util.Date; + +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.GeneratedValue; +import javax.persistence.Id; +import javax.persistence.PrePersist; +import javax.persistence.Table; + +/** + * @author Jason Song(song_s@ctrip.com) + */ +@Entity +@Table(name = "ReleaseMessage") +public class ReleaseMessage { + @Id + @GeneratedValue + @Column(name = "Id") + private long id; + + @Column(name = "Message", nullable = false) + private String message; + + @Column(name = "DataChange_LastTime") + private Date dataChangeLastModifiedTime; + + @PrePersist + protected void prePersist() { + if (this.dataChangeLastModifiedTime == null) { + dataChangeLastModifiedTime = new Date(); + } + } + + public ReleaseMessage() { + } + + public ReleaseMessage(String message) { + this.message = message; + } + + public long getId() { + return id; + } + + public void setId(long id) { + this.id = id; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } +} diff --git a/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/DatabaseMessageSender.java b/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/DatabaseMessageSender.java new file mode 100644 index 0000000000000000000000000000000000000000..13c1deb4d4ddd85e90b80f5ad5b25f3bcaeadb46 --- /dev/null +++ b/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/DatabaseMessageSender.java @@ -0,0 +1,46 @@ +package com.ctrip.apollo.biz.message; + +import com.ctrip.apollo.biz.entity.ReleaseMessage; +import com.ctrip.apollo.biz.repository.ReleaseMessageRepository; +import com.dianping.cat.Cat; +import com.dianping.cat.message.Message; +import com.dianping.cat.message.Transaction; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import java.util.Objects; + +/** + * @author Jason Song(song_s@ctrip.com) + */ +@Component +public class DatabaseMessageSender implements MessageSender { + private static final Logger logger = LoggerFactory.getLogger(DatabaseMessageSender.class); + + @Autowired + private ReleaseMessageRepository releaseMessageRepository; + + @Override + public void sendMessage(String message, String channel) { + logger.info("Sending message {} to channel {}", message, channel); + if (!Objects.equals(channel, Topics.APOLLO_RELEASE_TOPIC)) { + logger.warn("Channel {} not supported by DatabaseMessageSender!"); + return; + } + + Cat.logEvent("Apollo.AdminService.ReleaseMessage", message); + Transaction transaction = Cat.newTransaction("Apollo.AdminService", "sendMessage"); + try { + releaseMessageRepository.save(new ReleaseMessage(message)); + transaction.setStatus(Message.SUCCESS); + } catch (Throwable ex) { + logger.error("Sending message to database failed", ex); + transaction.setStatus(ex); + } finally { + transaction.complete(); + } + } +} diff --git a/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/DummyMessageSender.java b/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/DummyMessageSender.java deleted file mode 100644 index db2f8b2afd2f43c56a2ce330e78a5ae7554ee56e..0000000000000000000000000000000000000000 --- a/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/DummyMessageSender.java +++ /dev/null @@ -1,15 +0,0 @@ -package com.ctrip.apollo.biz.message; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * @author Jason Song(song_s@ctrip.com) - */ -public class DummyMessageSender implements MessageSender{ - private static final Logger logger = LoggerFactory.getLogger(DummyMessageSender.class); - @Override - public void sendMessage(String message, String channel) { - logger.warn("No message sender available! message: {}, channel: {}", message, channel); - } -} diff --git a/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/RedisMessageSender.java b/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/RedisMessageSender.java deleted file mode 100644 index 4d570a7b68aa0b05194b675e2a9720bdfc91dd3b..0000000000000000000000000000000000000000 --- a/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/RedisMessageSender.java +++ /dev/null @@ -1,37 +0,0 @@ -package com.ctrip.apollo.biz.message; - -import com.dianping.cat.Cat; -import com.dianping.cat.message.Message; -import com.dianping.cat.message.Transaction; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.data.redis.core.RedisTemplate; - -/** - * @author Jason Song(song_s@ctrip.com) - */ -public class RedisMessageSender implements MessageSender { - private static final Logger logger = LoggerFactory.getLogger(RedisMessageSender.class); - private RedisTemplate redisTemplate; - - public RedisMessageSender( - RedisTemplate redisTemplate) { - this.redisTemplate = redisTemplate; - } - - @Override - public void sendMessage(String message, String channel) { - logger.info("Sending message {} to channel {}", message, channel); - Transaction transaction = Cat.newTransaction("Apollo.AdminService", "RedisMessageSender"); - try { - redisTemplate.convertAndSend(channel, message); - transaction.setStatus(Message.SUCCESS); - } catch (Throwable ex) { - logger.error("Sending message to redis failed", ex); - transaction.setStatus(ex); - } finally { - transaction.complete(); - } - } -} diff --git a/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/ReleaseMessageScanner.java b/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/ReleaseMessageScanner.java new file mode 100644 index 0000000000000000000000000000000000000000..effcffcebc3864cde9a58210c8f371d162fa66bb --- /dev/null +++ b/apollo-biz/src/main/java/com/ctrip/apollo/biz/message/ReleaseMessageScanner.java @@ -0,0 +1,145 @@ +package com.ctrip.apollo.biz.message; + +import com.google.common.collect.Lists; + +import com.ctrip.apollo.biz.entity.ReleaseMessage; +import com.ctrip.apollo.biz.repository.ReleaseMessageRepository; +import com.ctrip.apollo.core.utils.ApolloThreadFactory; +import com.dianping.cat.Cat; +import com.dianping.cat.message.Message; +import com.dianping.cat.message.Transaction; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.core.env.Environment; +import org.springframework.util.CollectionUtils; + +import java.util.List; +import java.util.Objects; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * @author Jason Song(song_s@ctrip.com) + */ +public class ReleaseMessageScanner implements InitializingBean { + private static final Logger logger = LoggerFactory.getLogger(ReleaseMessageScanner.class); + private static final int DEFAULT_SCAN_INTERVAL_IN_MS = 1000; + @Autowired + private Environment env; + @Autowired + private ReleaseMessageRepository releaseMessageRepository; + private int databaseScanInterval; + private List listeners; + private ScheduledExecutorService executorService; + private long maxIdScanned; + + public ReleaseMessageScanner() { + listeners = Lists.newLinkedList(); + executorService = Executors.newScheduledThreadPool(1, ApolloThreadFactory + .create("ReleaseMessageScanner", true)); + } + + @Override + public void afterPropertiesSet() throws Exception { + populateDataBaseInterval(); + maxIdScanned = loadLargestMessageId(); + executorService.scheduleWithFixedDelay((Runnable) () -> { + Transaction transaction = Cat.newTransaction("Apollo.ReleaseMessageScanner", "scanMessage"); + try { + scanMessages(); + transaction.setStatus(Message.SUCCESS); + } catch (Throwable ex) { + transaction.setStatus(ex); + logger.error("Scan and send message failed", ex); + } finally { + transaction.complete(); + } + }, getDatabaseScanIntervalMs(), getDatabaseScanIntervalMs(), TimeUnit.MILLISECONDS); + + } + + /** + * add message listeners for release message + * @param listener + */ + public void addMessageListener(MessageListener listener) { + if (!listeners.contains(listener)) { + listeners.add(listener); + } + } + + /** + * Scan messages, continue scanning until there is no more messages + */ + private void scanMessages() { + boolean hasMoreMessages = true; + while (hasMoreMessages && !Thread.currentThread().isInterrupted()) { + hasMoreMessages = scanAndSendMessages(); + } + } + + /** + * scan messages and send + * + * @return whether there are more messages + */ + private boolean scanAndSendMessages() { + //current batch is 500 + List releaseMessages = + releaseMessageRepository.findFirst500ByIdGreaterThanOrderByIdAsc(maxIdScanned); + if (CollectionUtils.isEmpty(releaseMessages)) { + return false; + } + fireMessageScanned(releaseMessages); + int messageScanned = releaseMessages.size(); + maxIdScanned = releaseMessages.get(messageScanned - 1).getId(); + return messageScanned == 500; + } + + /** + * find largest message id as the current start point + * @return current largest message id + */ + private long loadLargestMessageId() { + ReleaseMessage releaseMessage = releaseMessageRepository.findTopByOrderByIdDesc(); + return releaseMessage == null ? 0 : releaseMessage.getId(); + } + + /** + * Notify listeners with messages loaded + * @param messages + */ + private void fireMessageScanned(List messages) { + for (ReleaseMessage message : messages) { + for (MessageListener listener : listeners) { + try { + listener.handleMessage(message.getMessage(), Topics.APOLLO_RELEASE_TOPIC); + } catch (Throwable ex) { + Cat.logError(ex); + logger.error("Failed to invoke message listener {}", listener.getClass(), ex); + } + } + } + } + + private void populateDataBaseInterval() { + databaseScanInterval = DEFAULT_SCAN_INTERVAL_IN_MS; + try { + String interval = env.getProperty("apollo.message-scan.interval"); + if (!Objects.isNull(interval)) { + databaseScanInterval = Integer.parseInt(interval); + } + } catch (Throwable ex) { + Cat.logError(ex); + logger.error("Load apollo message scan interval from system property failed", ex); + } + } + + private int getDatabaseScanIntervalMs() { + return databaseScanInterval; + } +} diff --git a/apollo-biz/src/main/java/com/ctrip/apollo/biz/repository/ReleaseMessageRepository.java b/apollo-biz/src/main/java/com/ctrip/apollo/biz/repository/ReleaseMessageRepository.java new file mode 100644 index 0000000000000000000000000000000000000000..1d0362e3b65f795a994fdad2a7314935c5170705 --- /dev/null +++ b/apollo-biz/src/main/java/com/ctrip/apollo/biz/repository/ReleaseMessageRepository.java @@ -0,0 +1,16 @@ +package com.ctrip.apollo.biz.repository; + +import com.ctrip.apollo.biz.entity.ReleaseMessage; + +import org.springframework.data.repository.PagingAndSortingRepository; + +import java.util.List; + +/** + * @author Jason Song(song_s@ctrip.com) + */ +public interface ReleaseMessageRepository extends PagingAndSortingRepository { + List findFirst500ByIdGreaterThanOrderByIdAsc(Long id); + + ReleaseMessage findTopByOrderByIdDesc(); +} diff --git a/apollo-biz/src/test/java/com/ctrip/apollo/biz/AllTests.java b/apollo-biz/src/test/java/com/ctrip/apollo/biz/AllTests.java index 49cfffc506dcae18a2513e62b0a736f09df56487..ec86b5f38ade60d641710e983f7299ae3967b7ab 100644 --- a/apollo-biz/src/test/java/com/ctrip/apollo/biz/AllTests.java +++ b/apollo-biz/src/test/java/com/ctrip/apollo/biz/AllTests.java @@ -1,5 +1,6 @@ package com.ctrip.apollo.biz; +import com.ctrip.apollo.biz.message.DatabaseMessageSenderTest; import com.ctrip.apollo.biz.repository.AppRepositoryTest; import com.ctrip.apollo.biz.service.AdminServiceTest; import com.ctrip.apollo.biz.service.AdminServiceTransactionTest; @@ -16,7 +17,8 @@ import org.junit.runners.Suite.SuiteClasses; AdminServiceTest.class, ConfigServiceTest.class, PrivilegeServiceTest.class, - AdminServiceTransactionTest.class}) + AdminServiceTransactionTest.class, + DatabaseMessageSenderTest.class}) public class AllTests { } diff --git a/apollo-biz/src/test/java/com/ctrip/apollo/biz/message/DatabaseMessageSenderTest.java b/apollo-biz/src/test/java/com/ctrip/apollo/biz/message/DatabaseMessageSenderTest.java new file mode 100644 index 0000000000000000000000000000000000000000..00f76c767141266f973eee56f81f92e66e6441d8 --- /dev/null +++ b/apollo-biz/src/test/java/com/ctrip/apollo/biz/message/DatabaseMessageSenderTest.java @@ -0,0 +1,55 @@ +package com.ctrip.apollo.biz.message; + +import com.ctrip.apollo.biz.entity.ReleaseMessage; +import com.ctrip.apollo.biz.repository.ReleaseMessageRepository; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import org.springframework.test.util.ReflectionTestUtils; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** + * @author Jason Song(song_s@ctrip.com) + */ +@RunWith(MockitoJUnitRunner.class) +public class DatabaseMessageSenderTest { + private DatabaseMessageSender messageSender; + @Mock + private ReleaseMessageRepository releaseMessageRepository; + + @Before + public void setUp() throws Exception { + messageSender = new DatabaseMessageSender(); + ReflectionTestUtils.setField(messageSender, "releaseMessageRepository", releaseMessageRepository); + } + + @Test + public void testSendMessage() throws Exception { + String someMessage = "some-message"; + ArgumentCaptor captor = ArgumentCaptor.forClass(ReleaseMessage.class); + + messageSender.sendMessage(someMessage, Topics.APOLLO_RELEASE_TOPIC); + + verify(releaseMessageRepository, times(1)).save(captor.capture()); + assertEquals(someMessage, captor.getValue().getMessage()); + } + + @Test + public void testSendUnsupportedMessage() throws Exception { + String someMessage = "some-message"; + String someUnsupportedTopic = "some-invalid-topic"; + + messageSender.sendMessage(someMessage, someUnsupportedTopic); + + verify(releaseMessageRepository, never()).save(any(ReleaseMessage.class)); + } +} diff --git a/apollo-biz/src/test/java/com/ctrip/apollo/biz/message/RedisMessageSenderTest.java b/apollo-biz/src/test/java/com/ctrip/apollo/biz/message/RedisMessageSenderTest.java deleted file mode 100644 index dc2cab878f9f8293fdc1710b114177be3e9e7a07..0000000000000000000000000000000000000000 --- a/apollo-biz/src/test/java/com/ctrip/apollo/biz/message/RedisMessageSenderTest.java +++ /dev/null @@ -1,49 +0,0 @@ -package com.ctrip.apollo.biz.message; - -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mock; -import org.mockito.runners.MockitoJUnitRunner; -import org.springframework.data.redis.core.RedisTemplate; - -import static org.mockito.Mockito.doThrow; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; - -/** - * @author Jason Song(song_s@ctrip.com) - */ -@RunWith(MockitoJUnitRunner.class) -public class RedisMessageSenderTest { - @Mock - private RedisTemplate redisTemplate; - private RedisMessageSender redisMessageSender; - - @Before - public void setUp() throws Exception { - redisMessageSender = new RedisMessageSender(redisTemplate); - } - - @Test - public void testSendMessage() throws Exception { - String someMessage = "someMessage"; - String someChannel = "someChannel"; - - redisMessageSender.sendMessage(someMessage, someChannel); - - verify(redisTemplate, times(1)).convertAndSend(someChannel, someMessage); - } - - @Test - public void testSendMessageWithError() throws Exception { - String someMessage = "someMessage"; - String someChannel = "someChannel"; - - doThrow(new RuntimeException()).when(redisTemplate).convertAndSend(someChannel, someMessage); - - redisMessageSender.sendMessage(someMessage, someChannel); - - } - -} diff --git a/apollo-biz/src/test/java/com/ctrip/apollo/biz/message/ReleaseMessageScannerTest.java b/apollo-biz/src/test/java/com/ctrip/apollo/biz/message/ReleaseMessageScannerTest.java new file mode 100644 index 0000000000000000000000000000000000000000..432c81cc9cf51fe9af3e652ad91eb864852d9473 --- /dev/null +++ b/apollo-biz/src/test/java/com/ctrip/apollo/biz/message/ReleaseMessageScannerTest.java @@ -0,0 +1,87 @@ +package com.ctrip.apollo.biz.message; + +import com.google.common.collect.Lists; +import com.google.common.util.concurrent.SettableFuture; + +import com.ctrip.apollo.biz.entity.ReleaseMessage; +import com.ctrip.apollo.biz.repository.ReleaseMessageRepository; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import org.springframework.core.env.Environment; +import org.springframework.test.util.ReflectionTestUtils; + +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.when; + +/** + * @author Jason Song(song_s@ctrip.com) + */ +@RunWith(MockitoJUnitRunner.class) +public class ReleaseMessageScannerTest { + private ReleaseMessageScanner releaseMessageScanner; + @Mock + private ReleaseMessageRepository releaseMessageRepository; + @Mock + private Environment env; + private int databaseScanInterval; + + @Before + public void setUp() throws Exception { + releaseMessageScanner = new ReleaseMessageScanner(); + ReflectionTestUtils + .setField(releaseMessageScanner, "releaseMessageRepository", releaseMessageRepository); + ReflectionTestUtils.setField(releaseMessageScanner, "env", env); + databaseScanInterval = 100; //100 ms + when(env.getProperty("apollo.message-scan.interval")).thenReturn(String.valueOf(databaseScanInterval)); + releaseMessageScanner.afterPropertiesSet(); + } + + @Test + public void testScanMessageAndNotifyMessageListener() throws Exception { + SettableFuture someListenerFuture = SettableFuture.create(); + MessageListener someListener = (message, channel) -> someListenerFuture.set(message); + releaseMessageScanner.addMessageListener(someListener); + + String someMessage = "someMessage"; + long someId = 100; + ReleaseMessage someReleaseMessage = assembleReleaseMessage(someId, someMessage); + + when(releaseMessageRepository.findFirst500ByIdGreaterThanOrderByIdAsc(0L)).thenReturn( + Lists.newArrayList(someReleaseMessage)); + + String someListenerMessage = + someListenerFuture.get(5000, TimeUnit.MILLISECONDS); + + assertEquals(someMessage, someListenerMessage); + + SettableFuture anotherListenerFuture = SettableFuture.create(); + MessageListener anotherListener = (message, channel) -> anotherListenerFuture.set(message); + releaseMessageScanner.addMessageListener(anotherListener); + + String anotherMessage = "anotherMessage"; + long anotherId = someId + 1; + ReleaseMessage anotherReleaseMessage = assembleReleaseMessage(anotherId, anotherMessage); + + when(releaseMessageRepository.findFirst500ByIdGreaterThanOrderByIdAsc(someId)).thenReturn( + Lists.newArrayList(anotherReleaseMessage)); + + String anotherListenerMessage = + anotherListenerFuture.get(5000, TimeUnit.MILLISECONDS); + + assertEquals(anotherMessage, anotherListenerMessage); + + } + + private ReleaseMessage assembleReleaseMessage(long id, String message) { + ReleaseMessage releaseMessage = new ReleaseMessage(); + releaseMessage.setId(id); + releaseMessage.setMessage(message); + return releaseMessage; + } +} diff --git a/apollo-client/src/main/java/com/ctrip/apollo/internals/AbstractConfigRepository.java b/apollo-client/src/main/java/com/ctrip/apollo/internals/AbstractConfigRepository.java index 886ebb366acf46e3fb296dd5010ba5e1136d3bed..c26348441a2e6c806b29c91dac2b862252d207e0 100644 --- a/apollo-client/src/main/java/com/ctrip/apollo/internals/AbstractConfigRepository.java +++ b/apollo-client/src/main/java/com/ctrip/apollo/internals/AbstractConfigRepository.java @@ -25,7 +25,7 @@ public abstract class AbstractConfigRepository implements ConfigRepository { } catch (Throwable ex) { Cat.logError(ex); logger - .warn("Sync config failed with repository {}, reason: {}", this.getClass(), ExceptionUtil + .warn("Sync config failed, will retry. Repository {}, reason: {}", this.getClass(), ExceptionUtil .getDetailMessage(ex)); } return false; diff --git a/apollo-client/src/main/java/com/ctrip/apollo/internals/LocalFileConfigRepository.java b/apollo-client/src/main/java/com/ctrip/apollo/internals/LocalFileConfigRepository.java index dc06d039d0077877c2bd832234d60d0827752deb..6459147bd690fe18efd1413923359bc2cfd1f2b7 100644 --- a/apollo-client/src/main/java/com/ctrip/apollo/internals/LocalFileConfigRepository.java +++ b/apollo-client/src/main/java/com/ctrip/apollo/internals/LocalFileConfigRepository.java @@ -1,7 +1,9 @@ package com.ctrip.apollo.internals; +import com.google.common.base.Joiner; import com.google.common.base.Preconditions; +import com.ctrip.apollo.core.ConfigConsts; import com.ctrip.apollo.util.ConfigUtil; import com.ctrip.apollo.util.ExceptionUtil; import com.dianping.cat.Cat; @@ -202,8 +204,10 @@ public class LocalFileConfigRepository extends AbstractConfigRepository } File assembleLocalCacheFile(File baseDir, String namespace) { - String fileName = String.format("%s-%s-%s.properties", m_configUtil.getAppId(), - m_configUtil.getCluster(), namespace); + + String fileName = + String.format("%s.properties", Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(m_configUtil.getAppId(), m_configUtil.getCluster(), namespace)); return new File(baseDir, fileName); } } diff --git a/apollo-client/src/main/java/com/ctrip/apollo/internals/RemoteConfigRepository.java b/apollo-client/src/main/java/com/ctrip/apollo/internals/RemoteConfigRepository.java index a22e3ef38ce563bc10132d8ae424d40ba0a68849..4fd923ff04ab344602bf79c9f894907617c88558 100644 --- a/apollo-client/src/main/java/com/ctrip/apollo/internals/RemoteConfigRepository.java +++ b/apollo-client/src/main/java/com/ctrip/apollo/internals/RemoteConfigRepository.java @@ -7,6 +7,7 @@ import com.google.common.collect.Maps; import com.google.common.escape.Escaper; import com.google.common.net.UrlEscapers; +import com.ctrip.apollo.core.ConfigConsts; import com.ctrip.apollo.core.dto.ApolloConfig; import com.ctrip.apollo.core.dto.ApolloConfigNotification; import com.ctrip.apollo.core.dto.ServiceDTO; @@ -43,6 +44,8 @@ import java.util.concurrent.atomic.AtomicReference; */ public class RemoteConfigRepository extends AbstractConfigRepository { private static final Logger logger = LoggerFactory.getLogger(RemoteConfigRepository.class); + private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR); + private static final Joiner.MapJoiner MAP_JOINER = Joiner.on("&").withKeyValueSeparator("="); private PlexusContainer m_container; private final ConfigServiceLocator m_serviceLocator; private final HttpUtil m_httpUtil; @@ -135,8 +138,7 @@ public class RemoteConfigRepository extends AbstractConfigRepository { String appId = m_configUtil.getAppId(); String cluster = m_configUtil.getCluster(); String dataCenter = m_configUtil.getDataCenter(); - Cat.logEvent("Apollo.Client.ConfigInfo", - String.format("%s-%s-%s", appId, cluster, m_namespace)); + Cat.logEvent("Apollo.Client.ConfigInfo", STRING_JOINER.join(appId, cluster, m_namespace)); int maxRetries = 2; Throwable exception = null; @@ -214,7 +216,7 @@ public class RemoteConfigRepository extends AbstractConfigRepository { String pathExpanded = String.format(path, pathParams.toArray()); if (!queryParams.isEmpty()) { - pathExpanded += "?" + Joiner.on("&").withKeyValueSeparator("=").join(queryParams); + pathExpanded += "?" + MAP_JOINER.join(queryParams); } if (!uri.endsWith("/")) { uri += "/"; @@ -276,7 +278,7 @@ public class RemoteConfigRepository extends AbstractConfigRepository { transaction.addData("StatusCode", response.getStatusCode()); transaction.setStatus(Message.SUCCESS); } catch (Throwable ex) { - logger.warn("Long polling failed for appId: {}, cluster: {}, namespace: {}, reason: {}", + logger.warn("Long polling failed, will retry. appId: {}, cluster: {}, namespace: {}, reason: {}", appId, cluster, m_namespace, ExceptionUtil.getDetailMessage(ex)); lastServiceDto = null; Cat.logError(ex); @@ -284,7 +286,7 @@ public class RemoteConfigRepository extends AbstractConfigRepository { transaction.setStatus(ex); } try { - TimeUnit.SECONDS.sleep(10); + TimeUnit.SECONDS.sleep(5); } catch (InterruptedException ie) { //ignore } @@ -314,7 +316,7 @@ public class RemoteConfigRepository extends AbstractConfigRepository { queryParams.put("releaseId", escaper.escape(previousConfig.getReleaseId())); } - String params = Joiner.on("&").withKeyValueSeparator("=").join(queryParams); + String params = MAP_JOINER.join(queryParams); if (!uri.endsWith("/")) { uri += "/"; } diff --git a/apollo-client/src/test/java/com/ctrip/apollo/integration/ConfigIntegrationTest.java b/apollo-client/src/test/java/com/ctrip/apollo/integration/ConfigIntegrationTest.java index 8a0e2bf6d43f146901fc14f0ad10fc007e28a85f..2cc67579130682cb51272e3ce059f344848ae519 100644 --- a/apollo-client/src/test/java/com/ctrip/apollo/integration/ConfigIntegrationTest.java +++ b/apollo-client/src/test/java/com/ctrip/apollo/integration/ConfigIntegrationTest.java @@ -1,5 +1,6 @@ package com.ctrip.apollo.integration; +import com.google.common.base.Joiner; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Lists; import com.google.common.collect.Maps; @@ -350,6 +351,7 @@ public class ConfigIntegrationTest extends BaseIntegrationTest { } private String assembleLocalCacheFileName() { - return String.format("%s-%s-%s.properties", someAppId, someClusterName, defaultNamespace); + return String.format("%s.properties", Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(someAppId, someClusterName, defaultNamespace)); } } diff --git a/apollo-client/src/test/java/com/ctrip/apollo/internals/LocalFileConfigRepositoryTest.java b/apollo-client/src/test/java/com/ctrip/apollo/internals/LocalFileConfigRepositoryTest.java index 9c9cdc9b960618ad25c9f3829da7351ce2af032c..84704d6b59bf003959d46c0b75af95b8bebb795f 100644 --- a/apollo-client/src/test/java/com/ctrip/apollo/internals/LocalFileConfigRepositoryTest.java +++ b/apollo-client/src/test/java/com/ctrip/apollo/internals/LocalFileConfigRepositoryTest.java @@ -1,8 +1,10 @@ package com.ctrip.apollo.internals; import com.google.common.base.Charsets; +import com.google.common.base.Joiner; import com.google.common.io.Files; +import com.ctrip.apollo.core.ConfigConsts; import com.ctrip.apollo.util.ConfigUtil; import org.junit.After; @@ -73,8 +75,8 @@ public class LocalFileConfigRepositoryTest extends ComponentTestCase { } private String assembleLocalCacheFileName() { - return String.format("%s-%s-%s.properties", someAppId, - someCluster, someNamespace); + return String.format("%s.properties", Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(someAppId, someCluster, someNamespace)); } @@ -144,7 +146,7 @@ public class LocalFileConfigRepositoryTest extends ComponentTestCase { assertThat( "LocalFileConfigRepository should persist local cache files and return that afterwards", - someProperties.entrySet(), equalTo(anotherProperties.entrySet())); + someProperties.entrySet(), equalTo(anotherProperties.entrySet())); } diff --git a/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/ConfigServiceAutoConfiguration.java b/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/ConfigServiceAutoConfiguration.java index 1081f078d01f0ec56ef78bebc39aa928a19a8aab..992c099297a5f0fdf3d94251e1a86dd98bc6713c 100644 --- a/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/ConfigServiceAutoConfiguration.java +++ b/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/ConfigServiceAutoConfiguration.java @@ -1,59 +1,25 @@ package com.ctrip.apollo.configservice; -import com.ctrip.apollo.biz.message.Topics; +import com.ctrip.apollo.biz.message.ReleaseMessageScanner; import com.ctrip.apollo.configservice.controller.NotificationController; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +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.data.redis.connection.jedis.JedisConnectionFactory; -import org.springframework.data.redis.listener.ChannelTopic; -import org.springframework.data.redis.listener.RedisMessageListenerContainer; -import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; /** * @author Jason Song(song_s@ctrip.com) */ @Configuration public class ConfigServiceAutoConfiguration { - - @ConditionalOnProperty(value = "apollo.redis.enabled", havingValue = "true", matchIfMissing = false) - public static class ConfigRedisConfiguration { - @Value("${apollo.redis.host}") - private String host; - @Value("${apollo.redis.port}") - private int port; - - @Bean - public JedisConnectionFactory redisConnectionFactory() { - JedisConnectionFactory factory = new JedisConnectionFactory(); - factory.setHostName(host); - factory.setPort(port); - return factory; - } - - @Bean - public RedisMessageListenerContainer redisMessageListenerContainer( - RedisConnectionFactory factory) { - RedisMessageListenerContainer container = new RedisMessageListenerContainer(); - container.setConnectionFactory(factory); - return container; - } - - @Bean - public ChannelTopic apolloReleaseTopic() { - return new ChannelTopic(Topics.APOLLO_RELEASE_TOPIC); - } - - @Bean - public MessageListenerAdapter apolloMessageListener(RedisMessageListenerContainer container, - NotificationController notification, - ChannelTopic topic) { - MessageListenerAdapter adapter = new MessageListenerAdapter(notification); - container.addMessageListener(adapter, topic); - return adapter; - } + @Autowired + private NotificationController notificationController; + + @Bean + public ReleaseMessageScanner releaseMessageScanner() { + ReleaseMessageScanner releaseMessageScanner = new ReleaseMessageScanner(); + releaseMessageScanner.addMessageListener(notificationController); + return releaseMessageScanner; } + } diff --git a/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/controller/ConfigController.java b/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/controller/ConfigController.java index 0a98ec19260cd327c3a95795c265f5f7ee2178d9..ac1a78a2ac5a19ed2a0ab6cf3250fd460fcf92ba 100644 --- a/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/controller/ConfigController.java +++ b/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/controller/ConfigController.java @@ -42,10 +42,11 @@ public class ConfigController { @Autowired private AppNamespaceService appNamespaceService; - private Gson gson = new Gson(); - private Type configurationTypeReference = + private static final Gson gson = new Gson(); + private static final Type configurationTypeReference = new TypeToken>() { }.getType(); + private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR); @RequestMapping(value = "/{appId}/{clusterName}", method = RequestMethod.GET) public ApolloConfig queryConfig(@PathVariable String appId, @PathVariable String clusterName, @@ -89,7 +90,7 @@ public class ConfigController { } String mergedReleaseId = FluentIterable.from(releases).transform( - input -> String.valueOf(input.getId())).join(Joiner.on("|")); + input -> String.valueOf(input.getId())).join(STRING_JOINER); if (mergedReleaseId.equals(clientSideReleaseId)) { // Client side configuration is the same with server side, return 304 @@ -148,11 +149,11 @@ public class ConfigController { } private String assembleKey(String appId, String cluster, String namespace, String datacenter) { - String key = String.format("%s-%s-%s", appId, cluster, namespace); + List keyParts = Lists.newArrayList(appId, cluster, namespace); if (!Strings.isNullOrEmpty(datacenter)) { - key += "-" + datacenter; + keyParts.add(datacenter); } - return key; + return STRING_JOINER.join(keyParts); } } diff --git a/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/controller/NotificationController.java b/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/controller/NotificationController.java index 432ccd84961c0c6073900eacd9f8b106be1f3e90..f4448db30ad241bef1e45f7467c7c5f6307883e8 100644 --- a/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/controller/NotificationController.java +++ b/apollo-configservice/src/main/java/com/ctrip/apollo/configservice/controller/NotificationController.java @@ -1,5 +1,7 @@ package com.ctrip.apollo.configservice.controller; +import com.google.common.base.Joiner; +import com.google.common.base.Splitter; import com.google.common.base.Strings; import com.google.common.collect.HashMultimap; import com.google.common.collect.Lists; @@ -29,8 +31,6 @@ import java.util.Collection; import java.util.List; import java.util.Objects; -import javax.servlet.http.HttpServletResponse; - /** * @author Jason Song(song_s@ctrip.com) */ @@ -40,8 +40,12 @@ public class NotificationController implements MessageListener { private static final Logger logger = LoggerFactory.getLogger(NotificationController.class); private static final long TIMEOUT = 360 * 60 * 1000;//6 hours private final Multimap>> - deferredResults = - Multimaps.synchronizedSetMultimap(HashMultimap.create()); + deferredResults = Multimaps.synchronizedSetMultimap(HashMultimap.create()); + private static final ResponseEntity + NOT_MODIFIED_RESPONSE = new ResponseEntity<>(HttpStatus.NOT_MODIFIED); + private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR); + private static final Splitter STRING_SPLITTER = + Splitter.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR); @Autowired private AppNamespaceService appNamespaceService; @@ -52,8 +56,7 @@ public class NotificationController implements MessageListener { @RequestParam(value = "cluster") String cluster, @RequestParam(value = "namespace", defaultValue = ConfigConsts.NAMESPACE_DEFAULT) String namespace, @RequestParam(value = "dataCenter", required = false) String dataCenter, - @RequestParam(value = "releaseId", defaultValue = "-1") String clientSideReleaseId, - HttpServletResponse response) { + @RequestParam(value = "releaseId", defaultValue = "-1") String clientSideReleaseId) { List watchedKeys = Lists.newArrayList(assembleKey(appId, cluster, namespace)); //Listen on more namespaces, since it's not the default namespace @@ -61,30 +64,32 @@ public class NotificationController implements MessageListener { watchedKeys.addAll(this.findPublicConfigWatchKey(appId, namespace, dataCenter)); } - ResponseEntity body = new ResponseEntity<>( - HttpStatus.NOT_MODIFIED); DeferredResult> deferredResult = - new DeferredResult<>(TIMEOUT, body); + new DeferredResult<>(TIMEOUT, NOT_MODIFIED_RESPONSE); //register all keys for (String key : watchedKeys) { this.deferredResults.put(key, deferredResult); } + deferredResult.onTimeout(() -> logWatchedKeysToCat(watchedKeys, "Apollo.LongPoll.TimeOutKeys")); + deferredResult.onCompletion(() -> { //unregister all keys for (String key : watchedKeys) { deferredResults.remove(key, deferredResult); } + logWatchedKeysToCat(watchedKeys, "Apollo.LongPoll.CompletedKeys"); }); + logWatchedKeysToCat(watchedKeys, "Apollo.LongPoll.RegisteredKeys"); logger.info("Listening {} from appId: {}, cluster: {}, namespace: {}, datacenter: {}", watchedKeys, appId, cluster, namespace, dataCenter); return deferredResult; } private String assembleKey(String appId, String cluster, String namespace) { - return String.format("%s-%s-%s", appId, cluster, namespace); + return STRING_JOINER.join(appId, cluster, namespace); } private List findPublicConfigWatchKey(String applicationId, String namespace, @@ -114,20 +119,20 @@ public class NotificationController implements MessageListener { @Override public void handleMessage(String message, String channel) { logger.info("message received - channel: {}, message: {}", channel, message); - Cat.logEvent("Apollo.LongPoll.Message", message); + Cat.logEvent("Apollo.LongPoll.Messages", message); if (!Topics.APOLLO_RELEASE_TOPIC.equals(channel) || Strings.isNullOrEmpty(message)) { return; } - String[] keys = message.split("-"); - //message should be appId-cluster-namespace - if (keys.length != 3) { + List keys = STRING_SPLITTER.splitToList(message); + //message should be appId|cluster|namespace + if (keys.size() != 3) { logger.error("message format invalid - {}", message); return; } ResponseEntity notification = new ResponseEntity<>( - new ApolloConfigNotification(keys[2]), HttpStatus.OK); + new ApolloConfigNotification(keys.get(2)), HttpStatus.OK); Collection>> results = deferredResults.get(message); @@ -137,5 +142,11 @@ public class NotificationController implements MessageListener { result.setResult(notification); } } + + private void logWatchedKeysToCat(List watchedKeys, String eventName) { + for (String watchedKey : watchedKeys) { + Cat.logEvent(eventName, watchedKey); + } + } } diff --git a/apollo-configservice/src/test/java/com/ctrip/apollo/ConfigServiceTestConfiguration.java b/apollo-configservice/src/test/java/com/ctrip/apollo/ConfigServiceTestConfiguration.java index cc849a1a3f3c36c783a1c0793c36806df2c57be9..588c6a453099503d85e460f7e28362bfe7450f18 100644 --- a/apollo-configservice/src/test/java/com/ctrip/apollo/ConfigServiceTestConfiguration.java +++ b/apollo-configservice/src/test/java/com/ctrip/apollo/ConfigServiceTestConfiguration.java @@ -1,5 +1,7 @@ package com.ctrip.apollo; +import com.ctrip.apollo.common.controller.WebSecurityConfig; + import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.ComponentScan.Filter; @@ -8,7 +10,7 @@ import org.springframework.context.annotation.FilterType; @Configuration @ComponentScan(excludeFilters = {@Filter(type = FilterType.ASSIGNABLE_TYPE, value = { - SampleConfigServiceApplication.class, ConfigServiceApplication.class})}) + SampleConfigServiceApplication.class, ConfigServiceApplication.class, WebSecurityConfig.class})}) @EnableAutoConfiguration public class ConfigServiceTestConfiguration { diff --git a/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/ConfigControllerTest.java b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/ConfigControllerTest.java index 31a08d1b739b0101d531278866e94557ea448ffa..65e45e63fd1910a30cd9dc8f73965086ed40e45b 100644 --- a/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/ConfigControllerTest.java +++ b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/ConfigControllerTest.java @@ -1,5 +1,6 @@ package com.ctrip.apollo.configservice.controller; +import com.google.common.base.Joiner; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Lists; import com.google.gson.Gson; @@ -243,7 +244,8 @@ public class ConfigControllerTest { .queryConfig(someAppId, someClusterName, somePublicNamespaceName, someDataCenter, someAppSideReleaseId, someResponse); - assertEquals(String.format("%s|%s", someAppSideReleaseId, somePublicAppSideReleaseId), + assertEquals(Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(someAppSideReleaseId, somePublicAppSideReleaseId), result.getReleaseId()); assertEquals(someAppId, result.getAppId()); assertEquals(someClusterName, result.getCluster()); diff --git a/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/NotificationControllerTest.java b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/NotificationControllerTest.java index 51e3a3c158539a80287e7092f4b992d1564ce5bd..323110703a1bb5a81fbaf32ea30ed1b4bee19cd3 100644 --- a/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/NotificationControllerTest.java +++ b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/NotificationControllerTest.java @@ -1,5 +1,6 @@ package com.ctrip.apollo.configservice.controller; +import com.google.common.base.Joiner; import com.google.common.collect.Lists; import com.google.common.collect.Multimap; @@ -40,8 +41,6 @@ public class NotificationControllerTest { private String someDataCenter; private String someReleaseId; @Mock - private HttpServletResponse response; - @Mock private AppNamespaceService appNamespaceService; private Multimap>> deferredResults; @@ -67,10 +66,11 @@ public class NotificationControllerTest { public void testPollNotificationWithDefaultNamespace() throws Exception { DeferredResult> deferredResult = controller - .pollNotification(someAppId, someCluster, defaultNamespace, someDataCenter, someReleaseId, - response); + .pollNotification(someAppId, someCluster, defaultNamespace, someDataCenter, someReleaseId); - String key = String.format("%s-%s-%s", someAppId, someCluster, defaultNamespace); + String key = + Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(someAppId, someCluster, defaultNamespace); assertEquals(1, deferredResults.size()); assertTrue(deferredResults.get(key).contains(deferredResult)); } @@ -87,18 +87,21 @@ public class NotificationControllerTest { DeferredResult> deferredResult = controller .pollNotification(someAppId, someCluster, somePublicNamespace, someDataCenter, - someReleaseId, - response); + someReleaseId); List publicClusters = Lists.newArrayList(someDataCenter, ConfigConsts.CLUSTER_NAME_DEFAULT); assertEquals(3, deferredResults.size()); - String key = String.format("%s-%s-%s", someAppId, someCluster, somePublicNamespace); + String key = + Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(someAppId, someCluster, somePublicNamespace); assertTrue(deferredResults.get(key).contains(deferredResult)); for (String cluster : publicClusters) { - String publicKey = String.format("%s-%s-%s", somePublicAppId, cluster, somePublicNamespace); + String publicKey = + Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(somePublicAppId, cluster, somePublicNamespace); assertTrue(deferredResults.get(publicKey).contains(deferredResult)); } } @@ -107,10 +110,11 @@ public class NotificationControllerTest { public void testPollNotificationWithDefaultNamespaceAndHandleMessage() throws Exception { DeferredResult> deferredResult = controller - .pollNotification(someAppId, someCluster, defaultNamespace, someDataCenter, someReleaseId, - response); + .pollNotification(someAppId, someCluster, defaultNamespace, someDataCenter, someReleaseId); - String key = String.format("%s-%s-%s", someAppId, someCluster, defaultNamespace); + String key = + Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(someAppId, someCluster, defaultNamespace); controller.handleMessage(key, Topics.APOLLO_RELEASE_TOPIC); @@ -133,10 +137,12 @@ public class NotificationControllerTest { DeferredResult> deferredResult = controller - .pollNotification(someAppId, someCluster, somePublicNamespace, someDataCenter, someReleaseId, - response); + .pollNotification(someAppId, someCluster, somePublicNamespace, someDataCenter, + someReleaseId); - String key = String.format("%s-%s-%s", somePublicAppId, someDataCenter, somePublicNamespace); + String key = + Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) + .join(somePublicAppId, someDataCenter, somePublicNamespace); controller.handleMessage(key, Topics.APOLLO_RELEASE_TOPIC); diff --git a/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/TestWebSecurityConfig.java b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/TestWebSecurityConfig.java new file mode 100644 index 0000000000000000000000000000000000000000..4f48616223cf221997b2a3c8c397b6266dd59a3c --- /dev/null +++ b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/controller/TestWebSecurityConfig.java @@ -0,0 +1,29 @@ +package com.ctrip.apollo.configservice.controller; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.annotation.Order; +import org.springframework.security.config.annotation.authentication.builders.AuthenticationManagerBuilder; +import org.springframework.security.config.annotation.web.builders.HttpSecurity; +import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter; + +@Configuration +@Order(99) +public class TestWebSecurityConfig extends WebSecurityConfigurerAdapter { + + @Override + protected void configure(HttpSecurity http) throws Exception { + http.httpBasic(); + http.csrf().disable(); + http.authorizeRequests().antMatchers("/").permitAll().and() + .authorizeRequests().antMatchers("/console/**").permitAll(); + + http.headers().frameOptions().disable(); + } + + @Autowired + public void configureGlobal(AuthenticationManagerBuilder auth) throws Exception { + auth.inMemoryAuthentication().withUser("user").password("").roles("USER"); + auth.inMemoryAuthentication().withUser("apollo").password("").roles("USER", "ADMIN"); + } +} diff --git a/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/integration/ConfigControllerIntegrationTest.java b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/integration/ConfigControllerIntegrationTest.java index b5eb5151156019ffb917e52be4fa42f91627af61..83993db7c9cfa8efe0d68f3e2bb4a31ebd1e6520 100644 --- a/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/integration/ConfigControllerIntegrationTest.java +++ b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/integration/ConfigControllerIntegrationTest.java @@ -83,7 +83,8 @@ public class ConfigControllerIntegrationTest extends AbstractBaseIntegrationTest public void testQueryConfigNotModified() throws Exception { String releaseId = String.valueOf(991); ResponseEntity response = restTemplate - .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?releaseId={releaseId}", ApolloConfig.class, + .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?releaseId={releaseId}", + ApolloConfig.class, getHostUrl(), someAppId, someCluster, someNamespace, releaseId); assertEquals(HttpStatus.NOT_MODIFIED, response.getStatusCode()); @@ -94,7 +95,8 @@ public class ConfigControllerIntegrationTest extends AbstractBaseIntegrationTest @Sql(scripts = "/integration-test/cleanup.sql", executionPhase = Sql.ExecutionPhase.AFTER_TEST_METHOD) public void testQueryPublicConfigWithDataCenterFoundAndNoOverride() throws Exception { ResponseEntity response = restTemplate - .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?dataCenter={dateCenter}", ApolloConfig.class, + .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?dataCenter={dateCenter}", + ApolloConfig.class, getHostUrl(), someAppId, someCluster, somePublicNamespace, someDC); ApolloConfig result = response.getBody(); @@ -111,11 +113,12 @@ public class ConfigControllerIntegrationTest extends AbstractBaseIntegrationTest @Sql(scripts = "/integration-test/cleanup.sql", executionPhase = Sql.ExecutionPhase.AFTER_TEST_METHOD) public void testQueryPublicConfigWithDataCenterFoundAndOverride() throws Exception { ResponseEntity response = restTemplate - .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?dataCenter={dateCenter}", ApolloConfig.class, + .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?dataCenter={dateCenter}", + ApolloConfig.class, getHostUrl(), someAppId, someDefaultCluster, somePublicNamespace, someDC); ApolloConfig result = response.getBody(); - assertEquals("994|993", result.getReleaseId()); + assertEquals("994" + ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR + "993", result.getReleaseId()); assertEquals(someAppId, result.getAppId()); assertEquals(someDefaultCluster, result.getCluster()); assertEquals(somePublicNamespace, result.getNamespace()); @@ -129,7 +132,8 @@ public class ConfigControllerIntegrationTest extends AbstractBaseIntegrationTest public void testQueryPublicConfigWithDataCenterNotFoundAndNoOverride() throws Exception { String someDCNotFound = "someDCNotFound"; ResponseEntity response = restTemplate - .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?dataCenter={dateCenter}", ApolloConfig.class, + .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?dataCenter={dateCenter}", + ApolloConfig.class, getHostUrl(), someAppId, someCluster, somePublicNamespace, someDCNotFound); ApolloConfig result = response.getBody(); @@ -147,11 +151,12 @@ public class ConfigControllerIntegrationTest extends AbstractBaseIntegrationTest public void testQueryPublicConfigWithDataCenterNotFoundAndOverride() throws Exception { String someDCNotFound = "someDCNotFound"; ResponseEntity response = restTemplate - .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?dataCenter={dateCenter}", ApolloConfig.class, + .getForEntity("{baseurl}/configs/{appId}/{clusterName}/{namespace}?dataCenter={dateCenter}", + ApolloConfig.class, getHostUrl(), someAppId, someDefaultCluster, somePublicNamespace, someDCNotFound); ApolloConfig result = response.getBody(); - assertEquals("994|992", result.getReleaseId()); + assertEquals("994" + ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR + "992", result.getReleaseId()); assertEquals(someAppId, result.getAppId()); assertEquals(someDefaultCluster, result.getCluster()); assertEquals(somePublicNamespace, result.getNamespace()); diff --git a/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/integration/NotificationControllerIntegrationTest.java b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/integration/NotificationControllerIntegrationTest.java index d950a227fc45b3261325fa7a37cc7fca13cd3a8f..30564ec3ff2509c0a1a1bd4687885fb58cb59042 100644 --- a/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/integration/NotificationControllerIntegrationTest.java +++ b/apollo-configservice/src/test/java/com/ctrip/apollo/configservice/integration/NotificationControllerIntegrationTest.java @@ -1,6 +1,9 @@ package com.ctrip.apollo.configservice.integration; -import com.ctrip.apollo.biz.message.Topics; +import com.google.common.base.Joiner; + +import com.ctrip.apollo.biz.entity.ReleaseMessage; +import com.ctrip.apollo.biz.repository.ReleaseMessageRepository; import com.ctrip.apollo.configservice.controller.NotificationController; import com.ctrip.apollo.core.ConfigConsts; import com.ctrip.apollo.core.dto.ApolloConfigNotification; @@ -14,7 +17,6 @@ import org.springframework.test.context.jdbc.Sql; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -26,6 +28,8 @@ import static org.junit.Assert.assertEquals; public class NotificationControllerIntegrationTest extends AbstractBaseIntegrationTest { @Autowired private NotificationController notificationController; + @Autowired + private ReleaseMessageRepository releaseMessageRepository; private String someAppId; private String someCluster; private String defaultNamespace; @@ -43,20 +47,16 @@ public class NotificationControllerIntegrationTest extends AbstractBaseIntegrati @Test public void testPollNotificationWithDefaultNamespace() throws Exception { - Future> future = - executorService.submit(() -> restTemplate - .getForEntity( - "{baseurl}/notifications?appId={appId}&cluster={clusterName}&namespace={namespace}", - ApolloConfigNotification.class, - getHostUrl(), someAppId, someCluster, defaultNamespace)); + AtomicBoolean stop = new AtomicBoolean(); + perodicSendMessage(assembleKey(someAppId, someCluster, defaultNamespace), stop); - //wait for the request connected to server - TimeUnit.MILLISECONDS.sleep(500); + ResponseEntity result = restTemplate.getForEntity( + "{baseurl}/notifications?appId={appId}&cluster={clusterName}&namespace={namespace}", + ApolloConfigNotification.class, + getHostUrl(), someAppId, someCluster, defaultNamespace); - notificationController.handleMessage(assembleKey(someAppId, someCluster, defaultNamespace), - Topics.APOLLO_RELEASE_TOPIC); + stop.set(true); - ResponseEntity result = future.get(500, TimeUnit.MILLISECONDS); ApolloConfigNotification notification = result.getBody(); assertEquals(HttpStatus.OK, result.getStatusCode()); assertEquals(defaultNamespace, notification.getNamespace()); @@ -69,19 +69,7 @@ public class NotificationControllerIntegrationTest extends AbstractBaseIntegrati String publicAppId = "somePublicAppId"; AtomicBoolean stop = new AtomicBoolean(); - executorService.submit((Runnable) () -> { - //wait for the request connected to server - while (!stop.get() && !Thread.currentThread().isInterrupted()) { - try { - TimeUnit.MILLISECONDS.sleep(100); - } catch (InterruptedException e) { - } - - notificationController.handleMessage( - assembleKey(publicAppId, ConfigConsts.CLUSTER_NAME_DEFAULT, somePublicNamespace), - Topics.APOLLO_RELEASE_TOPIC); - } - }); + perodicSendMessage(assembleKey(publicAppId, ConfigConsts.CLUSTER_NAME_DEFAULT, somePublicNamespace), stop); ResponseEntity result = restTemplate .getForEntity( @@ -104,19 +92,7 @@ public class NotificationControllerIntegrationTest extends AbstractBaseIntegrati String someDC = "someDC"; AtomicBoolean stop = new AtomicBoolean(); - executorService.submit((Runnable) () -> { - //wait for the request connected to server - while (!stop.get() && !Thread.currentThread().isInterrupted()) { - try { - TimeUnit.MILLISECONDS.sleep(100); - } catch (InterruptedException e) { - } - - notificationController.handleMessage( - assembleKey(publicAppId, someDC, somePublicNamespace), - Topics.APOLLO_RELEASE_TOPIC); - } - }); + perodicSendMessage(assembleKey(publicAppId, someDC, somePublicNamespace), stop); ResponseEntity result = restTemplate .getForEntity( @@ -131,8 +107,22 @@ public class NotificationControllerIntegrationTest extends AbstractBaseIntegrati assertEquals(somePublicNamespace, notification.getNamespace()); } - private String assembleKey(String appId, String cluster, String namespace) { - return String.format("%s-%s-%s", appId, cluster, namespace); + return Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR).join(appId, cluster, namespace); + } + + private void perodicSendMessage(String message, AtomicBoolean stop) { + executorService.submit((Runnable) () -> { + //wait for the request connected to server + while (!stop.get() && !Thread.currentThread().isInterrupted()) { + try { + TimeUnit.MILLISECONDS.sleep(100); + } catch (InterruptedException e) { + } + + ReleaseMessage releaseMessage = new ReleaseMessage(message); + releaseMessageRepository.save(releaseMessage); + } + }); } } diff --git a/apollo-configservice/src/test/resources/application.properties b/apollo-configservice/src/test/resources/application.properties index 30f9a1a8f2b37d59b47e7e1909ad46dcc7e6839f..3fad69d2ebc8a9729d988090233232b1c63a9d5e 100644 --- a/apollo-configservice/src/test/resources/application.properties +++ b/apollo-configservice/src/test/resources/application.properties @@ -3,3 +3,6 @@ spring.jpa.hibernate.naming_strategy=org.hibernate.cfg.EJB3NamingStrategy spring.h2.console.enabled = true spring.h2.console.settings.web-allow-others=true spring.jpa.properties.hibernate.show_sql=true + +# for ReleaseMessageScanner test +apollo.message-scan.interval=100 diff --git a/apollo-core/src/main/java/com/ctrip/apollo/core/ConfigConsts.java b/apollo-core/src/main/java/com/ctrip/apollo/core/ConfigConsts.java index 1813ae2d8b2ceb6aa2c925efe4cdc6616b435d5a..2c0c5aabc2eda2bd13cd11342785ecdd4617be7b 100644 --- a/apollo-core/src/main/java/com/ctrip/apollo/core/ConfigConsts.java +++ b/apollo-core/src/main/java/com/ctrip/apollo/core/ConfigConsts.java @@ -3,4 +3,5 @@ package com.ctrip.apollo.core; public interface ConfigConsts { String NAMESPACE_DEFAULT = "application"; String CLUSTER_NAME_DEFAULT = "default"; + String CLUSTER_NAMESPACE_SEPARATOR = "+"; }