Java 类com.amazonaws.services.sqs.AmazonSQS 实例源码

private static Flowable<SqsMessage> get(AmazonSQS sqs, String queueName, Optional<String> bucketName,
        Optional<AmazonS3> s3, Service service, int waitTimeSeconds) {
    return Flowable.defer(() -> {
        final String queueUrl = sqs.getQueueUrl(queueName).getQueueUrl();
        return Flowable.just(sqs.receiveMessage(request(queueName, waitTimeSeconds)) //
                .getMessages() //
                .stream() //
                .map(m -> Sqs.getNextMessage(m, queueUrl, bucketName, s3, sqs, service)) //
                .collect(Collectors.toList())) //
                .concatWith(Flowable.defer(() -> Flowable.just(sqs.receiveMessage(request(queueName, 0)) //
                        .getMessages() //
                        .stream() //
                        .map(m -> Sqs.getNextMessage(m, queueUrl, bucketName, s3, sqs, service)) //
                        .collect(Collectors.toList()))) //
                        .repeat())
                .takeWhile(list -> !list.isEmpty()) //
                .flatMapIterable(x -> x) //
                .filter(opt -> opt.isPresent()).map(opt -> opt.get());
    });//
}

/**
 * Tests the {@link BatchSQSMsgSender#send(org.apache.flume.Channel)} method. Tests the happy path scenario.
 *
 * @throws Exception
 */
@Test
public void testSend() throws Exception {
    BatchSQSMsgSender sqsMsgSender =
        new BatchSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey", 5, 10);
    AmazonSQS mockSqs = Mockito.mock(AmazonSQS.class);
    when(mockSqs.sendMessageBatch(any(SendMessageBatchRequest.class))).thenReturn(new SendMessageBatchResult());
    sqsMsgSender.setAmazonSQS(mockSqs);

    byte[] mockMsgPayload = {'A', 'b'};
    Event mockEvent = Mockito.mock(Event.class);
    when(mockEvent.getBody()).thenReturn(mockMsgPayload);

    Channel mockChannel = Mockito.mock(Channel.class);
    when(mockChannel.take()).thenReturn(mockEvent);

    sqsMsgSender.send(mockChannel);
}

/**
 * Tests the {@link BatchSQSMsgSender#send(org.apache.flume.Channel)} method. Tests the failure scenario when AWS
 * SQS API throws AmazonServiceException.
 *
 * @throws Exception
 */
@Test(expected = EventDeliveryException.class)
public void testSendFailureAmazonServiceException() throws Exception {
    BatchSQSMsgSender sqsMsgSender =
        new BatchSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey", 5, 10);
    AmazonSQS mockSqs = Mockito.mock(AmazonSQS.class);
    when(mockSqs.sendMessageBatch(any(SendMessageBatchRequest.class))).thenThrow(AmazonServiceException.class);
    sqsMsgSender.setAmazonSQS(mockSqs);

    byte[] mockMsgPayload = {'A', 'b'};
    Event mockEvent = Mockito.mock(Event.class);
    when(mockEvent.getBody()).thenReturn(mockMsgPayload);

    Channel mockChannel = Mockito.mock(Channel.class);
    when(mockChannel.take()).thenReturn(mockEvent);

    sqsMsgSender.send(mockChannel);
}

/**
 * Tests the {@link BatchSQSMsgSender#send(org.apache.flume.Channel)} method. Tests the failure scenario when AWS
 * SQS API throws AmazonClientException.
 *
 * @throws Exception
 */
@Test(expected = EventDeliveryException.class)
public void testSendFailureAmazonClientException() throws Exception {
    BatchSQSMsgSender sqsMsgSender =
        new BatchSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey", 5, 10);
    AmazonSQS mockSqs = Mockito.mock(AmazonSQS.class);
    when(mockSqs.sendMessageBatch(any(SendMessageBatchRequest.class))).thenThrow(AmazonClientException.class);
    sqsMsgSender.setAmazonSQS(mockSqs);

    byte[] mockMsgPayload = {'A', 'b'};
    Event mockEvent = Mockito.mock(Event.class);
    when(mockEvent.getBody()).thenReturn(mockMsgPayload);

    Channel mockChannel = Mockito.mock(Channel.class);
    when(mockChannel.take()).thenReturn(mockEvent);

    sqsMsgSender.send(mockChannel);
}

