Skip to content

R
ribbon

thythm / ribbon
与 Fork 源项目一致

从无法访问的项目Fork

通知 3
Star 0
Fork 0
R
ribbon

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 5c02176e 编写于 作者: T Tomasz Bak
浏览文件
操作

Merge branch '2.x-commonclient' into dynamic proxy prototype branch.

上级 6a47039b 559dd001
隐藏空白更改
内联 并排
Showing with 266 addition and 98 deletion
ribbon-client-extensions/src/main/java/com/netflix/ribbonclientextensions/http/HttpMetaRequest.java
浏览文件 @ 5c02176e
......@@ -11,7 +11,6 @@ import rx.Observer;
import rx.Observable.OnSubscribe;
import rx.Subscriber;
import rx.subjects.PublishSubject;
import rx.subjects.ReplaySubject;
import rx.subjects.Subject;
import com.netflix.hystrix.HystrixExecutableInfo;
......@@ -43,22 +42,22 @@ class HttpMetaRequest<T> implements RequestWithMetaData<T> {
}
}
private HttpRequestBuilder<T> requestBuilder;
private final HttpRequest<T> request;
HttpMetaRequest(HttpRequestBuilder<T> requestBuilder ) {
this.requestBuilder = requestBuilder;
HttpMetaRequest(HttpRequest<T> request) {
this.request = request;
}
@Override
public Observable<RibbonResponse<Observable<T>>> observe() {
RibbonHystrixObservableCommand<T> hystrixCommand = requestBuilder.createHystrixCommand();
RibbonHystrixObservableCommand<T> hystrixCommand = request.createHystrixCommand();
final Observable<T> output = hystrixCommand.observe();
return convertToRibbonResponse(output, hystrixCommand);
}
@Override
public Observable<RibbonResponse<Observable<T>>> toObservable() {
RibbonHystrixObservableCommand<T> hystrixCommand = requestBuilder.createHystrixCommand();
RibbonHystrixObservableCommand<T> hystrixCommand = request.createHystrixCommand();
final Observable<T> output = hystrixCommand.observe();
return convertToRibbonResponse(output, hystrixCommand);
}
......@@ -102,7 +101,7 @@ class HttpMetaRequest<T> implements RequestWithMetaData<T> {
@Override
public Future<RibbonResponse<T>> queue() {
final RibbonHystrixObservableCommand<T> hystrixCommand = requestBuilder.createHystrixCommand();
final RibbonHystrixObservableCommand<T> hystrixCommand = request.createHystrixCommand();
final Future<T> f = hystrixCommand.queue();
return new Future<RibbonResponse<T>>() {
@Override
......@@ -139,7 +138,7 @@ class HttpMetaRequest<T> implements RequestWithMetaData<T> {
@Override
public RibbonResponse<T> execute() {
RibbonHystrixObservableCommand<T> hystrixCommand = requestBuilder.createHystrixCommand();
RibbonHystrixObservableCommand<T> hystrixCommand = request.createHystrixCommand();
T obj = hystrixCommand.execute();
return new HttpMetaResponse<T>(obj, hystrixCommand);
}
......
ribbon-client-extensions/src/main/java/com/netflix/ribbonclientextensions/http/HttpRequest.java
浏览文件 @ 5c02176e
package com.netflix.ribbonclientextensions.http;
import io.netty.buffer.ByteBuf;
import io.reactivex.netty.protocol.http.client.HttpClient;
import io.reactivex.netty.protocol.http.client.HttpClientRequest;
import io.reactivex.netty.protocol.http.client.HttpClientResponse;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
import rx.Observable;
......@@ -7,41 +16,90 @@ import rx.Observable.OnSubscribe;
import rx.Subscriber;
import com.netflix.hystrix.HystrixExecutableInfo;
import com.netflix.hystrix.HystrixObservableCommand;
import com.netflix.ribbonclientextensions.CacheProvider;
import com.netflix.ribbonclientextensions.ResponseValidator;
import com.netflix.ribbonclientextensions.RibbonRequest;
import com.netflix.ribbonclientextensions.RequestWithMetaData;
import com.netflix.ribbonclientextensions.RibbonResponse;
import com.netflix.ribbonclientextensions.http.HttpRequestTemplate.CacheProviderWithKeyTemplate;
import com.netflix.ribbonclientextensions.hystrix.FallbackHandler;
import com.netflix.ribbonclientextensions.template.TemplateParser;
import com.netflix.ribbonclientextensions.template.TemplateParsingException;
class HttpRequest<T> implements RibbonRequest<T> {
static class CacheProviderWithKey<T> {
CacheProvider<T> cacheProvider;
String key;
public CacheProviderWithKey(CacheProvider<T> cacheProvider, String key) {
super();
this.cacheProvider = cacheProvider;
this.key = key;
}
public final CacheProvider<T> getCacheProvider() {
return cacheProvider;
}
public final String getKey() {
return key;
}
}
private final HttpClientRequest<ByteBuf> httpRequest;
private final String hystrixCacheKey;
private final List<CacheProviderWithKey<T>> cacheProviders;
private final Map<String, Object> requestProperties;
private final HttpClient<ByteBuf, ByteBuf> client;
private final HttpRequestTemplate<T> template;
private HttpRequestBuilder<T> requestBuilder;
HttpRequest(HttpRequestBuilder<T> requestBuilder) throws TemplateParsingException {
this.client = requestBuilder.template().getClient();
this.httpRequest = requestBuilder.createClientRequest();
this.hystrixCacheKey = requestBuilder.hystrixCacheKey();
this.requestProperties = new HashMap<String, Object>(requestBuilder.requestProperties());
this.cacheProviders = new LinkedList<CacheProviderWithKey<T>>();
this.template = requestBuilder.template();
addCacheProviders(requestBuilder.cacheProviders());
}
HttpRequest(HttpRequestBuilder<T> requestBuilder) {
this.requestBuilder = requestBuilder;
private void addCacheProviders(List<CacheProviderWithKeyTemplate<T>> providers) throws TemplateParsingException {
if (providers != null && providers.size() > 0) {
for (CacheProviderWithKeyTemplate<T> cacheProviderWithTemplate: providers) {
CacheProvider<T> provider = cacheProviderWithTemplate.getProvider();
String key = TemplateParser.toData(this.requestProperties, cacheProviderWithTemplate.getKeyTemplate());
cacheProviders.add(new CacheProviderWithKey<T>(provider, key));
}
}
}
RibbonHystrixObservableCommand<T> createHystrixCommand() {
return new RibbonHystrixObservableCommand<T>(client, httpRequest, hystrixCacheKey, cacheProviders, requestProperties, template.fallbackHandler(),
template.responseValidator(), template.getClassType(), template.hystrixProperties());
}
@Override
public T execute() {
return requestBuilder.createHystrixCommand().execute();
return createHystrixCommand().execute();
}
@Override
public Future<T> queue() {
return requestBuilder.createHystrixCommand().queue();
return createHystrixCommand().queue();
}
@Override
public Observable<T> observe() {
return requestBuilder.createHystrixCommand().observe();
return createHystrixCommand().observe();
}
@Override
public Observable<T> toObservable() {
return requestBuilder.createHystrixCommand().toObservable();
return createHystrixCommand().toObservable();
}
@Override
public RequestWithMetaData<T> withMetadata() {
return new HttpMetaRequest<T>(requestBuilder);
return new HttpMetaRequest<T>(this);
}
......
ribbon-client-extensions/src/main/java/com/netflix/ribbonclientextensions/http/HttpRequestBuilder.java
浏览文件 @ 5c02176e
package com.netflix.ribbonclientextensions.http;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import io.netty.buffer.ByteBuf;
import io.reactivex.netty.protocol.http.client.ContentSource;
import io.reactivex.netty.protocol.http.client.HttpClient;
import io.reactivex.netty.protocol.http.client.HttpClientRequest;
import io.reactivex.netty.protocol.http.client.RawContentSource;
import io.reactivex.netty.protocol.http.client.RepeatableContentHttpRequest;
import com.netflix.hystrix.HystrixObservableCommand;
import com.netflix.hystrix.exception.HystrixBadRequestException;
import com.netflix.ribbonclientextensions.RequestTemplate.RequestBuilder;
import com.netflix.ribbonclientextensions.RibbonRequest;
import com.netflix.ribbonclientextensions.http.HttpRequestTemplate.CacheProviderWithKeyTemplate;
import com.netflix.ribbonclientextensions.template.ParsedTemplate;
import com.netflix.ribbonclientextensions.template.TemplateParser;
import com.netflix.ribbonclientextensions.template.TemplateParsingException;
public class HttpRequestBuilder<T> extends RequestBuilder<T> {
private HttpRequestTemplate<T> requestTemplate;
private HttpClient<ByteBuf, ByteBuf> client;
private HystrixObservableCommand.Setter setter;
private Map<String, Object> vars;
private ParsedTemplate parsedUriTemplate;
private final HttpRequestTemplate<T> requestTemplate;
private final Map<String, Object> vars;
private final ParsedTemplate parsedUriTemplate;
private RawContentSource<?> rawContentSource;
HttpRequestBuilder(HttpClient<ByteBuf, ByteBuf> client, HttpRequestTemplate<T> requestTemplate, HystrixObservableCommand.Setter setter) {
HttpRequestBuilder(HttpRequestTemplate<T> requestTemplate) {
this.requestTemplate = requestTemplate;
this.client = client;
this.setter = setter;
this.parsedUriTemplate = requestTemplate.uriTemplate();
vars = new ConcurrentHashMap<String, Object>();
}
RibbonHystrixObservableCommand<T> createHystrixCommand() {
return new RibbonHystrixObservableCommand<T>(client, requestTemplate, this, setter);
}
@Override
public HttpRequestBuilder<T> withRequestProperty(
String key, Object value) {
......@@ -52,7 +46,11 @@ public class HttpRequestBuilder<T> extends RequestBuilder<T> {
@Override
public RibbonRequest<T> build() {
return new HttpRequest<T>(this);
try {
return new HttpRequest<T>(this);
} catch (TemplateParsingException e) {
throw new IllegalArgumentException(e);
}
}
HttpClientRequest<ByteBuf> createClientRequest() {
......@@ -63,20 +61,31 @@ public class HttpRequestBuilder<T> extends RequestBuilder<T> {
throw new HystrixBadRequestException("Problem parsing the URI template", e);
}
HttpClientRequest<ByteBuf> request = HttpClientRequest.create(requestTemplate.method(), uri);
for (Map.Entry<String, String> entry: requestTemplate.getHeaders().entries()) {
request.withHeader(entry.getKey(), entry.getValue());
}
if (rawContentSource != null) {
request.withRawContentSource(rawContentSource);
}
return request;
return new RepeatableContentHttpRequest<ByteBuf>(request);
}
String cacheKey() throws TemplateParsingException {
String hystrixCacheKey() throws TemplateParsingException {
if (requestTemplate.hystrixCacheKeyTemplate() == null) {
return null;
}
return TemplateParser.toData(vars, requestTemplate.hystrixCacheKeyTemplate());
}
Map<String, Object> requestProperties() {
return vars;
}
List<CacheProviderWithKeyTemplate<T>> cacheProviders() {
return requestTemplate.cacheProviders();
}
HttpRequestTemplate<T> template() {
return requestTemplate;
}
}
ribbon-client-extensions/src/main/java/com/netflix/ribbonclientextensions/http/HttpRequestTemplate.java
浏览文件 @ 5c02176e
package com.netflix.ribbonclientextensions.http;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.reactivex.netty.protocol.http.client.ContentSource;
import io.reactivex.netty.protocol.http.client.HttpClient;
import io.reactivex.netty.protocol.http.client.HttpClientResponse;
import io.reactivex.netty.protocol.http.client.RawContentSource;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import com.google.common.collect.Maps;
import com.netflix.client.netty.LoadBalancingRxClient;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandKey;
......@@ -34,14 +32,16 @@ public class HttpRequestTemplate<T> implements RequestTemplate<T, HttpClientResp
private HystrixObservableCommand.Setter setter;
private FallbackHandler<T> fallbackHandler;
private ParsedTemplate parsedUriTemplate;
private ResponseValidator<HttpClientResponse<ByteBuf>> transformer;
private ResponseValidator<HttpClientResponse<ByteBuf>> validator;
private HttpMethod method;
private String name;
private List<CacheProviderWithKeyTemplate<T>> cacheProviders;
private final String name;
private final List<CacheProviderWithKeyTemplate<T>> cacheProviders;
private ParsedTemplate hystrixCacheKeyTemplate;
private Map<String, ParsedTemplate> parsedTemplates;
private Class<? extends T> classType;
private int concurrentRequestLimit;
private final Class<? extends T> classType;
private final int concurrentRequestLimit;
private final HttpHeaders headers;
private final HttpResourceGroup group;
static class CacheProviderWithKeyTemplate<T> {
private ParsedTemplate keyTemplate;
......@@ -74,8 +74,10 @@ public class HttpRequestTemplate<T> implements RequestTemplate<T, HttpClientResp
concurrentRequestLimit = -1;
}
this.name = name;
// default method to GET
this.group = group;
method = HttpMethod.GET;
headers = new DefaultHttpHeaders();
headers.add(group.getHeaders());
cacheProviders = new LinkedList<CacheProviderWithKeyTemplate<T>>();
parsedTemplates = new HashMap<String, ParsedTemplate>();
}
......@@ -88,7 +90,6 @@ public class HttpRequestTemplate<T> implements RequestTemplate<T, HttpClientResp
@Override
public HttpRequestBuilder<T> requestBuilder() {
// TODO: apply hystrix properties passed in to the template
if (setter == null) {
setter = HystrixObservableCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(clientName))
.andCommandKey(HystrixCommandKey.Factory.asKey(name()));
......@@ -101,7 +102,7 @@ public class HttpRequestTemplate<T> implements RequestTemplate<T, HttpClientResp
HystrixCommandProperties.Setter().withExecutionIsolationSemaphoreMaxConcurrentRequests(concurrentRequestLimit));
}
}
return new HttpRequestBuilder<T>(client, this, setter);
return new HttpRequestBuilder<T>(this);
}
public HttpRequestTemplate<T> withMethod(String method) {
......@@ -124,6 +125,7 @@ public class HttpRequestTemplate<T> implements RequestTemplate<T, HttpClientResp
}
public HttpRequestTemplate<T> withHeader(String name, String value) {
headers.add(name, value);
return this;
}
......@@ -151,7 +153,7 @@ public class HttpRequestTemplate<T> implements RequestTemplate<T, HttpClientResp
}
ResponseValidator<HttpClientResponse<ByteBuf>> responseValidator() {
return transformer;
return validator;
}
FallbackHandler<T> fallbackHandler() {
......@@ -170,6 +172,10 @@ public class HttpRequestTemplate<T> implements RequestTemplate<T, HttpClientResp
return this.classType;
}
HttpHeaders getHeaders() {
return this.headers;
}
@Override
public String name() {
return name;
......@@ -178,14 +184,23 @@ public class HttpRequestTemplate<T> implements RequestTemplate<T, HttpClientResp
@Override
public HttpRequestTemplate<T> withResponseValidator(
ResponseValidator<HttpClientResponse<ByteBuf>> validator) {
this.transformer = validator;
this.validator = validator;
return this;
}
@Override
public HttpRequestTemplate<T> copy(String name) {
// TODO Auto-generated method stub
return null;
HttpRequestTemplate<T> newTemplate = new HttpRequestTemplate<T>(name, this.group, this.client, this.classType);
newTemplate.cacheProviders.addAll(this.cacheProviders);
newTemplate.method = this.method;
newTemplate.headers.add(this.headers);
newTemplate.parsedTemplates.putAll(this.parsedTemplates);
newTemplate.parsedUriTemplate = this.parsedUriTemplate;
newTemplate.setter = setter;
newTemplate.fallbackHandler = this.fallbackHandler;
newTemplate.validator = this.validator;
newTemplate.hystrixCacheKeyTemplate = this.hystrixCacheKeyTemplate;
return newTemplate;
}
@Override
......@@ -195,6 +210,12 @@ public class HttpRequestTemplate<T> implements RequestTemplate<T, HttpClientResp
return this;
}
Setter hystrixProperties() {
return this.setter;
}
HttpClient<ByteBuf, ByteBuf> getClient() {
return this.client;
}
}
ribbon-client-extensions/src/main/java/com/netflix/ribbonclientextensions/http/HttpResourceGroup.java
浏览文件 @ 5c02176e
package com.netflix.ribbonclientextensions.http;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpHeaders;
import io.reactivex.netty.protocol.http.client.HttpClient;
import com.netflix.client.config.ClientConfigBuilder;
import com.netflix.client.config.CommonClientConfigKey;
import com.netflix.client.config.DefaultClientConfigImpl;
import com.netflix.client.config.IClientConfig;
import com.netflix.client.config.IClientConfigKey;
import com.netflix.client.netty.RibbonTransport;
import com.netflix.loadbalancer.ILoadBalancer;
import com.netflix.ribbonclientextensions.ClientOptions;
import com.netflix.ribbonclientextensions.RequestTemplate;
import com.netflix.ribbonclientextensions.ResourceGroup;
public class HttpResourceGroup extends ResourceGroup<HttpRequestTemplate<?>> {
private final HttpClient<ByteBuf, ByteBuf> client;
private final HttpHeaders headers;
public HttpResourceGroup(String groupName) {
this(groupName, null);
......@@ -24,6 +22,7 @@ public class HttpResourceGroup extends ResourceGroup<HttpRequestTemplate<?>> {
public HttpResourceGroup(String groupName, ClientOptions options) {
super(groupName, options);
client = RibbonTransport.newHttpClient(getClientConfig());
headers = new DefaultHttpHeaders();
}
protected IClientConfig loadDefaultConfig(String groupName) {
......@@ -31,6 +30,7 @@ public class HttpResourceGroup extends ResourceGroup<HttpRequestTemplate<?>> {
}
public HttpResourceGroup withCommonHeader(String name, String value) {
headers.add(name, value);
return this;
}
......@@ -44,4 +44,8 @@ public class HttpResourceGroup extends ResourceGroup<HttpRequestTemplate<?>> {
return newRequestTemplate(name, ByteBuf.class);
}
HttpHeaders getHeaders() {
return headers;
}
}
ribbon-client-extensions/src/main/java/com/netflix/ribbonclientextensions/http/RibbonHystrixObservableCommand.java
浏览文件 @ 5c02176e
......@@ -6,15 +6,20 @@ import io.reactivex.netty.protocol.http.client.HttpClientRequest;
import io.reactivex.netty.protocol.http.client.HttpClientResponse;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import rx.Observable;
import rx.functions.Func1;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixObservableCommand;
import com.netflix.hystrix.exception.HystrixBadRequestException;
import com.netflix.ribbonclientextensions.CacheProvider;
import com.netflix.ribbonclientextensions.ResponseValidator;
import com.netflix.ribbonclientextensions.ServerError;
import com.netflix.ribbonclientextensions.UnsuccessfulResponseException;
import com.netflix.ribbonclientextensions.http.HttpRequest.CacheProviderWithKey;
import com.netflix.ribbonclientextensions.http.HttpRequestTemplate.CacheProviderWithKeyTemplate;
import com.netflix.ribbonclientextensions.hystrix.FallbackHandler;
import com.netflix.ribbonclientextensions.template.TemplateParser;
......@@ -22,70 +27,70 @@ import com.netflix.ribbonclientextensions.template.TemplateParsingException;
class RibbonHystrixObservableCommand<T> extends HystrixObservableCommand<T> {
private HttpClient<ByteBuf, ByteBuf> httpClient;
private HttpRequestTemplate<T> requestTemplate;
private HttpRequestBuilder<T> requestBuilder;
private final HttpClient<ByteBuf, ByteBuf> httpClient;
private final HttpClientRequest<ByteBuf> httpRequest;
private final String hystrixCacheKey;
private final List<CacheProviderWithKey<T>> cacheProviders;
private final Map<String, Object> requestProperties;
private final FallbackHandler<T> fallbackHandler;
private final Class<? extends T> classType;
private final ResponseValidator<HttpClientResponse<ByteBuf>> validator;
RibbonHystrixObservableCommand(
HttpClient<ByteBuf, ByteBuf> httpClient,
HttpRequestTemplate<T> requestTemplate,
HttpRequestBuilder<T> requestBuilder, RibbonHystrixObservableCommand.Setter setter) {
RibbonHystrixObservableCommand(HttpClient<ByteBuf, ByteBuf> httpClient,
HttpClientRequest<ByteBuf> httpRequest, String hystrixCacheKey,
List<CacheProviderWithKey<T>> cacheProviders,
Map<String, Object> requestProperties,
FallbackHandler<T> fallbackHandler,
ResponseValidator<HttpClientResponse<ByteBuf>> validator,
Class<? extends T> classType,
HystrixObservableCommand.Setter setter) {
super(setter);
this.httpClient = httpClient;
this.requestTemplate = requestTemplate;
this.requestBuilder = requestBuilder;
this.fallbackHandler = fallbackHandler;
this.validator = validator;
this.httpRequest = httpRequest;
this.hystrixCacheKey = hystrixCacheKey;
this.cacheProviders = cacheProviders;
this.classType = classType;
this.requestProperties = requestProperties;
}
@Override
protected String getCacheKey() {
try {
String key = requestBuilder.cacheKey();
if (key == null) {
return super.getCacheKey();
} else {
return key;
}
} catch (TemplateParsingException e) {
return super.getCacheKey();
if (hystrixCacheKey == null) {
return super.getCacheKey();
} else {
return hystrixCacheKey;
}
}
@Override
protected Observable<T> getFallback() {
FallbackHandler<T> handler = requestTemplate.fallbackHandler();
if (handler == null) {
if (fallbackHandler == null) {
return super.getFallback();
} else {
return handler.getFallback(this, requestBuilder.requestProperties());
return fallbackHandler.getFallback(this, this.requestProperties);
}
}
@Override
protected Observable<T> run() {
final Iterator<CacheProviderWithKeyTemplate<T>> cacheProviders = requestTemplate.cacheProviders().iterator();
Observable<T> cached = null;
if (cacheProviders.hasNext()) {
CacheProviderWithKeyTemplate<T> provider = cacheProviders.next();
String cacheKey;
try {
cacheKey = TemplateParser.toData(requestBuilder.requestProperties(), provider.getKeyTemplate());
} catch (TemplateParsingException e) {
return Observable.error(e);
}
cached = provider.getProvider().get(cacheKey, requestBuilder.requestProperties());
while (cacheProviders.hasNext()) {
final Observable<T> nextCache = cacheProviders.next().getProvider().get(cacheKey, requestBuilder.requestProperties());
cached = cached.onErrorResumeNext(nextCache);
for (CacheProviderWithKey<T> provider: cacheProviders) {
Observable<T> fromTheProvider = provider.getCacheProvider().get(provider.getKey(), this.requestProperties);
if (cached == null) {
cached = fromTheProvider;
} else {
cached = cached.onErrorResumeNext(fromTheProvider);
}
}
HttpClientRequest<ByteBuf> request = requestBuilder.createClientRequest();
Observable<HttpClientResponse<ByteBuf>> httpResponseObservable = httpClient.submit(request);
if (requestTemplate.responseValidator() != null) {
Observable<HttpClientResponse<ByteBuf>> httpResponseObservable = httpClient.submit(httpRequest);
if (this.validator != null) {
httpResponseObservable = httpResponseObservable.map(new Func1<HttpClientResponse<ByteBuf>, HttpClientResponse<ByteBuf>>(){
@Override
public HttpClientResponse<ByteBuf> call(HttpClientResponse<ByteBuf> t1) {
try {
requestTemplate.responseValidator().validate(t1);
validator.validate(t1);
} catch (UnsuccessfulResponseException e) {
throw new HystrixBadRequestException("Unsuccessful response", e);
} catch (ServerError e) {
......@@ -101,7 +106,7 @@ class RibbonHystrixObservableCommand<T> extends HystrixObservableCommand<T> {
return t1.getContent().map(new Func1<ByteBuf, T>(){
@Override
public T call(ByteBuf t1) {
return requestTemplate.getClassType().cast(t1);
return classType.cast(t1);
}
});
......
ribbon-client-extensions/src/test/java/com/netflix/ribbonclientextensions/RibbonTest.java
浏览文件 @ 5c02176e
......@@ -39,8 +39,11 @@ public class RibbonTest {
// LogManager.getRootLogger().setLevel((Level)Level.DEBUG);
MockWebServer server = new MockWebServer();
String content = "Hello world";
server.enqueue(new MockResponse().setResponseCode(200).setHeader("Content-type", "text/plain")
.setBody(content));
MockResponse response = new MockResponse().setResponseCode(200).setHeader("Content-type", "text/plain")
.setBody(content);
server.enqueue(response);
server.enqueue(response);
server.play();
HttpResourceGroup group = Ribbon.createHttpResourceGroup("myclient",
......@@ -51,6 +54,9 @@ public class RibbonTest {
RibbonRequest<ByteBuf> request = template.withUriTemplate("/").requestBuilder().build();
String result = request.execute().toString(Charset.defaultCharset());
assertEquals(content, result);
// repeat the same request
request.execute().toString(Charset.defaultCharset());
assertEquals(content, result);
}
......@@ -226,13 +232,13 @@ public class RibbonTest {
@Test
public void testCacheMiss() throws IOException {
public void testCacheMiss() throws IOException, InterruptedException {
MockWebServer server = new MockWebServer();
String content = "Hello world";
server.enqueue(new MockResponse().setResponseCode(200).setHeader("Content-type", "text/plain")
.setBody(content));
server.play();
HttpResourceGroup group = Ribbon.createHttpResourceGroup("myclient", ClientOptions.create()
.withConfigurationBasedServerList("localhost:" + server.getPort())
.withMaxAutoRetriesNextServer(1));
......
ribbon-client-extensions/src/test/java/com/netflix/ribbonclientextensions/http/TemplateBuilderTest.java
浏览文件 @ 5c02176e
package com.netflix.ribbonclientextensions.http;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import io.netty.buffer.ByteBuf;
import io.reactivex.netty.protocol.http.client.HttpClientRequest;
import io.reactivex.netty.protocol.http.client.HttpRequestHeaders;
import java.util.List;
import java.util.Map;
import org.junit.Test;
import rx.Observable;
import com.netflix.ribbonclientextensions.CacheProvider;
import com.netflix.ribbonclientextensions.Ribbon;
import com.netflix.ribbonclientextensions.RibbonRequest;
public class TemplateBuilderTest {
private static class FakeCacheProvider implements CacheProvider<String> {
String id;
FakeCacheProvider(String id) {
this.id = id;
}
@Override
public Observable<String> get(final String key,
Map<String, Object> requestProperties) {
if (key.equals(id)) {
return Observable.just(id);
} else {
return Observable.error(new IllegalArgumentException());
}
};
}
@Test
public void testVarReplacement() {
HttpResourceGroup group = Ribbon.createHttpResourceGroup("test");
......@@ -22,5 +50,43 @@ public class TemplateBuilderTest {
.createClientRequest();
assertEquals("/foo/3?name=netflix", request.getUri());
}
@Test
public void testCacheKeyTemplates() {
HttpResourceGroup group = Ribbon.createHttpResourceGroup("test");
HttpRequestTemplate<String> template = group.newRequestTemplate("resource1", String.class);
template.withUriTemplate("/foo/{id}")
.addCacheProvider("cache.{id}", new FakeCacheProvider("cache.3"))
.addCacheProvider("/cache/{id}", new FakeCacheProvider("/cache/5"));
RibbonRequest<String> request = template.requestBuilder().withRequestProperty("id", 3).build();
String result = request.execute();
assertEquals("cache.3", result);
request = template.requestBuilder().withRequestProperty("id", 5).build();
result = request.execute();
assertEquals("/cache/5", result);
}
@Test
public void testHttpHeaders() {
HttpResourceGroup group = Ribbon.createHttpResourceGroup("test");
group.withCommonHeader("header1", "group");
HttpRequestTemplate<String> template = group.newRequestTemplate("resource1", String.class);
template.withUriTemplate("/foo/bar")
.withHeader("header2", "template")
.withHeader("header1", "template");
HttpClientRequest<ByteBuf> request = template.requestBuilder().createClientRequest();
HttpRequestHeaders headers = request.getHeaders();
List<String> header1 = headers.getAll("header1");
assertEquals(2, header1.size());
assertEquals("group", header1.get(0));
assertEquals("template", header1.get(1));
List<String> header2 = headers.getAll("header2");
assertEquals(1, header2.size());
assertEquals("template", header2.get(0));
}
}
ribbon-loadbalancer/src/main/java/com/netflix/loadbalancer/LoadBalancerContext.java
浏览文件 @ 5c02176e
......@@ -461,7 +461,7 @@ public class LoadBalancerContext implements IClientConfigAware {
Server svc = lb.chooseServer(loadBalancerKey);
if (svc == null){
throw new ClientException(ClientException.ErrorType.GENERAL,
"LoadBalancer returned null Server for :"
"LoadBalancer does not have availble server for client: "
+ clientName);
}
host = svc.getHost();
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或