[GitHub] [pulsar-client-reactive] lhotari commented on a diff in pull request #11: Use org.reactivestreams Publisher type in ReactiveMessageSender API

[GitBox <gi...@apache.org>]

lhotari commented on code in PR #11:
URL: https://github.com/apache/pulsar-client-reactive/pull/11#discussion_r999734469


##########
pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/api/ReactiveMessageSender.java:
##########
@@ -17,13 +17,41 @@
 package org.apache.pulsar.reactive.client.api;
 
 import org.apache.pulsar.client.api.MessageId;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
+import org.reactivestreams.Publisher;
 
 public interface ReactiveMessageSender<T> {
 
-	Mono<MessageId> sendMessage(Mono<MessageSpec<T>> messageSpec);
+	/**
+	 * Send one message.
+	 * @param messageSpec the spec of the message to send
+	 * @return a publisher that will emit one message id and complete
+	 */
+	Publisher<MessageId> send(MessageSpec<T> messageSpec);
 
-	Flux<MessageId> sendMessages(Flux<MessageSpec<T>> messageSpecs);
+	/**
+	 * Send multiple messages and get the associated message ids in the same order as the
+	 * sent messages.
+	 * @param messageSpecs the specs of the messages to send
+	 * @return a publisher that will emit a message id per message successfully sent in
+	 * the order that they have been sent
+	 */
+	Publisher<MessageId> sendAll(Publisher<MessageSpec<T>> messageSpecs);
+
+	/**
+	 * Convert to a reactor based message sender.
+	 * @return the reactor based message sender instance
+	 */
+	default ReactorMessageSender<T> toReactor() {
+		return new ReactorMessageSender<>(this);
+	}
+
+	/**
+	 * Convert to a RxJava 3 based message sender. Use only if you have RxJava 3 on the
+	 * class path (not pulled by pulsar-client-reactive-api).
+	 * @return the RxJava 3 based message sender instance.
+	 */
+	default RxJavaMessageSender<T> toRxJava() {
+		return new RxJavaMessageSender<>(this);
+	}

Review Comment:
   I am not convinced that this is a scalable model. There are multiple Reactive Streams implementations and this couples them together. It would also clutter the main interface if the only way to add support is to add it to the interface. Did you consider an alternative approach that was proposed in #10, using `<R> R adapt(Class<R> adaptedSenderType);` instead ?
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org