@Test
public void testSend() throws Exception {
    BasicSQSMsgSender msgSender =
        new BasicSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey");

    Channel mockChannel = mock(Channel.class);
    Event mockEvent = mock(Event.class);
    when(mockEvent.getBody()).thenReturn("This is a test event message".getBytes());
    when(mockChannel.take()).thenReturn(mockEvent);

    AmazonSQS mockSqs = mock(AmazonSQS.class);
    when(mockSqs.sendMessage(any(SendMessageRequest.class))).thenReturn(new SendMessageResult());
    msgSender.setAmazonSQS(mockSqs);

    int eventCount = msgSender.send(mockChannel);
    assertEquals(1, eventCount);
}

@Test
public void testSendEventWithEmptyBody() throws Exception {
    BasicSQSMsgSender msgSender =
        new BasicSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey");

    Channel mockChannel = mock(Channel.class);
    Event mockEvent = mock(Event.class);
    when(mockEvent.getBody()).thenReturn("".getBytes());
    when(mockChannel.take()).thenReturn(mockEvent);

    AmazonSQS mockSqs = mock(AmazonSQS.class);
    when(mockSqs.sendMessage(any(SendMessageRequest.class))).thenReturn(new SendMessageResult());
    msgSender.setAmazonSQS(mockSqs);

    int eventCount = msgSender.send(mockChannel);
    assertEquals(0, eventCount);
}

@Test(expected = EventDeliveryException.class)
public void testSendFailureAmazonServiceException() throws Exception {
    BasicSQSMsgSender msgSender =
        new BasicSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey");

    Channel mockChannel = mock(Channel.class);
    Event mockEvent = mock(Event.class);
    when(mockEvent.getBody()).thenReturn("This is a test event message".getBytes());
    when(mockChannel.take()).thenReturn(mockEvent);

    AmazonSQS mockSqs = mock(AmazonSQS.class);
    when(mockSqs.sendMessage(any(SendMessageRequest.class)))
        .thenThrow(new AmazonServiceException("Mock AmazonServiceException"));
    msgSender.setAmazonSQS(mockSqs);

    msgSender.send(mockChannel);
}

@Test(expected = EventDeliveryException.class)
public void testSendFailureAmazonClientException() throws Exception {
    BasicSQSMsgSender msgSender =
        new BasicSQSMsgSender("https://some-fake/url", "us-east-1", "someAwsAccessKey", "someAwsSecretKey");

    Channel mockChannel = mock(Channel.class);
    Event mockEvent = mock(Event.class);
    when(mockEvent.getBody()).thenReturn("This is a test event message".getBytes());
    when(mockChannel.take()).thenReturn(mockEvent);

    AmazonSQS mockSqs = mock(AmazonSQS.class);
    when(mockSqs.sendMessage(any(SendMessageRequest.class)))
        .thenThrow(new AmazonClientException("Mock AmazonClientException"));
    msgSender.setAmazonSQS(mockSqs);

    msgSender.send(mockChannel);
}

private void pollMessages(AmazonSQS sqs) {
    log.info("Polling messages");
    while (true) {
        List<Message> messages = sqs.receiveMessage(QUEUE_URL).getMessages();
        messages.forEach(m -> {
            log.info("Message Received: " + m.getBody());
            System.out.println(m.getBody());
            DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest(QUEUE_URL, m.getReceiptHandle());
            sqs.deleteMessage(deleteMessageRequest);
        });
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            e.printStackTrace();
        }
    }
}

public static void changeMessageVisibilityMultiple(
        String queue_url, int timeout)
{
    AmazonSQS sqs = AmazonSQSClientBuilder.defaultClient();

    List<ChangeMessageVisibilityBatchRequestEntry> entries =
        new ArrayList<ChangeMessageVisibilityBatchRequestEntry>();

    entries.add(new ChangeMessageVisibilityBatchRequestEntry(
                "unique_id_msg1",
                sqs.receiveMessage(queue_url)
                   .getMessages()
                   .get(0)
                   .getReceiptHandle())
            .withVisibilityTimeout(timeout));

    entries.add(new ChangeMessageVisibilityBatchRequestEntry(
                "unique_id_msg2",
                sqs.receiveMessage(queue_url)
                   .getMessages()
                   .get(0)
                   .getReceiptHandle())
            .withVisibilityTimeout(timeout + 200));

    sqs.changeMessageVisibilityBatch(queue_url, entries);
}

