Skip to content

Commit ee5d70d

Browse files
debrucedebruce
committed
notification is still not working after 1.0.1-rc0 update
1 parent cd3d4f0 commit ee5d70d

4 files changed

Lines changed: 38 additions & 29 deletions

File tree

‎up_client_socket/cpp/conanfile.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
[requires]
22
up-core-api/1.6.0
33
protobuf/3.21.12
4-
up-cpp/1.0.0
4+
up-cpp/1.0.1-rc0
55
rapidjson/cci.20230929
66
spdlog/1.13.0
77
fmt/10.2.1

‎up_client_socket/cpp/include/SocketUTransport.h‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,8 +40,9 @@ class SocketUTransport : public uprotocol::transport::UTransport {
4040
const uprotocol::v1::UMessage& message) override;
4141

4242
[[nodiscard]] uprotocol::v1::UStatus registerListenerImpl(
43-
const uprotocol::v1::UUri& sink_filter, CallableConn&& listener,
44-
std::optional<uprotocol::v1::UUri>&& source_filter) override;
43+
CallableConn&& listener,
44+
const uprotocol::v1::UUri& source_filter,
45+
std::optional<uprotocol::v1::UUri>&& sink_filter) override;
4546

4647
void cleanupListener(CallableConn listener) override;
4748

‎up_client_socket/cpp/src/SocketUTransport.cpp‎

