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

增加ClientMessageListener功能

Browse Source
This commit is contained in:
redkale
2023-09-08 13:09:02 +08:00
parent ddd3d33edc
commit 4c3f919238
5 changed files with 137 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/redkale/net/client/Client.java
View File

@@ -303,7 +303,7 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
}
public final CompletableFuture<C> newConnection() {
return connect(true);
return connect(false);
}
private CompletableFuture<C> connect(final boolean pool) {

168
src/main/java/org/redkale/net/client/ClientCodec.java
View File

@@ -27,14 +27,16 @@ import org.redkale.util.*;
*/
public abstract class ClientCodec<R extends ClientRequest, P> implements CompletionHandler<Integer, ByteBuffer> {
protected final ClientConnection<R, P> connection;
private final List<ClientResponse<R, P>> respResults = new ArrayList<>();
private final ByteArray readArray = new ByteArray();
private final ObjectPool<ClientResponse<R, P>> respPool = ObjectPool.createUnsafePool(256, t -> new ClientResponse(), ClientResponse::prepare, ClientResponse::recycle);
protected final ClientConnection<R, P> connection;
protected ClientMessageListener messageListener;
public ClientCodec(ClientConnection<R, P> connection) {
Objects.requireNonNull(connection);
this.connection = connection;
@@ -42,6 +44,11 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
public abstract void decodeMessages(ByteBuffer buffer, ByteArray array);
public ClientCodec<R, P> withMessageListener(ClientMessageListener listener) {
this.messageListener = listener;
return this;
}
@Override
public final void completed(Integer count, ByteBuffer attachment) {
AsyncConnection channel = connection.channel;
@@ -69,8 +76,11 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
for (ClientResponse<R, P> cr : respResults) {
connection.doneResponseCounter.increment();
if (cr.isError()) {
connection.dispose(cr.exc);
connection.dispose(cr.cause);
return;
} else if (messageListener != null) {
messageListener.onMessage(connection, cr);
respPool.accept(cr);
} else {
ClientFuture<R, P> respFuture = connection.pollRespFuture(cr.getRequestid());
if (respFuture != null) {
@@ -78,7 +88,7 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
connection.dispose(new RedkaleException("request pipeline error"));
return;
}
responseComplete(false, respFuture, cr.message, cr.exc);
responseComplete(false, respFuture, cr.message, cr.cause);
}
respPool.accept(cr);
}
@@ -101,101 +111,99 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
}
void responseComplete(boolean halfCompleted, ClientFuture<R, P> respFuture, P message, Throwable exc) {
if (respFuture != null) {
R request = respFuture.request;
AsyncIOThread readThread = connection.channel.getReadIOThread();
final WorkThread workThread = request.workThread;
try {
if (!halfCompleted && !request.isCompleted()) {
if (exc == null) {
connection.sendHalfWriteInReadThread(request, exc);
//request没有发送完，respFuture需要再次接收
return;
} else {
connection.sendHalfWriteInReadThread(request, exc);
//异常了需要清掉半包
}
}
connection.respWaitingCounter.decrement();
if (connection.isAuthenticated()) {
connection.client.incrRespDoneCounter();
}
respFuture.cancelTimeout();
// if (connection.client.debug) {
// connection.client.logger.log(Level.FINEST, Utility.nowMillis() + ": " + Thread.currentThread().getName() + ": " + connection + ", 回调处理, req=" + request + ", message=" + message, exc);
// }
connection.preComplete(message, (R) request, exc);
R request = respFuture.request;
AsyncIOThread readThread = connection.channel.getReadIOThread();
final WorkThread workThread = request.workThread;
try {
if (!halfCompleted && !request.isCompleted()) {
if (exc == null) {
final P rs = request.respTransfer == null ? message : (P) request.respTransfer.apply(message);
if (workThread == null) {
readThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.complete(rs);
});
} else if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
if (workThread.inIO()) {
Traces.computeIfAbsent(request.traceid);
respFuture.complete(rs);
} else {
workThread.execute(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.complete(rs);
});
}
} else {
workThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.complete(rs);
});
}
} else { //异常
if (workThread == null) {
readThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
});
} else if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
if (workThread.inIO()) {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
} else {
workThread.execute(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
});
}
} else {
workThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
});
}
connection.sendHalfWriteInReadThread(request, exc);
//request没有发送完，respFuture需要再次接收
return;
} else {
connection.sendHalfWriteInReadThread(request, exc);
//异常了需要清掉半包
}
} catch (Throwable t) {
}
connection.respWaitingCounter.decrement();
if (connection.isAuthenticated()) {
connection.client.incrRespDoneCounter();
}
respFuture.cancelTimeout();
// if (connection.client.debug) {
// connection.client.logger.log(Level.FINEST, Utility.nowMillis() + ": " + Thread.currentThread().getName() + ": " + connection + ", 回调处理, req=" + request + ", message=" + message, cause);
// }
connection.preComplete(message, (R) request, exc);
if (exc == null) {
final P rs = request.respTransfer == null ? message : (P) request.respTransfer.apply(message);
if (workThread == null) {
readThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
respFuture.complete(rs);
});
} else if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
if (workThread.inIO()) {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
respFuture.complete(rs);
} else {
workThread.execute(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
respFuture.complete(rs);
});
}
} else {
workThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.complete(rs);
});
}
} else { //异常
if (workThread == null) {
readThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
});
} else if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
if (workThread.inIO()) {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
} else {
workThread.execute(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
});
}
} else {
workThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(exc);
});
}
}
} catch (Throwable t) {
if (workThread == null) {
readThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
});
} else if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
if (workThread.inIO()) {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
} else {
workThread.execute(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
});
}
connection.client.logger.log(Level.INFO, "Complete result error, request: " + respFuture.request, t);
} else {
workThread.runWork(() -> {
Traces.computeIfAbsent(request.traceid);
respFuture.completeExceptionally(t);
});
}
connection.client.logger.log(Level.INFO, "Complete result error, request: " + respFuture.request, t);
}
}
@@ -204,6 +212,10 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
connection.dispose(t);
}
public ClientMessageListener getMessageListener() {
return messageListener;
}
protected R nextRequest() {
return connection.findRequest(null);
}

