You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "cbornet (via GitHub)" <gi...@apache.org> on 2023/01/29 18:01:19 UTC
[GitHub] [pulsar-client-reactive] cbornet opened a new pull request, #117: Add utility wrappers for Mutiny
cbornet opened a new pull request, #117:
URL: https://github.com/apache/pulsar-client-reactive/pull/117
Fix #114
--
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
[GitHub] [pulsar-client-reactive] lhotari commented on a diff in pull request #117: Add utility wrappers for Mutiny
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #117:
URL: https://github.com/apache/pulsar-client-reactive/pull/117#discussion_r1091926119
##########
pulsar-client-reactive-mutiny/src/main/java/org/apache/pulsar/reactive/client/mutiny/MutinyConsumer.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.pulsar.reactive.client.mutiny;
+
+import java.util.function.Function;
+
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.reactive.client.api.MessageResult;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
+import org.reactivestreams.Publisher;
+
+/**
+ * A Mutiny wrapper for a {@link ReactiveMessageConsumer}.
+ */
+public class MutinyConsumer<T> {
+
+ private final ReactiveMessageConsumer<T> consumer;
+
+ /**
+ * Create a new {@link MutinyConsumer} wrapping the given
+ * {@link ReactiveMessageConsumer}.
+ * @param consumer the consumer to wrap
+ */
+ public MutinyConsumer(ReactiveMessageConsumer<T> consumer) {
+ this.consumer = consumer;
+ }
+
+ /**
+ * Consumes one message.
+ * @param messageHandler a {@link Function} to apply to the consumed message that
+ * returns a {@link MessageResult} which contains the acknowledgement or negative
+ * acknowledgement referencing the message id of the input message together with an
+ * optional return value object
+ * @param <R> the type of MessageResult returned by the message handler
+ * @return the value contained by the {@link MessageResult} returned by the message
+ * handler
+ */
+ public <R> Uni<R> consumeOne(Function<Message<T>, Uni<MessageResult<R>>> messageHandler) {
+ return Uni.createFrom().publisher(this.consumer.consumeOne((m) -> messageHandler.apply(m).toMulti()));
+ }
+
+ /**
+ * Consumes messages continuously.
+ * @param messageHandler a {@link Function} to apply to the consumed messages that
+ * returns {@link MessageResult}s, each containing the acknowledgement or negative
+ * acknowledgement referencing the message id of the corresponding input messages
+ * together with an optional return value object
+ * @param <R> the type of MessageResult returned by the message handler
+ * @return the values contained by the {@link MessageResult}s returned by the message
+ * handler
+ */
+ public <R> Multi<R> consumeMany(Function<Multi<Message<T>>, Publisher<MessageResult<R>>> messageHandler) {
Review Comment:
The suggestion was about making the interface Mutiny specific. My assumption is that when someone uses the Mutiny specific interface, they are working with Mutiny and not other RS implementations.
--
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
[GitHub] [pulsar-client-reactive] cbornet commented on a diff in pull request #117: Add utility wrappers for Mutiny
Posted by "cbornet (via GitHub)" <gi...@apache.org>.
cbornet commented on code in PR #117:
URL: https://github.com/apache/pulsar-client-reactive/pull/117#discussion_r1091998568
##########
pulsar-client-reactive-mutiny/src/main/java/org/apache/pulsar/reactive/client/mutiny/MutinyConsumer.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.pulsar.reactive.client.mutiny;
+
+import java.util.function.Function;
+
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.reactive.client.api.MessageResult;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
+import org.reactivestreams.Publisher;
+
+/**
+ * A Mutiny wrapper for a {@link ReactiveMessageConsumer}.
+ */
+public class MutinyConsumer<T> {
+
+ private final ReactiveMessageConsumer<T> consumer;
+
+ /**
+ * Create a new {@link MutinyConsumer} wrapping the given
+ * {@link ReactiveMessageConsumer}.
+ * @param consumer the consumer to wrap
+ */
+ public MutinyConsumer(ReactiveMessageConsumer<T> consumer) {
+ this.consumer = consumer;
+ }
+
+ /**
+ * Consumes one message.
+ * @param messageHandler a {@link Function} to apply to the consumed message that
+ * returns a {@link MessageResult} which contains the acknowledgement or negative
+ * acknowledgement referencing the message id of the input message together with an
+ * optional return value object
+ * @param <R> the type of MessageResult returned by the message handler
+ * @return the value contained by the {@link MessageResult} returned by the message
+ * handler
+ */
+ public <R> Uni<R> consumeOne(Function<Message<T>, Uni<MessageResult<R>>> messageHandler) {
+ return Uni.createFrom().publisher(this.consumer.consumeOne((m) -> messageHandler.apply(m).toMulti()));
+ }
+
+ /**
+ * Consumes messages continuously.
+ * @param messageHandler a {@link Function} to apply to the consumed messages that
+ * returns {@link MessageResult}s, each containing the acknowledgement or negative
+ * acknowledgement referencing the message id of the corresponding input messages
+ * together with an optional return value object
+ * @param <R> the type of MessageResult returned by the message handler
+ * @return the values contained by the {@link MessageResult}s returned by the message
+ * handler
+ */
+ public <R> Multi<R> consumeMany(Function<Multi<Message<T>>, Publisher<MessageResult<R>>> messageHandler) {
Review Comment:
With this interface the handler can return a Multi but also any other Publisher. You may get a Publisher from another lib (eg R2DBC) and want to return the Flux without translating to Multi.
--
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
Re: [PR] Add utility wrappers for Mutiny [pulsar-client-reactive]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari closed pull request #117: Add utility wrappers for Mutiny
URL: https://github.com/apache/pulsar-client-reactive/pull/117
--
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
Re: [PR] Add utility wrappers for Mutiny [pulsar-client-reactive]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #117:
URL: https://github.com/apache/pulsar-client-reactive/pull/117#issuecomment-2048172379
Closing this for now. We can come back to this later if there's more demand.
--
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
[GitHub] [pulsar-client-reactive] lhotari commented on a diff in pull request #117: Add utility wrappers for Mutiny
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #117:
URL: https://github.com/apache/pulsar-client-reactive/pull/117#discussion_r1090517546
##########
pulsar-client-reactive-mutiny/src/main/java/org/apache/pulsar/reactive/client/mutiny/MutinyConsumer.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.pulsar.reactive.client.mutiny;
+
+import java.util.function.Function;
+
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.reactive.client.api.MessageResult;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
+import org.reactivestreams.Publisher;
+
+/**
+ * A Mutiny wrapper for a {@link ReactiveMessageConsumer}.
+ */
+public class MutinyConsumer<T> {
+
+ private final ReactiveMessageConsumer<T> consumer;
+
+ /**
+ * Create a new {@link MutinyConsumer} wrapping the given
+ * {@link ReactiveMessageConsumer}.
+ * @param consumer the consumer to wrap
+ */
+ public MutinyConsumer(ReactiveMessageConsumer<T> consumer) {
+ this.consumer = consumer;
+ }
+
+ /**
+ * Consumes one message.
+ * @param messageHandler a {@link Function} to apply to the consumed message that
+ * returns a {@link MessageResult} which contains the acknowledgement or negative
+ * acknowledgement referencing the message id of the input message together with an
+ * optional return value object
+ * @param <R> the type of MessageResult returned by the message handler
+ * @return the value contained by the {@link MessageResult} returned by the message
+ * handler
+ */
+ public <R> Uni<R> consumeOne(Function<Message<T>, Uni<MessageResult<R>>> messageHandler) {
+ return Uni.createFrom().publisher(this.consumer.consumeOne((m) -> messageHandler.apply(m).toMulti()));
+ }
+
+ /**
+ * Consumes messages continuously.
+ * @param messageHandler a {@link Function} to apply to the consumed messages that
+ * returns {@link MessageResult}s, each containing the acknowledgement or negative
+ * acknowledgement referencing the message id of the corresponding input messages
+ * together with an optional return value object
+ * @param <R> the type of MessageResult returned by the message handler
+ * @return the values contained by the {@link MessageResult}s returned by the message
+ * handler
+ */
+ public <R> Multi<R> consumeMany(Function<Multi<Message<T>>, Publisher<MessageResult<R>>> messageHandler) {
Review Comment:
Since this interface is Mutiny specific, I wonder if replacing Publisher with Multi would be useful here?
```suggestion
public <R> Multi<R> consumeMany(Function<Multi<Message<T>>, Multi<MessageResult<R>>> messageHandler) {
```
--
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
Re: [PR] Add utility wrappers for Mutiny [pulsar-client-reactive]
Posted by "onobc (via GitHub)" <gi...@apache.org>.
onobc commented on PR #117:
URL: https://github.com/apache/pulsar-client-reactive/pull/117#issuecomment-2046465024
This one has been sitting here for a while. Should we close this?
--
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
[GitHub] [pulsar-client-reactive] cbornet commented on a diff in pull request #117: Add utility wrappers for Mutiny
Posted by "cbornet (via GitHub)" <gi...@apache.org>.
cbornet commented on code in PR #117:
URL: https://github.com/apache/pulsar-client-reactive/pull/117#discussion_r1090604359
##########
pulsar-client-reactive-mutiny/src/main/java/org/apache/pulsar/reactive/client/mutiny/MutinyConsumer.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.pulsar.reactive.client.mutiny;
+
+import java.util.function.Function;
+
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.reactive.client.api.MessageResult;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
+import org.reactivestreams.Publisher;
+
+/**
+ * A Mutiny wrapper for a {@link ReactiveMessageConsumer}.
+ */
+public class MutinyConsumer<T> {
+
+ private final ReactiveMessageConsumer<T> consumer;
+
+ /**
+ * Create a new {@link MutinyConsumer} wrapping the given
+ * {@link ReactiveMessageConsumer}.
+ * @param consumer the consumer to wrap
+ */
+ public MutinyConsumer(ReactiveMessageConsumer<T> consumer) {
+ this.consumer = consumer;
+ }
+
+ /**
+ * Consumes one message.
+ * @param messageHandler a {@link Function} to apply to the consumed message that
+ * returns a {@link MessageResult} which contains the acknowledgement or negative
+ * acknowledgement referencing the message id of the input message together with an
+ * optional return value object
+ * @param <R> the type of MessageResult returned by the message handler
+ * @return the value contained by the {@link MessageResult} returned by the message
+ * handler
+ */
+ public <R> Uni<R> consumeOne(Function<Message<T>, Uni<MessageResult<R>>> messageHandler) {
+ return Uni.createFrom().publisher(this.consumer.consumeOne((m) -> messageHandler.apply(m).toMulti()));
+ }
+
+ /**
+ * Consumes messages continuously.
+ * @param messageHandler a {@link Function} to apply to the consumed messages that
+ * returns {@link MessageResult}s, each containing the acknowledgement or negative
+ * acknowledgement referencing the message id of the corresponding input messages
+ * together with an optional return value object
+ * @param <R> the type of MessageResult returned by the message handler
+ * @return the values contained by the {@link MessageResult}s returned by the message
+ * handler
+ */
+ public <R> Multi<R> consumeMany(Function<Multi<Message<T>>, Publisher<MessageResult<R>>> messageHandler) {
Review Comment:
I'm not sure about this. Do you have examples where this would be a gain compared to using the broader Publisher ?
--
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