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

Add ReactivePulsarConsumerFactory (#178)

Browse Source
This commit is contained in:
Christophe Bornet
2022-10-29 01:44:10 +02:00
committed by Chris Bono
parent 978feb2d16
commit 47c9388bd9
4 changed files with 225 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarConsumerFactory.java Normal file
View File

@@ -0,0 +1,70 @@
/*
* Copyright 2022 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.core.reactive;
import java.util.Collections;
import java.util.List;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.reactive.client.api.ImmutableReactiveMessageConsumerSpec;
import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerSpec;
import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
import org.springframework.util.CollectionUtils;
/**
* Default implementation for {@link ReactivePulsarConsumerFactory}.
*
* @param <T> underlying payload type for the reactive consumer.
* @author Christophe Bornet
*/
public class DefaultReactivePulsarConsumerFactory<T> implements ReactivePulsarConsumerFactory<T> {
private final ReactiveMessageConsumerSpec consumerSpec;
private final ReactivePulsarClient reactivePulsarClient;
public DefaultReactivePulsarConsumerFactory(ReactivePulsarClient reactivePulsarClient,
ReactiveMessageConsumerSpec consumerSpec) {
this.consumerSpec = new ImmutableReactiveMessageConsumerSpec(
consumerSpec != null ? consumerSpec : new MutableReactiveMessageConsumerSpec());
this.reactivePulsarClient = reactivePulsarClient;
}
@Override
public ReactiveMessageConsumer<T> createConsumer(Schema<T> schema) {
return createConsumer(schema, Collections.emptyList());
}
@Override
public ReactiveMessageConsumer<T> createConsumer(Schema<T> schema,
List<ReactiveMessageConsumerBuilderCustomizer<T>> customizers) {
ReactiveMessageConsumerBuilder<T> consumer = this.reactivePulsarClient.messageConsumer(schema);
consumer.applySpec(this.consumerSpec);
if (!CollectionUtils.isEmpty(customizers)) {
customizers.forEach((c) -> c.customize(consumer));
}
return consumer.build();
}
}

36
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageConsumerBuilderCustomizer.java Normal file
View File

@@ -0,0 +1,36 @@
/*
* Copyright 2022 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.core.reactive;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
/**
* The interface to customize a {@link ReactiveMessageConsumerBuilder}.
*
* @param <T> The message payload type
* @author Christophe Bornet
*/
@FunctionalInterface
public interface ReactiveMessageConsumerBuilderCustomizer<T> {
/**
* Customizes a {@link ReactiveMessageConsumerBuilder}.
* @param reactiveMessageConsumerBuilder the builder to customize
*/
void customize(ReactiveMessageConsumerBuilder<T> reactiveMessageConsumerBuilder);
}

49
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarConsumerFactory.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* Copyright 2022 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.core.reactive;
import java.util.List;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
/**
* Pulsar reactive consumer factory interface.
*
* @param <T> payload type for the consumer.
* @author Christophe Bornet
*/
public interface ReactivePulsarConsumerFactory<T> {
/**
* Create a reactive message consumer.
* @param schema the schema of the messages to be consumed
* @return the reactive message consumer
*/
ReactiveMessageConsumer<T> createConsumer(Schema<T> schema);
/**
* Create a reactive message consumer.
* @param schema the schema of the messages to be consumed
* @param customizers the optional list of customizers to apply to the reactive
* message consumer builder
* @return the reactive message consumer
*/
ReactiveMessageConsumer<T> createConsumer(Schema<T> schema,
List<ReactiveMessageConsumerBuilderCustomizer<T>> customizers);
}

70
spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageConsumerFactoryTests.java Normal file
View File

@@ -0,0 +1,70 @@
/*
* Copyright 2022 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.core.reactive;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collections;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
import org.apache.pulsar.reactive.client.api.MutableReactiveMessageConsumerSpec;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerSpec;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Test;
/**
* Tests for {@link DefaultReactivePulsarConsumerFactory}
*
* @author Christophe Bornet
*/
class DefaultReactiveMessageConsumerFactoryTests {
private static final Schema<String> SCHEMA = Schema.STRING;
@Test
void createConsumer() {
MutableReactiveMessageConsumerSpec spec = new MutableReactiveMessageConsumerSpec();
spec.setConsumerName("test-consumer");
DefaultReactivePulsarConsumerFactory<String> consumerFactory = new DefaultReactivePulsarConsumerFactory<>(
AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec);
ReactiveMessageConsumer<String> consumer = consumerFactory.createConsumer(SCHEMA);
assertThat(consumer)
.extracting("consumerSpec", InstanceOfAssertFactories.type(ReactiveMessageConsumerSpec.class))
.extracting(ReactiveMessageConsumerSpec::getConsumerName).isEqualTo("test-consumer");
}
@Test
void createConsumerWithCustomizer() {
MutableReactiveMessageConsumerSpec spec = new MutableReactiveMessageConsumerSpec();
spec.setConsumerName("test-consumer");
DefaultReactivePulsarConsumerFactory<String> consumerFactory = new DefaultReactivePulsarConsumerFactory<>(
AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec);
ReactiveMessageConsumer<String> consumer = consumerFactory.createConsumer(SCHEMA,
Collections.singletonList(builder -> builder.consumerName("new-test-consumer")));
assertThat(consumer)
.extracting("consumerSpec", InstanceOfAssertFactories.type(ReactiveMessageConsumerSpec.class))
.extracting(ReactiveMessageConsumerSpec::getConsumerName).isEqualTo("new-test-consumer");
}
}