SQSEventQueueProvider implements EventQueueProvider { @Override public ObservableQueue getQueue(String queueURI) { return queues.computeIfAbsent(queueURI, q -> { Builder builder = new SQSObservableQueue.Builder(); return builder.withBatchSize(this.batchSize) .withClient(client) .withPollTimeInMS(this.pollTimeInMS) .withQueueName(queueURI) .withVisibilityTimeout(this.visibilityTimeoutInSeconds) .withScheduler(scheduler) .build(); }); } @Inject  SQSEventQueueProvider(AmazonSQSClient client, Configuration config, Scheduler scheduler); @Override ObservableQueue getQueue(String queueURI);  }
@Test public void testGetQueueWithDefaultConfiguration() { when(configuration.getIntProperty(anyString(), anyInt())).thenAnswer(invocation -> invocation.getArguments()[1]); ListQueuesResult listQueuesResult = new ListQueuesResult().withQueueUrls("test_queue_1"); when(amazonSQSClient.listQueues(any(ListQueuesRequest.class))).thenReturn(listQueuesResult); SQSEventQueueProvider sqsEventQueueProvider = new SQSEventQueueProvider(amazonSQSClient, configuration, Schedulers.from(Executors.newSingleThreadExecutor())); SQSObservableQueue sqsObservableQueue = (SQSObservableQueue) sqsEventQueueProvider.getQueue("test_queue_1"); assertNotNull(sqsObservableQueue); assertEquals(1, sqsObservableQueue.getBatchSize()); assertEquals(100, sqsObservableQueue.getPollTimeInMS()); assertEquals(60, sqsObservableQueue.getVisibilityTimeoutInSeconds()); }
@Test public void testGetQueueWithCustomConfiguration() { when(configuration.getIntProperty(eq("workflow.event.queues.sqs.batchSize"), anyInt())).thenReturn(10); when(configuration.getIntProperty(eq("workflow.event.queues.sqs.pollTimeInMS"), anyInt())).thenReturn(50); when(configuration.getIntProperty(eq("workflow.event.queues.sqs.visibilityTimeoutInSeconds"), anyInt())).thenReturn(30); ListQueuesResult listQueuesResult = new ListQueuesResult().withQueueUrls("test_queue_1"); when(amazonSQSClient.listQueues(any(ListQueuesRequest.class))).thenReturn(listQueuesResult); SQSEventQueueProvider sqsEventQueueProvider = new SQSEventQueueProvider(amazonSQSClient, configuration, Schedulers.from(Executors.newSingleThreadExecutor())); SQSObservableQueue sqsObservableQueue = (SQSObservableQueue) sqsEventQueueProvider.getQueue("test_queue_1"); assertNotNull(sqsObservableQueue); assertEquals(10, sqsObservableQueue.getBatchSize()); assertEquals(50, sqsObservableQueue.getPollTimeInMS()); assertEquals(30, sqsObservableQueue.getVisibilityTimeoutInSeconds()); }