natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test register callback once

This commit is contained in:
ken.lj 2019-06-26 15:56:20 +08:00
parent 6256ec303d
commit 13c01efd12
1 changed files with 59 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/ProtocolFilterWrapper.java
View File

@ -48,9 +48,12 @@ public class ProtocolFilterWrapper implements Protocol {
this.protocol = protocol;
}
private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
Invoker<T> last = invoker;
List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
if (!filters.isEmpty()) {
for (int i = filters.size() - 1; i >= 0; i--) {
final Filter filter = filters.get(i);
@ -87,18 +90,7 @@ public class ProtocolFilterWrapper implements Protocol {
}
throw e;
}
return asyncResult.thenApplyWithContext(r -> {
// onResponse callback
if (filter instanceof ListenableFilter) {
Filter.Listener listener = ((ListenableFilter) filter).listener();
if (listener != null) {
listener.onResponse(r, invoker, invocation);
}
} else {
filter.onResponse(r, invoker, invocation);
}
return r;
});
return asyncResult;
}
@Override
@ -113,7 +105,8 @@ public class ProtocolFilterWrapper implements Protocol {
};
}
}
return last;
return new CallbackRegistrationInvoker<>(last, filters);
}
@Override
@ -142,4 +135,57 @@ public class ProtocolFilterWrapper implements Protocol {
protocol.destroy();
}
static class CallbackRegistrationInvoker<T> implements Invoker<T> {
private final Invoker<T> filterInvoker;
private final List<Filter> filters;
public CallbackRegistrationInvoker(Invoker<T> filterInvoker, List<Filter> filters) {
this.filterInvoker = filterInvoker;
this.filters = filters;
}
@Override
public Result invoke(Invocation invocation) throws RpcException {
Result asyncResult = filterInvoker.invoke(invocation);
asyncResult.thenApplyWithContext(r -> {
for (int i = filters.size() - 1; i >= 0; i--) {
Filter filter = filters.get(i);
// onResponse callback
if (filter instanceof ListenableFilter) {
Filter.Listener listener = ((ListenableFilter) filter).listener();
if (listener != null) {
listener.onResponse(r, filterInvoker, invocation);
}
} else {
filter.onResponse(r, filterInvoker, invocation);
}
}
return r;
});
return asyncResult;
}
@Override
public Class<T> getInterface() {
return filterInvoker.getInterface();
}
@Override
public URL getUrl() {
return filterInvoker.getUrl();
}
@Override
public boolean isAvailable() {
return filterInvoker.isAvailable();
}
@Override
public void destroy() {
filterInvoker.destroy();
}
}
}