You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Nictec (via GitHub)" <gi...@apache.org> on 2024/09/20 08:37:07 UTC
[PR] Implementation of the asyncio consumer [pulsar-client-python]
Nictec opened a new pull request, #228:
URL: https://github.com/apache/pulsar-client-python/pull/228
Hi,
I implemented the asyncio consumer in a similar way as the producer. All missing async functions provided by the C++ library are now implemented in the pybind11 classes and I use futures in python to convert the functions with callbacks to async functions. The reason I started with this implementation is because I need the async consumer in a project with FastAPI myself.
Tests could be a little bit patchy, for that reason i would appreciate help if i missed something
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1827282524
##########
src/consumer.cc:
##########
@@ -4,7 +4,7 @@
* 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
Review Comment:
Could you revert this change?
##########
develop/install_manifest.txt:
##########
@@ -0,0 +1 @@
+/home/nl/PycharmProjects/pulsar-client-python/lib_pulsar.so
Review Comment:
Could you remove these two TXT files?
##########
src/client.cc:
##########
@@ -41,19 +42,39 @@ Consumer Client_subscribe(Client& client, const std::string& topic, const std::s
[&](SubscribeCallback callback) { client.subscribeAsync(topic, subscriptionName, conf, callback); });
}
+void Client_subscribeAsync(Client& client, const std::string& topic, const std::string& subscriptionName,
+ const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ py::gil_scoped_release release;
+ client.subscribeAsync(topic, subscriptionName, conf, callback);
+}
+
Consumer Client_subscribe_topics(Client& client, const std::vector<std::string>& topics,
const std::string& subscriptionName, const ConsumerConfiguration& conf) {
return waitForAsyncValue<Consumer>(
[&](SubscribeCallback callback) { client.subscribeAsync(topics, subscriptionName, conf, callback); });
}
+void Client_subscribe_topicsAsync(Client& client, const std::vector<std::string>& topics, const std::string& subscriptionName, const ConsumerConfiguration& conf, SubscribeCallback callback){
+ client.subscribeAsync(topics, subscriptionName, conf, [callback](Result result, pulsar::Consumer consumer){
Review Comment:
pybind11 will acquire the GIL when the wrapped callback of Python function is called. See https://pybind11.readthedocs.io/en/stable/advanced/misc.html#global-interpreter-lock-gil
> pybind11 will ensure that the GIL is held when it knows that it is calling Python code. For example, if a Python callback is passed to C++ code via std::function, when C++ code calls the function the built-in wrapper will acquire the GIL before calling the Python callback.
So you don't need to wrap the callback again with `py::gil_scoped_acquire`
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower closed pull request #228: Implementation of the asyncio consumer
URL: https://github.com/apache/pulsar-client-python/pull/228
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#issuecomment-3685982851
I will continue the work in https://github.com/apache/pulsar-client-python/pull/277
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#issuecomment-2446226595
Let me fix the broken CI first
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "Nictec (via GitHub)" <gi...@apache.org>.
Nictec commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1866591383
##########
src/client.cc:
##########
@@ -41,19 +42,39 @@ Consumer Client_subscribe(Client& client, const std::string& topic, const std::s
[&](SubscribeCallback callback) { client.subscribeAsync(topic, subscriptionName, conf, callback); });
}
+void Client_subscribeAsync(Client& client, const std::string& topic, const std::string& subscriptionName,
+ const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ py::gil_scoped_release release;
+ client.subscribeAsync(topic, subscriptionName, conf, callback);
+}
+
Consumer Client_subscribe_topics(Client& client, const std::vector<std::string>& topics,
const std::string& subscriptionName, const ConsumerConfiguration& conf) {
return waitForAsyncValue<Consumer>(
[&](SubscribeCallback callback) { client.subscribeAsync(topics, subscriptionName, conf, callback); });
}
+void Client_subscribe_topicsAsync(Client& client, const std::vector<std::string>& topics, const std::string& subscriptionName, const ConsumerConfiguration& conf, SubscribeCallback callback){
+ client.subscribeAsync(topics, subscriptionName, conf, [callback](Result result, pulsar::Consumer consumer){
Review Comment:
Nothing, I just forgot to remove my TODO. The following commit fixes the schema problems: [https://github.com/Nictec/pulsar-client-python/commit/656a287da2794ee4760baf9a09d53352f8f8520b](https://github.com/Nictec/pulsar-client-python/commit/656a287da2794ee4760baf9a09d53352f8f8520b)
Ill remove it.
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "Nictec (via GitHub)" <gi...@apache.org>.
Nictec commented on PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#issuecomment-2422218012
I found another issue with the schema system in my asyncio implementation, i will fix this too before the next commit.
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "Nictec (via GitHub)" <gi...@apache.org>.
Nictec commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1866591383
##########
src/client.cc:
##########
@@ -41,19 +42,39 @@ Consumer Client_subscribe(Client& client, const std::string& topic, const std::s
[&](SubscribeCallback callback) { client.subscribeAsync(topic, subscriptionName, conf, callback); });
}
+void Client_subscribeAsync(Client& client, const std::string& topic, const std::string& subscriptionName,
+ const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ py::gil_scoped_release release;
+ client.subscribeAsync(topic, subscriptionName, conf, callback);
+}
+
Consumer Client_subscribe_topics(Client& client, const std::vector<std::string>& topics,
const std::string& subscriptionName, const ConsumerConfiguration& conf) {
return waitForAsyncValue<Consumer>(
[&](SubscribeCallback callback) { client.subscribeAsync(topics, subscriptionName, conf, callback); });
}
+void Client_subscribe_topicsAsync(Client& client, const std::vector<std::string>& topics, const std::string& subscriptionName, const ConsumerConfiguration& conf, SubscribeCallback callback){
+ client.subscribeAsync(topics, subscriptionName, conf, [callback](Result result, pulsar::Consumer consumer){
Review Comment:
Nothing, I just forgot to remove my TODO. The following commit fixes the schema problems: [https://github.com/Nictec/pulsar-client-python/commit/656a287da2794ee4760baf9a09d53352f8f8520b](https://github.com/Nictec/pulsar-client-python/commit/656a287da2794ee4760baf9a09d53352f8f8520b)
Ill remove it.
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#issuecomment-2451137153
Could you rebase to master to resolve the conflicts and have the CI fixed?
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1869272997
##########
src/consumer.cc:
##########
@@ -16,22 +16,39 @@
* specific language governing permissions and limitations
* under the License.
*/
+
#include "utils.h"
#include <pulsar/Consumer.h>
+#include <pulsar/ConsumerConfiguration.h>
+#include <pulsar/Result.h>
+#include <pybind11/functional.h>
#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
+#include <memory>
namespace py = pybind11;
void Consumer_unsubscribe(Consumer& consumer) {
waitForAsyncResult([&consumer](ResultCallback callback) { consumer.unsubscribeAsync(callback); });
}
+void Consumer_unsubscribeAsync(Consumer& consumer, ResultCallback callback) {
+ consumer.unsubscribeAsync([callback] (Result result) {
+ py::gil_scoped_acquire acquire;
+ callback(result);
+ });
+}
+
Message Consumer_receive(Consumer& consumer) {
return waitForAsyncValue<Message>([&](ReceiveCallback callback) { consumer.receiveAsync(callback); });
}
+void Consumer_receiveAsync(Consumer& consumer, ReceiveCallback callback) {
+ py::gil_scoped_acquire acquire;
Review Comment:
It should release the GIL rather than acquire the GIL
##########
src/consumer.cc:
##########
@@ -16,22 +16,39 @@
* specific language governing permissions and limitations
* under the License.
*/
+
#include "utils.h"
#include <pulsar/Consumer.h>
+#include <pulsar/ConsumerConfiguration.h>
+#include <pulsar/Result.h>
+#include <pybind11/functional.h>
#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
+#include <memory>
namespace py = pybind11;
void Consumer_unsubscribe(Consumer& consumer) {
waitForAsyncResult([&consumer](ResultCallback callback) { consumer.unsubscribeAsync(callback); });
}
+void Consumer_unsubscribeAsync(Consumer& consumer, ResultCallback callback) {
+ consumer.unsubscribeAsync([callback] (Result result) {
+ py::gil_scoped_acquire acquire;
Review Comment:
This GIL acquire is not necessary
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "Nictec (via GitHub)" <gi...@apache.org>.
Nictec commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1869071160
##########
src/client.cc:
##########
@@ -41,19 +42,38 @@ Consumer Client_subscribe(Client& client, const std::string& topic, const std::s
[&](SubscribeCallback callback) { client.subscribeAsync(topic, subscriptionName, conf, callback); });
}
+void Client_subscribeAsync(Client& client, const std::string& topic, const std::string& subscriptionName,
+ const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ py::gil_scoped_release release;
+ client.subscribeAsync(topic, subscriptionName, conf, callback);
+}
+
Consumer Client_subscribe_topics(Client& client, const std::vector<std::string>& topics,
const std::string& subscriptionName, const ConsumerConfiguration& conf) {
return waitForAsyncValue<Consumer>(
[&](SubscribeCallback callback) { client.subscribeAsync(topics, subscriptionName, conf, callback); });
}
+void Client_subscribe_topicsAsync(Client& client, const std::vector<std::string>& topics, const std::string& subscriptionName, const ConsumerConfiguration& conf, SubscribeCallback callback){
+ client.subscribeAsync(topics, subscriptionName, conf, [callback](Result result, pulsar::Consumer consumer){
+ callback(result, consumer);
+ });
+}
+
Consumer Client_subscribe_pattern(Client& client, const std::string& topic_pattern,
const std::string& subscriptionName, const ConsumerConfiguration& conf) {
return waitForAsyncValue<Consumer>([&](SubscribeCallback callback) {
client.subscribeWithRegexAsync(topic_pattern, subscriptionName, conf, callback);
});
}
+void Client_subscribe_patternAsync(Client& client, const std::string& topic_pattern, const std::string& subscriptionName, const ConsumerConfiguration& conf, SubscribeCallback callback){
+ client.subscribeWithRegexAsync(topic_pattern, subscriptionName, conf, [callback](Result result, Consumer consumer){
+ py::gil_scoped_acquire acquire;
+ callback(result, consumer);
Review Comment:
Fixed and tested
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "Nictec (via GitHub)" <gi...@apache.org>.
Nictec commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1827477439
##########
src/consumer.cc:
##########
@@ -4,7 +4,7 @@
* 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
Review Comment:
Sure!
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "Nictec (via GitHub)" <gi...@apache.org>.
Nictec commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1866591383
##########
src/client.cc:
##########
@@ -41,19 +42,39 @@ Consumer Client_subscribe(Client& client, const std::string& topic, const std::s
[&](SubscribeCallback callback) { client.subscribeAsync(topic, subscriptionName, conf, callback); });
}
+void Client_subscribeAsync(Client& client, const std::string& topic, const std::string& subscriptionName,
+ const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ py::gil_scoped_release release;
+ client.subscribeAsync(topic, subscriptionName, conf, callback);
+}
+
Consumer Client_subscribe_topics(Client& client, const std::vector<std::string>& topics,
const std::string& subscriptionName, const ConsumerConfiguration& conf) {
return waitForAsyncValue<Consumer>(
[&](SubscribeCallback callback) { client.subscribeAsync(topics, subscriptionName, conf, callback); });
}
+void Client_subscribe_topicsAsync(Client& client, const std::vector<std::string>& topics, const std::string& subscriptionName, const ConsumerConfiguration& conf, SubscribeCallback callback){
+ client.subscribeAsync(topics, subscriptionName, conf, [callback](Result result, pulsar::Consumer consumer){
Review Comment:
Nothing, I just forgot to remove my TODO. The following commit fixes the schema problems: [](https://github.com/Nictec/pulsar-client-python/commit/656a287da2794ee4760baf9a09d53352f8f8520b)
Ill remove it.
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1866203221
##########
pulsar/asyncio.py:
##########
@@ -116,6 +123,190 @@ async def close(self) -> None:
self._producer.close_async(functools.partial(_set_future, future, value=None))
await future
+ async def flush(self):
+ """
+ Flush all the messages buffered in the client and wait until all messages have been successfully persisted.
+
+ Raises
+ ------
+ PulsarException
+ """
+ future = asyncio.get_running_loop().create_future()
+ self._producer.flush_async(functools.partial(_set_future, future, value=None))
+ await future
+
+ @property
+ def is_connected(self) -> bool:
+ """
+ Check if the producer is connected or not.
+ """
+
+ return self._producer.is_connected()
+
+ @property
+ def last_sequence_id(self) -> int:
+ """
+ Get the last sequence id
+ """
+ return self._producer.last_sequence_id()
+
+ @property
+ def name(self) -> str:
+ """
+ Get the name of the producer.
+ """
+ return self._producer.producer_name()
+
+ @property
+ def topic(self) -> str:
+ """
+ Get the topic name of the producer.
+ """
+ return self._producer.topic()
+
+
+class Consumer:
+ # BUG: schema stuff doesn´t work at all because 90% of the methods are missing
Review Comment:
What needs to be done here to make the schema work with the async consumer?
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1867009999
##########
src/client.cc:
##########
@@ -41,19 +42,38 @@ Consumer Client_subscribe(Client& client, const std::string& topic, const std::s
[&](SubscribeCallback callback) { client.subscribeAsync(topic, subscriptionName, conf, callback); });
}
+void Client_subscribeAsync(Client& client, const std::string& topic, const std::string& subscriptionName,
+ const ConsumerConfiguration& conf, SubscribeCallback callback) {
+ py::gil_scoped_release release;
+ client.subscribeAsync(topic, subscriptionName, conf, callback);
+}
+
Consumer Client_subscribe_topics(Client& client, const std::vector<std::string>& topics,
const std::string& subscriptionName, const ConsumerConfiguration& conf) {
return waitForAsyncValue<Consumer>(
[&](SubscribeCallback callback) { client.subscribeAsync(topics, subscriptionName, conf, callback); });
}
+void Client_subscribe_topicsAsync(Client& client, const std::vector<std::string>& topics, const std::string& subscriptionName, const ConsumerConfiguration& conf, SubscribeCallback callback){
+ client.subscribeAsync(topics, subscriptionName, conf, [callback](Result result, pulsar::Consumer consumer){
+ callback(result, consumer);
+ });
+}
+
Consumer Client_subscribe_pattern(Client& client, const std::string& topic_pattern,
const std::string& subscriptionName, const ConsumerConfiguration& conf) {
return waitForAsyncValue<Consumer>([&](SubscribeCallback callback) {
client.subscribeWithRegexAsync(topic_pattern, subscriptionName, conf, callback);
});
}
+void Client_subscribe_patternAsync(Client& client, const std::string& topic_pattern, const std::string& subscriptionName, const ConsumerConfiguration& conf, SubscribeCallback callback){
+ client.subscribeWithRegexAsync(topic_pattern, subscriptionName, conf, [callback](Result result, Consumer consumer){
+ py::gil_scoped_acquire acquire;
+ callback(result, consumer);
Review Comment:
You added three `subscribe` related async methods. However, their implementations are not consistent.
- `Client_subscribeAsync`: releases the GIL first, and then call the pulsar-client-cpp API directly
- `Client_subscribe_topicsAsync`: call the pulsar-client-cpp API and wrap the callback again
- `Client_subscribe_patternAsync`: call the pulsar-client-cpp API and wrap the callback with acquiring the GIL before executing the callback
IMO, the 1st way is correct. You can also take `Producer_sendAsync` as example. Though it does not use `py::gil_scoped_acquire` directly.
##########
tests/asyncio_test.py:
##########
@@ -19,17 +19,24 @@
#
import asyncio
+from typing import Iterable
+
+from _pulsar import ConsumerType
+
import pulsar
from pulsar.asyncio import (
Client,
PulsarException,
+ Consumer
)
from unittest import (
main,
IsolatedAsyncioTestCase,
)
-service_url = 'pulsar://localhost:6650'
+# TODO: Write tests for everything else
+
Review Comment:
What does the TODO here mean?
##########
src/consumer.cc:
##########
@@ -4,7 +4,7 @@
* 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
Review Comment:
This change is still not reverted
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#issuecomment-2512093461
@BewareMyPower I removed the binary files
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "Nictec (via GitHub)" <gi...@apache.org>.
Nictec commented on code in PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#discussion_r1866594726
##########
pulsar/asyncio.py:
##########
@@ -116,6 +123,190 @@ async def close(self) -> None:
self._producer.close_async(functools.partial(_set_future, future, value=None))
await future
+ async def flush(self):
+ """
+ Flush all the messages buffered in the client and wait until all messages have been successfully persisted.
+
+ Raises
+ ------
+ PulsarException
+ """
+ future = asyncio.get_running_loop().create_future()
+ self._producer.flush_async(functools.partial(_set_future, future, value=None))
+ await future
+
+ @property
+ def is_connected(self) -> bool:
+ """
+ Check if the producer is connected or not.
+ """
+
+ return self._producer.is_connected()
+
+ @property
+ def last_sequence_id(self) -> int:
+ """
+ Get the last sequence id
+ """
+ return self._producer.last_sequence_id()
+
+ @property
+ def name(self) -> str:
+ """
+ Get the name of the producer.
+ """
+ return self._producer.producer_name()
+
+ @property
+ def topic(self) -> str:
+ """
+ Get the topic name of the producer.
+ """
+ return self._producer.topic()
+
+
+class Consumer:
+ # BUG: schema stuff doesn´t work at all because 90% of the methods are missing
Review Comment:
This was resolved in [https://github.com/Nictec/pulsar-client-python/commit/656a287da2794ee4760baf9a09d53352f8f8520b](https://github.com/Nictec/pulsar-client-python/commit/656a287da2794ee4760baf9a09d53352f8f8520b). I simply forgot to remove the comment.
--
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] Implementation of the asyncio consumer [pulsar-client-python]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #228:
URL: https://github.com/apache/pulsar-client-python/pull/228#issuecomment-2379108093
```
Traceback (most recent call last):
File "asyncio_test.py", line 27, in <module>
from pulsar.asyncio import (
File "/opt/hostedtoolcache/Python/3.8.18/x64/lib/python3.8/site-packages/pulsar/asyncio.py", line 165, in <module>
class Consumer:
File "/opt/hostedtoolcache/Python/3.8.18/x64/lib/python3.8/site-packages/pulsar/asyncio.py", line 216, in Consumer
async def seek(self, position: tuple[int, int, int, int] | pulsar.MessageId):
TypeError: 'type' object is not subscriptable
```
Could you also make the tests pass for Python 3.8? Though Python 3.8 will reach EOL on the next month (2024-10)
--
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