SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polish Kinesis binder following previous commit

Browse Source
This commit is contained in:
Oleg Zhurakousky
2022-09-27 17:03:15 +02:00
parent 5035a40990
commit 60ddac5560
5 changed files with 19 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java
View File

@@ -215,6 +215,11 @@ public class KinesisMessageChannelBinder extends
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
@Override
public String getBinderIdentity() {
return "kinesis-" + super.getBinderIdentity();
}
@Override
protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
ExtendedProducerProperties<KinesisProducerProperties> producerProperties,

4
binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java
View File

@@ -264,6 +264,7 @@ public class KinesisBinderTests extends
ExtendedProducerProperties<KinesisProducerProperties> producerProps = createProducerProperties();
producerProps.setErrorChannelEnabled(true);
producerProps.populateBindingName("foobar");
DirectChannel moduleOutputChannel = createBindableChannel("output",
createProducerBindingProperties(producerProps));
Binding<MessageChannel> producerBinding = binder.bindProducer("ec.0",
@@ -271,7 +272,8 @@ public class KinesisBinderTests extends
ApplicationContext applicationContext = TestUtils.getPropertyValue(
binder.getBinder(), "applicationContext", ApplicationContext.class);
SubscribableChannel ec = applicationContext.getBean("ec.0.errors",
String s = testBinder.getBinder().getBinderIdentity() + "." + producerProps.getBindingName() + ".errors";
SubscribableChannel ec = applicationContext.getBean(s,
SubscribableChannel.class);
final AtomicReference<Message<?>> errorMessage = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);

3
binders/kinesis-binder/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/LocalstackContainerTest.java
View File

@@ -24,8 +24,6 @@ import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder;
import com.amazonaws.services.kinesis.AmazonKinesisAsync;
import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.testcontainers.containers.localstack.LocalStackContainer;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
@@ -37,7 +35,6 @@ import org.testcontainers.utility.DockerImageName;
* @since 4.0
*/
@Testcontainers(disabledWithoutDocker = true)
@DisabledOnOs(OS.MAC)
public interface LocalstackContainerTest {
LocalStackContainer LOCAL_STACK_CONTAINER =

1
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -266,7 +266,6 @@ public class RabbitBinderTests extends
final Message<?> message = MessageBuilder.withPayload("bad".getBytes())
.setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
String s = testBinder.getBinder().getBinderIdentity() + ".output.errors";
//return this.getBinderIdentity() + "-" + this.hashCode() + "." + bindingName + ".errors";
SubscribableChannel ec = binder.getApplicationContext().getBean(s,
SubscribableChannel.class);
final AtomicReference<Message<?>> errorMessage = new AtomicReference<>();

11
buildKinesisBinder.sh Executable file
View File

@@ -0,0 +1,11 @@
#!/bin/bash
if [ "$#" -eq 1 ]; then
ARGVALUE=$1
if [[ $ARGVALUE == *"skipTest"* ]]; then
echo "Skippping Tests"
./mvnw clean install -f binders/kinesis-binder/pom.xml -DskipTests
fi
else
./mvnw clean install -f binders/kinesis-binder/pom.xml
fi