Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove reflective cache clear on Reactive sender (#479)

Browse Source 
As it turns out the closeable() implementation of the reactive sender
cache is properly clearing the cache therefore the reflective
manual clear is not needed.
This commit is contained in:
Chris Bono
2023-11-03 10:21:03 -05:00
committed by GitHub
parent e972b52385
commit 728a74bcb0
1 changed files with 0 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java
View File

@@ -35,7 +35,6 @@ import org.springframework.pulsar.core.DefaultTopicResolver;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ReflectionUtils;
/**
* Default implementation of {@link ReactivePulsarSenderFactory}.
@@ -173,7 +172,6 @@ public final class DefaultReactivePulsarSenderFactory<T>
@Override
public void doStop() {
try {
this.reflectivelyClearCache();
this.reactiveMessageSenderCache.close();
}
@@ -182,68 +180,6 @@ public final class DefaultReactivePulsarSenderFactory<T>
}
}
/**
* Workaround to reflectively clear the underlying producer cache.
*
* TODO: Remove once this is supported in the Reactive client.
*/
private void reflectivelyClearCache() {
if (this.reactiveMessageSenderCache == null) {
this.logger.trace(() -> "Cache is null - nothing to clear");
return;
}
// reactiveMessageSenderCache
// (org.apache.pulsar.reactive.client.internal.adapter.ProducerCache)
var cacheProviderField = ReflectionUtils.findField(this.reactiveMessageSenderCache.getClass(), "cacheProvider");
if (cacheProviderField == null) {
this.logger.trace(
() -> "Could not locate 'cacheProvider' field on sender cache: " + this.reactiveMessageSenderCache);
return;
}
ReflectionUtils.makeAccessible(cacheProviderField);
// org.apache.pulsar.reactive.client.producercache.CaffeineShadedProducerCacheProvider
var cacheProvider = ReflectionUtils.getField(cacheProviderField, this.reactiveMessageSenderCache);
if (cacheProvider == null) {
this.logger.trace(() -> "Cache provider was null on sender cache: " + this.reactiveMessageSenderCache);
return;
}
// org.apache.pulsar.reactive.shade.com.github.benmanes.caffeine.cache.BoundedLocalCache$BoundedLocalAsyncCache
var cacheField = ReflectionUtils.findField(cacheProvider.getClass(), "cache");
if (cacheField == null) {
this.logger.trace(() -> "Could not locate 'cache' field on cache provider: " + cacheProvider);
return;
}
ReflectionUtils.makeAccessible(cacheField);
var cache = ReflectionUtils.getField(cacheField, cacheProvider);
if (cacheField == null) {
this.logger.trace(() -> "Cache impl was null on cache provider: " + cacheProvider);
return;
}
// org.apache.pulsar.reactive.shade.com.github.benmanes.caffeine.cache.SSLMSAW
var actualCacheField = ReflectionUtils.findField(cache.getClass(), "cache");
if (actualCacheField == null) {
this.logger.trace(() -> "Could not locate 'cache' field on cache impl: " + cache);
return;
}
ReflectionUtils.makeAccessible(actualCacheField);
var actualCache = ReflectionUtils.getField(actualCacheField, cache);
if (actualCache == null) {
this.logger.trace(() -> "Actual cache was null on cache impl: " + cache);
return;
}
var clearMethod = ReflectionUtils.findMethod(actualCache.getClass(), "clear");
if (clearMethod == null) {
this.logger.trace(() -> "Could not locate 'clear' method on actual cache: " + actualCache);
return;
}
ReflectionUtils.makeAccessible(clearMethod);
ReflectionUtils.invokeMethod(clearMethod, actualCache);
}
/**
* Builder for {@link DefaultReactivePulsarSenderFactory}.
*