lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WebSocket增加sendMap方法

Browse Source
This commit is contained in:
Redkale
2017-10-16 11:19:58 +08:00
parent 84b4eee7b5
commit e0041235fe
7 changed files with 54 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/org/redkale/convert/BinaryConvert.java
View File

@@ -31,4 +31,6 @@ public abstract class BinaryConvert<R extends Reader, W extends Writer> extends
public abstract byte[] convertTo(final Object value);
public abstract byte[] convertTo(final Type type, final Object value);
public abstract byte[] convertMapTo(final Object... values);
}

1
src/org/redkale/convert/TextConvert.java
View File

@@ -32,4 +32,5 @@ public abstract class TextConvert<R extends Reader, W extends Writer> extends Co
public abstract String convertTo(final Type type, final Object value);
public abstract String convertMapTo(final Object... values);
}

1
src/org/redkale/convert/bson/BsonConvert.java
View File

@@ -161,6 +161,7 @@ public final class BsonConvert extends BinaryConvert<BsonReader, BsonWriter> {
return result;
}
@Override
public byte[] convertMapTo(final Object... values) {
if (values == null) return null;
final BsonWriter out = writerPool.get().tiny(tiny);

1
src/org/redkale/convert/json/JsonConvert.java
View File

@@ -146,6 +146,7 @@ public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
return result;
}
@Override
public String convertMapTo(final Object... values) {
if (values == null) return "null";
final JsonWriter out = writerPool.get().tiny(tiny);

46
src/org/redkale/net/http/WebSocket.java
View File

@@ -118,7 +118,18 @@ public abstract class WebSocket<G extends Serializable, T> {
* @return 0表示成功， 非0表示错误码
*/
public final CompletableFuture<Integer> send(Object message) {
return send(message, true);
return send(false, message, true);
}
/**
* 给自身发送消息, 消息类型是key-value键值对
*
* @param messages key-value键值对
*
* @return 0表示成功， 非0表示错误码
*/
public final CompletableFuture<Integer> sendMap(Object... messages) {
return send(true, messages, true);
}
/**
@@ -130,6 +141,31 @@ public abstract class WebSocket<G extends Serializable, T> {
* @return 0表示成功， 非0表示错误码
*/
public final CompletableFuture<Integer> send(Object message, boolean last) {
return send(false, message, last);
}
/**
* 给自身发送消息, 消息类型是key-value键值对
*
* @param last 是否最后一条
* @param messages key-value键值对
*
* @return 0表示成功， 非0表示错误码
*/
public final CompletableFuture<Integer> sendMap(boolean last, Object... messages) {
return send(true, messages, last);
}
/**
* 给自身发送消息, 消息类型是Object[]
*
* @param mapconvable 是否convertMapTo
* @param message 不可为空, 只能是String或byte[]或可JavaBean对象，或Object[]
* @param last 是否最后一条
*
* @return 0表示成功， 非0表示错误码
*/
private CompletableFuture<Integer> send(boolean mapconvable, Object message, boolean last) {
if (message instanceof CompletableFuture) {
return ((CompletableFuture) message).thenCompose((json) -> {
if (json == null || json instanceof CharSequence || json instanceof byte[]) {
@@ -137,7 +173,7 @@ public abstract class WebSocket<G extends Serializable, T> {
} else if (message instanceof WebSocketPacket) {
return sendPacket((WebSocketPacket) message);
} else {
return sendPacket(new WebSocketPacket(getSendConvert(), json, last));
return sendPacket(new WebSocketPacket(getSendConvert(), mapconvable, json, last));
}
});
}
@@ -146,7 +182,7 @@ public abstract class WebSocket<G extends Serializable, T> {
} else if (message instanceof WebSocketPacket) {
return sendPacket((WebSocketPacket) message);
} else {
return sendPacket(new WebSocketPacket(getSendConvert(), message, last));
return sendPacket(new WebSocketPacket(getSendConvert(), mapconvable, message, last));
}
}
@@ -173,9 +209,9 @@ public abstract class WebSocket<G extends Serializable, T> {
*/
public final CompletableFuture<Integer> send(Convert convert, Object message, boolean last) {
if (message instanceof CompletableFuture) {
return ((CompletableFuture) message).thenCompose((json) -> sendPacket(new WebSocketPacket(convert == null ? getSendConvert() : convert, json, last)));
return ((CompletableFuture) message).thenCompose((json) -> sendPacket(new WebSocketPacket(convert == null ? getSendConvert() : convert, false, json, last)));
}
return sendPacket(new WebSocketPacket(convert == null ? getSendConvert() : convert, message, last));
return sendPacket(new WebSocketPacket(convert == null ? getSendConvert() : convert, false, message, last));
}
/**

4
src/org/redkale/net/http/WebSocketEngine.java
View File

@@ -147,7 +147,7 @@ public class WebSocketEngine {
//此处的WebSocketPacket只能是包含payload或bytes内容的，不能包含sendConvert、sendJson、sendBuffers
final WebSocketPacket packet = (message instanceof WebSocketPacket) ? (WebSocketPacket) message
: ((message == null || message instanceof CharSequence || message instanceof byte[])
? new WebSocketPacket((Serializable) message, last) : new WebSocketPacket(this.sendConvert, message, last));
? new WebSocketPacket((Serializable) message, last) : new WebSocketPacket(this.sendConvert, false, message, last));
packet.setSendBuffers(packet.encode(context.getBufferSupplier()));
CompletableFuture<Integer> future = null;
if (single) {
@@ -194,7 +194,7 @@ public class WebSocketEngine {
//此处的WebSocketPacket只能是包含payload或bytes内容的，不能包含sendConvert、sendJson、sendBuffers
final WebSocketPacket packet = (message instanceof WebSocketPacket) ? (WebSocketPacket) message
: ((message == null || message instanceof CharSequence || message instanceof byte[])
? new WebSocketPacket((Serializable) message, last) : new WebSocketPacket(this.sendConvert, message, last));
? new WebSocketPacket((Serializable) message, last) : new WebSocketPacket(this.sendConvert, false, message, last));
packet.setSendBuffers(packet.encode(context.getBufferSupplier()));
CompletableFuture<Integer> future = null;
if (single) {

8
src/org/redkale/net/http/WebSocketPacket.java
View File

@@ -61,6 +61,8 @@ public final class WebSocketPacket {
Convert sendConvert;
boolean mapconvable;
ByteBuffer[] sendBuffers;
ConvertMask receiveMasker;
@@ -114,11 +116,13 @@ public final class WebSocketPacket {
this.last = fin;
}
WebSocketPacket(Convert convert, Object json, boolean fin) {
WebSocketPacket(Convert convert, boolean mapconvable, Object json, boolean fin) {
this.type = (convert == null || !convert.isBinary()) ? FrameType.TEXT : FrameType.BINARY;
this.sendConvert = convert;
this.mapconvable = mapconvable;
this.sendJson = json;
this.last = fin;
if (mapconvable && !(json instanceof Object[])) throw new IllegalArgumentException();
}
WebSocketPacket(ByteBuffer[] sendBuffers, FrameType type, boolean fin) {
@@ -207,7 +211,7 @@ public final class WebSocketPacket {
return supplier.get();
}
};
ByteBuffer[] buffers = this.sendConvert.convertTo(newsupplier, sendJson);
ByteBuffer[] buffers = this.mapconvable ? this.sendConvert.convertMapTo(supplier, (Object[]) sendJson) : this.sendConvert.convertTo(newsupplier, sendJson);
int len = 0;
for (ByteBuffer buf : buffers) {
len += buf.remaining();