Set replication client filters in RefreshablePeerEurekaNodes

spring-cloud-netflix-eureka-server/pom.xml

@@ -61,6 +61,10 @@
 			<groupId>com.sun.jersey</groupId>
 			<artifactId>jersey-server</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.sun.jersey</groupId>
+			<artifactId>jersey-client</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>com.netflix.eureka</groupId>
 			<artifactId>eureka-core</artifactId>

spring-cloud-netflix-eureka-server/src/main/java/org/springframework/cloud/netflix/eureka/server/EurekaServerAutoConfiguration.java

@@ -36,10 +36,12 @@
 import com.netflix.eureka.DefaultEurekaServerContext;
 import com.netflix.eureka.EurekaServerConfig;
 import com.netflix.eureka.EurekaServerContext;
+import com.netflix.eureka.cluster.PeerEurekaNode;
 import com.netflix.eureka.cluster.PeerEurekaNodes;
 import com.netflix.eureka.registry.PeerAwareInstanceRegistry;
 import com.netflix.eureka.resources.DefaultServerCodecs;
 import com.netflix.eureka.resources.ServerCodecs;
+import com.netflix.eureka.transport.JerseyReplicationClient;
 import com.sun.jersey.api.core.DefaultResourceConfig;
 import com.sun.jersey.spi.container.servlet.ServletContainer;
 
@@ -139,6 +141,12 @@ private static CodecWrapper getFullXml(EurekaServerConfig serverConfig) {
 				: codec;
 	}
 