@Test
public void testGetAmazonSQSClientCacheHitMiss()
{
    // Create an AWS parameters DTO that contains proxy information.
    AwsParamsDto awsParamsDto = new AwsParamsDto(NO_AWS_ACCESS_KEY, NO_AWS_SECRET_KEY, NO_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT);

    // Get an Amazon SQS client.
    AmazonSQS amazonSQS = awsClientFactory.getAmazonSQSClient(awsParamsDto);

    // Confirm a cache hit.
    assertEquals(amazonSQS,
        awsClientFactory.getAmazonSQSClient(new AwsParamsDto(NO_AWS_ACCESS_KEY, NO_AWS_SECRET_KEY, NO_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT)));

    // Confirm a cache miss due to http proxy information.
    assertNotEquals(amazonSQS, awsClientFactory
        .getAmazonSQSClient(new AwsParamsDto(NO_AWS_ACCESS_KEY, NO_AWS_SECRET_KEY, NO_SESSION_TOKEN, HTTP_PROXY_HOST_2, HTTP_PROXY_PORT_2)));

    // Clear the cache.
    cacheManager.getCache(DaoSpringModuleConfig.HERD_CACHE_NAME).clear();

    // Confirm a cache miss due to cleared cache.
    assertNotEquals(amazonSQS, awsClientFactory.getAmazonSQSClient(awsParamsDto));
}

@Override
public SendMessageResult sendMessage(String queueName, String messageText, Map<String, MessageAttributeValue> messageAttributes, AmazonSQS amazonSQS)
{
    // Throw a throttling exception for a specific queue name for testing purposes.
    if (queueName.equals(MockAwsOperationsHelper.AMAZON_THROTTLING_EXCEPTION))
    {
        AmazonServiceException throttlingException = new AmazonServiceException("test throttling exception");
        throttlingException.setErrorCode("ThrottlingException");
        throw throttlingException;
    }

    // Throw an illegal state exception for a specific queue name for testing purposes.
    if (queueName.equals(MOCK_SQS_QUEUE_NOT_FOUND_NAME))
    {
        throw new IllegalStateException(String.format("AWS SQS queue with \"%s\" name not found.", queueName));
    }

    // Nothing else to do in the normal case since our unit tests aren't reading messages once they have been published.
    return new SendMessageResult().withMessageId(AbstractDaoTest.MESSAGE_ID);
}

@Test
public void shouldGetCorrectQueueUrl() throws Exception {
    //GIVEN
    AmazonSQS sqs = mock(AmazonSQS.class);
    field("sqs").ofType(AmazonSQS.class).in(bundle).set(sqs);

    String queueUrl = "https://eu-central-1/queue.amazonaws.com/123456/test-queue";
    when(sqs.getQueueUrl("test-queue")).thenReturn(new GetQueueUrlResult()
            .withQueueUrl(queueUrl));

    //WHEN
    Optional<String> urlForQueue = bundle.getUrlForQueue("test-queue");

    //THEN
    assertThat(urlForQueue.isPresent()).isTrue();
    assertThat(urlForQueue.get()).isEqualTo(queueUrl);
}

@Test
public void shouldCreateNewQueueWhenNoQueueUrlIsFound() throws Exception {
    //GIVEN
    AmazonSQS sqs = mock(AmazonSQS.class);
    field("sqs").ofType(AmazonSQS.class).in(bundle).set(sqs);

    String queueUrl = "https://eu-central-1/queue.amazonaws.com/123456/test-queue";
    when(sqs.getQueueUrl("test-queue")).thenThrow(new QueueDoesNotExistException("Simulates that queue does not exist"));
    when(sqs.createQueue(new CreateQueueRequest("test-queue"))).thenReturn(new CreateQueueResult().withQueueUrl(queueUrl));

    //WHEN
    Optional<String> urlForQueue = bundle.getUrlForQueue("test-queue");

    //THEN
    assertThat(urlForQueue.isPresent()).isTrue();
    assertThat(urlForQueue.get()).isEqualTo(queueUrl);
}

