@Test public void clearOnTimeout() { AtomicReference<SerializedCommandResponse> responseAtomicReference = new AtomicReference<>(); testSubject.put("1234", new CommandInformation("1234", "Source", "Target", responseAtomicReference::set, new ClientStreamIdentification("context", "client"), "component")); clock.timeElapses(100000); testSubject.clearOnTimeout(); assertNotNull(responseAtomicReference.get()); }
@Scheduled(fixedDelayString = "${axoniq.axonserver.cache-close-rate:5000}") public void clearOnTimeout() { logger.debug("Checking timed out commands"); long minTimestamp = clock.millis() - defaultCommandTimeout; Set<Entry<String, CommandInformation>> toDelete = entrySet().stream().filter(e -> e.getValue().getTimestamp() < minTimestamp).collect( Collectors.toSet()); if( ! toDelete.isEmpty()) { logger.warn("Found {} waiting commands to delete", toDelete.size()); toDelete.forEach(e -> { logger.warn("Cancelling command {} sent by {}, waiting for reply from {}", e.getValue().getRequestIdentifier(), e.getValue().getSourceClientId(), e.getValue().getClientStreamIdentification()); remove(e.getKey()); e.getValue().cancel(); }); } }
CommandCache extends ConcurrentHashMap<String, CommandInformation> { @Scheduled(fixedDelayString = "${axoniq.axonserver.cache-close-rate:5000}") public void clearOnTimeout() { logger.debug("Checking timed out commands"); long minTimestamp = clock.millis() - defaultCommandTimeout; Set<Entry<String, CommandInformation>> toDelete = entrySet().stream().filter(e -> e.getValue().getTimestamp() < minTimestamp).collect( Collectors.toSet()); if( ! toDelete.isEmpty()) { logger.warn("Found {} waiting commands to delete", toDelete.size()); toDelete.forEach(e -> { logger.warn("Cancelling command {} sent by {}, waiting for reply from {}", e.getValue().getRequestIdentifier(), e.getValue().getSourceClientId(), e.getValue().getClientStreamIdentification()); remove(e.getKey()); e.getValue().cancel(); }); } } }
CommandCache extends ConcurrentHashMap<String, CommandInformation> { @Scheduled(fixedDelayString = "${axoniq.axonserver.cache-close-rate:5000}") public void clearOnTimeout() { logger.debug("Checking timed out commands"); long minTimestamp = clock.millis() - defaultCommandTimeout; Set<Entry<String, CommandInformation>> toDelete = entrySet().stream().filter(e -> e.getValue().getTimestamp() < minTimestamp).collect( Collectors.toSet()); if( ! toDelete.isEmpty()) { logger.warn("Found {} waiting commands to delete", toDelete.size()); toDelete.forEach(e -> { logger.warn("Cancelling command {} sent by {}, waiting for reply from {}", e.getValue().getRequestIdentifier(), e.getValue().getSourceClientId(), e.getValue().getClientStreamIdentification()); remove(e.getKey()); e.getValue().cancel(); }); } } @Autowired  CommandCache(@Value("${axoniq.axonserver.default-command-timeout:300000}") long defaultCommandTimeout,
                        Clock clock, @Value("${axoniq.axonserver.command-cache-capacity:0}") long cacheCapacity);  CommandCache(Clock clock); }
CommandCache extends ConcurrentHashMap<String, CommandInformation> { @Scheduled(fixedDelayString = "${axoniq.axonserver.cache-close-rate:5000}") public void clearOnTimeout() { logger.debug("Checking timed out commands"); long minTimestamp = clock.millis() - defaultCommandTimeout; Set<Entry<String, CommandInformation>> toDelete = entrySet().stream().filter(e -> e.getValue().getTimestamp() < minTimestamp).collect( Collectors.toSet()); if( ! toDelete.isEmpty()) { logger.warn("Found {} waiting commands to delete", toDelete.size()); toDelete.forEach(e -> { logger.warn("Cancelling command {} sent by {}, waiting for reply from {}", e.getValue().getRequestIdentifier(), e.getValue().getSourceClientId(), e.getValue().getClientStreamIdentification()); remove(e.getKey()); e.getValue().cancel(); }); } } @Autowired  CommandCache(@Value("${axoniq.axonserver.default-command-timeout:300000}") long defaultCommandTimeout,
                        Clock clock, @Value("${axoniq.axonserver.command-cache-capacity:0}") long cacheCapacity);  CommandCache(Clock clock); @Scheduled(fixedDelayString = "${axoniq.axonserver.cache-close-rate:5000}") void clearOnTimeout(); @Override CommandInformation put(@Nonnull String key, @Nonnull CommandInformation value); @Override CommandInformation putIfAbsent(String key, CommandInformation value); @Override void putAll(Map<? extends String, ? extends CommandInformation> m); }
CommandCache extends ConcurrentHashMap<String, CommandInformation> { @Scheduled(fixedDelayString = "${axoniq.axonserver.cache-close-rate:5000}") public void clearOnTimeout() { logger.debug("Checking timed out commands"); long minTimestamp = clock.millis() - defaultCommandTimeout; Set<Entry<String, CommandInformation>> toDelete = entrySet().stream().filter(e -> e.getValue().getTimestamp() < minTimestamp).collect( Collectors.toSet()); if( ! toDelete.isEmpty()) { logger.warn("Found {} waiting commands to delete", toDelete.size()); toDelete.forEach(e -> { logger.warn("Cancelling command {} sent by {}, waiting for reply from {}", e.getValue().getRequestIdentifier(), e.getValue().getSourceClientId(), e.getValue().getClientStreamIdentification()); remove(e.getKey()); e.getValue().cancel(); }); } } @Autowired  CommandCache(@Value("${axoniq.axonserver.default-command-timeout:300000}") long defaultCommandTimeout,
                        Clock clock, @Value("${axoniq.axonserver.command-cache-capacity:0}") long cacheCapacity);  CommandCache(Clock clock); @Scheduled(fixedDelayString = "${axoniq.axonserver.cache-close-rate:5000}") void clearOnTimeout(); @Override CommandInformation put(@Nonnull String key, @Nonnull CommandInformation value); @Override CommandInformation putIfAbsent(String key, CommandInformation value); @Override void putAll(Map<? extends String, ? extends CommandInformation> m);  }