+	@Bean
+	@ConditionalOnMissingBean
+	public ReplicationClientAdditionalFilters replicationClientAdditionalFilters() {
+		return new ReplicationClientAdditionalFilters(Collections.emptySet());
+	}
+
 	@Bean
 	public PeerAwareInstanceRegistry peerAwareInstanceRegistry(
 			ServerCodecs serverCodecs) {
@@ -152,9 +160,11 @@ public PeerAwareInstanceRegistry peerAwareInstanceRegistry(
 	@Bean
 	@ConditionalOnMissingBean
 	public PeerEurekaNodes peerEurekaNodes(PeerAwareInstanceRegistry registry,
-			ServerCodecs serverCodecs) {
+			ServerCodecs serverCodecs,
+			ReplicationClientAdditionalFilters replicationClientAdditionalFilters) {
 		return new RefreshablePeerEurekaNodes(registry, this.eurekaServerConfig,
-				this.eurekaClientConfig, serverCodecs, this.applicationInfoManager);
+				this.eurekaClientConfig, serverCodecs, this.applicationInfoManager,
+				replicationClientAdditionalFilters);
 	}
 
 	@Bean
@@ -273,12 +283,33 @@ public EurekaServerConfig eurekaServerConfig(EurekaClientConfig clientConfig) {
 	static class RefreshablePeerEurekaNodes extends PeerEurekaNodes
 			implements ApplicationListener<EnvironmentChangeEvent> {
 
+		private ReplicationClientAdditionalFilters replicationClientAdditionalFilters;
+
 		RefreshablePeerEurekaNodes(final PeerAwareInstanceRegistry registry,
 				final EurekaServerConfig serverConfig,
 				final EurekaClientConfig clientConfig, final ServerCodecs serverCodecs,
-				final ApplicationInfoManager applicationInfoManager) {
+				final ApplicationInfoManager applicationInfoManager,
+				final ReplicationClientAdditionalFilters replicationClientAdditionalFilters) {
 			super(registry, serverConfig, clientConfig, serverCodecs,
 					applicationInfoManager);
+			this.replicationClientAdditionalFilters = replicationClientAdditionalFilters;
+		}
+
+		@Override
+		protected PeerEurekaNode createPeerEurekaNode(String peerEurekaNodeUrl) {
+			JerseyReplicationClient replicationClient = JerseyReplicationClient
+					.createReplicationClient(serverConfig, serverCodecs,
+							peerEurekaNodeUrl);
+
+			this.replicationClientAdditionalFilters.getFilters()
+					.forEach(replicationClient::addReplicationClientFilter);
+
+			String targetHost = hostFromUrl(peerEurekaNodeUrl);
+			if (targetHost == null) {
+				targetHost = "host";
+			}
+			return new PeerEurekaNode(registry, targetHost, peerEurekaNodeUrl,
+					replicationClient, serverConfig);
 		}
 
 		@Override

spring-cloud-netflix-eureka-server/src/main/java/org/springframework/cloud/netflix/eureka/server/ReplicationClientAdditionalFilters.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2013-2019 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.netflix.eureka.server;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+
+import com.sun.jersey.api.client.filter.ClientFilter;
+
+/**
+ * @author Yuxin Bai
+ */
+public class ReplicationClientAdditionalFilters {
+
+	private Collection<ClientFilter> filters;
+
+	public ReplicationClientAdditionalFilters(Collection<ClientFilter> filters) {
+		this.filters = new LinkedHashSet<>(filters);
+	}
+
+	public Collection<ClientFilter> getFilters() {
+		return this.filters;
+	}
+
+}

spring-cloud-netflix-eureka-server/src/test/java/org/springframework/cloud/netflix/eureka/server/RefreshablePeerEurekaNodesTests.java

@@ -159,7 +159,8 @@ class VerifyablePeerEurekNode extends RefreshablePeerEurekaNodes {
 					ServerCodecs serverCodecs,
 					ApplicationInfoManager applicationInfoManager) {
 				super(registry, serverConfig, clientConfig, serverCodecs,
-						applicationInfoManager);
+						applicationInfoManager,
+						new ReplicationClientAdditionalFilters(Collections.emptySet()));
 			}
 
 			protected void updatePeerEurekaNodes(List<String> newPeerUrls) {

spring-cloud-netflix-eureka-server/src/test/java/org/springframework/cloud/netflix/eureka/server/RefreshablePeerEurekaNodesWithCustomFiltersTests.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2013-2019 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.netflix.eureka.server;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+
+import com.netflix.eureka.cluster.PeerEurekaNodes;
+import com.sun.jersey.api.client.ClientHandlerException;
+import com.sun.jersey.api.client.ClientRequest;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.filter.ClientFilter;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.netflix.eureka.server.EurekaServerAutoConfiguration.RefreshablePeerEurekaNodes;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.ReflectionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Yuxin Bai
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(
+		classes = RefreshablePeerEurekaNodesWithCustomFiltersTests.Application.class,
+		webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+		value = { "spring.application.name=eureka", "server.contextPath=/context",
+				"management.security.enabled=false" })
+public class RefreshablePeerEurekaNodesWithCustomFiltersTests {
+
+	@Autowired
+	private PeerEurekaNodes peerEurekaNodes;
+
+	@Test
+	public void testCustomPeerNodesShouldTakePrecedenceOverDefault() {
+		assertThat(peerEurekaNodes instanceof RefreshablePeerEurekaNodes)
+				.as("PeerEurekaNodes should be an instance of RefreshablePeerEurekaNodes")
+				.isTrue();
+
+		ReplicationClientAdditionalFilters filters = getField(
+				RefreshablePeerEurekaNodes.class,
+				(RefreshablePeerEurekaNodes) peerEurekaNodes,
+				"replicationClientAdditionalFilters");
+		assertThat(filters.getFilters()).as(
+				"PeerEurekaNodes'should have only one filter set on replicationClientAdditionalFilters")
+				.hasSize(1);
+		assertThat(filters.getFilters().iterator()
+				.next() instanceof Application.CustomClientFilter).as(
+						"The type of the filter should be CustomClientFilter as user declared so")
+						.isTrue();
+	}
+
+	private static <T, R> R getField(Class<T> clazz, T target, String fieldName) {
+		Field field = ReflectionUtils.findField(clazz, fieldName);
+		ReflectionUtils.makeAccessible(field);
+		@SuppressWarnings("unchecked")
+		R value = (R) ReflectionUtils.getField(field, target);
+		return value;
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	@EnableEurekaServer
+	protected static class Application {
+
+		@Bean
+		public ReplicationClientAdditionalFilters customFilters() {
+			return new ReplicationClientAdditionalFilters(
+					Collections.singletonList(new CustomClientFilter()));
+		}
+
+		protected class CustomClientFilter extends ClientFilter {
+
+			@Override
+			public ClientResponse handle(ClientRequest cr) throws ClientHandlerException {
+				return getNext().handle(cr);
+			}
+
+		}
+
+	}
+
+}