@Test
public void shouldRegisterHealthCheck() throws Exception {
    //GIVEN
    AmazonSQS sqs = mock(AmazonSQS.class);

    SqsBundle spiedBundle = spy(bundle);
    doReturn(sqs).when(spiedBundle).getAmazonSQS();

    LifecycleEnvironment lifecycle = mock(LifecycleEnvironment.class);
    doNothing().when(lifecycle).manage((Managed) anyObject());
    when(environment.lifecycle()).thenReturn(lifecycle);

    HealthCheckRegistry healthChecks = mock(HealthCheckRegistry.class);
    doNothing().when(healthChecks).register(anyObject(), anyObject());
    when(environment.healthChecks()).thenReturn(healthChecks);

    //WHEN
    spiedBundle.run(configurationHolder, environment);

    //THEN
    verify(healthChecks, times(1)).register(eq("SqsBundle"), any(SqsBundleHealthCheck.class));
}

@Test
public void shouldCorrectlyCreateSenderIfQueueExists() throws Exception, CannotCreateSenderException {
    //GIVEN
    AmazonSQS sqs = mock(AmazonSQS.class);
    field("sqs").ofType(AmazonSQS.class).in(bundle).set(sqs);

    String queueUrl = "https://eu-central-1/queue.amazonaws.com/123456/test-queue";
    when(sqs.getQueueUrl("test-queue")).thenReturn(new GetQueueUrlResult()
            .withQueueUrl(queueUrl));

    //WHEN
    SqsSender sender = bundle.createSender("test-queue");

    //THEN
    assertThat(sender).isNotNull();
}

@Override
public Properties send(Properties properties, Object message)
        throws ConnectorException {


    String access_key_id = properties.getProperty("AccessKeyId");
       String secret_access_key = properties.getProperty("SecretAccessKey");
       BasicAWSCredentials credentials = new BasicAWSCredentials(access_key_id, secret_access_key);

       AmazonSQS sqs = new AmazonSQSClient(credentials);
       //System.out.println(properties.getProperty("region"));
       // Region selection
       Region region = Region.getRegion(Regions.fromName(properties.getProperty("region")));
       sqs.setRegion(region);

       GetQueueUrlResult queueUrl = sqs.getQueueUrl(properties.getProperty("Queue"));
       String messageStr = new String((byte[])message);

       sqs.sendMessage(new SendMessageRequest(queueUrl.getQueueUrl(), messageStr));

    return properties;
}

/**
 * Returns a client instance for AWS SQS.
 * @return a client that talks to SQS
 */
public static AmazonSQS getClient() {
    if (sqsClient != null) {
        return sqsClient;
    }
    if (Config.IN_PRODUCTION) {
        sqsClient = AmazonSQSClientBuilder.standard().withCredentials(new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(Config.AWS_ACCESSKEY, Config.AWS_SECRETKEY))).
                withRegion(Config.AWS_REGION).build();
    } else {
        sqsClient = AmazonSQSClientBuilder.standard().
                withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("x", "x"))).
                withEndpointConfiguration(new EndpointConfiguration(LOCAL_ENDPOINT, "")).build();
    }

    Para.addDestroyListener(new DestroyListener() {
        public void onDestroy() {
            sqsClient.shutdown();
        }
    });
    return sqsClient;
}

private SQSConnectionFactory(final Builder builder) {
    this.providerConfiguration = builder.providerConfiguration;
    this.amazonSQSClientSupplier = new AmazonSQSClientSupplier() {
        @Override
        public AmazonSQS get() {
            AmazonSQSClient amazonSQSClient = new AmazonSQSClient(builder.awsCredentialsProvider, builder.clientConfiguration);
            if (builder.region != null) {
                amazonSQSClient.setRegion(builder.region);
            }
            if (builder.endpoint != null) {
                amazonSQSClient.setEndpoint(builder.endpoint);
            }
            if (builder.signerRegionOverride != null) {
                amazonSQSClient.setSignerRegionOverride(builder.signerRegionOverride);
            }
            return amazonSQSClient;
        }
    };
}

/**
 * Sets up a queue so that it subscribes to all messages that gets published on the
 * specified topic. The queue and topic must be created before-hand, or an exception
 * will be thrown.
 *
 * This method will also configure the queue so that it has permissions to receive
 * messages from the topic.
 *
 * @param queueURL The queue that should receive the messages posted to the topic.
 * @param topicARN The topic whose messages should be posted to the queue.
 * @param amazonSQS Amazon SQS client.
 * @param amazonSNS Amazon SNS client.
 */
