下面列出了org.springframework.http.server.reactive.ServerHttpResponse#writeWith ( ) 实例代码,或者点击链接到github查看源代码,也可以在右侧发表评论。
/**
* 生成返回数据
*
* @param exchange ServerWebExchange
* @return Mono
*/
private Mono<Void> initResponse(ServerWebExchange exchange) {
//设置headers
ServerHttpResponse response = exchange.getResponse();
HttpHeaders httpHeaders = response.getHeaders();
httpHeaders.add("Content-Type", "application/json; charset=UTF-8");
httpHeaders.add("Cache-Control", "no-store, no-cache, must-revalidate, max-age=0");
//设置body
String json = Json.toJson(reply);
logger.warn("返回数据: {}", json);
DataBuffer body = response.bufferFactory().wrap(json.getBytes());
Long startTime = exchange.getAttribute(COUNT_START_TIME);
if (startTime != null) {
long duration = (System.currentTimeMillis() - startTime);
logger.info("处理时间: {} ms", duration);
}
return response.writeWith(Mono.just(body));
}
@Override
public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
String path = request.getURI().getPath();
switch (path) {
case "/write-and-flush":
return response.writeAndFlushWith(
testInterval(Duration.ofMillis(50), 2)
.map(longValue -> wrap("data" + longValue + "\n", response))
.map(Flux::just)
.mergeWith(Flux.never()));
case "/write-and-complete":
return response.writeWith(
chunks1K().take(64).map(s -> wrap(s, response)));
case "/write-and-never-complete":
// Reactor requires at least 50 to flush, Tomcat/Undertow 8, Jetty 1
return response.writeWith(
chunks1K().take(64).map(s -> wrap(s, response)).mergeWith(Flux.never()));
default:
return response.writeWith(Flux.empty());
}
}
@Override
public Mono<Void> render(@Nullable Map<String, ?> model, @Nullable MediaType contentType,
ServerWebExchange exchange) {
StringBuilder builder = new StringBuilder();
builder.append("name=").append(this.name).append('\n');
for (Map.Entry<String, ?> entry : model.entrySet()) {
builder.append(entry.getKey()).append('=').append(entry.getValue()).append('\n');
}
builder.setLength(builder.length() - 1);
byte[] bytes = builder.toString().getBytes(StandardCharsets.UTF_8);
ServerHttpResponse response = exchange.getResponse();
DataBuffer buffer = response.bufferFactory().wrap(bytes);
response.getHeaders().setContentType(MediaType.TEXT_PLAIN);
return response.writeWith(Mono.just(buffer));
}
protected Mono<Void> handleAuthenticationFailure(ServerWebExchange exchange, String errorMsg) {
CommonResponseDto responseDto = CommonResponseDto.error(errorMsg);
ServerHttpResponse response = exchange.getResponse();
try {
byte[] bits = objectMapper.writeValueAsBytes(responseDto);
DataBuffer buffer = response.bufferFactory().wrap(bits);
response.setStatusCode(HttpStatus.UNAUTHORIZED);
response.getHeaders().add("Content-Type", "application/json;charset=UTF-8");
response.getHeaders().add(HttpHeaders.WWW_AUTHENTICATE, headerValue);
return response.writeWith(Mono.just(buffer));
} catch (JsonProcessingException e) {
log.debug("failed to process json", e);
response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
return response.setComplete();
}
}
@RequestMapping(path = "/gzip", produces = MediaType.APPLICATION_JSON_VALUE)
public Mono<Void> gzip(ServerWebExchange exchange) throws IOException {
if (log.isDebugEnabled()) {
log.debug("httpbin /gzip");
}
String jsonResponse = OBJECT_MAPPER.writeValueAsString("httpbin compatible home");
byte[] bytes = jsonResponse.getBytes(StandardCharsets.UTF_8);
ServerHttpResponse response = exchange.getResponse();
response.getHeaders().add(HttpHeaders.CONTENT_ENCODING, "gzip");
DataBufferFactory dataBufferFactory = response.bufferFactory();
response.setStatusCode(HttpStatus.OK);
ByteArrayOutputStream bos = new ByteArrayOutputStream();
GZIPOutputStream is = new GZIPOutputStream(bos);
FileCopyUtils.copy(bytes, is);
byte[] gzippedResponse = bos.toByteArray();
DataBuffer wrap = dataBufferFactory.wrap(gzippedResponse);
return response.writeWith(Flux.just(wrap));
}
@Override
public Mono<Void> render(@Nullable Map<String, ?> model, @Nullable MediaType mediaType, ServerWebExchange exchange) {
ServerHttpResponse response = exchange.getResponse();
if (mediaType != null) {
response.getHeaders().setContentType(mediaType);
}
model = new TreeMap<>(model);
String value = this.name + ": " + model.toString();
ByteBuffer byteBuffer = ByteBuffer.wrap(value.getBytes(UTF_8));
DataBuffer dataBuffer = new DefaultDataBufferFactory().wrap(byteBuffer);
return response.writeWith(Flux.just(dataBuffer));
}
private Mono<Void> unAuth(ServerHttpResponse resp, String msg) {
resp.setStatusCode(HttpStatus.UNAUTHORIZED);
resp.getHeaders().add("Content-Type", "application/json;charset=UTF-8");
String result = "";
try {
result = objectMapper.writeValueAsString(ResponseProvider.unAuth(msg));
} catch (JsonProcessingException e) {
log.error(e.getMessage(), e);
}
DataBuffer buffer = resp.bufferFactory().wrap(result.getBytes(StandardCharsets.UTF_8));
return resp.writeWith(Flux.just(buffer));
}
/**
* Process the web request and validate the API version in the header. If the API version does not match, then set
* an HTTP 412 status and write the error message to the response.
*
* @param exchange {@inheritDoc}
* @param chain {@inheritDoc}
* @return {@inheritDoc}
*/
@Override
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
PathPattern p = new PathPatternParser().parse(V2_API_PATH_PATTERN);
Mono<Void> filterMono = chain.filter(exchange);
if (p.matches(exchange.getRequest().getPath()) && version != null && !anyVersionAllowed()) {
String apiVersion = exchange.getRequest().getHeaders().getFirst(version.getBrokerApiVersionHeader());
ServerHttpResponse response = exchange.getResponse();
String message = null;
if (apiVersion == null) {
response.setStatusCode(HttpStatus.BAD_REQUEST);
message = ServiceBrokerApiVersionErrorMessage.from(version.getApiVersion(), "null").toString();
}
else if (!version.getApiVersion().equals(apiVersion)) {
response.setStatusCode(HttpStatus.PRECONDITION_FAILED);
message = ServiceBrokerApiVersionErrorMessage.from(version.getApiVersion(), apiVersion)
.toString();
}
if (message != null) {
String json;
try {
json = new ObjectMapper().writeValueAsString(new ErrorMessage(message));
}
catch (JsonProcessingException e) {
json = "{}";
}
Flux<DataBuffer> responseBody =
Flux.just(json)
.map(s -> toDataBuffer(s, response.bufferFactory()));
filterMono = response.writeWith(responseBody);
}
}
return filterMono;
}
@Override
public Mono<Void> handle(ServerWebExchange exchange, Throwable ex) {
ServerHttpResponse response = exchange.getResponse();
if (response.isCommitted()) {
return Mono.error(ex);
}
// header set
response.getHeaders().setContentType(MediaType.APPLICATION_JSON);
if (ex instanceof ResponseStatusException) {
response.setStatusCode(((ResponseStatusException) ex).getStatus());
}
return response
.writeWith(Mono.fromSupplier(() -> {
DataBufferFactory bufferFactory = response.bufferFactory();
try {
Map<String,Object> map = new HashMap<>();
map.put("code",500);
map.put("msg",ex.getMessage());
return bufferFactory.wrap(objectMapper.writeValueAsBytes(map));
} catch (JsonProcessingException e) {
log.error("Error writing response", ex);
return bufferFactory.wrap(new byte[0]);
}
}));
}
@Override
public GatewayFilter apply(Object config) {
return (exchange, chain) -> {
ServerHttpRequest request = exchange.getRequest();
// 不是登录请求,直接向下执行
if (!StrUtil.containsAnyIgnoreCase(request.getURI().getPath()
, OAUTH_TOKEN_URL)) {
return chain.filter(exchange);
}
// 刷新token,直接向下执行
String grantType = request.getQueryParams().getFirst("grant_type");
if (StrUtil.equals(REFRESH_TOKEN, grantType)) {
return chain.filter(exchange);
}
// 终端设置不校验, 直接向下执行
try {
//校验验证码
checkCode(request);
} catch (Exception e) {
ServerHttpResponse response = exchange.getResponse();
response.setStatusCode(HttpStatus.PRECONDITION_REQUIRED);
try {
Map map= new HashMap<>();
map.put("code",-1);
map.put("msg","失败");
return response.writeWith(Mono.just(response.bufferFactory()
.wrap(objectMapper.writeValueAsBytes(map))));
} catch (JsonProcessingException e1) {
log.error("对象输出异常", e1);
}
}
return chain.filter(exchange);
};
}
@Override
public Mono<Void> handle(ServerWebExchange exchange, Throwable ex) {
ServerHttpResponse response = exchange.getResponse();
if (response.isCommitted()) {
return Mono.error(ex);
}
// header set
response.getHeaders().setContentType(MediaType.APPLICATION_JSON);
if (ex instanceof ResponseStatusException) {
response.setStatusCode(((ResponseStatusException) ex).getStatus());
}
return response
.writeWith(Mono.fromSupplier(() -> {
DataBufferFactory bufferFactory = response.bufferFactory();
try {
Map<String,Object> map = new HashMap<>();
map.put("code",500);
map.put("msg",ex.getMessage());
return bufferFactory.wrap(objectMapper.writeValueAsBytes(map));
} catch (JsonProcessingException e) {
log.error("Error writing response", ex);
return bufferFactory.wrap(new byte[0]);
}
}));
}
@Override
public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
response.setStatusCode(HttpStatus.OK);
response.getHeaders().add(HttpHeaderNames.CONTENT_TYPE.toString(),
MediaType.PLAIN_TEXT_UTF_8.toString());
return response.writeWith(request.getBody());
}
/**
* return json result
* @param exchange the current server exchange
* @param result json result
* @return {@code Mono<Void>}
*/
private Mono<Void> jsonResult(ServerWebExchange exchange ,JsonResult result){
ServerHttpResponse response = exchange.getResponse();
response.getHeaders().setContentType(MediaType.APPLICATION_JSON_UTF8);
return response.writeWith(Mono.just(exchange.getResponse()
.bufferFactory()
.wrap(Objects.requireNonNull(JSON.toJSONString(result)).getBytes(Charset.forName("utf8")))));
}
private Mono<Void> handleNormalResponse(ServerHttpResponse serverHttpResponse, ExecutionResult executionResult) throws IOException {
Map<String, Object> result = executionResult.toSpecification();
serverHttpResponse.setStatusCode(HttpStatus.OK);
HttpHeaders headers = serverHttpResponse.getHeaders();
headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
String body = objectMapper.writeValueAsString(result);
return serverHttpResponse.writeWith(strToDataBuffer(body));
}
@Override
public Mono<Void> render(@Nullable Map<String, ?> model, @Nullable MediaType mediaType, ServerWebExchange exchange) {
ServerHttpResponse response = exchange.getResponse();
if (mediaType != null) {
response.getHeaders().setContentType(mediaType);
}
model = new TreeMap<>(model);
String value = this.name + ": " + model.toString();
ByteBuffer byteBuffer = ByteBuffer.wrap(value.getBytes(UTF_8));
DataBuffer dataBuffer = new DefaultDataBufferFactory().wrap(byteBuffer);
return response.writeWith(Flux.just(dataBuffer));
}
@Override
public GatewayFilter apply(Object config) {
return (exchange, chain) -> {
ServerHttpRequest request = exchange.getRequest();
// 不是登录请求,直接向下执行
if (!StrUtil.containsAnyIgnoreCase(request.getURI().getPath()
, SecurityConstants.OAUTH_TOKEN_URL)) {
return chain.filter(exchange);
}
// 刷新token,直接向下执行
String grantType = request.getQueryParams().getFirst("grant_type");
if (StrUtil.equals(SecurityConstants.REFRESH_TOKEN, grantType)) {
return chain.filter(exchange);
}
// 终端设置不校验, 直接向下执行
try {
String[] clientInfos = WebUtils.getClientId(request);
if (filterIgnorePropertiesConfig.getClients().contains(clientInfos[0])) {
return chain.filter(exchange);
}
//校验验证码
checkCode(request);
} catch (Exception e) {
ServerHttpResponse response = exchange.getResponse();
response.setStatusCode(HttpStatus.PRECONDITION_REQUIRED);
try {
return response.writeWith(Mono.just(response.bufferFactory()
.wrap(objectMapper.writeValueAsBytes(
SmakerResult.builder().msg(e.getMessage())
.code(CommonConstants.FAIL).build()))));
} catch (JsonProcessingException e1) {
log.error("对象输出异常", e1);
}
}
return chain.filter(exchange);
};
}
private Mono<Void> setBody(ClientResponse clientResponse, ServerHttpResponse response) {
Mono<DataBuffer> body = clientResponse.bodyToMono(byte[].class)
.map(bytes -> response.bufferFactory().wrap(bytes));
return response.writeWith(body);
}
private Mono<Void> getVoidMono(ServerHttpResponse response, String jsonString) {
response.getHeaders().add("Content-Type", "application/json;charset=UTF-8");
byte[] datas = jsonString.getBytes(StandardCharsets.UTF_8);
DataBuffer buffer = response.bufferFactory().wrap(datas);
return response.writeWith(Mono.just(buffer));
}
private static Mono<Void> handler(final ServerWebExchange serverWebExchange) {
final ServerHttpResponse response = serverWebExchange.getResponse();
response.setStatusCode(HttpStatus.OK);
return response.writeWith(Mono.just(new DefaultDataBufferFactory().wrap("Hello World!\n".getBytes())));
}
/**
* 设置webflux模型响应
*
* @param response ServerHttpResponse
* @param contentType content-type
* @param status http状态码
* @param value 响应内容
* @return Mono<Void>
*/
public static Mono<Void> makeWebFluxResponse(ServerHttpResponse response, String contentType,
HttpStatus status, Object value) {
response.setStatusCode(status);
response.getHeaders().add(HttpHeaders.CONTENT_TYPE, contentType);
DataBuffer dataBuffer = response.bufferFactory().wrap(JSONObject.toJSONString(value).getBytes());
return response.writeWith(Mono.just(dataBuffer));
}