[#9666] Durable atc for delayed agent list query

Author: smilu97

realtime/realtime-web/src/main/java/com/navercorp/pinpoint/web/realtime/RealtimeWebConfig.java

@@ -30,7 +30,8 @@
 @Import({
         WebActiveThreadCountConfig.class,
         WebActiveThreadDumpConfig.class,
-        WebEchoConfig.class
+        WebEchoConfig.class,
+        RealtimeWebPropertySources.class
 })
 public class RealtimeWebConfig {
 

realtime/realtime-web/src/main/java/com/navercorp/pinpoint/web/realtime/RealtimeWebPropertySources.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2023 NAVER Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.navercorp.pinpoint.web.realtime;
+
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.context.annotation.PropertySources;
+
+/**
+ * @author youngjin.kim2
+ */
+@PropertySources({
+        @PropertySource(name = "RealtimeWebPropertySourceRoot", value = {RealtimeWebPropertySources.ROOT}),
+})
+public class RealtimeWebPropertySources {
+
+    public static final String ROOT = "classpath:pinpoint-web-realtime.properties";
+
+}

realtime/realtime-web/src/main/java/com/navercorp/pinpoint/web/realtime/activethread/count/dao/ConstantFetcher.java renamed to realtime/realtime-web/src/main/java/com/navercorp/pinpoint/web/realtime/activethread/count/dao/ActiveThreadCountDao.java

@@ -15,20 +15,15 @@
  */
 package com.navercorp.pinpoint.web.realtime.activethread.count.dao;
 
+import com.navercorp.pinpoint.common.server.cluster.ClusterKey;
+import com.navercorp.pinpoint.realtime.dto.ATCSupply;
+import reactor.core.publisher.Flux;
+
 /**
  * @author youngjin.kim2
  */
-class ConstantFetcher<T> implements Fetcher<T>{
-
-    private final T value;
-
-    ConstantFetcher(T value) {
-        this.value = value;
-    }
+public interface ActiveThreadCountDao {
 
-    @Override
-    public T fetch() {
-        return value;
-    }
+    Flux<ATCSupply> getSupplies(ClusterKey agentKey);
 
 }

realtime/realtime-web/src/main/java/com/navercorp/pinpoint/web/realtime/activethread/count/dao/ActiveThreadCountFetcherFactory.java

@@ -15,60 +15,32 @@
  */
 package com.navercorp.pinpoint.web.realtime.activethread.count.dao;
 
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import com.navercorp.pinpoint.common.server.cluster.ClusterKey;
 import com.navercorp.pinpoint.pubsub.endpoint.PubSubClientFactory;
 import com.navercorp.pinpoint.pubsub.endpoint.PubSubFluxClient;
 import com.navercorp.pinpoint.realtime.RealtimePubSubServiceDescriptors;
 import com.navercorp.pinpoint.realtime.dto.ATCDemand;
 import com.navercorp.pinpoint.realtime.dto.ATCSupply;
 import com.navercorp.pinpoint.redis.pubsub.RedisPubSubConfig;
 import com.navercorp.pinpoint.web.realtime.RealtimeWebCommonConfig;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
-import java.util.concurrent.TimeUnit;
-
 /**
  * @author youngjin.kim2
  */
 @Configuration
 @Import({ RealtimeWebCommonConfig.class, RedisPubSubConfig.class })
 public class ActiveThreadCountWebDaoConfig {
 
-    @Value("${pinpoint.web.realtime.atc.supply.expireInMs:3000}")
-    private long supplyExpireInMs;
-
-    @Value("${pinpoint.web.realtime.atc.supply.prepareInMs:10000}")
-    private long prepareInMs;
-
     @Bean
     PubSubFluxClient<ATCDemand, ATCSupply> atcEndpoint(PubSubClientFactory clientFactory) {
         return clientFactory.build(RealtimePubSubServiceDescriptors.ATC);
     }
 
     @Bean
-    Cache<ClusterKey, Fetcher<ATCSupply>> fetcherCache() {
-        return CacheBuilder.newBuilder()
-                .expireAfterAccess(1, TimeUnit.HOURS)
-                .initialCapacity(512)
-                .maximumSize(65536)
-                .build();
-    }
-
-    @Bean
-    FetcherFactory<ClusterKey, ATCSupply> atcSupplyFetcherFactory(
-            PubSubFluxClient<ATCDemand, ATCSupply> endpoint,
-            Cache<ClusterKey, Fetcher<ATCSupply>> fetcherCache
-    ) {
-        final long recordMaxAgeNanos = TimeUnit.MILLISECONDS.toNanos(supplyExpireInMs);
-        final long prepareInNanos = TimeUnit.MILLISECONDS.toNanos(prepareInMs);
-        final ActiveThreadCountFetcherFactory fetcherFactory
-                = new ActiveThreadCountFetcherFactory(endpoint, recordMaxAgeNanos, prepareInNanos);
-        return new CachedFetcherFactory<>(fetcherFactory, fetcherCache);
+    ActiveThreadCountDao activeThreadCountDao(PubSubFluxClient<ATCDemand, ATCSupply> client) {
+        return new PubSubActiveThreadCountDao(client);
     }
 
 }