public static void subscribeSQSQueueToSNSTopic(
    String queueURL,
    String topicARN,
    AmazonSQS amazonSQS,
    AmazonSNS amazonSNS
) {
    // Verify that the queue exists, and get its ARN
    String queueARN = getSQSQueueARN(amazonSQS, queueURL);

    // The "is already subscribing?"-check has a dual purpose: it will also verify that the
    // topic is already created. If the topic is not created beforehand, this will throw a
    // com.amazonaws.services.sns.model.NotFoundException
    if (isSQSQueueSubscribingToSNSTopic(amazonSNS, queueARN, topicARN)) {
        // Subscription already configured. Do nothing.
        return;
    }

    allowSQSQueueToReceiveMessagesFromSNSTopic(amazonSQS, queueURL, queueARN, topicARN);
    subscribeSQSQueueToSNSTopic(amazonSNS, queueARN, topicARN);
}

public static MessageQueueProcessor createQueueProcessor(
    AmazonSQS amazonSQS,
    String name,
    String queueUrl,
    String deadLetterQueueUrl,
    MessageHandler<Message> messageHandler
) {
    return new QueueProcessor(
        name,
        queueUrl,
        deadLetterQueueUrl,
        amazonSQS,
        messageHandler,
        null
    );
}

public static MessageQueueProcessor createQueueProcessor(
    AmazonSQS amazonSQS,
    String name,
    String queueUrl,
    String deadLetterQueueUrl,
    MessageHandler<Message> messageHandler,
    ExecutorService executorService
) {
    return new QueueProcessor(
        name,
        queueUrl,
        deadLetterQueueUrl,
        amazonSQS,
        messageHandler,
        executorService
    );
}

public static <M> MessageQueueProcessor createQueueProcessor(
    AmazonSQS amazonSQS,
    Class<M> classType,
    String name,
    String queueUrl,
    String deadLetterQueueUrl,
    MessageHandler<M> messageHandler
) {
    return new QueueProcessor(
        name,
        queueUrl,
        deadLetterQueueUrl,
        amazonSQS,
        new MessageHandlerForSingleMessageType<>(messageHandler, classType),
        null
    );
}

public static <M> MessageQueueProcessor createQueueProcessor(
    AmazonSQS amazonSQS,
    Class<M> classType,
    String name,
    String queueUrl,
    String deadLetterQueueUrl,
    MessageHandler<M> messageHandler,
    ExecutorService executorService
) {
    return new QueueProcessor(
        name,
        queueUrl,
        deadLetterQueueUrl,
        amazonSQS,
        new MessageHandlerForSingleMessageType<>(messageHandler, classType),
        executorService
    );
}

private QueueServiceSender(
        String queueUrl,
        AmazonSQS amazonSQS,
        MessageSerializer messageSerializer
) {
    if (anyEmpty(queueUrl, amazonSQS, messageSerializer)) {
        throw new IllegalArgumentException(
                "None of queueUrl, amazonSQS or messageSerializer can be empty!\n"
                        + "queueUrl = " + queueUrl + "\n"
                        + "amazonSQS = " + amazonSQS + "\n"
                        + "messageSerializer = " + messageSerializer
        );
    }
    this.queueUrl = queueUrl;
    this.amazonSQS = amazonSQS;
    this.messageSerializer = messageSerializer;
}

public static <T> MessageQueueConsumer<T> encryptedMessageQueueConsumer(
        final Class<T> messageClass,
        final String queueUrl,
        final AmazonSQS amazonSQSClient,
        byte[] privatePgpKey,
        final String privatePgpKeyPassphrase
) throws MessagingException {
    if (empty(privatePgpKey) || empty(privatePgpKeyPassphrase)) {
        throw new MessagingException("Can't create encryptedQueueServicePoller with private PGP key as null or privatePgpKeyPassphrase as null");
    }
    return new QueueServicePoller<>(messageClass,
            queueUrl,
            amazonSQSClient,
            privatePgpKey,
            privatePgpKeyPassphrase);
}

