forked from apache/pulsar-client-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathreader.cc
More file actions
69 lines (58 loc) · 2.42 KB
/
Copy pathreader.cc
File metadata and controls
69 lines (58 loc) · 2.42 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
/**
* 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
*
* http://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.
*/
#include "utils.h"
#include <pybind11/pybind11.h>
namespace py = pybind11;
Message Reader_readNext(Reader& reader) {
return waitForAsyncValue<Message>([&](ReadNextCallback callback) { reader.readNextAsync(callback); });
}
Message Reader_readNextTimeout(Reader& reader, int timeoutMs) {
Message msg;
Result res;
Py_BEGIN_ALLOW_THREADS res = reader.readNext(msg, timeoutMs);
Py_END_ALLOW_THREADS
CHECK_RESULT(res);
return msg;
}
bool Reader_hasMessageAvailable(Reader& reader) {
return waitForAsyncValue<bool>(
[&](HasMessageAvailableCallback callback) { reader.hasMessageAvailableAsync(callback); });
}
void Reader_close(Reader& reader) {
waitForAsyncResult([&](ResultCallback callback) { reader.closeAsync(callback); });
}
void Reader_seek(Reader& reader, const MessageId& msgId) {
waitForAsyncResult([&](ResultCallback callback) { reader.seekAsync(msgId, callback); });
}
void Reader_seek_timestamp(Reader& reader, uint64_t timestamp) {
waitForAsyncResult([&](ResultCallback callback) { reader.seekAsync(timestamp, callback); });
}
bool Reader_is_connected(Reader& reader) { return reader.isConnected(); }
void export_reader(py::module_& m) {
using namespace py;
class_<Reader>(m, "Reader")
.def("topic", &Reader::getTopic, return_value_policy::copy)
.def("read_next", &Reader_readNext)
.def("read_next", &Reader_readNextTimeout)
.def("has_message_available", &Reader_hasMessageAvailable)
.def("close", &Reader_close)
.def("seek", &Reader_seek)
.def("seek", &Reader_seek_timestamp)
.def("is_connected", &Reader_is_connected);
}