Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Pulsar binder subscription name improvements

Browse Source 
Provide a default subscription name that is anonymous each time the app starts.
This can be overridden to a fixed subscription name by providing it as a property.
This commit is contained in:
Soby Chacko
2023-01-20 16:21:38 -05:00
parent ada87dbeee
commit ea1597a0a1
5 changed files with 106 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml
View File

@@ -15,7 +15,6 @@ spring:
bindings:
timeLogger-in-0:
consumer:
subscription-name: my-scst-sub1
schema-type: STRING
timeSupplier-out-0:
producer:

51
spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* Copyright 2023 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.pulsar.spring.cloud.stream.binder;
import java.util.UUID;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties;
/**
* Utility methods for the binder.
*
* @author Soby Chacko
*/
public final class PulsarBinderUtils {
static final String ANON_SUBSCRIPTION = "anon-subscription";
static final char SUBSCRIPTION_NAME_SEPARATOR = '-';
private PulsarBinderUtils() {
}
static String subscriptionName(PulsarConsumerProperties pulsarConsumerProperties,
ConsumerDestination consumerDestination) {
String subscriptionName = pulsarConsumerProperties.getSubscriptionName();
if (subscriptionName == null) {
// if subscription name is not provided, each time the app starts, it will be
// an anonymous subscription
subscriptionName = consumerDestination.getName() + SUBSCRIPTION_NAME_SEPARATOR + ANON_SUBSCRIPTION
+ SUBSCRIPTION_NAME_SEPARATOR + UUID.randomUUID();
}
return subscriptionName;
}
}

3
spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java
View File

@@ -110,7 +110,8 @@ public class PulsarMessageChannelBinder extends
else {
pulsarContainerProperties.setSchema(Schema.BYTES);
}
pulsarContainerProperties.setSubscriptionName(properties.getExtension().getSubscriptionName());
String subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination);
pulsarContainerProperties.setSubscriptionName(subscriptionName);
DefaultPulsarMessageListenerContainer<?> container = new DefaultPulsarMessageListenerContainer<>(
this.pulsarConsumerFactory, pulsarContainerProperties);
pulsarMessageDrivenChannelAdapter.setMessageListenerContainer(container);

6
spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java
View File

@@ -19,7 +19,6 @@ package org.springframework.pulsar.spring.cloud.stream.binder.properties;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
/**
@@ -29,8 +28,8 @@ import org.springframework.lang.Nullable;
*/
public class PulsarConsumerProperties {
@NonNull
private String subscriptionName = "DEFAULT-SUBSCRIPTION-TODO-CHANGE-THIS";
@Nullable
private String subscriptionName;
@Nullable
private SchemaType schemaType;
@@ -38,6 +37,7 @@ public class PulsarConsumerProperties {
@Nullable
private SubscriptionType subscriptionType;
@Nullable
public String getSubscriptionName() {
return this.subscriptionName;
}

50
spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2023 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.pulsar.spring.cloud.stream.binder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import org.junit.jupiter.api.Test;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties;
public class PulsarBinderUtilsTests {
@Test
void subscriptionNameIsNotNullWhenProvidedAsProperty() {
ConsumerDestination consumerDestination = mock(ConsumerDestination.class);
PulsarConsumerProperties pulsarConsumerProperties = mock(PulsarConsumerProperties.class);
when(pulsarConsumerProperties.getSubscriptionName()).thenReturn("my-subscription");
String subscriptionName = PulsarBinderUtils.subscriptionName(pulsarConsumerProperties, consumerDestination);
assertThat(subscriptionName).isEqualTo("my-subscription");
}
@Test
void subscriptionNameIsNotNullWhenPropertyIsMissing() {
ConsumerDestination consumerDestination = mock(ConsumerDestination.class);
PulsarConsumerProperties pulsarConsumerProperties = mock(PulsarConsumerProperties.class);
when(pulsarConsumerProperties.getSubscriptionName()).thenReturn(null);
when(consumerDestination.getName()).thenReturn("my-topic");
String subscriptionName = PulsarBinderUtils.subscriptionName(pulsarConsumerProperties, consumerDestination);
assertThat(subscriptionName).startsWith("my-topic" + PulsarBinderUtils.SUBSCRIPTION_NAME_SEPARATOR
+ PulsarBinderUtils.ANON_SUBSCRIPTION + PulsarBinderUtils.SUBSCRIPTION_NAME_SEPARATOR);
}
}