public static void main(String... args) throws IOException {
    AWSCredentialsProvider credentials = CredentialsUtils.getAwsCredentials();

    AmazonS3 storage = new AmazonS3Client(credentials);
    storage.setRegion(Region.getRegion(Regions.EU_WEST_1));

    AmazonSQS queue = new AmazonSQSClient(credentials);
    queue.setRegion(Region.getRegion(Regions.EU_WEST_1));

    AmazonEC2 machines = new AmazonEC2Client(credentials);
    machines.setRegion(Region.getRegion(Regions.EU_WEST_1));

    AmazonIdentityManagement identityManagement = new AmazonIdentityManagementClient(credentials);
    identityManagement.setRegion(Region.getRegion(Regions.EU_WEST_1));

    if ("launch".equals(args[0])) {
        setup(storage, queue, machines, identityManagement);

        manageAnswers(queue);
    } else if ("run".equals(args[0])) {
        run(storage, queue);
    } else {
        System.out.println("Huh hoh! Don't know what you intended to do...");
    }
}

private static void run(AmazonS3 storage, AmazonSQS queue) throws FileNotFoundException, UnknownHostException {
    while (true) {
        String[] message = SQSUtils.readMessageFrom(queue, INPUT_QUEUE);

        String url = message[0];
        String receipt = message[1];

        if ("SUICIDE".equals(url)) {
            SQSUtils.acknowledge(queue, INPUT_QUEUE, receipt);
            break;
        }

        try {
            processQueries(storage, queue, url, receipt);

            SQSUtils.acknowledge(queue, INPUT_QUEUE, receipt);
        } catch (Exception e) {
            // Because message is not acknowledged, it will be reprocessed by someone else soon
            e.printStackTrace();
        }
    }
}

@Test
public void oneQueue() {
    AmazonSQS amazonSQS = mock(AmazonSQS.class);

    // return one queue
    when(amazonSQS.listQueues(any(ListQueuesRequest.class)))
            .thenReturn(new ListQueuesResult().withQueueUrls("test-foo"));

    // return 3 messages from the queue
    when(amazonSQS.receiveMessage(any(ReceiveMessageRequest.class)))
            .thenReturn(new ReceiveMessageResult().withMessages(newMessage("foo"), newMessage("foo"), newMessage("foo")));

    AmazonSQSPrioritizedMessageProvider provider = new AmazonSQSPrioritizedMessageProvider(amazonSQS, "test", 1, 60 * 1000);
    List<Message> messages = provider.next();
    assertMessages(messages, 3, "foo");

    verify(amazonSQS).listQueues(any(ListQueuesRequest.class));
    verify(amazonSQS).receiveMessage(any(ReceiveMessageRequest.class));
}

public void deleteMessage(Optional<AmazonS3> s3, AmazonSQS sqs) {
    Preconditions.checkArgument(!s3.isPresent() || s3Id.isPresent(), "s3Id must be present");
    if (s3Id.isPresent()) {
        s3.get().deleteObject(service.bucketName.get(), s3Id.get());
    }
    sqs.deleteMessage(service.queueName, messageReceiptHandle);
}

Service(Optional<Callable<AmazonS3>> s3Factory, Callable<AmazonSQS> sqsFactory, Optional<AmazonS3> s3,
        AmazonSQS sqs, String queueName, Optional<String> bucketName) {
    Preconditions.checkNotNull(s3Factory);
    Preconditions.checkNotNull(sqsFactory);
    Preconditions.checkNotNull(s3);
    Preconditions.checkNotNull(sqs);
    Preconditions.checkNotNull(queueName);
    Preconditions.checkNotNull(bucketName);
    this.s3Factory = s3Factory;
    this.sqsFactory = sqsFactory;
    this.s3 = s3;
    this.sqs = sqs;
    this.queueName = queueName;
    this.bucketName = bucketName;
}

public static String sendToQueueUsingS3(AmazonSQS sqs, String queueUrl, AmazonS3 s3, String bucketName,
        Map<String, String> headers, byte[] message, Callable<String> s3IdFactory) {
    Preconditions.checkNotNull(sqs);
    Preconditions.checkNotNull(s3);
    Preconditions.checkNotNull(queueUrl);
    Preconditions.checkNotNull(bucketName);
    Preconditions.checkNotNull(message);
    String s3Id;
    try {
        s3Id = s3IdFactory.call();
    } catch (final Exception e1) {
        throw new RuntimeException(e1);
    }
    final ObjectMetadata metadata = new ObjectMetadata();
    metadata.setContentLength(message.length);
    for (final Entry<String, String> header : headers.entrySet()) {
        metadata.setHeader(header.getKey(), header.getValue());
    }
    s3.putObject(bucketName, s3Id, new ByteArrayInputStream(message), metadata);
    try {
        sqs.sendMessage(queueUrl, s3Id);
    } catch (final RuntimeException e) {
        try {
            s3.deleteObject(bucketName, s3Id);
            throw e;
        } catch (final RuntimeException e2) {
            throw new io.reactivex.exceptions.CompositeException(e, e2);
        }
    }
    return s3Id;
}