Lines changed: 15 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,7 @@ struct SocketUTransport::Impl {
209209
auto& attributes = umsg.attributes();
210210
auto key = makeCallbackKey(attributes.source(), attributes.sink());
211211
auto patterns = generateOptionals(key);
212+
size_t match_count = 0;
212213
for (const auto& pattern : patterns) {
213214
// cout << "looking up " << pattern << endl;
214215
auto ptr = callback_data_.find(pattern);
@@ -218,9 +219,15 @@ struct SocketUTransport::Impl {
218219
unique_lock<mutex> lock(ptr->mtx);
219220
for (auto callback : ptr->listeners) {
220221
callback(umsg);
222+
match_count++;
221223
}
222224
}
223225
}
226+
if (match_count == 0) {
227+
for (const auto& pattern : patterns) {
228+
cout << "no match " << pattern << endl;
229+
}
230+
}
224231

225232
} catch (const system_error& e) {
226233
if (e.code() == errc::io_error) {
@@ -234,12 +241,12 @@ struct SocketUTransport::Impl {
234241
}
235242
}
236243

237-
UStatus registerListenerImpl(const UUri& sink_filter,
238-
CallableConn& listener,
239-
optional<UUri>& source_filter) {
244+
UStatus registerListenerImpl(CallableConn& listener,
245+
const UUri& source_filter,
246+
optional<UUri>& sink_filter) {
240247
UStatus retval;
241248
retval.set_code(UCode::OK);
242-
auto key = makeCallbackKey(sink_filter, source_filter);
249+
auto key = makeCallbackKey(source_filter, sink_filter);
243250
cout << "inserting " << key << endl;
244251
auto ptr = callback_data_.find(key, true);
245252
unique_lock<mutex> lock(ptr->mtx);
@@ -264,10 +271,10 @@ UStatus SocketUTransport::sendImpl(const UMessage& umsg) {
264271
return pImpl->sendImpl(umsg);
265272
}
266273

267-
UStatus SocketUTransport::registerListenerImpl(const UUri& sink_filter,
268-
CallableConn&& listener,
269-
optional<UUri>&& source_filter) {
270-
return pImpl->registerListenerImpl(sink_filter, listener, source_filter);
274+
UStatus SocketUTransport::registerListenerImpl(CallableConn&& listener,
275+
const UUri& source_filter,
276+
optional<UUri>&& sink_filter) {
277+
return pImpl->registerListenerImpl(listener, source_filter, sink_filter);
271278
}
272279

273280
void SocketUTransport::cleanupListener(CallableConn listener) {

‎up_client_socket/cpp/src/test.cpp‎

Lines changed: 19 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -49,9 +49,10 @@ struct TestUUri {
4949
void test_pub_sub(shared_ptr<SocketUTransport> transport)
5050
{
5151
TestUUri src{"10.0.0.1", 0x10001, 1, 0x8000};
52-
auto action_exact = [&](const uprotocol::v1::UMessage& msg) { cout << "#### got sub from pub" << endl; };
5352

54-
auto lhandle1 = transport->registerListener(src, action_exact);
53+
auto lhandle1 = transport->registerListener([&](const uprotocol::v1::UMessage& msg) {
54+
cout << "#### got sub from pub" << endl;
55+
}, src);
5556

5657
for (auto i = 0; i < 2; i++) {
5758
{
@@ -75,19 +76,22 @@ void test_pub_sub(shared_ptr<SocketUTransport> transport)
7576
void test_rpc_req(shared_ptr<SocketUTransport> transport)
7677
{
7778
TestUUri src{"10.0.0.1", 0x10001, 1, 0};
78-
auto action_exact = [&](const uprotocol::v1::UMessage& msg) { cout << "#### got rpc req" << endl; };
79-
8079
TestUUri sink{"10.0.0.2", 0x10002, 2, 2};
8180

82-
auto lhandle0 = transport->registerListener(src, action_exact, sink);
83-
auto lhandle1 = transport->registerListener(src, action_exact);
81+
auto lhandle0 = transport->registerListener([&](const uprotocol::v1::UMessage& msg) {
82+
cout << "#### got rpc req exact" << endl;
83+
}, src, sink);
84+
85+
// auto lhandle1 = transport->registerListener([&](const uprotocol::v1::UMessage& msg) {
86+
// cout << "#### got rpc req without sink" << endl;
87+
// }, src);
8488

8589
for (auto i = 0; i < 2; i++) {
8690
{
8791
uprotocol::v1::UAttributes attr;
8892
attr.set_type(uprotocol::v1::UMESSAGE_TYPE_REQUEST);
8993
*attr.mutable_id() = make_uuid();
90-
*attr.mutable_source() = src.withReqIdZero();
94+
*attr.mutable_source() = src;
9195
*attr.mutable_sink() = sink;
9296
attr.set_priority(uprotocol::v1::UPRIORITY_CS4);
9397
attr.set_payload_format(uprotocol::v1::UPAYLOAD_FORMAT_TEXT);
@@ -106,12 +110,10 @@ void test_rpc_req(shared_ptr<SocketUTransport> transport)
106110
void test_rpc_resp(shared_ptr<SocketUTransport> transport)
107111
{
108112
TestUUri src{"10.0.0.1", 0x10001, 1, 0};
109-
auto action_exact = [&](const uprotocol::v1::UMessage& msg) { cout << "#### got rpc resp" << endl; };
110-
111113
TestUUri sink{"10.0.0.2", 0x10002, 2, 2};
112114

113-
auto lhandle0 = transport->registerListener(sink, action_exact, src);
114-
// auto lhandle1 = transport->registerListener(src, action_exact);
115+
auto lhandle0 = transport->registerListener([&](const uprotocol::v1::UMessage& msg) {
116+
cout << "#### got rpc resp" << endl; }, sink, src);
115117

116118
for (auto i = 0; i < 2; i++) {
117119
{
@@ -136,12 +138,11 @@ void test_rpc_resp(shared_ptr<SocketUTransport> transport)
136138

137139
void test_notification(shared_ptr<SocketUTransport> transport)
138140
{
139-
TestUUri src{"10.0.0.1", 0x18001, 1, 1};
140-
auto action_exact = [&](const uprotocol::v1::UMessage& msg) { cout << "#### got notification" << endl; };
141-
142-
TestUUri sink{"10.0.0.2", 0x10002, 2, 1};
141+
TestUUri src{"10.0.0.1", 0x8001, 1, 1};
142+
TestUUri sink{"10.0.0.2", 0x10002, 2, 2};
143143

144-
auto lhandle0 = transport->registerListener(sink, action_exact, src);
144+
auto lhandle0 = transport->registerListener([&](const uprotocol::v1::UMessage& msg) {
145+
cout << "#### got notification" << endl; }, sink, src);
145146
cout << "after registerListener" << endl;
146147
// auto lhandle1 = transport->registerListener(src, action_exact);
147148

@@ -150,8 +151,8 @@ void test_notification(shared_ptr<SocketUTransport> transport)
150151
uprotocol::v1::UAttributes attr;
151152
attr.set_type(uprotocol::v1::UMESSAGE_TYPE_NOTIFICATION);
152153
*attr.mutable_id() = make_uuid();
153-
*attr.mutable_source() = src;
154-
*attr.mutable_sink() = sink.withReqIdZero();
154+
*attr.mutable_source() = src; // .withReqIdZero();
155+
*attr.mutable_sink() = sink;
155156
// attr.set_priority(uprotocol::v1::UPRIORITY_CS4);
156157
attr.set_payload_format(uprotocol::v1::UPAYLOAD_FORMAT_TEXT);
157158
// *attr.mutable_reqid() = make_uuid();

0 commit comments

Comments
 (0)