SpringCloud/spring-cloud-openfeign
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch '1.3.x'

Browse Source
This commit is contained in:
Dave Syer
2017-06-12 16:15:44 +01:00
parent 1a0428da97 4783b07407
commit 310f558e9c
2 changed files with 40 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-cloud-netflix-turbine-stream/src/main/java/org/springframework/cloud/netflix/turbine/stream/HystrixStreamAggregator.java
View File

@@ -17,6 +17,7 @@
package org.springframework.cloud.netflix.turbine.stream;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
@@ -56,18 +57,33 @@ public class HystrixStreamAggregator {
payload = payload.replace("\\\"", "\"");
}
try {
@SuppressWarnings("unchecked")
Map<String, Object> map = this.objectMapper.readValue(payload, Map.class);
Map<String, Object> data = getPayloadData(map);
log.debug("Received hystrix stream payload: " + data);
this.subject.onNext(data);
if (payload.startsWith("[")) {
@SuppressWarnings("unchecked")
List<Map<String, Object>> list = this.objectMapper.readValue(payload,
List.class);
for (Map<String, Object> map : list) {
sendMap(map);
}
}
else {
@SuppressWarnings("unchecked")
Map<String, Object> map = this.objectMapper.readValue(payload, Map.class);
sendMap(map);
}
}
catch (IOException ex) {
log.error("Error receiving hystrix stream payload: " + payload, ex);
}
}
private void sendMap(Map<String, Object> map) {
Map<String, Object> data = getPayloadData(map);
if (log.isDebugEnabled()) {
log.debug("Received hystrix stream payload: " + data);
}
this.subject.onNext(data);
}
public static Map<String, Object> getPayloadData(Map<String, Object> jsonMap) {
@SuppressWarnings("unchecked")
Map<String, Object> origin = (Map<String, Object>) jsonMap.get("origin");

26
spring-cloud-netflix-turbine-stream/src/test/java/org/springframework/cloud/netflix/turbine/stream/HystrixStreamAggregatorTests.java
View File

@@ -16,19 +16,20 @@
package org.springframework.cloud.netflix.turbine.stream;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.boot.test.rule.OutputCapture;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.not;
import static org.junit.Assert.assertThat;
import java.util.Map;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.boot.test.rule.OutputCapture;
import com.fasterxml.jackson.databind.ObjectMapper;
import rx.subjects.PublishSubject;
public class HystrixStreamAggregatorTests {
@@ -52,6 +53,15 @@ public class HystrixStreamAggregatorTests {
this.output.expect(not(containsString("ERROR")));
}
@Test
public void messageWrappedInArray() throws Exception {
this.publisher.subscribe(map -> {
assertThat(map.get("type"), equalTo("HystrixCommand"));
});
this.aggregator.sendToSubject("[" + PAYLOAD + "]");
this.output.expect(not(containsString("ERROR")));
}
@Test
public void doubleEncodedMessage() throws Exception {
this.publisher.subscribe(map -> {