static Flowable<SqsMessage> messages(Callable<AmazonSQS> sqsFactory, Optional<Callable<AmazonS3>> s3Factory,
        String queueName, Optional<String> bucketName, Optional<Flowable<Integer>> waitTimesSeconds) {
    Preconditions.checkNotNull(sqsFactory);
    Preconditions.checkNotNull(s3Factory);
    Preconditions.checkNotNull(queueName);
    Preconditions.checkNotNull(bucketName);
    Preconditions.checkNotNull(waitTimesSeconds);
    return Flowable.using(sqsFactory,
            sqs -> createObservableWithSqs(sqs, s3Factory, sqsFactory, queueName, bucketName, waitTimesSeconds),
            sqs -> sqs.shutdown());
}

private static Flowable<SqsMessage> createObservableWithSqs(AmazonSQS sqs, Optional<Callable<AmazonS3>> s3Factory,
        Callable<AmazonSQS> sqsFactory, String queueName, Optional<String> bucketName,
        Optional<Flowable<Integer>> waitTimesSeconds) {

    return Flowable.using(() -> s3Factory.map(x -> {
        try {
            return x.call();
        } catch (final Exception e) {
            throw new RuntimeException(e);
        }
    }), //
            s3 -> createObservableWithS3(sqs, s3Factory, sqsFactory, queueName, bucketName, s3, waitTimesSeconds),
            s3 -> s3.ifPresent(Util::shutdown));
}

private static Flowable<SqsMessage> createObservableWithS3(AmazonSQS sqs, Optional<Callable<AmazonS3>> s3Factory,
        Callable<AmazonSQS> sqsFactory, String queueName, Optional<String> bucketName, Optional<AmazonS3> s3,
        Optional<Flowable<Integer>> waitTimesSeconds) {
    final Service service = new Service(s3Factory, sqsFactory, s3, sqs, queueName, bucketName);
    if (waitTimesSeconds.isPresent()) {
        return createObservablePolling(sqs, s3Factory, sqsFactory, queueName, bucketName, s3,
                waitTimesSeconds.get());
    } else {
        return createObservableContinousLongPolling(sqs, queueName, bucketName, s3, service);
    }
}

public ContinuousLongPollingSyncOnSubscribe(AmazonSQS sqs, String queueName, Optional<AmazonS3> s3,
        Optional<String> bucketName, Service service) {
    this.sqs = sqs;
    this.queueName = queueName;
    this.s3 = s3;
    this.bucketName = bucketName;
    this.service = service;
}

public static AmazonSQS createSqsClient(AWSCredentials credentials, ClientConfiguration cc, Regions region) {
    return AmazonSQSClientBuilder //
            .standard() //
            .withCredentials(new AWSStaticCredentialsProvider(credentials)) //
            .withRegion(region) //
            .build();
}

static void shutdown(AmazonSQS client) {
    try {
        client.shutdown();
    } catch (final RuntimeException e) {
        // ignore
    }
}

public ListBoxModel doFillRegionItems() {
    ListBoxModel items = new ListBoxModel();
    items.add("", "");
    for (Regions region : Regions.values()) {
        Region r = Region.getRegion(region);
        if (r.isServiceSupported(AmazonSQS.ENDPOINT_PREFIX) && r.isServiceSupported("codecommit")) {
            items.add(region.getName(), region.name());
        }
    }
    return items;
}

@Override
public AmazonSQS createSQSAsync(final SQSQueue queue) {
    AWSCredentialsProvider credentials = queue.hasCredentials() ? queue.lookupAwsCredentials() : DefaultAWSCredentialsProviderChain.getInstance();
    AmazonSQSAsyncClientBuilder sqsAsyncBuilder = createStandardAsyncClientBuilder(queue, credentials);
    final QueueBufferConfig queueBufferConfig = this.getQueueBufferConfig(queue);
    final AmazonSQSBufferedAsyncClient sqsBufferedAsync = new AmazonSQSBufferedAsyncClient(sqsAsyncBuilder.build(), queueBufferConfig);
    return sqsBufferedAsync;
}