readHandler) {
@@ -43,7 +44,7 @@ public void accept(int[] data) {
if (ptr > prev) {
sendChunk(data, prev, ptr);
}
- readHandler.accept(new int[]{'\r','\n'});
+ readHandler.accept(CRLF);
prev = ++ptr;
} else {
ptr++;
diff --git a/src/main/java/io/termd/core/util/ByteBufPool.java b/src/main/java/io/termd/core/util/ByteBufPool.java
new file mode 100644
index 0000000..92e1d36
--- /dev/null
+++ b/src/main/java/io/termd/core/util/ByteBufPool.java
@@ -0,0 +1,124 @@
+package io.termd.core.util;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * 简单的ByteBuf池，比netty的PooledByteBufAllocator更轻量，内存碎片少
+ *
+ * //get buf
+ * final ByteBuf byteBuf = byteBufPool.get(50, TimeUnit.MILLISECONDS);
+ * boolean done = false;
+ * try {
+ * //write bytes
+ * byteBuf.writeBytes(buffer, start, size);
+ * if (context != null) {
+ * context.writeAndFlush(new TextWebSocketFrame(byteBuf)).addListener(new ChannelFutureListener() {
+ * public void operationComplete(ChannelFuture future) throws Exception {
+ * //give back
+ * byteBufPool.put(byteBuf);
+ * }
+ * });
+ * done = true;
+ * }
+ * } finally {
+ * if (!done) {
+ * //discard
+ * byteBufPool.discard(byteBuf);
+ * }
+ * }
+ *
+ * //onClose
+ * protected void onClose() {
+ * //...
+ * byteBufPool.release();
+ * }
+ *
+ * @author gongdewei 2020/5/19
+ */
+public class ByteBufPool {
+
+ private int byteBufCapacity;// = 128;
+
+ //pool size: 128 * 1024 = 128K
+ private int poolSize;// = 1024;
+ private AtomicInteger allocSize = new AtomicInteger();
+ private BlockingQueue byteBufPool;
+ private AtomicBoolean closed = new AtomicBoolean(false);
+
+ public ByteBufPool() {
+ this(1024, 128);
+ }
+
+ public ByteBufPool(int poolSize, int byteBufCapacity) {
+ this.byteBufCapacity = byteBufCapacity;
+ this.poolSize = poolSize;
+ byteBufPool = new ArrayBlockingQueue(poolSize);
+ }
+
+ /**
+ * Apply a fixed capacity ByteBuf
+ * @return
+ */
+ public ByteBuf get(long timeout, TimeUnit unit) {
+ if (closed.get()) {
+ throw new IllegalStateException("ByteBufPool is closed");
+ }
+ ByteBuf byteBuf = byteBufPool.poll();
+ if (byteBuf == null && allocSize.get() >= poolSize) {
+ //wait if pool is full alloc
+ try {
+ byteBuf = byteBufPool.poll(timeout, unit);
+ } catch (InterruptedException e) {
+ //ignore
+ }
+ }
+ if (byteBuf == null) {
+ //heap buffer
+ byteBuf = Unpooled.buffer(byteBufCapacity);
+ allocSize.incrementAndGet();
+ }
+ return byteBuf.retain();
+ }
+
+ /**
+ * Give back a ByteBuf
+ * @param byteBuf
+ */
+ public void put(ByteBuf byteBuf) {
+ // give back
+ byteBuf.clear();
+ if (byteBuf.capacity() != byteBufCapacity || !byteBufPool.offer(byteBuf)) {
+ //buf pool is full or capacity not match
+ byteBuf.release();
+ allocSize.decrementAndGet();
+ }
+ }
+
+ /**
+ * Discard a ByteBuf, call it when there is a problem/exception and you don’t know how to recycle it safely.
+ * @param byteBuf
+ */
+ public void discard(ByteBuf byteBuf) {
+ //just release refCnt
+ byteBuf.release();
+ allocSize.decrementAndGet();
+ }
+
+ /**
+ * Release all cached ByteBuf, call it when shutdown/stop service
+ */
+ public void release() {
+ closed.set(true);
+ while (!byteBufPool.isEmpty()) {
+ ByteBuf byteBuf = byteBufPool.poll();
+ byteBuf.release();
+ }
+ }
+}
diff --git a/src/main/java/io/termd/core/util/Helper.java b/src/main/java/io/termd/core/util/Helper.java
index 7e85f1d..bc5da38 100644
--- a/src/main/java/io/termd/core/util/Helper.java
+++ b/src/main/java/io/termd/core/util/Helper.java
@@ -19,10 +19,10 @@
import io.termd.core.function.Consumer;
import io.termd.core.function.IntConsumer;
+import java.nio.IntBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
-import java.util.LinkedList;
import java.util.List;
import java.util.ServiceLoader;
@@ -53,21 +53,55 @@ public static void noop() {}
* @return the code points
*/
public static int[] toCodePoints(String s) {
- List codePoints = new ArrayList();
- for (int offset = 0; offset < s.length();) {
+ int count = Character.codePointCount(s, 0, s.length());
+ int[] codePoints = new int[count];
+ for (int offset = 0, i = 0; i < count && offset < s.length();i++) {
int cp = s.codePointAt(offset);
- codePoints.add(cp);
+ codePoints[i] = cp;
offset += Character.charCount(cp);
}
- return convert(codePoints);
+ return codePoints;
+
+// List codePoints = new ArrayList();
+// for (int offset = 0; offset < s.length();) {
+// int cp = s.codePointAt(offset);
+// codePoints.add(cp);
+// offset += Character.charCount(cp);
+// }
+// return convert(codePoints);
}
/**
- * Code point to string conversion.
+ * Convert the string to an array of code points.
+ * Notice: Ensure out buffer capacity >= Helper.codePointCount(s)
*
- * @param codePoints the code points
- * @return the corresponding string
+ * @param s the string to convert
+ * @param out the buffer to write code points
+ */
+ public static void toCodePoints(String s, IntBuffer out) {
+ //Assert out.remaining() >= codePointCount
+ for (int offset = 0; offset < s.length(); ) {
+ int cp = s.codePointAt(offset);
+ out.put(cp);
+ offset += Character.charCount(cp);
+ }
+ }
+
+ /**
+ * Get code points count of string
+ * @param s
+ * @return
*/
+ public static int codePointCount(String s) {
+ return Character.codePointCount(s, 0, s.length());
+ }
+
+ /**
+ * Code point to string conversion.
+ *
+ * @param codePoints the code points
+ * @return the corresponding string
+ */
public static String fromCodePoints(int[] codePoints) {
return new String(codePoints, 0, codePoints.length);
}
diff --git a/src/test/java/io/termd/core/http/websocket/server/WebSocketTtyConnection.java b/src/test/java/io/termd/core/http/websocket/server/WebSocketTtyConnection.java
index b54c080..5e18000 100644
--- a/src/test/java/io/termd/core/http/websocket/server/WebSocketTtyConnection.java
+++ b/src/test/java/io/termd/core/http/websocket/server/WebSocketTtyConnection.java
@@ -45,17 +45,17 @@ public class WebSocketTtyConnection extends HttpTtyConnection {
private Set readonlyChannels = new HashSet();
@Override
- protected void write(byte[] buffer) {
+ protected void write(byte[] buffer, int offset, int length) {
if (isOpen()) {
- sendBinary(buffer, webSocketChannel);
+ sendBinary(ByteBuffer.wrap(buffer, offset, length), webSocketChannel);
}
for (WebSocketChannel channel : readonlyChannels) {
- sendBinary(buffer, channel);
+ sendBinary(ByteBuffer.wrap(buffer, offset, length), channel);
}
}
- private void sendBinary(byte[] buffer, WebSocketChannel webSocketChannel) {
- WebSockets.sendBinary(ByteBuffer.wrap(buffer), webSocketChannel, null);
+ private void sendBinary(ByteBuffer buffer, WebSocketChannel webSocketChannel) {
+ WebSockets.sendBinary(buffer, webSocketChannel, null);
}
@Override