diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java index 726e9b4..b55e0b9 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java @@ -105,9 +105,9 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { private InitialPositionInStream streamInitialSequence = InitialPositionInStream.LATEST; - private int idleBetweenPolls; + private int idleBetweenPolls = 1000; - private int consumerBackoff; + private int consumerBackoff = 1000; private Converter converter = new DeserializingConverter(); @@ -298,7 +298,9 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { @Override public void destroy() { super.destroy(); - this.scheduler.shutdown(); + if (isRunning()) { + this.scheduler.shutdown(); + } } @Override diff --git a/src/test/java/org/springframework/integration/aws/EnvironmentHostNameResolver.java b/src/test/java/org/springframework/integration/aws/EnvironmentHostNameResolver.java index e614022..b8588c2 100644 --- a/src/test/java/org/springframework/integration/aws/EnvironmentHostNameResolver.java +++ b/src/test/java/org/springframework/integration/aws/EnvironmentHostNameResolver.java @@ -20,6 +20,11 @@ import cloud.localstack.docker.annotation.IHostNameResolver; import com.amazonaws.SDKGlobalConfiguration; /** + * An {@link IHostNameResolver} implementation for {@value EnvironmentHostNameResolver#DOCKER_HOST_NAME} + * environment variable to resolve for Local Stack Docker instance. + * Also this class places an {@value SDKGlobalConfiguration#AWS_CBOR_DISABLE_SYSTEM_PROPERTY} + * system property to disable CBOR for services requests. + * * @author Artem Bilan * * @since 2.3 diff --git a/src/test/java/org/springframework/integration/aws/ExtendedDockerTestUtils.java b/src/test/java/org/springframework/integration/aws/ExtendedDockerTestUtils.java index 76eea57..0f4f0d9 100644 --- a/src/test/java/org/springframework/integration/aws/ExtendedDockerTestUtils.java +++ b/src/test/java/org/springframework/integration/aws/ExtendedDockerTestUtils.java @@ -25,12 +25,18 @@ import cloud.localstack.docker.LocalstackDocker; import com.amazonaws.ClientConfiguration; import com.amazonaws.client.builder.AwsAsyncClientBuilder; import com.amazonaws.client.builder.AwsClientBuilder; +import com.amazonaws.services.cloudwatch.AmazonCloudWatchAsync; +import com.amazonaws.services.cloudwatch.AmazonCloudWatchAsyncClientBuilder; import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder; import com.amazonaws.services.kinesis.AmazonKinesisAsync; import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder; /** + * An utility class for providing AWS {@code async} clients based on the Local Stack Docker instance. + * In addition it provides SSL-based async clients. + * + * * @author Artem Bilan * * @since 2.3 @@ -38,23 +44,61 @@ import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder; public final class ExtendedDockerTestUtils { public static AmazonKinesisAsync getClientKinesisAsync() { + return doGetClientKinesisAsync(false); + } + + public static AmazonKinesisAsync getClientKinesisAsyncSsl() { + return doGetClientKinesisAsync(true); + } + + private static AmazonKinesisAsync doGetClientKinesisAsync(boolean ssl) { AmazonKinesisAsyncClientBuilder amazonKinesisAsyncClientBuilder = AmazonKinesisAsyncClientBuilder.standard() .withEndpointConfiguration( - createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointKinesis)); + createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointKinesis, ssl)); return applyConfigurationAndBuild(amazonKinesisAsyncClientBuilder); } public static AmazonDynamoDBAsync getClientDynamoDbAsync() { + return doClientDynamoDbAsync(false); + } + + public static AmazonDynamoDBAsync getClientDynamoDbAsyncSsl() { + return doClientDynamoDbAsync(true); + } + + private static AmazonDynamoDBAsync doClientDynamoDbAsync(boolean ssl) { AmazonDynamoDBAsyncClientBuilder dynamoDBAsyncClientBuilder = AmazonDynamoDBAsyncClientBuilder.standard() .withEndpointConfiguration( - createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointDynamoDB)); + createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointDynamoDB, ssl)); return applyConfigurationAndBuild(dynamoDBAsyncClientBuilder); } - private static AwsClientBuilder.EndpointConfiguration createEndpointConfiguration(Supplier supplier) { - return new AwsClientBuilder.EndpointConfiguration(supplier.get(), DEFAULT_REGION); + public static AmazonCloudWatchAsync getClientCloudWatchAsync() { + return doClientCloudWatchAsync(false); + } + + public static AmazonCloudWatchAsync getClientCloudWatchAsyncSsl() { + return doClientCloudWatchAsync(true); + } + + private static AmazonCloudWatchAsync doClientCloudWatchAsync(boolean ssl) { + AmazonCloudWatchAsyncClientBuilder cloudWatchAsyncClientBuilder = + AmazonCloudWatchAsyncClientBuilder.standard() + .withEndpointConfiguration( + createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointCloudWatch, ssl)); + return applyConfigurationAndBuild(cloudWatchAsyncClientBuilder); + } + + private static AwsClientBuilder.EndpointConfiguration createEndpointConfiguration(Supplier supplier, + boolean ssl) { + + String serviceEndpoint = supplier.get(); + if (ssl) { + serviceEndpoint = serviceEndpoint.replaceFirst("http", "https"); + } + return new AwsClientBuilder.EndpointConfiguration(serviceEndpoint, DEFAULT_REGION); } private static > T applyConfigurationAndBuild(C builder) { diff --git a/src/test/java/org/springframework/integration/aws/LocalStackSslEnvironmentProvider.java b/src/test/java/org/springframework/integration/aws/LocalStackSslEnvironmentProvider.java new file mode 100644 index 0000000..dee547a --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/LocalStackSslEnvironmentProvider.java @@ -0,0 +1,46 @@ +/* + * Copyright 2019 the original author or authors. + * + * 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 + * + * https://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 org.springframework.integration.aws; + +import java.util.Collections; +import java.util.Map; + +import cloud.localstack.docker.annotation.IEnvironmentVariableProvider; +import com.amazonaws.SDKGlobalConfiguration; + +/** + * An {@link IEnvironmentVariableProvider} implementation to provide a {@code USE_SSL} + * environment variable for docker to start a Local Stack in TLS mode. + * Also this class populates a {@value SDKGlobalConfiguration#DISABLE_CERT_CHECKING_SYSTEM_PROPERTY} + * system property to disable SSL certificates validation. + * + * @author Artem Bilan + * + * @since 2.3 + */ +public class LocalStackSslEnvironmentProvider implements IEnvironmentVariableProvider { + + static { + System.setProperty(SDKGlobalConfiguration.DISABLE_CERT_CHECKING_SYSTEM_PROPERTY, "true"); + } + + @Override + public Map getEnvironmentVariables() { + return Collections.singletonMap("USE_SSL", "true"); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java index 6cacdcc..dd5aa9b 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java @@ -67,7 +67,6 @@ import com.amazonaws.services.kinesis.AmazonKinesisAsync; * * @since 1.1 */ -//@Disabled("Looks like Kinesis is not supported well in Local Stack") @SpringJUnitConfig @EnabledIfEnvironmentVariable(named = EnvironmentHostNameResolver.DOCKER_HOST_NAME, matches = ".+") @ExtendWith(LocalstackDockerExtension.class) @@ -123,21 +122,21 @@ public class KinesisIntegrationTests { .contains("Channel 'kinesisReceiveChannel' expected one of the following data types " + "[class java.util.Date], but received [class java.lang.String]"); - for (int i = 0; i < 10; i++) { + for (int i = 0; i < 2; i++) { this.kinesisSendChannel .send(MessageBuilder.withPayload(new Date()).setHeader(AwsHeaders.STREAM, TEST_STREAM).build()); } Set receivedSequences = new HashSet<>(); - for (int i = 0; i < 10; i++) { + for (int i = 0; i < 2; i++) { receive = this.kinesisReceiveChannel.receive(20_000); assertThat(receive).isNotNull(); String sequenceNumber = receive.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, String.class); assertThat(receivedSequences.add(sequenceNumber)).isTrue(); } - assertThat(receivedSequences.size()).isEqualTo(10); + assertThat(receivedSequences.size()).isEqualTo(2); receive = this.kinesisReceiveChannel.receive(10); assertThat(receive).isNull(); diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java new file mode 100644 index 0000000..201431e --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java @@ -0,0 +1,212 @@ +/* + * Copyright 2017-2019 the original author or authors. + * + * 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 + * + * https://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 org.springframework.integration.aws.kinesis; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.entry; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.Date; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable; +import org.junit.jupiter.api.extension.ExtendWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.aws.EnvironmentHostNameResolver; +import org.springframework.integration.aws.ExtendedDockerTestUtils; +import org.springframework.integration.aws.LocalStackSslEnvironmentProvider; +import org.springframework.integration.aws.inbound.kinesis.KclMessageDrivenChannelAdapter; +import org.springframework.integration.aws.inbound.kinesis.KinesisMessageHeaderErrorMessageStrategy; +import org.springframework.integration.aws.outbound.KplMessageHandler; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import cloud.localstack.TestUtils; +import cloud.localstack.docker.LocalstackDocker; +import cloud.localstack.docker.LocalstackDockerExtension; +import cloud.localstack.docker.annotation.LocalstackDockerProperties; +import com.amazonaws.services.cloudwatch.AmazonCloudWatch; +import com.amazonaws.services.dynamodbv2.AmazonDynamoDB; +import com.amazonaws.services.kinesis.AmazonKinesis; +import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream; +import com.amazonaws.services.kinesis.producer.KinesisProducer; +import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration; + +/** + * @author Artem Bilan + * + * @since 1.1 + */ +@SpringJUnitConfig +@EnabledIfEnvironmentVariable(named = EnvironmentHostNameResolver.DOCKER_HOST_NAME, matches = ".+") +@ExtendWith(LocalstackDockerExtension.class) +@LocalstackDockerProperties(randomizePorts = true, + hostNameResolver = EnvironmentHostNameResolver.class, + environmentVariableProvider = LocalStackSslEnvironmentProvider.class, + services = { "kinesis", "dynamodb", "cloudwatch" }) +@DirtiesContext +public class KplKclIntegrationTests { + + private static final String TEST_STREAM = "TestStream"; + + private static AmazonKinesis AMAZON_KINESIS; + + private static AmazonDynamoDB DYNAMO_DB; + + private static AmazonCloudWatch CLOUD_WATCH; + + @Autowired + private MessageChannel kinesisSendChannel; + + @Autowired + private PollableChannel kinesisReceiveChannel; + + @Autowired + private PollableChannel errorChannel; + + @BeforeAll + static void setup() { + AMAZON_KINESIS = ExtendedDockerTestUtils.getClientKinesisAsyncSsl(); + DYNAMO_DB = ExtendedDockerTestUtils.getClientDynamoDbAsyncSsl(); + CLOUD_WATCH = ExtendedDockerTestUtils.getClientCloudWatchAsyncSsl(); + AMAZON_KINESIS.createStream(TEST_STREAM, 1); + } + + @AfterAll + static void tearDown() { + AMAZON_KINESIS.deleteStream(TEST_STREAM); + } + + @Test + void testKinesisInboundOutbound() { + this.kinesisSendChannel + .send(MessageBuilder.withPayload("foo").setHeader(AwsHeaders.STREAM, TEST_STREAM).build()); + + Date now = new Date(); + this.kinesisSendChannel.send(MessageBuilder.withPayload(now).setHeader(AwsHeaders.STREAM, TEST_STREAM) + .setHeader("foo", "BAR").build()); + + Message receive = this.kinesisReceiveChannel.receive(20_000); + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo(now); + assertThat(receive.getHeaders()).contains(entry("foo", "BAR")); + assertThat(receive.getHeaders()).containsKey(IntegrationMessageHeaderAccessor.SOURCE_DATA); + + Message errorMessage = this.errorChannel.receive(10_000); + assertThat(errorMessage).isNotNull(); + assertThat(errorMessage.getHeaders().get(AwsHeaders.RAW_RECORD)).isNotNull(); + assertThat(((Exception) errorMessage.getPayload()).getMessage()) + .contains("Channel 'kinesisReceiveChannel' expected one of the following data types " + + "[class java.util.Date], but received [class java.lang.String]"); + + this.kinesisSendChannel + .send(MessageBuilder.withPayload(new Date()).setHeader(AwsHeaders.STREAM, TEST_STREAM).build()); + + receive = this.kinesisReceiveChannel.receive(20_000); + assertThat(receive).isNotNull(); + assertThat(receive.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, String.class)).isNotEmpty(); + + receive = this.kinesisReceiveChannel.receive(10); + assertThat(receive).isNull(); + } + + @Configuration + @EnableIntegration + public static class TestConfiguration { + + @Bean + public KinesisProducerConfiguration kinesisProducerConfiguration() throws URISyntaxException { + URI kinesisUri = new URI(LocalstackDocker.INSTANCE.getEndpointKinesis()); + URI cloudWatchUri = new URI(LocalstackDocker.INSTANCE.getEndpointCloudWatch()); + return new KinesisProducerConfiguration() + .setCredentialsProvider(TestUtils.getCredentialsProvider()) + .setRegion(TestUtils.DEFAULT_REGION) + .setKinesisEndpoint(kinesisUri.getHost()) + .setKinesisPort(kinesisUri.getPort()) + .setCloudwatchEndpoint(cloudWatchUri.getHost()) + .setCloudwatchPort(cloudWatchUri.getPort()) + .setVerifyCertificate(false); + } + + @Bean + @ServiceActivator(inputChannel = "kinesisSendChannel") + public MessageHandler kplMessageHandler(KinesisProducerConfiguration kinesisProducerConfiguration) { + KplMessageHandler kinesisMessageHandler = + new KplMessageHandler(new KinesisProducer(kinesisProducerConfiguration)); + kinesisMessageHandler.setPartitionKey("1"); + kinesisMessageHandler.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo")); + return kinesisMessageHandler; + } + + @Bean + public KclMessageDrivenChannelAdapter kclMessageDrivenChannelAdapter() { + KclMessageDrivenChannelAdapter adapter = new KclMessageDrivenChannelAdapter( + TEST_STREAM, AMAZON_KINESIS, CLOUD_WATCH, DYNAMO_DB, TestUtils.getCredentialsProvider()); + adapter.setOutputChannel(kinesisReceiveChannel()); + adapter.setErrorChannel(errorChannel()); + adapter.setErrorMessageStrategy(new KinesisMessageHeaderErrorMessageStrategy()); + adapter.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo")); + adapter.setStreamInitialSequence(InitialPositionInStream.TRIM_HORIZON); + adapter.setBindSourceRecord(true); + return adapter; + } + + @Bean + public PollableChannel kinesisReceiveChannel() { + QueueChannel queueChannel = new QueueChannel(); + queueChannel.setDatatypes(Date.class); + return queueChannel; + } + + @Bean + public PollableChannel errorChannel() { + QueueChannel queueChannel = new QueueChannel(); + queueChannel.addInterceptor(new ChannelInterceptor() { + + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + if (message instanceof ErrorMessage) { + throw (RuntimeException) ((ErrorMessage) message).getPayload(); + } + } + + }); + return queueChannel; + } + + } + +} diff --git a/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java b/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java index 55b8af3..2f71393 100644 --- a/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java +++ b/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java @@ -86,7 +86,6 @@ class DynamoDbLockRegistryLeaderInitiatorTests { DYNAMO_DB.deleteTable(DynamoDbLockRegistry.DEFAULT_TABLE_NAME); } - // @Disabled("Doesn't work properly against Local Stack when two instances try to lock in table") @Test void testDistributedLeaderElection() throws Exception { CountDownLatch granted = new CountDownLatch(1);