22
src/main/java/org/redkale/net/client/ClientConnection.java
View File

@@ -278,6 +278,10 @@ public abstract class ClientConnection<R extends ClientRequest, P> implements Co
} else if (connEntry != null) { //index=-1
connEntry.connOpenState.set(false);
}
ClientMessageListener listener = getCodec().getMessageListener();
if (listener != null) {
listener.onClose(this);
}
}
public void dispose(Throwable exc) {
@@ -330,15 +334,6 @@ public abstract class ClientConnection<R extends ClientRequest, P> implements Co
}
}
//只会被ClientCodec在ReadIOThread中调用
ClientFuture<R, P> pollRespFuture(Serializable requestid) {
if (requestid == null) {
return respFutureQueue.poll();
} else {
return respFutureMap.remove(requestid);
}
}
//只会被ClientCodec在ReadIOThread中调用
R findRequest(Serializable requestid) {
if (requestid == null) {
@@ -353,6 +348,15 @@ public abstract class ClientConnection<R extends ClientRequest, P> implements Co
}
}
//只会被ClientCodec在ReadIOThread中调用
protected ClientFuture<R, P> pollRespFuture(Serializable requestid) {
if (requestid == null) {
return respFutureQueue.poll();
} else {
return respFutureMap.remove(requestid);
}
}
public boolean isAuthenticated() {
return authenticated;
}

22
src/main/java/org/redkale/net/client/ClientMessageListener.java Normal file
View File

@@ -0,0 +1,22 @@
/*
*
*/
package org.redkale.net.client;
/**
* 接收消息事件
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*
* @since 2.8.0
*/
public abstract class ClientMessageListener {
public abstract void onMessage(ClientConnection conn, ClientResponse resp);
public void onClose(ClientConnection conn) {
}
}

22
src/main/java/org/redkale/net/client/ClientResponse.java
View File

@@ -25,7 +25,7 @@ public class ClientResponse<R extends ClientRequest, P> {
protected P message;
protected Throwable exc;
protected Throwable cause;
public ClientResponse() {
}
@@ -37,7 +37,7 @@ public class ClientResponse<R extends ClientRequest, P> {
public ClientResponse(R request, Throwable exc) {
this.request = request;
this.exc = exc;
this.cause = exc;
}
public Serializable getRequestid() {
@@ -52,20 +52,20 @@ public class ClientResponse<R extends ClientRequest, P> {
public ClientResponse<R, P> set(R request, Throwable exc) {
this.request = request;
this.exc = exc;
this.cause = exc;
return this;
}
protected void prepare() {
this.request = null;
this.message = null;
this.exc = null;
this.cause = null;
}
protected boolean recycle() {
this.request = null;
this.message = null;
this.exc = null;
this.cause = null;
return true;
}
@@ -85,18 +85,18 @@ public class ClientResponse<R extends ClientRequest, P> {
this.message = message;
}
public Throwable getExc() {
return exc;
public Throwable getCause() {
return cause;
}
public void setExc(Throwable exc) {
this.exc = exc;
public void setCause(Throwable cause) {
this.cause = cause;
}
@Override
public String toString() {
if (exc != null) {
return "{\"exc\":" + exc + "}";
if (cause != null) {
return "{\"exc\":" + cause + "}";
}
return "{\"message\":" + message + "}";
}