You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by li...@apache.org on 2024/06/25 01:01:06 UTC
(servicecomb-java-chassis) branch master updated (25c6b3455 -> 45be436d4)
This is an automated email from the ASF dual-hosted git repository.
liubao pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git
from 25c6b3455 Bump com.github.spotbugs:spotbugs-maven-plugin from 4.8.2.0 to 4.8.5.0 (#4372)
new 050a77d32 [SCB-2885]generate content-type text/event-stream
new 45be436d4 [SCB-2885]refactoring to using vert.x Buffer instead of Netty ByteBuffer
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../rest/codec/produce/ProduceProcessor.java | 2 +-
.../rest/filter/inner/RestServerCodecFilter.java | 6 +--
.../common/rest/codec/param/TestBodyProcessor.java | 12 ++---
demo/demo-zookeeper/provider/pom.xml | 4 ++
.../samples/ReactiveStreamController.java} | 61 ++++++++++++++++------
dependencies/default/pom.xml | 13 +++++
.../servicecomb/foundation/vertx/VertxUtils.java | 5 +-
.../vertx/client/tcp/TcpClientConnection.java | 2 +-
.../vertx/http/HttpServletResponseEx.java | 16 +++++-
.../vertx/http/StandardHttpServletRequestEx.java | 6 +--
.../VertxServerRequestToHttpServletRequest.java | 2 +-
.../foundation/vertx/stream/BufferInputStream.java | 51 +++++++++++-------
.../vertx/stream/BufferOutputStream.java | 36 ++++++-------
.../foundation/vertx/tcp/TcpConnection.java | 21 ++------
.../servicecomb/foundation/vertx/TestStream.java | 28 +++++-----
.../foundation/vertx/TestVertxUtils.java | 5 +-
.../vertx/client/tcp/TestTcpClientConnection.java | 10 ++--
.../http/TestStandardHttpServletRequestEx.java | 2 +-
.../vertx/stream/TestBufferInputStream.java | 9 ++--
swagger/swagger-generator/generator-core/pom.xml | 4 ++
.../generator/core/SwaggerGeneratorContext.java | 3 +-
.../PublisherProcessor.java} | 26 +++++----
...icecomb.swagger.generator.ResponseTypeProcessor | 1 +
.../highway/HighwayProducerInvocationFlow.java | 2 +-
24 files changed, 198 insertions(+), 129 deletions(-)
copy demo/{demo-nacos/consumer/src/main/java/org/apache/servicecomb/samples/ConsumerController.java => demo-zookeeper/provider/src/main/java/org/apache/servicecomb/samples/ReactiveStreamController.java} (53%)
copy swagger/swagger-generator/generator-core/src/main/java/org/apache/servicecomb/swagger/generator/core/processor/{annotation/RequestBodyMethodAnnotationProcessor.java => response/PublisherProcessor.java} (62%)
(servicecomb-java-chassis) 02/02: [SCB-2885]refactoring to using vert.x Buffer instead of Netty ByteBuffer
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
liubao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git
commit 45be436d442341d9b9e121dc41ecb8c67f69d762
Author: liubao <bi...@qq.com>
AuthorDate: Mon Jun 24 20:15:04 2024 +0800
[SCB-2885]refactoring to using vert.x Buffer instead of Netty ByteBuffer
---
.../rest/codec/produce/ProduceProcessor.java | 2 +-
.../rest/filter/inner/RestServerCodecFilter.java | 6 +--
.../common/rest/codec/param/TestBodyProcessor.java | 12 +++--
.../servicecomb/foundation/vertx/VertxUtils.java | 5 ++-
.../vertx/client/tcp/TcpClientConnection.java | 2 +-
.../vertx/http/HttpServletResponseEx.java | 16 ++++++-
.../vertx/http/StandardHttpServletRequestEx.java | 6 +--
.../VertxServerRequestToHttpServletRequest.java | 2 +-
.../foundation/vertx/stream/BufferInputStream.java | 51 ++++++++++++++--------
.../vertx/stream/BufferOutputStream.java | 36 +++++++--------
.../foundation/vertx/tcp/TcpConnection.java | 21 ++-------
.../servicecomb/foundation/vertx/TestStream.java | 28 +++++++-----
.../foundation/vertx/TestVertxUtils.java | 5 ++-
.../vertx/client/tcp/TestTcpClientConnection.java | 10 ++---
.../http/TestStandardHttpServletRequestEx.java | 2 +-
.../vertx/stream/TestBufferInputStream.java | 9 ++--
.../highway/HighwayProducerInvocationFlow.java | 2 +-
17 files changed, 112 insertions(+), 103 deletions(-)
diff --git a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceProcessor.java b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceProcessor.java
index bcf7cb74b..574c1e1b6 100644
--- a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceProcessor.java
+++ b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceProcessor.java
@@ -72,7 +72,7 @@ public interface ProduceProcessor {
return null;
}
- try (BufferInputStream input = new BufferInputStream(buffer.getByteBuf())) {
+ try (BufferInputStream input = new BufferInputStream(buffer)) {
return doDecodeResponse(input, type);
}
}
diff --git a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilter.java b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilter.java
index 18a5d9d94..348ee7f19 100644
--- a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilter.java
+++ b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilter.java
@@ -50,8 +50,8 @@ import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import io.netty.buffer.Unpooled;
import io.vertx.core.MultiMap;
+import io.vertx.core.buffer.Buffer;
import jakarta.servlet.http.Part;
import jakarta.ws.rs.core.HttpHeaders;
@@ -132,7 +132,7 @@ public class RestServerCodecFilter extends AbstractFilter implements ProviderFil
}
responseEx.setContentType(produceProcessor.getName());
- try (BufferOutputStream output = new BufferOutputStream(Unpooled.compositeBuffer())) {
+ try (BufferOutputStream output = new BufferOutputStream(Buffer.buffer())) {
if (failed) {
produceProcessor.encodeResponse(output, ((InvocationException) response.getResult()).getErrorData());
} else {
@@ -144,7 +144,7 @@ public class RestServerCodecFilter extends AbstractFilter implements ProviderFil
return CompletableFuture.completedFuture(response);
} catch (Throwable e) {
LOGGER.error("internal service error must be fixed.", e);
- try (BufferOutputStream output = new BufferOutputStream(Unpooled.compositeBuffer())) {
+ try (BufferOutputStream output = new BufferOutputStream(Buffer.buffer())) {
responseEx.setStatus(500);
produceProcessor.encodeResponse(output, new CommonExceptionData("500", "Internal Server Error"));
responseEx.setBodyBuffer(output.getBuffer());
diff --git a/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/codec/param/TestBodyProcessor.java b/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/codec/param/TestBodyProcessor.java
index 3debd1645..784150324 100644
--- a/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/codec/param/TestBodyProcessor.java
+++ b/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/codec/param/TestBodyProcessor.java
@@ -41,8 +41,6 @@ import org.springframework.core.env.Environment;
import com.fasterxml.jackson.databind.type.TypeFactory;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
import io.swagger.v3.oas.models.OpenAPI;
import io.swagger.v3.oas.models.media.Content;
import io.swagger.v3.oas.models.parameters.RequestBody;
@@ -66,7 +64,7 @@ public class TestBodyProcessor {
ParamValueProcessor processor;
- final ByteBuf inputBodyByteBuf = Unpooled.buffer();
+ final Buffer inputBodyByteBuf = Buffer.buffer();
final BufferInputStream inputStream = new BufferInputStream(inputBodyByteBuf);
@@ -142,7 +140,7 @@ public class TestBodyProcessor {
@Test
public void testGetValueTextPlain() throws Exception {
setupGetValue(String.class);
- inputBodyByteBuf.writeCharSequence("abc", StandardCharsets.UTF_8);
+ inputBodyByteBuf.appendString("abc", StandardCharsets.UTF_8.toString());
Mockito.when(request.getContentType()).thenReturn(MediaType.TEXT_PLAIN);
@@ -152,7 +150,7 @@ public class TestBodyProcessor {
@Test
public void testGetValueContextTypeJson() throws Exception {
setupGetValue(Integer.class);
- inputBodyByteBuf.writeCharSequence("\"1\"", StandardCharsets.UTF_8);
+ inputBodyByteBuf.appendString("\"1\"", StandardCharsets.UTF_8.toString());
Mockito.when(request.getContentType()).thenReturn(MediaType.APPLICATION_JSON);
@@ -162,7 +160,7 @@ public class TestBodyProcessor {
@Test
public void testGetValueDefaultJson() throws Exception {
setupGetValue(Integer.class);
- inputBodyByteBuf.writeCharSequence("\"1\"", StandardCharsets.UTF_8);
+ inputBodyByteBuf.appendString("\"1\"", StandardCharsets.UTF_8.toString());
Assertions.assertEquals(1, processor.getValue(request));
}
@@ -220,7 +218,7 @@ public class TestBodyProcessor {
public void testGetValueRawJson() throws Exception {
createRawJsonProcessor();
initInputStream();
- inputBodyByteBuf.writeCharSequence("\"1\"", StandardCharsets.UTF_8);
+ inputBodyByteBuf.appendString("\"1\"", StandardCharsets.UTF_8.toString());
Assertions.assertEquals("\"1\"", processor.getValue(request));
}
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/VertxUtils.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/VertxUtils.java
index 331c502ff..43094de2a 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/VertxUtils.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/VertxUtils.java
@@ -153,8 +153,9 @@ public final class VertxUtils {
}
public static byte[] getBytesFast(Buffer buffer) {
- ByteBuf byteBuf = buffer.getByteBuf();
- return getBytesFast(byteBuf);
+ byte[] arr = new byte[buffer.length()];
+ buffer.getBytes(arr, 0);
+ return arr;
}
public static byte[] getBytesFast(ByteBuf byteBuf) {
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/tcp/TcpClientConnection.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/tcp/TcpClientConnection.java
index 5d363e874..d8800800f 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/tcp/TcpClientConnection.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/tcp/TcpClientConnection.java
@@ -144,7 +144,7 @@ public class TcpClientConnection extends TcpConnection {
if (Status.WORKING.equals(status)) {
// encode in sender thread
try (TcpOutputStream os = tcpClientPackage.createStream()) {
- write(os.getByteBuf());
+ write(os.getBuffer());
tcpClientPackage.finishWriteToBuffer();
}
return true;
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java
index 265ee658e..e6ec15dc0 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java
@@ -17,14 +17,16 @@
package org.apache.servicecomb.foundation.vertx.http;
+import java.io.IOException;
+import java.io.PrintWriter;
import java.util.concurrent.CompletableFuture;
+import io.vertx.core.http.HttpHeaders;
+import jakarta.servlet.ServletOutputStream;
import jakarta.servlet.http.HttpServletResponse;
import jakarta.servlet.http.Part;
import jakarta.ws.rs.core.Response.StatusType;
-import io.vertx.core.http.HttpHeaders;
-
public interface HttpServletResponseEx extends HttpServletResponse, BodyBufferSupport {
StatusType getStatusType();
@@ -37,4 +39,14 @@ public interface HttpServletResponseEx extends HttpServletResponse, BodyBufferSu
default void setChunked(boolean chunked) {
setHeader(HttpHeaders.TRANSFER_ENCODING.toString(), HttpHeaders.CHUNKED.toString());
}
+
+ @Override
+ default ServletOutputStream getOutputStream() throws IOException {
+ throw new IOException("Not allowed");
+ }
+
+ @Override
+ default PrintWriter getWriter() throws IOException {
+ throw new IOException("Not allowed");
+ }
}
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/StandardHttpServletRequestEx.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/StandardHttpServletRequestEx.java
index 21fd45309..82b46b475 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/StandardHttpServletRequestEx.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/StandardHttpServletRequestEx.java
@@ -38,8 +38,6 @@ import org.apache.servicecomb.foundation.vertx.stream.BufferInputStream;
import com.google.common.annotations.VisibleForTesting;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
import io.vertx.core.buffer.Buffer;
import jakarta.servlet.ServletInputStream;
import jakarta.servlet.http.HttpServletRequest;
@@ -77,9 +75,9 @@ public class StandardHttpServletRequestEx extends HttpServletRequestWrapper impl
if (this.inputStream == null) {
if (cacheRequest) {
byte[] inputBytes = IOUtils.toByteArray(getRequest().getInputStream());
- ByteBuf byteBuf = Unpooled.wrappedBuffer(inputBytes);
+ Buffer byteBuf = Buffer.buffer(inputBytes);
this.inputStream = new BufferInputStream(byteBuf);
- setBodyBuffer(Buffer.buffer(Unpooled.wrappedBuffer(byteBuf)));
+ setBodyBuffer(byteBuf);
} else {
this.inputStream = getRequest().getInputStream();
}
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerRequestToHttpServletRequest.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerRequestToHttpServletRequest.java
index 737e400bd..da55f5ff6 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerRequestToHttpServletRequest.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerRequestToHttpServletRequest.java
@@ -268,7 +268,7 @@ public class VertxServerRequestToHttpServletRequest extends AbstractHttpServletR
if (context.body().buffer() == null) {
return null;
}
- inputStream = new BufferInputStream(context.body().buffer().getByteBuf());
+ inputStream = new BufferInputStream(context.body().buffer());
return inputStream;
}
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/BufferInputStream.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/BufferInputStream.java
index 09840be11..af0d4acf1 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/BufferInputStream.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/BufferInputStream.java
@@ -20,57 +20,69 @@ package org.apache.servicecomb.foundation.vertx.stream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
-import io.netty.buffer.ByteBuf;
+import io.vertx.core.buffer.Buffer;
import jakarta.servlet.ReadListener;
import jakarta.servlet.ServletInputStream;
public class BufferInputStream extends ServletInputStream {
- private final ByteBuf byteBuf;
+ private int readIndex = 0;
- public BufferInputStream(ByteBuf buffer) {
+ private final Buffer byteBuf;
+
+ public BufferInputStream(Buffer buffer) {
this.byteBuf = buffer;
}
@Override
public long skip(long len) {
int skipLen = Math.min((int) len, available());
- byteBuf.skipBytes(skipLen);
+ this.readIndex += skipLen;
return skipLen;
}
public byte readByte() {
- return byteBuf.readByte();
+ int index = readIndex;
+ readIndex = index + 1;
+ return byteBuf.getByte(index);
}
@Override
public int read() {
- return byteBuf.readUnsignedByte();
+ return this.readByte() & 255;
}
public boolean readBoolean() {
- return byteBuf.readBoolean();
+ return this.readByte() != 0;
}
public short readShort() {
- return byteBuf.readShort();
+ int index = readIndex;
+ readIndex = index + 2;
+ return byteBuf.getShort(index);
}
public int readInt() {
- return byteBuf.readInt();
+ int index = readIndex;
+ readIndex = index + 4;
+ return byteBuf.getInt(index);
}
public long readLong() {
- return byteBuf.readLong();
+ int index = readIndex;
+ readIndex = index + 8;
+ return byteBuf.getLong(index);
}
public int getIndex() {
- return byteBuf.readerIndex();
+ return readIndex;
}
public String readString() {
int length = readInt();
byte[] bytes = new byte[length];
- byteBuf.readBytes(bytes);
+ int index = readIndex;
+ readIndex = index + length;
+ byteBuf.getBytes(index, readIndex, bytes);
return new String(bytes, StandardCharsets.UTF_8);
}
@@ -94,32 +106,35 @@ public class BufferInputStream extends ServletInputStream {
len = avail;
}
- byteBuf.readBytes(b, off, len);
+ int index = readIndex;
+ readIndex = index + len;
+
+ byteBuf.getBytes(index, readIndex, b);
return len;
}
@Override
public int available() {
- return byteBuf.readableBytes();
+ return byteBuf.length() - readIndex;
}
@Override
public void close() {
- byteBuf.release();
+ // nothing to do
}
@Override
public void reset() throws IOException {
- byteBuf.resetReaderIndex();
+ readIndex = 0;
}
- public ByteBuf getByteBuf() {
+ public Buffer getByteBuf() {
return byteBuf;
}
@Override
public boolean isFinished() {
- return !byteBuf.isReadable();
+ return byteBuf.length() > readIndex;
}
@Override
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/BufferOutputStream.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/BufferOutputStream.java
index f063d5d06..42c9b3126 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/BufferOutputStream.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/BufferOutputStream.java
@@ -20,9 +20,7 @@ package org.apache.servicecomb.foundation.vertx.stream;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
-import io.netty.buffer.ByteBuf;
import io.vertx.core.buffer.Buffer;
-import io.vertx.core.buffer.impl.VertxByteBufAllocator;
/**
* BufferOutputStream.
@@ -32,40 +30,36 @@ import io.vertx.core.buffer.impl.VertxByteBufAllocator;
public class BufferOutputStream extends OutputStream {
private static final int DIRECT_BUFFER_SIZE = 1024;
- protected ByteBuf byteBuf;
+ protected Buffer byteBuf;
public BufferOutputStream() {
- this(VertxByteBufAllocator.DEFAULT.heapBuffer(DIRECT_BUFFER_SIZE, Integer.MAX_VALUE));
+ this(Buffer.buffer(DIRECT_BUFFER_SIZE));
}
- public BufferOutputStream(ByteBuf buffer) {
+ public BufferOutputStream(Buffer buffer) {
this.byteBuf = buffer;
}
- public ByteBuf getByteBuf() {
- return byteBuf;
- }
-
public Buffer getBuffer() {
- return Buffer.buffer(byteBuf);
+ return byteBuf;
}
public int length() {
- return byteBuf.readableBytes();
+ return byteBuf.length();
}
public void writeByte(byte value) {
- byteBuf.writeByte(value);
+ byteBuf.appendByte(value);
}
// 实际是写byte
@Override
public void write(int byteValue) {
- byteBuf.writeByte((byte) byteValue);
+ byteBuf.appendByte((byte) byteValue);
}
public void write(boolean value) {
- byteBuf.writeBoolean(value);
+ byteBuf.appendByte(value ? (byte) 1 : (byte) 0);
}
public void writeInt(int pos, int value) {
@@ -73,20 +67,20 @@ public class BufferOutputStream extends OutputStream {
}
public void writeShort(short value) {
- byteBuf.writeShort(value);
+ byteBuf.appendShort(value);
}
public void writeInt(int value) {
- byteBuf.writeInt(value);
+ byteBuf.appendInt(value);
}
public void writeLong(long value) {
- byteBuf.writeLong(value);
+ byteBuf.appendLong(value);
}
public void writeString(String value) {
- byteBuf.writeInt(value.length());
- byteBuf.writeCharSequence(value, StandardCharsets.UTF_8);
+ writeInt(value.length());
+ byteBuf.appendString(value, StandardCharsets.UTF_8.toString());
}
@Override
@@ -96,7 +90,7 @@ public class BufferOutputStream extends OutputStream {
@Override
public void write(byte[] bytes, int offset, int len) {
- byteBuf.writeBytes(bytes, offset, len);
+ byteBuf.appendBytes(bytes, offset, len);
}
@Override
@@ -105,6 +99,6 @@ public class BufferOutputStream extends OutputStream {
}
public int writerIndex() {
- return byteBuf.writerIndex();
+ return byteBuf.length();
}
}
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/tcp/TcpConnection.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/tcp/TcpConnection.java
index 95f460013..15e6977d2 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/tcp/TcpConnection.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/tcp/TcpConnection.java
@@ -20,9 +20,6 @@ import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
-import io.netty.buffer.CompositeByteBuf;
import io.vertx.core.Context;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetSocket;
@@ -49,7 +46,7 @@ public class TcpConnection {
// so this optimization:
// 1.avoid vertx's lock
// 2.reduce netty's task schedule
- private final Queue<ByteBuf> writeQueue = new ConcurrentLinkedQueue<>();
+ private final Queue<Buffer> writeQueue = new ConcurrentLinkedQueue<>();
private final AtomicLong writeQueueSize = new AtomicLong();
@@ -83,7 +80,7 @@ public class TcpConnection {
this.context = netSocket.getContext();
}
- public void write(ByteBuf buf) {
+ public void write(Buffer buf) {
writeQueue.add(buf);
long oldSize = writeQueueSize.getAndIncrement();
if (oldSize == 0) {
@@ -97,25 +94,15 @@ public class TcpConnection {
}
protected void writeInContext() {
- CompositeByteBuf cbb = ByteBufAllocator.DEFAULT.compositeBuffer();
for (; ; ) {
- ByteBuf buf = writeQueue.poll();
+ Buffer buf = writeQueue.poll();
if (buf == null) {
break;
}
writeQueueSize.decrementAndGet();
- cbb.addComponent(true, buf);
- if (cbb.numComponents() == cbb.maxNumComponents()) {
- CompositeByteBuf last = cbb;
- netSocket.write(Buffer.buffer(last)).onComplete(r -> last.release());
- cbb = ByteBufAllocator.DEFAULT.compositeBuffer();
- }
- }
- if (cbb.isReadable()) {
- CompositeByteBuf last = cbb;
- netSocket.write(Buffer.buffer(last)).onComplete(r -> last.release());
+ netSocket.write(buf);
}
}
}
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestStream.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestStream.java
index 2c9fdc895..ea3a797da 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestStream.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestStream.java
@@ -22,8 +22,7 @@ import org.apache.servicecomb.foundation.vertx.stream.BufferOutputStream;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import io.netty.buffer.ByteBuf;
-import io.vertx.core.buffer.impl.VertxByteBufAllocator;
+import io.vertx.core.buffer.Buffer;
public class TestStream {
@@ -31,16 +30,21 @@ public class TestStream {
@Test
public void testBufferInputStream() {
- ByteBuf obuf = VertxByteBufAllocator.DEFAULT.heapBuffer(DIRECT_BUFFER_SIZE, Integer.MAX_VALUE);
- obuf.writeBytes(("testss").getBytes());
+ Buffer obuf = Buffer.buffer(DIRECT_BUFFER_SIZE);
+ obuf.appendBytes(("testss").getBytes());
@SuppressWarnings("resource")
BufferInputStream oBufferInputStream = new BufferInputStream(obuf);
- Assertions.assertNotEquals(1234, oBufferInputStream.skip(0));
- Assertions.assertNotEquals(obuf.readByte(), oBufferInputStream.readByte());
- Assertions.assertEquals(obuf.readByte() + 1, oBufferInputStream.read());
- Assertions.assertEquals(obuf.readBoolean(), oBufferInputStream.readBoolean());
- Assertions.assertEquals(obuf.readerIndex(), oBufferInputStream.getIndex());
- Assertions.assertEquals(obuf.readableBytes(), oBufferInputStream.available());
+
+ byte test = oBufferInputStream.readByte();
+ Assertions.assertEquals((byte) 't', test);
+ Assertions.assertEquals((byte) 'e', oBufferInputStream.read());
+
+ Assertions.assertEquals(2, oBufferInputStream.skip(2));
+ Assertions.assertEquals((byte) 's', oBufferInputStream.read());
+ Assertions.assertTrue(oBufferInputStream.readBoolean());
+
+ Assertions.assertEquals(6, oBufferInputStream.getIndex());
+ Assertions.assertEquals(0, oBufferInputStream.available());
}
@Test
@@ -53,9 +57,9 @@ public class TestStream {
Assertions.assertTrue((1 < oBufferOutputStream.length()));
@SuppressWarnings("resource")
- BufferInputStream oBufferInputStream = new BufferInputStream(oBufferOutputStream.getByteBuf());
+ BufferInputStream oBufferInputStream = new BufferInputStream(oBufferOutputStream.getBuffer());
Assertions.assertEquals("test", oBufferInputStream.readString());
Assertions.assertEquals(1, oBufferInputStream.readByte());
- Assertions.assertEquals(true, oBufferInputStream.readBoolean());
+ Assertions.assertTrue(oBufferInputStream.readBoolean());
}
}
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestVertxUtils.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestVertxUtils.java
index 399d0b1f6..f3762fe5f 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestVertxUtils.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestVertxUtils.java
@@ -131,11 +131,12 @@ public class TestVertxUtils {
@Test
public void testgetBytesFastBufferInputStream() throws IOException {
byte[] bytes = new byte[] {1};
- ByteBuf byteBuf = Unpooled.wrappedBuffer(bytes);
+ Buffer byteBuf = Buffer.buffer(bytes);
try (BufferInputStream inputStream = new BufferInputStream(byteBuf)) {
byte[] result = VertxUtils.getBytesFast(inputStream);
- Assertions.assertSame(bytes, result);
+ Assertions.assertEquals(bytes.length, result.length);
+ Assertions.assertEquals(bytes[0], result[0]);
}
}
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestTcpClientConnection.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestTcpClientConnection.java
index 0767898f2..de87122f1 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestTcpClientConnection.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestTcpClientConnection.java
@@ -25,13 +25,14 @@ import org.apache.servicecomb.foundation.vertx.client.tcp.TcpClientConnection.St
import org.apache.servicecomb.foundation.vertx.tcp.TcpOutputStream;
import org.junit.Before;
import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
import io.vertx.core.AsyncResult;
import io.vertx.core.Context;
import io.vertx.core.Handler;
import io.vertx.core.Promise;
+import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetSocket;
import io.vertx.core.net.impl.NetSocketImpl;
import mockit.Deencapsulation;
@@ -39,7 +40,6 @@ import mockit.Expectations;
import mockit.Mock;
import mockit.MockUp;
import mockit.Mocked;
-import org.junit.jupiter.api.Assertions;
public class TestTcpClientConnection {
@Mocked
@@ -90,14 +90,14 @@ public class TestTcpClientConnection {
Deencapsulation.setField(tcpClientConnection, "status", Status.WORKING);
long msgId = 1;
- ByteBuf byteBuf = Unpooled.buffer();
+ Buffer byteBuf = Buffer.buffer();
new Expectations(tcpClientConnection) {
{
tcpClientPackage.getMsgId();
result = msgId;
tcpClientPackage.createStream();
result = tcpOutputStream;
- tcpOutputStream.getByteBuf();
+ tcpOutputStream.getBuffer();
result = byteBuf;
}
};
@@ -172,7 +172,7 @@ public class TestTcpClientConnection {
{
tcpClientPackage.getMsgId();
result = msgId;
- tcpClientConnection.write((ByteBuf) any);
+ tcpClientConnection.write((Buffer) any);
}
};
new MockUp<Context>(context) {
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestStandardHttpServletRequestEx.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestStandardHttpServletRequestEx.java
index d0adef2ce..15e52aa5f 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestStandardHttpServletRequestEx.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestStandardHttpServletRequestEx.java
@@ -117,7 +117,7 @@ public class TestStandardHttpServletRequestEx {
inherited.put("p1", v1);
Buffer buffer = Buffer.buffer("p1=v1-3;p2=v2");
- BufferInputStream inputStream = new BufferInputStream(buffer.getByteBuf());
+ BufferInputStream inputStream = new BufferInputStream(buffer);
new Expectations() {
{
request.getParameterMap();
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/stream/TestBufferInputStream.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/stream/TestBufferInputStream.java
index 070fe207d..40be5d486 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/stream/TestBufferInputStream.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/stream/TestBufferInputStream.java
@@ -29,8 +29,7 @@ import org.junit.Test;
import org.junit.jupiter.api.Assertions;
import org.mockito.Mockito;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
+import io.vertx.core.buffer.Buffer;
public class TestBufferInputStream {
@@ -38,7 +37,7 @@ public class TestBufferInputStream {
@Before
public void setUp() throws Exception {
- ByteBuf buffer = Mockito.mock(ByteBuf.class);
+ Buffer buffer = Mockito.mock(Buffer.class);
instance = new BufferInputStream(buffer);
}
@@ -60,8 +59,8 @@ public class TestBufferInputStream {
gzipOutputStream.write(text.getBytes());
gzipOutputStream.close();
- ByteBuf buffer = Unpooled.buffer();
- buffer.writeBytes(out.toByteArray());
+ Buffer buffer = Buffer.buffer();
+ buffer.appendBytes(out.toByteArray());
out.close();
BufferInputStream bufferInputStream = new BufferInputStream(buffer);
GZIPInputStream gzipInputStream = new GZIPInputStream(bufferInputStream);
diff --git a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayProducerInvocationFlow.java b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayProducerInvocationFlow.java
index 36e0af598..f973f9e1c 100644
--- a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayProducerInvocationFlow.java
+++ b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayProducerInvocationFlow.java
@@ -58,6 +58,6 @@ public class HighwayProducerInvocationFlow extends ProducerInvocationFlow {
@Override
protected void sendResponse(Invocation invocation, Response response) {
HighwayTransportContext transportContext = invocation.getTransportContext();
- connection.write(transportContext.getResponseBuffer().getByteBuf());
+ connection.write(transportContext.getResponseBuffer());
}
}
(servicecomb-java-chassis) 01/02: [SCB-2885]generate content-type text/event-stream
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
liubao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git
commit 050a77d320a11a8e108145f45306a10799198f33
Author: liubao <bi...@qq.com>
AuthorDate: Fri Jun 21 16:19:02 2024 +0800
[SCB-2885]generate content-type text/event-stream
---
demo/demo-zookeeper/provider/pom.xml | 4 ++
.../samples/ReactiveStreamController.java | 73 ++++++++++++++++++++++
dependencies/default/pom.xml | 13 ++++
swagger/swagger-generator/generator-core/pom.xml | 4 ++
.../generator/core/SwaggerGeneratorContext.java | 3 +-
.../processor/response/PublisherProcessor.java | 46 ++++++++++++++
...icecomb.swagger.generator.ResponseTypeProcessor | 1 +
7 files changed, 143 insertions(+), 1 deletion(-)
diff --git a/demo/demo-zookeeper/provider/pom.xml b/demo/demo-zookeeper/provider/pom.xml
index 8f32376ae..0bd66e9ca 100644
--- a/demo/demo-zookeeper/provider/pom.xml
+++ b/demo/demo-zookeeper/provider/pom.xml
@@ -46,6 +46,10 @@
<groupId>org.apache.servicecomb</groupId>
<artifactId>config-zookeeper</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.reactivex.rxjava3</groupId>
+ <artifactId>rxjava</artifactId>
+ </dependency>
</dependencies>
<build>
diff --git a/demo/demo-zookeeper/provider/src/main/java/org/apache/servicecomb/samples/ReactiveStreamController.java b/demo/demo-zookeeper/provider/src/main/java/org/apache/servicecomb/samples/ReactiveStreamController.java
new file mode 100644
index 000000000..b72187b09
--- /dev/null
+++ b/demo/demo-zookeeper/provider/src/main/java/org/apache/servicecomb/samples/ReactiveStreamController.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.servicecomb.samples;
+
+
+import org.apache.servicecomb.provider.rest.common.RestSchema;
+import org.reactivestreams.Publisher;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import io.reactivex.rxjava3.core.Flowable;
+
+@RestSchema(schemaId = "ReactiveStreamController")
+@RequestMapping(path = "/")
+public class ReactiveStreamController {
+ public static class Model {
+ private String name;
+
+ private int age;
+
+ public Model() {
+
+ }
+
+ public Model(String name, int age) {
+ this.name = name;
+ this.age = age;
+ }
+
+ public int getAge() {
+ return age;
+ }
+
+ public Model setAge(int age) {
+ this.age = age;
+ return this;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public Model setName(String name) {
+ this.name = name;
+ return this;
+ }
+ }
+
+ @GetMapping("/sseString")
+ public Publisher<String> sseString() {
+ return Flowable.fromArray("a", "b", "c");
+ }
+
+ @GetMapping("/sseModel")
+ public Publisher<Model> sseModel() {
+ return Flowable.fromArray(new Model("a", 1), new Model("b", 2),
+ new Model("c", 3));
+ }
+}
diff --git a/dependencies/default/pom.xml b/dependencies/default/pom.xml
index d6f708c3e..ec9d645d1 100644
--- a/dependencies/default/pom.xml
+++ b/dependencies/default/pom.xml
@@ -77,8 +77,10 @@
<protobuf.version>3.23.4</protobuf.version>
<protostuff.version>1.8.0</protostuff.version>
<protostuff-parser.version>2.2.27</protostuff-parser.version>
+ <reactive-streams.version>1.0.4</reactive-streams.version>
<resilience4j.versioin>1.7.0</resilience4j.versioin>
<ribbon.version>2.7.18</ribbon.version>
+ <rxjava.version>3.1.8</rxjava.version>
<seanyinx.version>1.0.0</seanyinx.version>
<servo.version>0.13.2</servo.version>
<servlet-api.version>6.0.0</servlet-api.version>
@@ -285,6 +287,17 @@
<version>${swagger.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.reactivex.rxjava3</groupId>
+ <artifactId>rxjava</artifactId>
+ <version>${rxjava.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.reactivestreams</groupId>
+ <artifactId>reactive-streams</artifactId>
+ <version>${reactive-streams.version}</version>
+ </dependency>
+
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-codegen</artifactId>
diff --git a/swagger/swagger-generator/generator-core/pom.xml b/swagger/swagger-generator/generator-core/pom.xml
index 5f4c902c8..b48dd2281 100644
--- a/swagger/swagger-generator/generator-core/pom.xml
+++ b/swagger/swagger-generator/generator-core/pom.xml
@@ -43,6 +43,10 @@
<groupId>jakarta.servlet</groupId>
<artifactId>jakarta.servlet-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.reactivestreams</groupId>
+ <artifactId>reactive-streams</artifactId>
+ </dependency>
<dependency>
<groupId>org.mockito</groupId>
diff --git a/swagger/swagger-generator/generator-core/src/main/java/org/apache/servicecomb/swagger/generator/core/SwaggerGeneratorContext.java b/swagger/swagger-generator/generator-core/src/main/java/org/apache/servicecomb/swagger/generator/core/SwaggerGeneratorContext.java
index 837499dad..63d2259e6 100644
--- a/swagger/swagger-generator/generator-core/src/main/java/org/apache/servicecomb/swagger/generator/core/SwaggerGeneratorContext.java
+++ b/swagger/swagger-generator/generator-core/src/main/java/org/apache/servicecomb/swagger/generator/core/SwaggerGeneratorContext.java
@@ -30,7 +30,8 @@ import jakarta.ws.rs.core.MediaType;
public class SwaggerGeneratorContext {
protected static final List<String> SUPPORTED_CONTENT_TYPE
= Arrays.asList(MediaType.APPLICATION_JSON, SwaggerConst.PROTOBUF_TYPE, MediaType.TEXT_PLAIN,
- MediaType.MULTIPART_FORM_DATA, MediaType.APPLICATION_FORM_URLENCODED);
+ MediaType.MULTIPART_FORM_DATA, MediaType.APPLICATION_FORM_URLENCODED,
+ MediaType.SERVER_SENT_EVENTS);
protected static final List<String> SUPPORTED_BODY_CONTENT_TYPE
= Arrays.asList(MediaType.APPLICATION_JSON, SwaggerConst.PROTOBUF_TYPE, MediaType.TEXT_PLAIN);
diff --git a/swagger/swagger-generator/generator-core/src/main/java/org/apache/servicecomb/swagger/generator/core/processor/response/PublisherProcessor.java b/swagger/swagger-generator/generator-core/src/main/java/org/apache/servicecomb/swagger/generator/core/processor/response/PublisherProcessor.java
new file mode 100644
index 000000000..8388477dc
--- /dev/null
+++ b/swagger/swagger-generator/generator-core/src/main/java/org/apache/servicecomb/swagger/generator/core/processor/response/PublisherProcessor.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.servicecomb.swagger.generator.core.processor.response;
+
+import java.lang.reflect.Type;
+import java.util.List;
+
+import org.apache.servicecomb.swagger.generator.OperationGenerator;
+import org.apache.servicecomb.swagger.generator.SwaggerGenerator;
+import org.reactivestreams.Publisher;
+
+import io.swagger.v3.oas.models.media.Schema;
+import jakarta.ws.rs.core.MediaType;
+
+public class PublisherProcessor extends DefaultResponseTypeProcessor {
+ public PublisherProcessor() {
+ extractActualType = true;
+ }
+
+ @Override
+ public Type getProcessType() {
+ return Publisher.class;
+ }
+
+ @Override
+ public Schema<?> process(SwaggerGenerator swaggerGenerator, OperationGenerator operationGenerator,
+ Type genericResponseType) {
+ operationGenerator.getOperationGeneratorContext()
+ .updateProduces(List.of(MediaType.SERVER_SENT_EVENTS));
+ return super.process(swaggerGenerator, operationGenerator, genericResponseType);
+ }
+}
diff --git a/swagger/swagger-generator/generator-core/src/main/resources/META-INF/services/org.apache.servicecomb.swagger.generator.ResponseTypeProcessor b/swagger/swagger-generator/generator-core/src/main/resources/META-INF/services/org.apache.servicecomb.swagger.generator.ResponseTypeProcessor
index 53678b702..9abf37e8c 100644
--- a/swagger/swagger-generator/generator-core/src/main/resources/META-INF/services/org.apache.servicecomb.swagger.generator.ResponseTypeProcessor
+++ b/swagger/swagger-generator/generator-core/src/main/resources/META-INF/services/org.apache.servicecomb.swagger.generator.ResponseTypeProcessor
@@ -17,3 +17,4 @@
org.apache.servicecomb.swagger.generator.core.processor.response.CompletableFutureProcessor
org.apache.servicecomb.swagger.generator.core.processor.response.OptionalProcessor
+org.apache.servicecomb.swagger.generator.core.processor.response.PublisherProcessor