parent
997d1b3d7e
commit
bd59c7a691
|
@ -524,7 +524,6 @@ class DataBufferTests extends AbstractDataBufferAllocatingTests {
|
||||||
}
|
}
|
||||||
|
|
||||||
@ParameterizedDataBufferAllocatingTest
|
@ParameterizedDataBufferAllocatingTest
|
||||||
@SuppressWarnings("deprecation")
|
|
||||||
void asByteBufferIndexLength(DataBufferFactory bufferFactory) {
|
void asByteBufferIndexLength(DataBufferFactory bufferFactory) {
|
||||||
super.bufferFactory = bufferFactory;
|
super.bufferFactory = bufferFactory;
|
||||||
|
|
||||||
|
@ -534,12 +533,10 @@ class DataBufferTests extends AbstractDataBufferAllocatingTests {
|
||||||
ByteBuffer result = buffer.asByteBuffer(1, 2);
|
ByteBuffer result = buffer.asByteBuffer(1, 2);
|
||||||
assertThat(result.capacity()).isEqualTo(2);
|
assertThat(result.capacity()).isEqualTo(2);
|
||||||
|
|
||||||
boolean isNetty5DataBufferFactory = bufferFactory instanceof Netty5DataBufferFactory;
|
assumeFalse(bufferFactory instanceof Netty5DataBufferFactory, () -> {
|
||||||
if (isNetty5DataBufferFactory) {
|
|
||||||
DataBufferUtils.release(buffer);
|
DataBufferUtils.release(buffer);
|
||||||
}
|
return "Netty 5 does share the internal buffer";
|
||||||
|
});
|
||||||
assumeFalse(isNetty5DataBufferFactory, "Netty 5 does share the internal buffer");
|
|
||||||
|
|
||||||
buffer.write((byte) 'c');
|
buffer.write((byte) 'c');
|
||||||
assertThat(result.remaining()).isEqualTo(2);
|
assertThat(result.remaining()).isEqualTo(2);
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
/*
|
/*
|
||||||
* Copyright 2002-2019 the original author or authors.
|
* Copyright 2002-2022 the original author or authors.
|
||||||
*
|
*
|
||||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
* you may not use this file except in compliance with the License.
|
* you may not use this file except in compliance with the License.
|
||||||
|
@ -208,12 +208,12 @@ public abstract class AbstractDecoderTests<D extends Decoder<?>> extends Abstrac
|
||||||
protected void testDecodeError(Publisher<DataBuffer> input, ResolvableType outputType,
|
protected void testDecodeError(Publisher<DataBuffer> input, ResolvableType outputType,
|
||||||
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
|
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
|
||||||
|
|
||||||
Flux<DataBuffer> buffer = Mono.from(input).concatWith(Flux.error(new InputException()));
|
Flux<DataBuffer> flux = Mono.from(input).concatWith(Flux.error(new InputException()));
|
||||||
assertThatExceptionOfType(InputException.class).isThrownBy(() ->
|
assertThatExceptionOfType(InputException.class).isThrownBy(() ->
|
||||||
this.decoder.decode(buffer, outputType, mimeType, hints)
|
this.decoder.decode(flux, outputType, mimeType, hints)
|
||||||
.doOnNext(o -> {
|
.doOnNext(object -> {
|
||||||
if (o instanceof Buffer buf) {
|
if (object instanceof Buffer buffer) {
|
||||||
buf.close();
|
buffer.close();
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
.blockLast(Duration.ofSeconds(5)));
|
.blockLast(Duration.ofSeconds(5)));
|
||||||
|
@ -234,9 +234,9 @@ public abstract class AbstractDecoderTests<D extends Decoder<?>> extends Abstrac
|
||||||
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
|
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
|
||||||
|
|
||||||
Flux<?> result = this.decoder.decode(input, outputType, mimeType, hints)
|
Flux<?> result = this.decoder.decode(input, outputType, mimeType, hints)
|
||||||
.doOnNext(o -> {
|
.doOnNext(object -> {
|
||||||
if (o instanceof Buffer buf) {
|
if (object instanceof Buffer buffer) {
|
||||||
buf.close();
|
buffer.close();
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
StepVerifier.create(result).expectNextCount(1).thenCancel().verify();
|
StepVerifier.create(result).expectNextCount(1).thenCancel().verify();
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
/*
|
/*
|
||||||
* Copyright 2002-2021 the original author or authors.
|
* Copyright 2002-2022 the original author or authors.
|
||||||
*
|
*
|
||||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
* you may not use this file except in compliance with the License.
|
* you may not use this file except in compliance with the License.
|
||||||
|
@ -24,6 +24,7 @@ import org.springframework.core.io.buffer.DataBufferUtils;
|
||||||
import org.springframework.core.io.buffer.Netty5DataBufferFactory;
|
import org.springframework.core.io.buffer.Netty5DataBufferFactory;
|
||||||
import org.springframework.lang.Nullable;
|
import org.springframework.lang.Nullable;
|
||||||
import org.springframework.util.Assert;
|
import org.springframework.util.Assert;
|
||||||
|
import org.springframework.util.ClassUtils;
|
||||||
import org.springframework.util.ObjectUtils;
|
import org.springframework.util.ObjectUtils;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -37,6 +38,10 @@ import org.springframework.util.ObjectUtils;
|
||||||
*/
|
*/
|
||||||
public class WebSocketMessage {
|
public class WebSocketMessage {
|
||||||
|
|
||||||
|
private static final boolean reactorNetty2Present = ClassUtils.isPresent(
|
||||||
|
"io.netty5.handler.codec.http.websocketx.WebSocketFrame", WebSocketMessage.class.getClassLoader());
|
||||||
|
|
||||||
|
|
||||||
private final Type type;
|
private final Type type;
|
||||||
|
|
||||||
private final DataBuffer payload;
|
private final DataBuffer payload;
|
||||||
|
@ -129,15 +134,11 @@ public class WebSocketMessage {
|
||||||
* @see DataBufferUtils#retain(DataBuffer)
|
* @see DataBufferUtils#retain(DataBuffer)
|
||||||
*/
|
*/
|
||||||
public WebSocketMessage retain() {
|
public WebSocketMessage retain() {
|
||||||
if (!(this.nativeMessage instanceof io.netty5.handler.codec.http.websocketx.WebSocketFrame frame) ) {
|
if (reactorNetty2Present) {
|
||||||
DataBufferUtils.retain(this.payload);
|
return ReactorNetty2Helper.retain(this);
|
||||||
return this;
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
io.netty5.handler.codec.http.websocketx.WebSocketFrame newFrame = frame.send().receive();
|
|
||||||
DataBuffer newPayload = ((Netty5DataBufferFactory) this.payload.factory()).wrap(newFrame.binaryData());
|
|
||||||
return new WebSocketMessage(this.type, newPayload, newFrame);
|
|
||||||
}
|
}
|
||||||
|
DataBufferUtils.retain(this.payload);
|
||||||
|
return this;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -199,4 +200,20 @@ public class WebSocketMessage {
|
||||||
PONG
|
PONG
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
private static class ReactorNetty2Helper {
|
||||||
|
|
||||||
|
static WebSocketMessage retain(WebSocketMessage message) {
|
||||||
|
if (message.nativeMessage instanceof io.netty5.handler.codec.http.websocketx.WebSocketFrame netty5Frame) {
|
||||||
|
io.netty5.handler.codec.http.websocketx.WebSocketFrame frame = netty5Frame.send().receive();
|
||||||
|
DataBuffer payload = ((Netty5DataBufferFactory) message.payload.factory()).wrap(frame.binaryData());
|
||||||
|
return new WebSocketMessage(message.type, payload, frame);
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
DataBufferUtils.retain(message.payload);
|
||||||
|
return message;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
/*
|
/*
|
||||||
* Copyright 2002-2020 the original author or authors.
|
* Copyright 2002-2022 the original author or authors.
|
||||||
*
|
*
|
||||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
* you may not use this file except in compliance with the License.
|
* you may not use this file except in compliance with the License.
|
||||||
|
|
Loading…
Reference in New Issue