PollingCorrelationManager extends DefaultCorrelationManager<T> { public T find(String correlationKey) { return find(correlationKey, endpointConfiguration.getTimeout()); } PollingCorrelationManager(PollableEndpointConfiguration endpointConfiguration, String retryLogMessage); T find(String correlationKey); @Override String getCorrelationKey(String correlationKeyName, TestContext context); @Override T find(String correlationKey, long timeout); String getRetryLogMessage();  }
@Test public void testFind() throws Exception { DirectSyncEndpointConfiguration pollableEndpointConfiguration = new DirectSyncEndpointConfiguration(); pollableEndpointConfiguration.setPollingInterval(100L); pollableEndpointConfiguration.setTimeout(500L); PollingCorrelationManager<String> correlationManager = new PollingCorrelationManager<>(pollableEndpointConfiguration, "Try again"); Assert.assertNull(correlationManager.find("")); correlationManager.store("foo", "bar"); Assert.assertNull(correlationManager.find("bar")); Assert.assertEquals(correlationManager.find("foo"), "bar"); Assert.assertNull(correlationManager.find("foo")); for (String key : new String[]{"1", "2", "3", "4", "5"}) { correlationManager.store(key, "value" + key); } for (String key : new String[]{"1", "5", "3", "2", "4"}) { Assert.assertEquals(correlationManager.find(key), "value" + key); Assert.assertNull(correlationManager.find(key)); } }
@Test public void testFindWithRetry() { DirectSyncEndpointConfiguration pollableEndpointConfiguration = new DirectSyncEndpointConfiguration(); pollableEndpointConfiguration.setPollingInterval(100L); pollableEndpointConfiguration.setTimeout(500L); PollingCorrelationManager<String> correlationManager = new PollingCorrelationManager<>(pollableEndpointConfiguration, "Try again"); correlationManager.setObjectStore(objectStore); reset(objectStore); when(objectStore.remove("foo")).thenReturn(null).thenReturn("bar"); Assert.assertEquals(correlationManager.find("foo"), "bar"); }
@Test public void testNotFindWithRetry() { DirectSyncEndpointConfiguration pollableEndpointConfiguration = new DirectSyncEndpointConfiguration(); pollableEndpointConfiguration.setPollingInterval(100L); pollableEndpointConfiguration.setTimeout(300L); PollingCorrelationManager<String> correlationManager = new PollingCorrelationManager<>(pollableEndpointConfiguration, "Try again"); correlationManager.setObjectStore(objectStore); reset(objectStore); when(objectStore.remove("foo")).thenReturn(null); Assert.assertNull(correlationManager.find("foo")); }